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>]
debug_log_seconds: Annotated[float | None, <efro.dataclassio.IOAttrs object at 0x1042ee4e0>] = None
error: Annotated[str | None, <efro.dataclassio.IOAttrs object at 0x1042ee9f0>] = None
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.

PrivateHostingState( unavailable_error: str | None = None, party_code: str | None = None, tickets_to_host_now: int = 0, tokens_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
tokens_to_host_now: int = 0
minutes_until_free_host: float | None = None
free_host_minutes_remaining: float | None = None
@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)
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)
error: str | None = None
address4: Annotated[str | None, <efro.dataclassio.IOAttrs object at 0x10436dd30>] = None
address6: Annotated[str | None, <efro.dataclassio.IOAttrs object at 0x10436e150>] = None
port: int | None = None
password: str | None = None