Coverage for server/parts/game.py : 79%
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 typing_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:
157 return None
159 if ev.gid != core.room.gid_of(g):
160 return
162 pkt = Ag(self, g)['data'][u].feed_recv(ev.tag, ev.data)
163 if pkt:
164 core.events.game_data_recv.emit((g, u, pkt))
166 # ----- Private Methods -----
167 def _setup_game(self, g: ServerGame) -> None:
168 core = self.core
169 users = core.room.users_of(g)
170 gid = core.room.gid_of(g)
172 Ag(self, g)['data'] = {
173 u: GameData(gid) for u in users
174 }
175 Ag(self, g)['players'] = self._build_players(g, users)
177 def _build_players(self, g: ServerGame, users: List[Client]) -> BatchList[Player]:
178 core = self.core
179 pl: BatchList[Player] = BatchList([HumanPlayer(g, core.auth.uid_of(u), u) for u in users])
180 pl[:0] = [NPCPlayer(g, i.name, i.input_handler) for i in g.npc_players]
182 return pl
184 # ----- Public Methods -----
185 def create_game(self, cls: Type[ServerGame]) -> ServerGame:
186 g = cls()
188 seed = random.getrandbits(63)
189 g.random = random.Random(seed)
191 g._[self] = GamePartAssocOnGame({
192 'params': {k: v[0] for k, v in cls.params_def.items()},
193 'players': BatchList(),
194 'fleed': defaultdict(bool),
195 'aborted': False,
196 'crashed': False,
197 'rngseed': seed,
198 'data': {},
199 'winners': [],
200 'halt': None,
201 })
203 return g
205 def replay(self, u: Client, to: Client) -> None:
206 core = self.core
207 g = Au(self, u)['game']
208 assert g
209 pkts = Ag(self, g)['data'][u].get_sent()
210 if not pkts:
211 return
213 gid = core.room.gid_of(g)
215 to.write_bulk([
216 wire.GameData(gid, p.tag, p.data)
217 for p in pkts
218 ])
219 to.write(wire.GameData(gid, "__game_live", None))
221 def mark_crashed(self, g: ServerGame) -> None:
222 Ag(self, g)['crashed'] = True
224 def is_crashed(self, g: ServerGame) -> bool:
225 return Ag(self, g)['crashed']
227 def mark_aborted(self, g: ServerGame) -> None:
228 Ag(self, g)['aborted'] = True
230 def is_aborted(self, g: ServerGame) -> bool:
231 return Ag(self, g)['aborted']
233 def is_fleed(self, g: ServerGame, u: Client) -> bool:
234 return Ag(self, g)['fleed'][u]
236 def get_gamedata_archive(self, g: ServerGame) -> List[GameArchive]:
237 core = self.core
238 ul = core.room.users_of(g)
239 return [
240 Ag(self, g)['data'][u].archive() for u in ul
241 ]
243 def write(self, g: ServerGame, u: Client, tag: str, data: object) -> None:
244 core = self.core
245 assert Au(self, u)['game'] is g
246 pkt = Ag(self, g)['data'][u].feed_send(tag, data)
247 gid = core.room.gid_of(g)
248 u.write(wire.GameData(gid=gid, tag=tag, data=data))
249 core.events.game_data_send.emit((g, u, pkt))
251 def current(self, u: Client) -> Optional[ServerGame]:
252 g = Au(self, u)['game']
253 return g
255 def set_winners(self, g: ServerGame, winners: List[Player]) -> None:
256 Ag(self, g)['winners'] = winners
258 def set_param(self, g: ServerGame, key: str, value: Any) -> bool:
259 if Ag(self, g)['params'][key] == value: 259 ↛ 260line 259 didn't jump to line 260, because the condition on line 259 was never true
260 return False
262 Ag(self, g)['params'][key] = value
263 return True
265 def params_of(self, g: ServerGame) -> Dict[str, Any]:
266 return Ag(self, g)['params']
268 def winners_of(self, g: ServerGame) -> List[Player]:
269 return Ag(self, g)['winners']
271 def rngseed_of(self, g: ServerGame) -> int:
272 return Ag(self, g)['rngseed']
274 def gamedata_of(self, g: ServerGame, u: Client) -> GameData:
275 return Ag(self, g)['data'][u]
277 def players_of(self, g: ServerGame) -> BatchList[Player]:
278 return Ag(self, g)['players']