# Released under the MIT License. See LICENSE for details.
#
"""Functionality related to the server manager script."""
from __future__ import annotations
from enum import Enum
from dataclasses import field, dataclass
from typing import TYPE_CHECKING, Any
from efro.dataclassio import ioprepped
if TYPE_CHECKING:
pass
[docs]
@ioprepped
@dataclass
class ServerConfig:
"""Configuration for the server manager app (<appname>_server)."""
# Name of our server in the public parties list.
party_name: str = 'FFA'
# If True, your party will show up in the global public party list
# Otherwise it will still be joinable via LAN or connecting by IP
# address.
party_is_public: bool = True
# If True, all connecting clients will be authenticated through the
# master server to screen for fake account info. Generally this
# should always be enabled unless you are hosting on a LAN with no
# internet connection.
authenticate_clients: bool = True
# IDs of server admins. Server admins are not kickable through the
# default kick vote system and they are able to kick players without
# a vote. To get your account id, enter 'getaccountid' in
# settings->advanced->enter-code.
admins: list[str] = field(default_factory=list)
# Whether the default kick-voting system is enabled.
enable_default_kick_voting: bool = True
# To be included in the public server list, your server MUST be
# accessible via an ipv4 address. By default, the master server will
# try to use the address your server contacts it from, but this may
# be an ipv6 address these days so you may need to provide an ipv4
# address explicitly.
public_ipv4_address: str | None = None
# You can optionally provide an ipv6 address for your server for the
# public server list. Unlike ipv4, a server is not required to have
# an ipv6 address to appear in the list, but is still good to
# provide when available since more and more devices are using ipv6
# these days. Your server's ipv6 address will be autodetected if
# your server uses ipv6 when communicating with the master server. You
# can pass an empty string here to explicitly disable the ipv6
# address.
public_ipv6_address: str | None = None
# UDP port to host on. Change this to work around firewalls or run
# multiple servers on one machine.
#
# 43210 is the default and the only port that will show up in the
# LAN browser tab.
port: int = 43210
# Max devices in the party. Note that this does *NOT* mean max
# players. Any device in the party can have more than one player on
# it if they have multiple controllers. Also, this number currently
# includes the server so generally make it 1 bigger than you need.
max_party_size: int = 6
# Max players that can join a session. If present this will override
# the session's preferred max_players. if a value below 0 is given
# player limit will be removed.
session_max_players_override: int | None = None
# Options here are 'ffa' (free-for-all), 'teams' and 'coop'
# (cooperative) This value is ignored if you supply a playlist_code
# (see below).
session_type: str = 'ffa'
# Playlist-code for teams or free-for-all mode sessions. To host
# your own custom playlists, use the 'share' functionality in the
# playlist editor in the regular version of the game. This will give
# you a numeric code you can enter here to host that playlist.
playlist_code: int | None = None
# Alternately, you can embed playlist data here instead of using
# codes. Make sure to set session_type to the correct type for the
# data here.
playlist_inline: list[dict[str, Any]] | None = None
# Whether to shuffle the playlist or play its games in designated
# order.
playlist_shuffle: bool = True
# If True, keeps team sizes equal by disallowing joining the largest
# team (teams mode only).
auto_balance_teams: bool = True
# The campaign used when in co-op session mode. Do
# print(ba.app.campaigns) to see available campaign names.
coop_campaign: str = 'Easy'
# The level name within the campaign used in co-op session mode. For
# campaign name FOO, do print(ba.app.campaigns['FOO'].levels) to see
# available level names.
coop_level: str = 'Onslaught Training'
# Whether to enable telnet access.
#
# IMPORTANT: This option is no longer available, as it was being
# used for exploits. Live access to the running server is still
# possible through the mgr.cmd() function in the server script. Run
# your server through tools such as 'screen' or 'tmux' and you can
# reconnect to it remotely over a secure ssh connection.
enable_telnet: bool = False
# Series length in teams mode (7 == 'best-of-7' series; a team must
# get 4 wins)
teams_series_length: int = 7
# Points to win in free-for-all mode (Points are awarded per game
# based on performance)
ffa_series_length: int = 24
# If you have a custom stats webpage for your server, you can use
# this to provide a convenient in-game link to it in the
# server-browser alongside the server name.
#
# if ${ACCOUNT} is present in the string, it will be replaced by the
# currently-signed-in account's id. To fetch info about an account,
# your back-end server can use the following url:
# https://legacy.ballistica.net/accountquery?id=ACCOUNT_ID_HERE
stats_url: str | None = None
# If present, the server subprocess will attempt to gracefully exit
# after this amount of time. A graceful exit can occur at the end of
# a series or other opportune time. Server-managers set to
# auto-restart (the default) will then spin up a fresh subprocess.
# This mechanism can be useful to clear out any memory leaks or
# other accumulated bad state in the server subprocess.
clean_exit_minutes: float | None = None
# If present, the server subprocess will shut down immediately after
# this amount of time. This can be useful as a fallback for
# clean_exit_time. The server manager will then spin up a fresh
# server subprocess if auto-restart is enabled (the default).
unclean_exit_minutes: float | None = None
# If present, the server subprocess will shut down immediately if
# this amount of time passes with no activity from any players. The
# server manager will then spin up a fresh server subprocess if
# auto-restart is enabled (the default).
idle_exit_minutes: float | None = None
# Should the tutorial be shown at the beginning of games?
show_tutorial: bool = False
# Team names (teams mode only).
team_names: tuple[str, str] | None = None
# Team colors (teams mode only).
team_colors: (
tuple[tuple[float, float, float], tuple[float, float, float]] | None
) = None
# Whether to enable the queue where players can line up before
# entering your server. Disabling this can be used as a workaround
# to deal with queue spamming attacks.
enable_queue: bool = True
# Protocol version we host with. Currently the default is 33 which
# still allows older 1.4 game clients to connect. Explicitly setting
# to 35 no longer allows those clients but adds/fixes a few things
# such as making camera shake properly work in net games.
protocol_version: int | None = None
# (internal) stress-testing mode.
stress_test_players: int | None = None
# How many seconds individual players from a given account must wait
# before rejoining the game. This can help suppress exploits
# involving leaving and rejoining or switching teams rapidly.
player_rejoin_cooldown: float = 10.0
# Log levels for particular loggers, overriding the engine's
# defaults. Valid values are NOTSET, DEBUG, INFO, WARNING, ERROR, or
# CRITICAL.
log_levels: dict[str, str] | None = None
# NOTE: as much as possible, communication from the server-manager to
# the child-process should go through these and not ad-hoc Python string
# commands since this way is type safe.
[docs]
class ServerCommand:
"""Base class for commands that can be sent to the server."""
[docs]
@dataclass
class StartServerModeCommand(ServerCommand):
"""Tells the app to switch into 'server' mode."""
config: ServerConfig
[docs]
class ShutdownReason(Enum):
"""Reason a server is shutting down."""
NONE = 'none'
RESTARTING = 'restarting'
[docs]
@dataclass
class ShutdownCommand(ServerCommand):
"""Tells the server to shut down."""
reason: ShutdownReason
immediate: bool
[docs]
@dataclass
class ChatMessageCommand(ServerCommand):
"""Chat message from the server."""
message: str
clients: list[int] | None
[docs]
@dataclass
class ScreenMessageCommand(ServerCommand):
"""Screen-message from the server."""
message: str
color: tuple[float, float, float] | None
clients: list[int] | None
[docs]
@dataclass
class ClientListCommand(ServerCommand):
"""Print a list of clients."""
[docs]
@dataclass
class KickCommand(ServerCommand):
"""Kick a client."""
client_id: int
ban_time: int | None
# Docs-generation hack; import some stuff that we likely only forward-declared
# in our actual source code so that docs tools can find it.
from typing import (Coroutine, Any, Literal, Callable,
Generator, Awaitable, Sequence, Self)
import asyncio
from concurrent.futures import Future
from pathlib import Path
from enum import Enum