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

## 158 statements 142 run 16 missing 2 excluded 6 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 ↛ 67 line 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 gv['items'],

118 )

119 self.games[gid] = g

120 core = self.core

121 core.events.game_joined.emit(g)

122 return ev

123

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

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

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

127 return ev

128

129 self.kill_game(g)

130

131 core = self.core

132 core.events.game_left.emit(g)

133

134 return ev

135

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

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

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

139 return ev

140

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

142 core = self.core

143 core.events.game_ended.emit(g)

144

145 return ev

146

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

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

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

150 return ev

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

152 return ev

153

154 # ----- Private Methods -----

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

156 from thb import modes

157 g = modes[mode]()

158 assert isinstance(g, Game)

159

160 assoc: GamePartAssocOnGame = {

161 'gid': gid,

162 'name': name,

163 'users': users,

164 'presence': {},

165 'params': params,

166 'players': [],

167 'items': items,

168 'data': GameData(gid),

169 'observe': False,

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 if 'greenlet' in A(self, g): 222 ↛ 223 line 222 didn't jump to line 223, because the condition on line 222 was never true

223 A(self, g)['greenlet'].kill(ForcedKill)

224

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

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

227

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

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

230

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

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

233

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

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

236

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

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

239

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

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

242

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

244 for p in A(self, g)['players']: 244 ↛ 248 line 244 didn't jump to line 248, because the loop on line 244 didn't complete

245 if isinstance(p, Theone):

246 return p

247 else:

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

249

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

251 assert isinstance(p, Player), p

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