Coverage for client/parts/replay.py : 58%
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 typing import Any, Dict, List, TYPE_CHECKING
6import zlib
8# -- third party --
9from mypy_extensions import TypedDict
10import msgpack
12# -- own --
13from client.base import Game
15# -- typing --
16if TYPE_CHECKING:
17 from client.core import Core # noqa: F401
20# -- code --
21class 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
34class Replay(object):
35 def __init__(self, core: Core):
36 self.core = core
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)]
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 }
59 return zlib.compress(msgpack.packb(rep, use_bin_type=True))
61 def loads(self, s: bytes) -> ReplayFile:
62 s = msgpack.unpackb(zlib.decompress(s), encoding='utf-8')
63 return s
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)