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    minutes_until_free_host: float | None = None
68    free_host_minutes_remaining: float | None = None
69
70
71@ioprepped
72@dataclass
73class PrivateHostingConfig:
74    """Config provided when hosting a private party."""
75
76    session_type: str = 'ffa'
77    playlist_name: str = 'Unknown'
78    randomize: bool = False
79    tutorial: bool = False
80    custom_team_names: tuple[str, str] | None = None
81    custom_team_colors: (
82        tuple[tuple[float, float, float], tuple[float, float, float]] | None
83    ) = None
84    playlist: list[dict[str, Any]] | None = None
85    exit_minutes: float = 120.0
86    exit_minutes_unclean: float = 180.0
87    exit_minutes_idle: float = 10.0
88
89
90@ioprepped
91@dataclass
92class PrivatePartyConnectResult:
93    """Info about a server we get back when connecting."""
94
95    error: str | None = None
96    address4: Annotated[str | None, IOAttrs('addr')] = None
97    address6: Annotated[str | None, IOAttrs('addr6')] = None
98    port: int | None = None
99    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._base.IOAttrs object>], latlong: Annotated[tuple[float, float] | None, <efro.dataclassio._base.IOAttrs object>], address: Annotated[str, <efro.dataclassio._base.IOAttrs object>], port: Annotated[int, <efro.dataclassio._base.IOAttrs object>])
zone: Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x104913a40>]
latlong: Annotated[tuple[float, float] | None, <efro.dataclassio._base.IOAttrs object at 0x103de49e0>]
address: Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x10399ec90>]
port: Annotated[int, <efro.dataclassio._base.IOAttrs object at 0x104922ed0>]
@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._base.IOAttrs object>], latlong: Annotated[tuple[float, float] | None, <efro.dataclassio._base.IOAttrs object>], ping_per_dist: Annotated[float, <efro.dataclassio._base.IOAttrs object>], max_dist: Annotated[float, <efro.dataclassio._base.IOAttrs object>], debug_log_seconds: Annotated[float | None, <efro.dataclassio._base.IOAttrs object>] = None, error: Annotated[str | None, <efro.dataclassio._base.IOAttrs object>] = None, servers: Annotated[list[ServerNodeEntry], <efro.dataclassio._base.IOAttrs object>] = <factory>)
time: Annotated[datetime.datetime, <efro.dataclassio._base.IOAttrs object at 0x104923ad0>]
latlong: Annotated[tuple[float, float] | None, <efro.dataclassio._base.IOAttrs object at 0x104923d10>]
ping_per_dist: Annotated[float, <efro.dataclassio._base.IOAttrs object at 0x104923ec0>]
max_dist: Annotated[float, <efro.dataclassio._base.IOAttrs object at 0x104923e90>]
debug_log_seconds: Annotated[float | None, <efro.dataclassio._base.IOAttrs object at 0x104988740>] = None
error: Annotated[str | None, <efro.dataclassio._base.IOAttrs object at 0x10498a120>] = None
servers: Annotated[list[ServerNodeEntry], <efro.dataclassio._base.IOAttrs object at 0x104988c80>]
@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    minutes_until_free_host: float | None = None
69    free_host_minutes_remaining: float | None = None

Combined state of whether we're hosting, whether we can, etc.

PrivateHostingState( unavailable_error: str | None = None, party_code: str | None = None, tickets_to_host_now: int = 0, minutes_until_free_host: float | None = None, free_host_minutes_remaining: float | None = None)
unavailable_error: str | None = None
party_code: str | None = None
tickets_to_host_now: int = 0
minutes_until_free_host: float | None = None
free_host_minutes_remaining: float | None = None
@ioprepped
@dataclass
class PrivateHostingConfig:
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

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)
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:
 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

Info about a server we get back when connecting.

PrivatePartyConnectResult( error: str | None = None, address4: Annotated[str | None, <efro.dataclassio._base.IOAttrs object>] = None, address6: Annotated[str | None, <efro.dataclassio._base.IOAttrs object>] = None, port: int | None = None, password: str | None = None)
error: str | None = None
address4: Annotated[str | None, <efro.dataclassio._base.IOAttrs object at 0x10498b440>] = None
address6: Annotated[str | None, <efro.dataclassio._base.IOAttrs object at 0x10498b590>] = None
port: int | None = None
password: str | None = None