# Coverage for client/parts/replay.py : 58%

## 34 statements 23 run 11 missing 2 excluded 0 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 typing import Any, Dict, List, TYPE_CHECKING

6 import zlib

7

8# -- third party --

9 from mypy_extensions import TypedDict

10 import msgpack

11

12# -- own --

13 from client.base import Game

14

15# -- typing --

16 if TYPE_CHECKING:

17 from client.core import Core # noqa: F401

18

19

20# -- code --

21 class ReplayFile(TypedDict):

22 version: int

23 cliver: str

24 mode: str

25 name: str

26 params: Dict[str, Any]

27 items: Dict[int, List[str]]

28 users: List[int]

29 me: int

30 data: Any # FIXME

31 gid: int

32

33

34 class Replay(object):

35 def __init__(self, core: Core):

36 self.core = core

37

38 def dumps(self, g: Game) -> bytes:

39 core = self.core

40 uid = core.auth.uid

41 uids = [p.uid for p in core.game.players_of(g)]

42

43 rep: ReplayFile = {

44 'version': 1,

45 'cliver': core.warpgate.current_git_version(),

46 'mode': g.__class__.__name__,

47 'name': core.game.name_of(g),

48 'params': core.game.params_of(g),

49 'items': {

50 p.uid: [i.sku for i in items]

51 for p, items in core.game.items_of(g).items()

52 },

53 'users': uids,

54 'me': uid,

55 'data': core.game.gamedata_of(g).archive(),

56 'gid': core.game.gid_of(g),

57 }

58

59 return zlib.compress(msgpack.packb(rep, use_bin_type=True))

60

61 def loads(self, s: bytes) -> ReplayFile:

62 s = msgpack.unpackb(zlib.decompress(s), encoding='utf-8')

63 return s

64

65 def start_replay(self, rep: ReplayFile) -> None:

66 core = self.core

67 g = core.game.create_game(

68 rep['gid'],

69 rep['mode'],

70 rep['name'],

71 [{'uid': i, 'state': 'game'} for i in rep['users']],

72 rep['params'],

73 rep['items'],

74 )

75 core.game.gamedata_of(g).feed_archive(rep['data'])

76 core.events.game_started.emit(g)