""" All adapter capability Protocol definitions. Core code only imports from here — never from adapter internals. """ from __future__ import annotations from pathlib import Path from typing import Any, Callable, Protocol, runtime_checkable from pydantic import BaseModel @runtime_checkable class ConfigGenerator(Protocol): """ Merged protocol: config schema definition + file generation + launch args. Always implement all methods. Return empty dict/list where not applicable. """ game_type: str def get_sections(self) -> dict[str, type[BaseModel]]: """Return {section_name: PydanticModelClass} for all config sections.""" ... def get_defaults(self, section: str) -> dict[str, Any]: """Return default values dict for the given section.""" ... def get_sensitive_fields(self, section: str) -> list[str]: """ Return JSON keys in this section that need Fernet encryption. Core's ConfigRepository encrypts/decrypts these transparently. Example: ["password", "password_admin"] for section "server". """ ... def get_config_version(self) -> str: """ Current adapter schema version string (e.g. "1.0.0"). Stored in game_configs.schema_version. When this changes, core calls migrate_config() automatically. """ ... def migrate_config( self, old_version: str, config_json: dict[str, dict] ) -> dict[str, dict]: """ Transform config JSON from old_version to current version. Called by ConfigRepository when stored schema_version differs. Returns migrated config dict. Raises ConfigMigrationError on failure — core keeps original. """ ... def write_configs( self, server_id: int, server_dir: Path, config_sections: dict[str, dict], ) -> list[Path]: """ Write all config files to disk using atomic write pattern: 1. Write to .tmp files 2. os.replace() each .tmp to final path 3. On any failure: clean up .tmp files, raise ConfigWriteError Returns list of written file paths. """ ... def build_launch_args( self, config_sections: dict[str, dict], mod_args: list[str] | None = None, ) -> list[str]: """ Return full CLI argument list for the game executable. Raises LaunchArgsError if required values are missing/invalid. """ ... def preview_config( self, server_id: int, server_dir: Path, config_sections: dict[str, dict], ) -> dict[str, str]: """ Render config files as strings WITHOUT writing to disk. Returns {label: content}. Label = filename for file-based games, var name for env-var games. """ ... def normalize_section(self, section: str, data: dict) -> dict: """ Optional: backfill / migrate a stored section dict before returning it to callers. Called by service.get_config_section() via hasattr guard. Default: return data unchanged. Implement to add new optional fields with defaults. """ return data @runtime_checkable class RemoteAdminClient(Protocol): """A connected client instance. Not required to be thread-safe — core wraps calls.""" def send_command(self, command: str, timeout: float = 5.0) -> str | None: ... def get_players(self) -> list[dict]: ... def kick_player(self, identifier: str, reason: str = "") -> bool: ... def ban_player(self, identifier: str, duration_minutes: int, reason: str) -> bool: ... def say_all(self, message: str) -> bool: ... def shutdown(self) -> bool: ... def keepalive(self) -> None: ... def disconnect(self) -> None: ... @runtime_checkable class RemoteAdmin(Protocol): """Factory for remote admin clients. One per adapter, creates clients on demand.""" def create_client(self, host: str, port: int, password: str) -> RemoteAdminClient: ... def get_startup_delay(self) -> float: """Seconds to wait after server start before connecting. Default: 30.""" ... def get_poll_interval(self) -> float: """Seconds between player list polls. Default: 10.""" ... def get_player_data_schema(self) -> type[BaseModel] | None: """Pydantic model for players.game_data JSON. None = no validation.""" ... @runtime_checkable class LogParser(Protocol): """Parses game-specific log lines into standard format.""" def parse_line(self, line: str) -> dict | None: """ Parse one log line. Returns: {"timestamp": ISO str, "level": "info"|"warning"|"error", "message": str} Returns None to skip the line (e.g. blank lines, binary garbage). """ ... def get_log_file_resolver(self, server_id: int) -> Callable[[Path], Path | None]: """ Return a callable(server_dir: Path) -> Path | None. Called by LogTailThread to find the current log file. Return None if log file not yet created. """ ... @runtime_checkable class MissionManager(Protocol): """Handles mission/scenario file format and rotation.""" file_extension: str # e.g. ".pbo" def parse_mission_filename(self, filename: str) -> dict: ... def get_rotation_config(self, rotation_entries: list[dict]) -> str: ... def get_missions_dir(self, server_dir: Path) -> Path: ... def get_mission_data_schema(self) -> type[BaseModel] | None: """Pydantic model for missions.game_data. None = no validation.""" ... @runtime_checkable class ModManager(Protocol): """Handles mod folder conventions and CLI argument building.""" def get_mod_folder_pattern(self) -> str: ... def build_mod_args(self, server_mods: list[dict]) -> list[str]: ... def validate_mod_folder(self, path: Path) -> bool: ... def get_mod_data_schema(self) -> type[BaseModel] | None: """Pydantic model for mods.game_data. None = no validation.""" ... @runtime_checkable class ProcessConfig(Protocol): """Game-specific process and directory conventions.""" def get_allowed_executables(self) -> list[str]: ... def get_port_conventions(self, game_port: int) -> dict[str, int]: ... def get_default_game_port(self) -> int: ... def get_default_rcon_port(self, game_port: int) -> int | None: ... def get_server_dir_layout(self) -> list[str]: ... @runtime_checkable class BanManager(Protocol): """Bidirectional sync between DB bans and game ban file.""" def get_ban_file_path(self, server_dir: Path) -> Path: ... def sync_bans_to_file(self, bans: list[dict], ban_file: Path) -> None: ... def read_bans_from_file(self, ban_file: Path) -> list[dict]: ... def get_ban_data_schema(self) -> type[BaseModel] | None: """Pydantic model for bans.game_data. None = no validation.""" ... @runtime_checkable class GameAdapter(Protocol): """ Composite adapter. Every game must implement this. Optional capabilities return None — core degrades gracefully. Use has_capability(name) instead of None checks throughout. """ game_type: str # e.g. "arma3" display_name: str # e.g. "Arma 3" version: str # e.g. "1.0.0" def get_config_generator(self) -> ConfigGenerator: ... def get_process_config(self) -> ProcessConfig: ... def get_log_parser(self) -> LogParser: ... def get_remote_admin(self) -> RemoteAdmin | None: ... def get_mission_manager(self) -> MissionManager | None: ... def get_mod_manager(self) -> ModManager | None: ... def get_ban_manager(self) -> BanManager | None: ... def has_capability(self, name: str) -> bool: """ Explicit capability probe. Use this instead of: if adapter.get_remote_admin() is not None: Use this instead: if adapter.has_capability("remote_admin"): Valid names: "config_generator", "process_config", "log_parser", "remote_admin", "mission_manager", "mod_manager", "ban_manager" """ ... def get_additional_routers(self) -> list: """List of FastAPI APIRouter instances for game-specific routes.""" ... def get_custom_thread_factories(self) -> list[Callable]: """List of callables(server_id, db) -> BaseServerThread for extra threads.""" ...