3025c2021c
- Add per-mission params to rotation (MissionRotationItem.params); falls back to default_mission_params, then omits entirely if both empty - Add key-value widget to ConfigEditor for default_mission_params field - Add MissionParamsEditor component for editing param key/value/type rows - Bump config schema to 1.1.0 with migration from 1.0.0 - Add normalize_section() to Protocol and ArmaConfigGenerator for read-time backfill of old stored rows - Set Arma3 BasicConfig and ProfileConfig defaults from basic.cfg / Administrator.Arma3Profile - Document 3 known Mods tab bugs in CLAUDE.md for next-session fix
246 lines
8.3 KiB
Python
246 lines
8.3 KiB
Python
"""
|
|
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."""
|
|
... |