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

## 177 statements 147 run 30 missing 4 excluded 3 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 typing_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 ↛ 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

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 ↛ exitline 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:

157 return None

158

159 if ev.gid != core.room.gid_of(g):

160 return

161

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

165

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)

171

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

173 u: GameData(gid) for u in users

174 }

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

176

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]

181

182 return pl

183

184 # ----- Public Methods -----

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

186 g = cls()

187

188 seed = random.getrandbits(63)

189 g.random = random.Random(seed)

190

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

202

203 return g

204

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

212

213 gid = core.room.gid_of(g)

214

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

220

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

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

223

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

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

226

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

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

229

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

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

232

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

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

235

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 ]

242

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

250

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

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

253 return g

254

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

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

257

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

261

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

263 return True

264

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

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

267

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

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

270

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

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

273

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

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

276

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

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