Coverage for client/parts/game.py : 87%
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 -*-
2from __future__ import annotations
4# -- stdlib --
5from typing import Any, Dict, List, Sequence, TYPE_CHECKING
6import logging
8# -- third party --
9from gevent import Greenlet
10from mypy_extensions import TypedDict
12# -- own --
13from client.base import ClientGameRunner, ForcedKill, Someone, Theone
14from game.base import Game, GameData, GameItem, Player
15from utils.events import EventHub
16from utils.misc import BatchList
17import wire
19# -- typing --
20if TYPE_CHECKING:
21 from client.core import Core # noqa: F401
24# -- code --
25log = logging.getLogger('client.parts.Game')
26STOP = EventHub.STOP_PROPAGATION
29class 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
42def A(self: GamePart, v: Game) -> GamePartAssocOnGame:
43 return v._[self]
46class GamePart(object):
47 def __init__(self, core: Core):
48 self.core = core
49 self.games: Dict[int, Game] = {}
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
61 # ----- Reactions -----
62 def _room_users(self, ev: wire.RoomUsers) -> wire.RoomUsers:
63 core = self.core
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
69 A(self, g)['users'] = ev.users
70 core.events.room_users.emit((g, ev.users))
72 return ev
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
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
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
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
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
124 def _game_left(self, ev: wire.GameLeft) -> wire.GameLeft:
125 g = self.games.get(ev.gid)
126 if not g: 126 ↛ 127line 126 didn't jump to line 127, because the condition on line 126 was never true
127 return ev
129 self.kill_game(g)
131 core = self.core
132 core.events.game_left.emit(g)
134 return ev
136 def _game_ended(self, ev: wire.GameEnded) -> wire.GameEnded:
137 g = self.games.get(ev.gid)
138 if not g: 138 ↛ 139line 138 didn't jump to line 139, because the condition on line 138 was never true
139 return ev
141 log.info('=======GAME ENDED=======')
142 core = self.core
143 core.events.game_ended.emit(g)
145 return ev
147 def _game_data(self, ev: wire.GameData) -> wire.GameData:
148 g = self.games.get(ev.gid)
149 if not g: 149 ↛ 150line 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
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)
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
173 return g
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]
178 me = Theone(g, me_uid)
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)]
186 return pl
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 }
195 # ----- Public Methods -----
196 def is_observe(self, g: Game) -> bool:
197 return A(self, g)['observe']
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)
206 A(self, g)['greenlet'] = gr
207 core.runner.start(gr)
208 log.info('----- GAME STARTED: %d -----' % A(self, g)['gid'])
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))
221 def kill_game(self, g: Game) -> None:
222 if 'greenlet' in A(self, g): 222 ↛ 223line 222 didn't jump to line 223, because the condition on line 222 was never true
223 A(self, g)['greenlet'].kill(ForcedKill)
225 def gid_of(self, g: Game) -> int:
226 return A(self, g)['gid']
228 def name_of(self, g: Game) -> str:
229 return A(self, g)['name']
231 def gamedata_of(self, g: Game) -> GameData:
232 return A(self, g)['data']
234 def items_of(self, g: Game) -> Dict[Player, List[GameItem]]:
235 return A(self, g)['items']
237 def params_of(self, g: Game) -> dict:
238 return A(self, g)['params']
240 def players_of(self, g: Game) -> BatchList[Player]:
241 return A(self, g)['players']
243 def theone_of(self, g: Game) -> Theone:
244 for p in A(self, g)['players']: 244 ↛ 248line 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!")
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)