bacommon.net
Network related data and functionality.
1# Released under the MIT License. See LICENSE for details. 2# 3"""Network related data and functionality.""" 4 5from __future__ import annotations 6 7import datetime 8from typing import TYPE_CHECKING, Any, Annotated 9from dataclasses import dataclass, field 10 11from efro.dataclassio import ioprepped, IOAttrs 12 13if TYPE_CHECKING: 14 pass 15 16 17@ioprepped 18@dataclass 19class ServerNodeEntry: 20 """Information about a specific server.""" 21 22 zone: Annotated[str, IOAttrs('r')] 23 24 # TODO: Remove soft_default after all master-servers upgraded. 25 latlong: Annotated[ 26 tuple[float, float] | None, IOAttrs('ll', soft_default=None) 27 ] 28 address: Annotated[str, IOAttrs('a')] 29 port: Annotated[int, IOAttrs('p')] 30 31 32@ioprepped 33@dataclass 34class ServerNodeQueryResponse: 35 """A response to a query about server-nodes.""" 36 37 # The current utc time on the master server. 38 time: Annotated[datetime.datetime, IOAttrs('t')] 39 40 # Where the master server sees the query as coming from. 41 latlong: Annotated[tuple[float, float] | None, IOAttrs('ll')] 42 43 ping_per_dist: Annotated[float, IOAttrs('ppd')] 44 max_dist: Annotated[float, IOAttrs('md')] 45 46 debug_log_seconds: Annotated[ 47 float | None, IOAttrs('d', store_default=False) 48 ] = None 49 50 # If present, something went wrong, and this describes it. 51 error: Annotated[str | None, IOAttrs('e', store_default=False)] = None 52 53 # The set of servernodes. 54 servers: Annotated[ 55 list[ServerNodeEntry], IOAttrs('s', store_default=False) 56 ] = field(default_factory=list) 57 58 59@ioprepped 60@dataclass 61class PrivateHostingState: 62 """Combined state of whether we're hosting, whether we can, etc.""" 63 64 unavailable_error: str | None = None 65 party_code: str | None = None 66 tickets_to_host_now: int = 0 67 tokens_to_host_now: int = 0 68 minutes_until_free_host: float | None = None 69 free_host_minutes_remaining: float | None = None 70 71 72@ioprepped 73@dataclass 74class PrivateHostingConfig: 75 """Config provided when hosting a private party.""" 76 77 session_type: str = 'ffa' 78 playlist_name: str = 'Unknown' 79 randomize: bool = False 80 tutorial: bool = False 81 custom_team_names: tuple[str, str] | None = None 82 custom_team_colors: ( 83 tuple[tuple[float, float, float], tuple[float, float, float]] | None 84 ) = None 85 playlist: list[dict[str, Any]] | None = None 86 exit_minutes: float = 120.0 87 exit_minutes_unclean: float = 180.0 88 exit_minutes_idle: float = 10.0 89 90 91@ioprepped 92@dataclass 93class PrivatePartyConnectResult: 94 """Info about a server we get back when connecting.""" 95 96 error: str | None = None 97 address4: Annotated[str | None, IOAttrs('addr')] = None 98 address6: Annotated[str | None, IOAttrs('addr6')] = None 99 port: int | None = None 100 password: str | None = None
@ioprepped
@dataclass
class
ServerNodeEntry:
18@ioprepped 19@dataclass 20class ServerNodeEntry: 21 """Information about a specific server.""" 22 23 zone: Annotated[str, IOAttrs('r')] 24 25 # TODO: Remove soft_default after all master-servers upgraded. 26 latlong: Annotated[ 27 tuple[float, float] | None, IOAttrs('ll', soft_default=None) 28 ] 29 address: Annotated[str, IOAttrs('a')] 30 port: Annotated[int, IOAttrs('p')]
Information about a specific server.
ServerNodeEntry( zone: Annotated[str, <efro.dataclassio.IOAttrs object>], latlong: Annotated[tuple[float, float] | None, <efro.dataclassio.IOAttrs object>], address: Annotated[str, <efro.dataclassio.IOAttrs object>], port: Annotated[int, <efro.dataclassio.IOAttrs object>])
zone: Annotated[str, <efro.dataclassio.IOAttrs object at 0x1035bf710>]
latlong: Annotated[tuple[float, float] | None, <efro.dataclassio.IOAttrs object at 0x10430bd10>]
address: Annotated[str, <efro.dataclassio.IOAttrs object at 0x10430b8f0>]
port: Annotated[int, <efro.dataclassio.IOAttrs object at 0x10430a0f0>]
@ioprepped
@dataclass
class
ServerNodeQueryResponse:
33@ioprepped 34@dataclass 35class ServerNodeQueryResponse: 36 """A response to a query about server-nodes.""" 37 38 # The current utc time on the master server. 39 time: Annotated[datetime.datetime, IOAttrs('t')] 40 41 # Where the master server sees the query as coming from. 42 latlong: Annotated[tuple[float, float] | None, IOAttrs('ll')] 43 44 ping_per_dist: Annotated[float, IOAttrs('ppd')] 45 max_dist: Annotated[float, IOAttrs('md')] 46 47 debug_log_seconds: Annotated[ 48 float | None, IOAttrs('d', store_default=False) 49 ] = None 50 51 # If present, something went wrong, and this describes it. 52 error: Annotated[str | None, IOAttrs('e', store_default=False)] = None 53 54 # The set of servernodes. 55 servers: Annotated[ 56 list[ServerNodeEntry], IOAttrs('s', store_default=False) 57 ] = field(default_factory=list)
A response to a query about server-nodes.
ServerNodeQueryResponse( time: Annotated[datetime.datetime, <efro.dataclassio.IOAttrs object>], latlong: Annotated[tuple[float, float] | None, <efro.dataclassio.IOAttrs object>], ping_per_dist: Annotated[float, <efro.dataclassio.IOAttrs object>], max_dist: Annotated[float, <efro.dataclassio.IOAttrs object>], debug_log_seconds: Annotated[float | None, <efro.dataclassio.IOAttrs object>] = None, error: Annotated[str | None, <efro.dataclassio.IOAttrs object>] = None, servers: Annotated[list[ServerNodeEntry], <efro.dataclassio.IOAttrs object>] = <factory>)
time: Annotated[datetime.datetime, <efro.dataclassio.IOAttrs object at 0x1042ede20>]
latlong: Annotated[tuple[float, float] | None, <efro.dataclassio.IOAttrs object at 0x1042ee150>]
ping_per_dist: Annotated[float, <efro.dataclassio.IOAttrs object at 0x1042ee330>]
max_dist: Annotated[float, <efro.dataclassio.IOAttrs object at 0x1042ee450>]
servers: Annotated[list[ServerNodeEntry], <efro.dataclassio.IOAttrs object at 0x1042eed50>]
@ioprepped
@dataclass
class
PrivateHostingState:
60@ioprepped 61@dataclass 62class PrivateHostingState: 63 """Combined state of whether we're hosting, whether we can, etc.""" 64 65 unavailable_error: str | None = None 66 party_code: str | None = None 67 tickets_to_host_now: int = 0 68 tokens_to_host_now: int = 0 69 minutes_until_free_host: float | None = None 70 free_host_minutes_remaining: float | None = None
Combined state of whether we're hosting, whether we can, etc.
@ioprepped
@dataclass
class
PrivateHostingConfig:
73@ioprepped 74@dataclass 75class PrivateHostingConfig: 76 """Config provided when hosting a private party.""" 77 78 session_type: str = 'ffa' 79 playlist_name: str = 'Unknown' 80 randomize: bool = False 81 tutorial: bool = False 82 custom_team_names: tuple[str, str] | None = None 83 custom_team_colors: ( 84 tuple[tuple[float, float, float], tuple[float, float, float]] | None 85 ) = None 86 playlist: list[dict[str, Any]] | None = None 87 exit_minutes: float = 120.0 88 exit_minutes_unclean: float = 180.0 89 exit_minutes_idle: float = 10.0
Config provided when hosting a private party.
PrivateHostingConfig( session_type: str = 'ffa', playlist_name: str = 'Unknown', randomize: bool = False, tutorial: bool = False, custom_team_names: tuple[str, str] | None = None, custom_team_colors: tuple[tuple[float, float, float], tuple[float, float, float]] | None = None, playlist: list[dict[str, typing.Any]] | None = None, exit_minutes: float = 120.0, exit_minutes_unclean: float = 180.0, exit_minutes_idle: float = 10.0)
@ioprepped
@dataclass
class
PrivatePartyConnectResult:
92@ioprepped 93@dataclass 94class PrivatePartyConnectResult: 95 """Info about a server we get back when connecting.""" 96 97 error: str | None = None 98 address4: Annotated[str | None, IOAttrs('addr')] = None 99 address6: Annotated[str | None, IOAttrs('addr6')] = None 100 port: int | None = None 101 password: str | None = None
Info about a server we get back when connecting.
PrivatePartyConnectResult( error: str | None = None, address4: Annotated[str | None, <efro.dataclassio.IOAttrs object>] = None, address6: Annotated[str | None, <efro.dataclassio.IOAttrs object>] = None, port: int | None = None, password: str | None = None)