# Coverage for client/parts/game.py : 81%

## 159 statements 134 run 25 missing 2 excluded 5 partial

Hot-keys on this page

r m x p toggle line displays

j k next/prev highlighted chunk

0 (zero) top of page

1 (one) first highlighted chunk

1# -*- coding: utf-8 -*-

2 from __future__ import annotations

3

4# -- stdlib --

5 from typing import Any, Dict, List, Sequence, TYPE_CHECKING

6 import logging

7

8# -- third party --

9 from gevent import Greenlet

10 from mypy_extensions import TypedDict

11

12# -- own --

13 from client.base import ClientGameRunner, ForcedKill, Someone, Theone

14 from game.base import Game, GameData, GameItem, Player

15 from utils.events import EventHub

16 from utils.misc import BatchList

17 import wire

18

19# -- typing --

20 if TYPE_CHECKING:

21 from client.core import Core # noqa: F401

22

23

24# -- code --

25 log = logging.getLogger('client.parts.Game')

26 STOP = EventHub.STOP_PROPAGATION

27

28

29 class GamePartAssocOnGame(TypedDict):

30 gid: int

31 name: str

32 users: List[wire.model.User]

33 presence: Dict[Player, bool]

34 params: Dict[str, Any]

35 players: BatchList[Player]

36 items: Dict[Player, List[GameItem]]

37 data: GameData

38 observe: bool

39 greenlet: Greenlet

40

41

42 def A(self: GamePart, v: Game) -> GamePartAssocOnGame:

43 return v._[self]

44

45

46 class GamePart(object):

47 def __init__(self, core: Core):

48 self.core = core

49 self.games: Dict[int, Game] = {}

50

51 D = core.events.server_command

52 D[wire.RoomUsers] += self._room_users

53 D[wire.GameStarted] += self._game_started

54 D[wire.PlayerPresence] += self._player_presence

55 D[wire.GameJoined] += self._game_joined

56 D[wire.ObserveStarted] += self._observe_started

57 D[wire.GameLeft] += self._game_left

58 D[wire.GameEnded] += self._game_ended

59 D[wire.GameData] += self._game_data

60

61 # ----- Reactions -----

62 def _room_users(self, ev: wire.RoomUsers) -> wire.RoomUsers:

63 core = self.core

64

65 g = self.games.get(ev.gid)

66 if not g: 66 ↛ 67line 66 didn't jump to line 67, because the condition on line 66 was never true

67 return ev

68

69 A(self, g)['users'] = ev.users

70 core.events.room_users.emit((g, ev.users))

71

72 return ev

73

74 def _do_start_game(self, gv: wire.model.GameDetail, me_uid: int) -> None:

75 core = self.core

76 g = self.games[gv['gid']]

77 A(self, g)['users'] = gv['users']

78 A(self, g)['params'] = gv['params']

79 players = self._build_players(g, A(self, g)['users'], me_uid)

80 A(self, g)['players'] = players

81 items = self._build_items(g, players, gv['items'])

82 A(self, g)['items'] = items

83 core.events.game_started.emit(g)

84 # Game not actually running now,

85 # UI should start it by core.game.start_game(g) manually

86

87 def _game_started(self, ev: wire.GameStarted) -> wire.GameStarted:

88 core = self.core

89 self._do_start_game(ev.game, core.auth.uid)

90 return ev

91

92 def _observe_started(self, ev: wire.ObserveStarted) -> wire.ObserveStarted:

93 g = self.games[ev.game['gid']]

94 A(self, g)['observe'] = True

95 self._do_start_game(ev.game, ev.observee)

96 return ev

97

98 def _player_presence(self, ev: wire.PlayerPresence) -> wire.PlayerPresence:

99 core = self.core

100 g = self.games[ev.gid]

101 pl = A(self, g)['players']

102 tbl = ev.presence

103 presence = {p: tbl[p.uid] for p in pl}

104 A(self, g)['presence'] = presence

105 core.events.player_presence.emit((g, presence))

106 return ev

107

108 def _game_joined(self, ev: wire.GameJoined) -> wire.GameJoined:

109 gv = ev.game

110 gid = gv['gid']

111 g = self._make_game(

112 gid,

113 gv['type'],

114 gv['name'],

115 gv['users'],

116 gv['params'],

117 )

118 self.games[gid] = g

119 core = self.core

120 core.events.game_joined.emit(g)

121 return ev

122

123 def _game_left(self, ev: wire.GameLeft) -> wire.GameLeft:

124 g = self.games.get(ev.gid)

125 if not g: 125 ↛ 126line 125 didn't jump to line 126, because the condition on line 125 was never true

126 return ev

127

128 self.kill_game(g)

129

130 core = self.core

131 core.events.game_left.emit(g)

132

133 return ev

134

135 def _game_ended(self, ev: wire.GameEnded) -> wire.GameEnded:

136 g = self.games.get(ev.gid)

137 if not g: 137 ↛ 138line 137 didn't jump to line 138, because the condition on line 137 was never true

138 return ev

139

140 log.info('=======GAME ENDED=======')

141 core = self.core

142 core.events.game_ended.emit(g)

143

144 return ev

145

146 def _game_data(self, ev: wire.GameData) -> wire.GameData:

147 g = self.games.get(ev.gid)

148 if not g: 148 ↛ 149line 148 didn't jump to line 149, because the condition on line 148 was never true

149 return ev

150 A(self, g)['data'].feed_recv(ev.tag, ev.data)

151 return ev

152

153 # ----- Private Methods -----

154 def _make_game(self, gid: int, mode: str, name: str, users: List[wire.model.User], params: Dict[str, Any]) -> Game:

155 from thb import modes

156 g = modes[mode]()

157 assert isinstance(g, Game)

158

159 assoc: GamePartAssocOnGame = {

160 'gid': gid,

161 'name': name,

162 'users': users,

163 'presence': {},

164 'params': params,

165 'players': BatchList[Player](),

166 'items': {},

167 'data': GameData(gid),

168 'observe': False,

169 'greenlet': None,

170 }

171 g._[self] = assoc

172

173 return g

174

175 def _build_players(self, g: Game, uvl: Sequence[wire.model.User], me_uid: int) -> BatchList[Player]:

176 assert me_uid in [uv['uid'] for uv in uvl]

177

178 me = Theone(g, me_uid)

179

180 pl = BatchList[Player]([

181 me if uv['uid'] == me_uid else Someone(g, uv['uid'])

182 for uv in uvl

183 ])

184 pl[:0] = [Someone(g, 0) for i, npc in enumerate(g.npc_players)]

185

186 return pl

187

188 def _build_items(self, g: Game, players: Sequence[Player], items: Dict[int, List[str]]) -> Dict[Player, List[GameItem]]:

189 m = {p.uid: p for p in players}

190 return {

191 m[uid]: [GameItem.from_sku(i) for i in skus]

192 for uid, skus in items.items()

193 }

194

195 # ----- Public Methods -----

196 def is_observe(self, g: Game) -> bool:

197 return A(self, g)['observe']

198

199 def start_game(self, g: Game) -> None:

200 core = self.core

201 gr = ClientGameRunner(core, g)

202 @gr.link_exception

203 def crash(gr: Greenlet) -> None:

204 core.events.game_crashed.emit(g)

205

206 A(self, g)['greenlet'] = gr

207 core.runner.start(gr)

208 log.info('----- GAME STARTED: %d -----' % A(self, g)['gid'])

209

210 def write(self, g: Game, tag: str, data: object) -> None:

211 core = self.core

212 pkt = A(self, g)['data'].feed_send(tag, data)

213 gid = A(self, g)['gid']

214 core.server.write(wire.GameData(

215 gid=gid,

216 tag=pkt.tag,

217 data=pkt.data,

218 ))

219 # core.events.game_data_send.emit((g, pkt))

220

221 def kill_game(self, g: Game) -> None:

222 gr = A(self, g)['greenlet']

223 if gr: 223 ↛ 224line 223 didn't jump to line 224, because the condition on line 223 was never true

224 gr.kill(ForcedKill)

225

226 def gid_of(self, g: Game) -> int:

227 return A(self, g)['gid']

228

229 def name_of(self, g: Game) -> str:

230 return A(self, g)['name']

231

232 def gamedata_of(self, g: Game) -> GameData:

233 return A(self, g)['data']

234

235 def items_of(self, g: Game) -> Dict[Player, List[GameItem]]:

236 return A(self, g)['items']

237

238 def params_of(self, g: Game) -> dict:

239 return A(self, g)['params']

240

241 def players_of(self, g: Game) -> BatchList[Player]:

242 return A(self, g)['players']

243

244 def theone_of(self, g: Game) -> Theone:

245 for p in A(self, g)['players']:

246 if isinstance(p, Theone):

247 return p

248 else:

249 raise Exception("Couldn't find Theone!")

250

251 def is_dropped(self, g: Game, p: Player) -> bool:

252 assert isinstance(p, Player), p

253 return not A(self, g)['presence'].get(p, True)