Coverage for client/parts/game.py : 81%
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 )
118 self.games[gid] = g
119 core = self.core
120 core.events.game_joined.emit(g)
121 return ev
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
128 self.kill_game(g)
130 core = self.core
131 core.events.game_left.emit(g)
133 return ev
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
140 log.info('=======GAME ENDED=======')
141 core = self.core
142 core.events.game_ended.emit(g)
144 return ev
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
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)
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
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 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)
226 def gid_of(self, g: Game) -> int:
227 return A(self, g)['gid']
229 def name_of(self, g: Game) -> str:
230 return A(self, g)['name']
232 def gamedata_of(self, g: Game) -> GameData:
233 return A(self, g)['data']
235 def items_of(self, g: Game) -> Dict[Player, List[GameItem]]:
236 return A(self, g)['items']
238 def params_of(self, g: Game) -> dict:
239 return A(self, g)['params']
241 def players_of(self, g: Game) -> BatchList[Player]:
242 return A(self, g)['players']
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!")
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)