8 logger = logging.getLogger(__name__)
14 sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
16 connections: Dict[Connection, Union[
17 Generator[Optional[master.SVMessage], Tuple[Connection, bytes], None],
18 Generator[Tuple[Optional[game.SVMessage], game.SequenceInfo], Tuple[Connection, bytes], None],
20 count_inforesponse = 0
22 q_master = master.CLGetServersExt(game="Xonotic", protocol=3)
23 conn = (socket.gethostbyname("dpmaster.deathmask.net"), 27950)
24 connections[conn] = master.sv_parse()
25 sock.sendto(q_master.encode(), conn)
27 logger.debug("recv(...)")
29 data, conn = sock.recvfrom(1400)
30 except KeyboardInterrupt:
32 logger.debug(f"recv({conn}): {data}")
33 msg = connections[conn].send((conn, data))
34 if isinstance(msg, tuple):
37 logger.info(f"recv({conn}): {msg}")
38 if isinstance(msg, master.SVGetServersExtResponse):
39 logger.info(f"servers: {len(msg.servers)}")
40 for srv in msg.servers:
41 conn = (str(srv.addr), srv.port)
42 q_server = game.CLGetInfo()
43 connections[conn] = game.sv_parse()
45 sock.sendto(q_server.encode(), conn)
46 except socket.gaierror:
48 if isinstance(msg, game.SVGetInfoResponse):
49 count_inforesponse += 1
50 logger.info(f"status-{count_inforesponse}: {msg}")
53 if __name__ == "__main__":