misc/infrastructure/python/slist/game.py

   1 import uuid
   2
   3 import attr
   4
   5 from .utils import *
   6
   7 HEADER = b"\xFF\xFF\xFF\xFF"
   8
   9
  10 @attr.s(auto_attribs=True, frozen=True, slots=True)
  11 class CLGetInfo(Writable):
  12     def encode(self) -> bytes:
  13         return HEADER + f"getinfo {uuid.uuid4()}".encode(UTF_8)
  14
  15
  16 @attr.s(auto_attribs=True, frozen=True, slots=True)
  17 class SVGetInfoResponse(Readable):
  18     gamename: str
  19     modname: str
  20     gameversion: int
  21     sv_maxclients: int
  22     clients: int
  23     bots: int
  24     mapname: str
  25     hostname: str
  26     protocol: int
  27     qcstatus: Optional[str]
  28     challenge: Optional[str]
  29     d0_blind_id: Optional[str] = None
  30
  31     @classmethod
  32     @generator
  33     def decode(cls) -> Generator[Optional["SVGetInfoResponse"], bytes, None]:
  34         ret: Optional[SVGetInfoResponse] = None
  35         while True:
  36             buf: bytes
  37             buf = yield ret
  38             parts = buf.decode(UTF_8).split("\\")[1:]
  39             pairs = zip(*[iter(parts)] * 2)
  40             args = dict(pairs)
  41             for k in ("gameversion", "sv_maxclients", "clients", "bots", "protocol"):
  42                 args[k] = int(args[k])
  43             ret = SVGetInfoResponse(**args)
  44
  45
  46 SVMessage = Union[SVGetInfoResponse]
  47
  48
  49 @generator
  50 def sv_parse() -> Generator[Optional[SVMessage], bytes, None]:
  51     getinfo_response = b"infoResponse\n"
  52     ret: Optional[SVMessage] = None
  53     while True:
  54         buf: bytes
  55         buf = yield ret
  56         ret = None
  57         if buf.startswith(HEADER):
  58             buf = buf[len(HEADER):]
  59             if buf.startswith(getinfo_response):
  60                 buf = buf[len(getinfo_response):]
  61                 ret = SVGetInfoResponse.decode().send(buf)
  62                 continue