Coverage for server/parts/game.py : 84%
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 collections import defaultdict
6from typing import Any, Dict, List, Optional, TYPE_CHECKING, Tuple, Type
7import logging
8import random
10# -- third party --
11from gevent.event import AsyncResult
12from mypy_extensions import TypedDict
14# -- own --
15from game.base import GameArchive, GameData, Player
16from server.base import Game as ServerGame, HumanPlayer, NPCPlayer
17from server.endpoint import Client
18from server.utils import command
19from utils.misc import BatchList
20import wire
22# -- typing --
23if TYPE_CHECKING:
24 from server.core import Core # noqa: F401
27# -- code --
28log = logging.getLogger('Game')
31class GamePartAssocOnClient(TypedDict):
32 game: Optional[ServerGame]
33 params: Dict[str, Any]
36def Au(self: GamePart, u: Client) -> GamePartAssocOnClient:
37 return u._[self]
40class GamePartAssocOnGame(TypedDict):
41 params: Dict[str, Any]
42 players: BatchList[Player]
43 fleed: Dict[Client, bool]
44 aborted: bool
45 crashed: bool
46 rngseed: int
47 data: Dict[Client, GameData]
48 winners: List[Player]
49 halt: Optional[AsyncResult]
52def Ag(self: GamePart, g: ServerGame) -> GamePartAssocOnGame:
53 return g._[self]
56class GamePart(object):
57 def __init__(self, core: Core):
58 self.core = core
60 core.events.game_started += self.handle_game_started
61 core.events.game_ended += self.handle_game_ended
62 core.events.game_joined += self.handle_game_joined
63 core.events.game_left += self.handle_game_left
64 core.events.game_successive_create += self.handle_game_successive_create
65 core.events.user_state_transition += self.handle_user_state_transition
66 core.events.client_pivot += self.handle_client_pivot
68 D = core.events.client_command
69 D[wire.GameData] += self._gamedata
71 def __repr__(self) -> str:
72 return self.__class__.__name__
74 def handle_user_state_transition(self, ev: Tuple[Client, str, str]) -> Tuple[Client, str, str]:
75 u, f, t = ev
76 if t == 'lobby':
77 u._[self] = GamePartAssocOnClient(game=None, params={})
79 return ev
81 def handle_client_pivot(self, u: Client) -> Client:
82 core = self.core
83 if core.lobby.state_of(u) == 'game':
84 g = Au(self, u)['game']
85 assert g
86 assert u in core.room.users_of(g)
88 u.write(wire.GameJoined(core.view.GameDetail(g)))
89 u.write(wire.GameStarted(core.view.GameDetail(g)))
91 self.replay(u, u)
93 return u
95 def handle_game_started(self, g: ServerGame) -> ServerGame:
96 self._setup_game(g)
98 core = self.core
99 users = core.room.users_of(g)
100 for u in users:
101 u.write(wire.GameStarted(core.view.GameDetail(g)))
103 return g
105 def handle_game_ended(self, g: ServerGame) -> ServerGame:
106 core = self.core
107 users = core.room.online_users_of(g)
108 for u in users:
109 u.write(wire.GameEnded(core.room.gid_of(g)))
111 return g
113 def handle_game_joined(self, ev: Tuple[ServerGame, Client]) -> Tuple[ServerGame, Client]:
114 g, u = ev
115 core = self.core
116 if core.room.is_started(g): 116 ↛ 117line 116 didn't jump to line 117, because the condition on line 116 was never true
117 Ag(self, g)['data'][u].revive()
118 Ag(self, g)['fleed'][u] = False
120 Au(self, u)['game'] = g
122 return ev
124 def handle_game_left(self, ev: Tuple[ServerGame, Client]) -> Tuple[ServerGame, Client]:
125 g, u = ev
126 core = self.core
127 if core.room.is_started(g):
128 Ag(self, g)['data'][u].die()
129 uid = core.auth.uid_of(u)
130 p = next(i for i in Ag(self, g)['players'] if i.uid == uid) 130 ↛ exitline 130 didn't finish the generator expression on line 130
131 Ag(self, g)['fleed'][u] = bool(g.can_leave(p))
133 Au(self, u)['game'] = None
135 return ev
137 def handle_game_successive_create(self, ev: Tuple[ServerGame, ServerGame]) -> Tuple[ServerGame, ServerGame]:
138 old, g = ev
139 core = self.core
141 assert old.ended
142 assert not g.ended
144 params = Ag(self, old)['params']
145 Ag(self, g)['params'] = params
146 gid = core.room.gid_of(g)
147 core.room.online_users_of(old).write(wire.GameParams(gid=gid, params=params))
149 return ev
151 # ----- Commands -----
152 @command('game')
153 def _gamedata(self, u: Client, ev: wire.GameData) -> None:
154 core = self.core
155 g = Au(self, u)['game']
156 if not g: 156 ↛ 157line 156 didn't jump to line 157, because the condition on line 156 was never true
157 return None
159 if ev.gid != core.room.gid_of(g): 159 ↛ 160line 159 didn't jump to line 160, because the condition on line 159 was never true
160 return
162 pkt = Ag(self, g)['data'][u].feed_recv(ev.tag, ev.data)
163 core.events.game_data_recv.emit((g, u, pkt))
165 # ----- Private Methods -----
166 def _setup_game(self, g: ServerGame) -> None:
167 core = self.core
168 users = core.room.users_of(g)
169 gid = core.room.gid_of(g)
171 Ag(self, g)['data'] = {
172 u: GameData(gid) for u in users
173 }
174 Ag(self, g)['players'] = self._build_players(g, users)
176 def _build_players(self, g: ServerGame, users: List[Client]) -> BatchList[Player]:
177 core = self.core
178 pl: BatchList[Player] = BatchList([HumanPlayer(g, core.auth.uid_of(u), u) for u in users])
179 pl[:0] = [NPCPlayer(g, i.name, i.input_handler) for i in g.npc_players]
181 return pl
183 # ----- Public Methods -----
184 def create_game(self, cls: Type[ServerGame]) -> ServerGame:
185 g = cls()
187 seed = random.getrandbits(63)
188 g.random = random.Random(seed)
190 g._[self] = GamePartAssocOnGame({
191 'params': {k: v[0] for k, v in cls.params_def.items()},
192 'players': BatchList(),
193 'fleed': defaultdict(bool),
194 'aborted': False,
195 'crashed': False,
196 'rngseed': seed,
197 'data': {},
198 'winners': [],
199 'halt': None,
200 })
202 return g
204 def replay(self, u: Client, to: Client) -> None:
205 core = self.core
206 g = Au(self, u)['game']
207 assert g
208 pkts = Ag(self, g)['data'][u].get_sent()
209 if not pkts:
210 return
212 gid = core.room.gid_of(g)
214 to.write_bulk([
215 wire.GameData(gid, p.tag, p.data)
216 for p in pkts
217 ])
218 to.write(wire.GameData(gid, "__game_live", None))
220 def mark_crashed(self, g: ServerGame) -> None:
221 Ag(self, g)['crashed'] = True
223 def is_crashed(self, g: ServerGame) -> bool:
224 return Ag(self, g)['crashed']
226 def mark_aborted(self, g: ServerGame) -> None:
227 Ag(self, g)['aborted'] = True
229 def is_aborted(self, g: ServerGame) -> bool:
230 return Ag(self, g)['aborted']
232 def is_fleed(self, g: ServerGame, u: Client) -> bool:
233 return Ag(self, g)['fleed'][u]
235 def get_gamedata_archive(self, g: ServerGame) -> List[GameArchive]:
236 core = self.core
237 ul = core.room.users_of(g)
238 return [
239 Ag(self, g)['data'][u].archive() for u in ul
240 ]
242 def write(self, g: ServerGame, u: Client, tag: str, data: object) -> None:
243 core = self.core
244 assert Au(self, u)['game'] is g
245 pkt = Ag(self, g)['data'][u].feed_send(tag, data)
246 gid = core.room.gid_of(g)
247 u.write(wire.GameData(gid=gid, tag=tag, data=data))
248 core.events.game_data_send.emit((g, u, pkt))
250 def current(self, u: Client) -> Optional[ServerGame]:
251 g = Au(self, u)['game']
252 return g
254 def set_winners(self, g: ServerGame, winners: List[Player]) -> None:
255 Ag(self, g)['winners'] = winners
257 def set_param(self, g: ServerGame, key: str, value: Any) -> bool:
258 if Ag(self, g)['params'][key] == value: 258 ↛ 259line 258 didn't jump to line 259, because the condition on line 258 was never true
259 return False
261 Ag(self, g)['params'][key] = value
262 return True
264 def params_of(self, g: ServerGame) -> Dict[str, Any]:
265 return Ag(self, g)['params']
267 def winners_of(self, g: ServerGame) -> List[Player]:
268 return Ag(self, g)['winners']
270 def rngseed_of(self, g: ServerGame) -> int:
271 return Ag(self, g)['rngseed']
273 def gamedata_of(self, g: ServerGame, u: Client) -> GameData:
274 return Ag(self, g)['data'][u]
276 def players_of(self, g: ServerGame) -> BatchList[Player]:
277 return Ag(self, g)['players']