|
7 | 7 | import trio
|
8 | 8 |
|
9 | 9 |
|
10 |
| -def unpack_string(data: bytes) -> tuple[str, bytes]: |
11 |
| - str_len, data = *struct.unpack_from('<I', data), data[4:] |
| 10 | +def unpack_string(data: bytes, len_type: str = 'I') -> tuple[str, bytes]: |
| 11 | + format = f'<{len_type}' |
| 12 | + size = struct.calcsize(format) |
| 13 | + str_len, data = *struct.unpack_from(format, data), data[size:] |
12 | 14 | string, data = data[:str_len], data[str_len:]
|
13 | 15 | encoding: str = chardet.detect(string)['encoding']
|
14 | 16 | return string.decode(encoding), data
|
@@ -43,6 +45,42 @@ def from_data(cls, data: bytes) -> ServerInfo:
|
43 | 45 | )
|
44 | 46 |
|
45 | 47 |
|
| 48 | +@dataclass |
| 49 | +class PlayerInfo: |
| 50 | + name: str |
| 51 | + score: int |
| 52 | + |
| 53 | + @classmethod |
| 54 | + def from_data(cls, data: bytes) -> tuple[PlayerInfo, bytes]: |
| 55 | + name, data = unpack_string(data, 'B') |
| 56 | + score = struct.unpack_from('<i', data)[0] |
| 57 | + data = data[4:] # int, see above |
| 58 | + |
| 59 | + return cls( |
| 60 | + name=name, |
| 61 | + score=score, |
| 62 | + ), data |
| 63 | + |
| 64 | + |
| 65 | +@dataclass |
| 66 | +class PlayerList: |
| 67 | + players: list[PlayerInfo] |
| 68 | + |
| 69 | + @classmethod |
| 70 | + def from_data(cls, data: bytes) -> PlayerList: |
| 71 | + player_count = struct.unpack_from('<H', data)[0] |
| 72 | + data = data[2:] # short, see above |
| 73 | + players = [] |
| 74 | + |
| 75 | + for _ in range(player_count): |
| 76 | + player, data = PlayerInfo.from_data(data) |
| 77 | + players.append(player) |
| 78 | + |
| 79 | + assert not data # We consumed all the buffer |
| 80 | + |
| 81 | + return cls(players=players) |
| 82 | + |
| 83 | + |
46 | 84 | @dataclass
|
47 | 85 | class Client:
|
48 | 86 | ip: str
|
@@ -98,3 +136,9 @@ async def info(self) -> ServerInfo:
|
98 | 136 | assert self._prefix
|
99 | 137 | data = await self.receive(header=self._prefix + b'i')
|
100 | 138 | return ServerInfo.from_data(data)
|
| 139 | + |
| 140 | + async def players(self) -> PlayerList: |
| 141 | + await self.send(b'c') |
| 142 | + assert self._prefix |
| 143 | + data = await self.receive(header=self._prefix + b'c') |
| 144 | + return PlayerList.from_data(data) |
0 commit comments