# Coverage for client/base.py : 89%

## 161 statements 146 run 15 missing 2 excluded 6 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 copy import copy

6 from random import Random

7 from typing import Any, Optional, Sequence, TYPE_CHECKING, cast

8 import logging

9

10# -- third party --

11 from gevent import Greenlet

12 import gevent

13

14# -- own --

15 from game.base import Game, GameAbort, GameEnded, GameRunner, InputTransaction, Inputlet, Player

16 from game.base import TimeLimitExceeded

17

18# -- typing --

19 if TYPE_CHECKING:

20 from client.core import Core # noqa: F401

21

22

23# -- code --

24 log = logging.getLogger('client.base')

25

26

27 class ForcedKill(gevent.GreenletExit):

28 pass

29

30

31 class Theone(Player):

32

33 def __init__(self, game: Game, uid: int):

34 Player.__init__(self)

35 self.game = game

36 self.uid = uid

37

38 def reveal(self, obj: Any) -> None:

39 # It's me, server will tell me what the hell these is.

40 g = self.game

41 core = cast(ClientGameRunner, g.runner).core

42 st = g.get_synctag()

43 _, raw = core.game.gamedata_of(g).gexpect([f'Sync:{st}'])

44 if isinstance(obj, (list, tuple)):

45 for o, rd in zip(obj, raw):

46 o.sync(rd)

47 else:

48 obj.sync(raw)

49

50

51 class Someone(Player):

52

53 def __init__(self, game: Game, uid: int):

54 Player.__init__(self)

55 self.game = game

56 self.uid = uid

57

58 def reveal(self, ol: Any) -> None:

59 # Peer player, won't reveal.

60 self.game.get_synctag() # must sync

61

62

63 class ClientGameRunner(GameRunner):

64 game: Game

65 core: Core

66

67 def __init__(self, core: Core, g: Game):

68 self.core = core

69 self.game = g

70 super().__init__()

71

72 def _run(self) -> None:

73 g = self.game

74

75 import base64

76 s = (id(self) % 1099511627689).to_bytes(8, byteorder='little')[:5]

77 s = base64.b32encode(s).decode('utf-8')

78 self.gr_name = f'{repr(self.game)}:{s[:5]}'

79

80 g.runner = self

81 g.synctag = 0

82 g.random = Random()

83 core = self.core

84 core.events.game_started.emit(g)

85 params = core.game.params_of(g)

86 items = core.game.items_of(g)

87 players = core.game.players_of(g)

88

89 try:

90 g.process_action(g.bootstrap(params, items, players))

91 except GameEnded as e: 91 ↛ 93 line 91 didn't jump to line 93

92 g.winners = e.winners

93 except GameAbort:

94 pass

95 finally:

96 g.ended = True

97

98 core.events.client_game_finished.emit(g)

99

100 def pause(self, time: float) -> None:

101 core = self.core

102 if not core.options.testing: 102 ↛ 103 line 102 didn't jump to line 103, because the condition on line 102 was never true

103 core.runner.sleep(time)

104

105 def is_dropped(self, p: Player) -> bool:

106 core = self.core

107 return core.game.is_dropped(self.game, p)

108

109 def is_aborted(self) -> bool:

110 return False

111

112 def get_side(self) -> str:

113 return 'client'

114

115 def user_input(

116 self,

117 entities: Sequence[Any],

118 inputlet: Inputlet,

119 timeout: int = 25,

120 type: str = 'single',

121 trans: Optional[InputTransaction] = None,

122 ):

123

124 assert type in ('single', 'all', 'any')

125 assert not type == 'single' or len(entities) == 1

126

127 timeout = max(0, timeout)

128

129 inputlet.timeout = timeout

130 entities = list(entities)

131

132 if not trans:

133 with InputTransaction(inputlet.tag(), entities) as trans:

134 return self.user_input(entities, inputlet, timeout, type, trans)

135

136 g = self.game

137 assert isinstance(g, Game)

138 core = self.core

139

140 t = {'single': '', 'all': '&', 'any': '|'}[type]

141 tag = 'I{0}:{1}:'.format(t, inputlet.tag())

142

143 ilets = {e: copy(inputlet) for e in entities}

144 for e in entities:

145 ilets[e].actor = e

146

147 inputproc: Optional[Greenlet] = None

148

149 me = core.game.theone_of(g)

150

151 results = {e: None for e in entities}

152

153 synctags = {e: g.get_synctag() for e in entities}

154 synctags_r = {v: k for k, v in synctags.items()}

155

156 def get_player(e):

157 if isinstance(e, Player):

158 return e

159 elif hasattr(e, 'get_player'): 159 ↛ 164 line 159 didn't jump to line 164, because the condition on line 159 was never false

160 p = e.get_player()

161 assert isinstance(p, Player), f'{e}.get_player() == {p}, not a Player'

162 return p

163 else:

164 raise Exception(f'Do not know how to process {e}')

165

166 e2p = {e: get_player(e) for e in entities}

167 p2e = {p: e for e, p in e2p.items()}

168

169 def input_func(st: str) -> None:

170 gevent.getcurrent().gr_name = 'InputFunc'

171 my = ilets[p2e[me]]

172 with TimeLimitExceeded(timeout + 1, False):

173 _, my = g.emit_event('user_input', (trans, my))

174

175 core.game.write(g, f'{tag}{st}', my.data())

176

177 try:

178 for e in entities:

179 g.emit_event('user_input_start', (trans, ilets[e]))

180

181 if me in p2e: # me involved

182 if not core.game.is_observe(g): 182 ↛ 193 line 182 didn't jump to line 193, because the condition on line 182 was never false

183 inputproc = core.runner.spawn(input_func, synctags[p2e[me]])

184 inputproc.game = g

185 gr_current = gevent.getcurrent()

186

187 @inputproc.link_exception

188 def chain_failure(gr):

189 exc = Exception("input_func failed")

190 exc.__cause__ = gr.exception

191 gr_current.kill(exc)

192

193 orig_entities = entities[:]

194 inputany_entity = None

195

196 g.emit_event('user_input_begin_wait_resp', trans) # for replay speed control

197 while entities:

198 # should be [tag, <Data for Inputlet.parse>]

199 # tag likes 'RI?:ChooseOption:2345'

200 tag_, data = core.game.gamedata_of(g).gexpect([f'R{tag}{st}' for st in synctags_r])

201 st = int(tag_.split(':')[2])

202 if st not in synctags_r: 202 ↛ 203 line 202 didn't jump to line 203, because the condition on line 202 was never true

203 log.warning('Unexpected sync tag: %d, expecting %s', st, list(synctags_r))

204 continue

205

206 e = synctags_r[st]

207

208 my = ilets[e]

209

210 try:

211 rst = my.parse(data)

212 except Exception:

213 if core.options.testing:

214 raise

215 log.exception('user_input: exception in .process()')

216 rst = None

217

218 rst = my.post_process(e, rst)

219

220 g.emit_event('user_input_finish', (trans, my, rst))

221

222 entities.remove(e)

223 results[e] = rst

224

225 # also remove from synctags

226 del synctags_r[st]

227 del synctags[e]

228

229 if type == 'any' and rst is not None:

230 assert not inputany_entity

231 inputany_entity = e

232

233 g.emit_event('user_input_end_wait_resp', trans) # for replay speed control

234

235 finally:

236 inputproc and [inputproc.kill(), inputproc.join()]

237

238 if type == 'single':

239 return results[orig_entities[0]]

240

241 elif type == 'any':

242 if not inputany_entity:

243 return None, None

244

245 return inputany_entity, results[inputany_entity]

246

247 elif type == 'all': 247 ↛ 251 line 247 didn't jump to line 251, because the condition on line 247 was never false

248 # return OrderedDict([(i, results[i]) for i in orig_entities])

249 return {e: results[e] for e in orig_entities}

250

251 assert False, 'WTF?!'