# Coverage for server/parts/game.py : 84%

## 176 statements 153 run 23 missing 4 excluded 5 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 collections import defaultdict

6 from typing import Any, Dict, List, Optional, TYPE_CHECKING, Tuple, Type

7 import logging

8 import random

9

10# -- third party --

11 from gevent.event import AsyncResult

12 from mypy_extensions import TypedDict

13

14# -- own --

15 from game.base import GameArchive, GameData, Player

16 from server.base import Game as ServerGame, HumanPlayer, NPCPlayer

17 from server.endpoint import Client

18 from server.utils import command

19 from utils.misc import BatchList

20 import wire

21

22# -- typing --

23 if TYPE_CHECKING:

24 from server.core import Core # noqa: F401

25

26

27# -- code --

28 log = logging.getLogger('Game')

29

30

31 class GamePartAssocOnClient(TypedDict):

32 game: Optional[ServerGame]

33 params: Dict[str, Any]

34

35

36 def Au(self: GamePart, u: Client) -> GamePartAssocOnClient:

37 return u._[self]

38

39

40 class 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]

50

51

52 def Ag(self: GamePart, g: ServerGame) -> GamePartAssocOnGame:

53 return g._[self]

54

55

56 class GamePart(object):

57 def __init__(self, core: Core):

58 self.core = core

59

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

67

68 D = core.events.client_command

69 D[wire.GameData] += self._gamedata

70

71 def __repr__(self) -> str:

72 return self.__class__.__name__

73

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={})

78

79 return ev

80

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)

87

88 u.write(wire.GameJoined(core.view.GameDetail(g)))

89 u.write(wire.GameStarted(core.view.GameDetail(g)))

90

91 self.replay(u, u)

92

93 return u

94

95 def handle_game_started(self, g: ServerGame) -> ServerGame:

96 self._setup_game(g)

97

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)))

102

103 return g

104

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)))

110

111 return g

112

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 ↛ 117 line 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

119

120 Au(self, u)['game'] = g

121

122 return ev

123

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 ↛ exit line 130 didn't finish the generator expression on line 130

131 Ag(self, g)['fleed'][u] = bool(g.can_leave(p))

132

133 Au(self, u)['game'] = None

134

135 return ev

136

137 def handle_game_successive_create(self, ev: Tuple[ServerGame, ServerGame]) -> Tuple[ServerGame, ServerGame]:

138 old, g = ev

139 core = self.core

140

141 assert old.ended

142 assert not g.ended

143

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))

148

149 return ev

150

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 ↛ 157 line 156 didn't jump to line 157, because the condition on line 156 was never true

157 return None

158

159 if ev.gid != core.room.gid_of(g): 159 ↛ 160 line 159 didn't jump to line 160, because the condition on line 159 was never true

160 return

161

162 pkt = Ag(self, g)['data'][u].feed_recv(ev.tag, ev.data)

163 core.events.game_data_recv.emit((g, u, pkt))

164

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)

170

171 Ag(self, g)['data'] = {

172 u: GameData(gid) for u in users

173 }

174 Ag(self, g)['players'] = self._build_players(g, users)

175

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]

180

181 return pl

182

183 # ----- Public Methods -----

184 def create_game(self, cls: Type[ServerGame]) -> ServerGame:

185 g = cls()

186

187 seed = random.getrandbits(63)

188 g.random = random.Random(seed)

189

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 })

201

202 return g

203

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

211

212 gid = core.room.gid_of(g)

213

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))

219

220 def mark_crashed(self, g: ServerGame) -> None:

221 Ag(self, g)['crashed'] = True

222

223 def is_crashed(self, g: ServerGame) -> bool:

224 return Ag(self, g)['crashed']

225

226 def mark_aborted(self, g: ServerGame) -> None:

227 Ag(self, g)['aborted'] = True

228

229 def is_aborted(self, g: ServerGame) -> bool:

230 return Ag(self, g)['aborted']

231

232 def is_fleed(self, g: ServerGame, u: Client) -> bool:

233 return Ag(self, g)['fleed'][u]

234

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 ]

241

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))

249

250 def current(self, u: Client) -> Optional[ServerGame]:

251 g = Au(self, u)['game']

252 return g

253

254 def set_winners(self, g: ServerGame, winners: List[Player]) -> None:

255 Ag(self, g)['winners'] = winners

256

257 def set_param(self, g: ServerGame, key: str, value: Any) -> bool:

258 if Ag(self, g)['params'][key] == value: 258 ↛ 259 line 258 didn't jump to line 259, because the condition on line 258 was never true

259 return False

260

261 Ag(self, g)['params'][key] = value

262 return True

263

264 def params_of(self, g: ServerGame) -> Dict[str, Any]:

265 return Ag(self, g)['params']

266

267 def winners_of(self, g: ServerGame) -> List[Player]:

268 return Ag(self, g)['winners']

269

270 def rngseed_of(self, g: ServerGame) -> int:

271 return Ag(self, g)['rngseed']

272

273 def gamedata_of(self, g: ServerGame, u: Client) -> GameData:

274 return Ag(self, g)['data'][u]

275

276 def players_of(self, g: ServerGame) -> BatchList[Player]:

277 return Ag(self, g)['players']