""" Arma 3 config generator. Merged protocol: Pydantic models (schema) + file generation + launch args. """ from __future__ import annotations import os from pathlib import Path from typing import Any from pydantic import BaseModel, Field # ─── Pydantic Models (config schema) ───────────────────────────────────────── class ServerConfig(BaseModel): hostname: str = "My Arma 3 Server" password: str | None = None password_admin: str = "" server_command_password: str | None = None max_players: int = Field(default=40, gt=0, le=1000) kick_duplicate: int = Field(default=1, ge=0, le=1) persistent: int = Field(default=1, ge=0, le=1) vote_threshold: float = Field(default=0.33, ge=0.0, le=1.0) vote_mission_players: int = Field(default=1, ge=0) vote_timeout: int = Field(default=60, ge=0) role_timeout: int = Field(default=90, ge=0) briefing_timeout: int = Field(default=60, ge=0) debriefing_timeout: int = Field(default=45, ge=0) lobby_idle_timeout: int = Field(default=300, ge=0) disable_von: int = Field(default=0, ge=0, le=1) von_codec: int = Field(default=1, ge=0, le=1) von_codec_quality: int = Field(default=20, ge=0, le=30) max_ping: int = Field(default=250, gt=0) max_packet_loss: int = Field(default=50, ge=0, le=100) max_desync: int = Field(default=200, ge=0) disconnect_timeout: int = Field(default=15, ge=0) kick_on_ping: int = Field(default=1, ge=0, le=1) kick_on_packet_loss: int = Field(default=1, ge=0, le=1) kick_on_desync: int = Field(default=1, ge=0, le=1) kick_on_timeout: int = Field(default=1, ge=0, le=1) battleye: int = Field(default=1, ge=0, le=1) verify_signatures: int = Field(default=2, ge=0, le=2) allowed_file_patching: int = Field(default=0, ge=0, le=2) forced_difficulty: str = "Regular" timestamp_format: str = "short" auto_select_mission: int = Field(default=0, ge=0, le=1) random_mission_order: int = Field(default=0, ge=0, le=1) log_file: str = "server_console.log" skip_lobby: int = Field(default=0, ge=0, le=1) drawing_in_map: int = Field(default=1, ge=0, le=1) upnp: int = Field(default=0, ge=0, le=1) loopback: int = Field(default=0, ge=0, le=1) statistics_enabled: int = Field(default=1, ge=0, le=1) motd_lines: list[str] = Field(default_factory=lambda: ["Welcome!", "Have fun"]) motd_interval: float = Field(default=5.0, gt=0) headless_clients: list[str] = Field(default_factory=list) local_clients: list[str] = Field(default_factory=list) admin_uids: list[str] = Field(default_factory=list) class BasicConfig(BaseModel): min_bandwidth: int = Field(default=800000, gt=0) max_bandwidth: int = Field(default=25000000, gt=0) max_msg_send: int = Field(default=384, gt=0) max_size_guaranteed: int = Field(default=512, gt=0) max_size_non_guaranteed: int = Field(default=256, gt=0) min_error_to_send: float = Field(default=0.003, gt=0) max_custom_file_size: int = Field(default=100000, ge=0) class ProfileConfig(BaseModel): reduced_damage: int = Field(default=0, ge=0, le=1) group_indicators: int = Field(default=0, ge=0, le=3) friendly_tags: int = Field(default=0, ge=0, le=3) enemy_tags: int = Field(default=0, ge=0, le=3) detected_mines: int = Field(default=0, ge=0, le=3) commands: int = Field(default=1, ge=0, le=3) waypoints: int = Field(default=1, ge=0, le=3) tactical_ping: int = Field(default=0, ge=0, le=1) weapon_info: int = Field(default=2, ge=0, le=3) stance_indicator: int = Field(default=2, ge=0, le=3) stamina_bar: int = Field(default=0, ge=0, le=1) weapon_crosshair: int = Field(default=0, ge=0, le=1) vision_aid: int = Field(default=0, ge=0, le=1) third_person_view: int = Field(default=0, ge=0, le=1) camera_shake: int = Field(default=1, ge=0, le=1) score_table: int = Field(default=1, ge=0, le=1) death_messages: int = Field(default=1, ge=0, le=1) von_id: int = Field(default=1, ge=0, le=1) map_content_friendly: int = Field(default=0, ge=0, le=3) map_content_enemy: int = Field(default=0, ge=0, le=3) map_content_mines: int = Field(default=0, ge=0, le=3) auto_report: int = Field(default=0, ge=0, le=1) multiple_saves: int = Field(default=0, ge=0, le=1) ai_level_preset: int = Field(default=3, ge=0, le=4) skill_ai: float = Field(default=0.5, ge=0.0, le=1.0) precision_ai: float = Field(default=0.5, ge=0.0, le=1.0) class LaunchConfig(BaseModel): world: str = "empty" extra_params: str = "" limit_fps: int = Field(default=50, gt=0, le=1000) auto_init: int = Field(default=0, ge=0, le=1) load_mission_to_memory: int = Field(default=0, ge=0, le=1) enable_ht: int = Field(default=0, ge=0, le=1) huge_pages: int = Field(default=0, ge=0, le=1) cpu_count: int | None = None ex_threads: int = Field(default=7, ge=0) max_mem: int | None = None no_logs: int = Field(default=0, ge=0, le=1) netlog: int = Field(default=0, ge=0, le=1) class RConConfig(BaseModel): rcon_password: str = "" max_ping: int = Field(default=200, gt=0) enabled: int = Field(default=1, ge=0, le=1) # ─── Config Generator ───────────────────────────────────────────────────────── class Arma3ConfigGenerator: game_type = "arma3" SECTIONS: dict[str, type[BaseModel]] = { "server": ServerConfig, "basic": BasicConfig, "profile": ProfileConfig, "launch": LaunchConfig, "rcon": RConConfig, } SENSITIVE_FIELDS: dict[str, list[str]] = { "server": ["password", "password_admin", "server_command_password"], "rcon": ["rcon_password"], } def get_sections(self) -> dict[str, type[BaseModel]]: return self.SECTIONS def get_defaults(self, section: str) -> dict[str, Any]: model_cls = self.SECTIONS.get(section) if model_cls is None: return {} return model_cls().model_dump() def get_sensitive_fields(self, section: str) -> list[str]: return self.SENSITIVE_FIELDS.get(section, []) def get_config_version(self) -> str: return "1.0.0" def migrate_config(self, old_version: str, config_json: dict) -> dict: """ For version 1.0.0 there is nothing to migrate. Future versions: add migration logic here. """ from adapters.exceptions import ConfigMigrationError raise ConfigMigrationError( old_version, f"No migration path from {old_version} to {self.get_config_version()}" ) # ── Config file writers ─────────────────────────────────────────────────── @staticmethod def _escape(value: str) -> str: """ Escape a string for use inside Arma 3 double-quoted config values. Order matters: escape backslashes FIRST. """ value = value.replace("\\", "\\\\") value = value.replace('"', '\\"') value = value.replace('\n', '\\n') return value @staticmethod def _atomic_write(path: Path, content: str) -> None: """Write content to path atomically via tmp file + os.replace().""" from adapters.exceptions import ConfigWriteError tmp_path = path.with_suffix(path.suffix + ".tmp") try: path.parent.mkdir(parents=True, exist_ok=True) tmp_path.write_text(content, encoding="utf-8") os.replace(str(tmp_path), str(path)) except OSError as e: # Clean up tmp file if it exists try: tmp_path.unlink(missing_ok=True) except OSError as exc: logger.debug("Could not clean up temp file %s: %s", tmp_path, exc) raise ConfigWriteError(str(path), str(e)) from e def _render_server_cfg(self, cfg: ServerConfig) -> str: """Render server.cfg content string.""" motd_items = ", ".join(f'"{self._escape(l)}"' for l in cfg.motd_lines) headless = ", ".join(f'"{h}"' for h in cfg.headless_clients) local = ", ".join(f'"{l}"' for l in cfg.local_clients) admin_uids = ", ".join(f'"{u}"' for u in cfg.admin_uids) lines = [ f'hostname = "{self._escape(cfg.hostname)}";', ] if cfg.password: lines.append(f'password = "{self._escape(cfg.password)}";') if cfg.password_admin: lines.append(f'passwordAdmin = "{self._escape(cfg.password_admin)}";') if cfg.server_command_password: lines.append(f'serverCommandPassword = "{self._escape(cfg.server_command_password)}";') lines += [ f"maxPlayers = {cfg.max_players};", f"kickDuplicate = {cfg.kick_duplicate};", f"persistent = {cfg.persistent};", f"voteThreshold = {cfg.vote_threshold};", f"voteMissionPlayers = {cfg.vote_mission_players};", f"voteTimeout = {cfg.vote_timeout};", f"roleTimeout = {cfg.role_timeout};", f"briefingTimeOut = {cfg.briefing_timeout};", f"debriefingTimeOut = {cfg.debriefing_timeout};", f"lobbyIdleTimeout = {cfg.lobby_idle_timeout};", f"disableVoN = {cfg.disable_von};", f"vonCodec = {cfg.von_codec};", f"vonCodecQuality = {cfg.von_codec_quality};", f"maxPing = {cfg.max_ping};", f"maxPacketLoss = {cfg.max_packet_loss};", f"maxDesync = {cfg.max_desync};", f"disconnectTimeout = {cfg.disconnect_timeout};", f"kickOnPing = {cfg.kick_on_ping};", f"kickOnPacketLoss = {cfg.kick_on_packet_loss};", f"kickOnDesync = {cfg.kick_on_desync};", f"kickOnTimeout = {cfg.kick_on_timeout};", f"BattlEye = {cfg.battleye};", f"verifySignatures = {cfg.verify_signatures};", f"allowedFilePatching = {cfg.allowed_file_patching};", f'forcedDifficulty = "{cfg.forced_difficulty}";', f'timeStampFormat = "{cfg.timestamp_format}";', f"autoSelectMission = {cfg.auto_select_mission};", f"randomMissionOrder = {cfg.random_mission_order};", f'logFile = "{cfg.log_file}";', f"skipLobby = {cfg.skip_lobby};", f"drawingInMap = {cfg.drawing_in_map};", f"upnp = {cfg.upnp};", f"loopback = {cfg.loopback};", f"statisticsEnabled = {cfg.statistics_enabled};", f"motd[] = {{{motd_items}}};", f"motdInterval = {cfg.motd_interval};", ] if cfg.headless_clients: lines.append(f"headlessClients[] = {{{headless}}};") if cfg.local_clients: lines.append(f"localClient[] = {{{local}}};") if cfg.admin_uids: lines.append(f"admins[] = {{{admin_uids}}};") return "\n".join(lines) + "\n" def _render_basic_cfg(self, cfg: BasicConfig) -> str: return ( f"MinBandwidth = {cfg.min_bandwidth};\n" f"MaxBandwidth = {cfg.max_bandwidth};\n" f"MaxMsgSend = {cfg.max_msg_send};\n" f"MaxSizeGuaranteed = {cfg.max_size_guaranteed};\n" f"MaxSizeNonguaranteed = {cfg.max_size_non_guaranteed};\n" f"MinErrorToSend = {cfg.min_error_to_send};\n" f"MaxCustomFileSize = {cfg.max_custom_file_size};\n" ) def _render_arma3profile(self, cfg: ProfileConfig) -> str: return ( "class DifficultyPresets {\n" " class CustomDifficulty {\n" " class Options {\n" f" reducedDamage = {cfg.reduced_damage};\n" f" groupIndicators = {cfg.group_indicators};\n" f" friendlyTags = {cfg.friendly_tags};\n" f" enemyTags = {cfg.enemy_tags};\n" f" detectedMines = {cfg.detected_mines};\n" f" commands = {cfg.commands};\n" f" waypoints = {cfg.waypoints};\n" f" tacticalPing = {cfg.tactical_ping};\n" f" weaponInfo = {cfg.weapon_info};\n" f" stanceIndicator = {cfg.stance_indicator};\n" f" staminaBar = {cfg.stamina_bar};\n" f" weaponCrosshair = {cfg.weapon_crosshair};\n" f" visionAid = {cfg.vision_aid};\n" f" thirdPersonView = {cfg.third_person_view};\n" f" cameraShake = {cfg.camera_shake};\n" f" scoreTable = {cfg.score_table};\n" f" deathMessages = {cfg.death_messages};\n" f" vonID = {cfg.von_id};\n" f" mapContentFriendly = {cfg.map_content_friendly};\n" f" mapContentEnemy = {cfg.map_content_enemy};\n" f" mapContentMines = {cfg.map_content_mines};\n" f" autoReport = {cfg.auto_report};\n" f" multipleSaves = {cfg.multiple_saves};\n" " };\n" f" aiLevelPreset = {cfg.ai_level_preset};\n" " };\n" " class CustomAILevel {\n" f" skillAI = {cfg.skill_ai};\n" f" precisionAI = {cfg.precision_ai};\n" " };\n" "};\n" ) def _render_beserver_cfg(self, cfg: RConConfig) -> str: return ( f"RConPassword {cfg.rcon_password}\n" f"MaxPing {cfg.max_ping}\n" ) # ── Public interface ────────────────────────────────────────────────────── def write_configs( self, server_id: int, server_dir: Path, config_sections: dict[str, dict], ) -> list[Path]: server_cfg = ServerConfig(**config_sections.get("server", {})) basic_cfg = BasicConfig(**config_sections.get("basic", {})) profile_cfg = ProfileConfig(**config_sections.get("profile", {})) rcon_cfg = RConConfig(**config_sections.get("rcon", {})) written = [] pairs = [ (server_dir / "server.cfg", self._render_server_cfg(server_cfg)), (server_dir / "basic.cfg", self._render_basic_cfg(basic_cfg)), (server_dir / "server" / "server.Arma3Profile", self._render_arma3profile(profile_cfg)), (server_dir / "battleye" / "beserver.cfg", self._render_beserver_cfg(rcon_cfg)), ] for path, content in pairs: self._atomic_write(path, content) written.append(path) # Restrict permissions on files containing passwords (Unix only) if os.name != "nt": for path in [server_dir / "server.cfg", server_dir / "battleye" / "beserver.cfg"]: if path.exists(): os.chmod(path, 0o600) return written def build_launch_args( self, config_sections: dict[str, dict], mod_args: list[str] | None = None, ) -> list[str]: from adapters.exceptions import LaunchArgsError launch = LaunchConfig(**config_sections.get("launch", {})) server = ServerConfig(**config_sections.get("server", {})) args = [ f"-port={config_sections.get('_port', 2302)}", "-config=server.cfg", "-cfg=basic.cfg", "-profiles=./server", "-name=server", f"-world={launch.world}", f"-limitFPS={launch.limit_fps}", "-bepath=./battleye", ] if launch.auto_init: args.append("-autoInit") if launch.enable_ht: args.append("-enableHT") if launch.huge_pages: args.append("-hugePages") if launch.cpu_count is not None: args.append(f"-cpuCount={launch.cpu_count}") if launch.max_mem is not None: args.append(f"-maxMem={launch.max_mem}") if launch.no_logs: args.append("-noLogs") if launch.netlog: args.append("-netlog") if launch.extra_params: args.extend(launch.extra_params.split()) if mod_args: args.extend(mod_args) return args def get_ui_schema(self) -> dict: return { "server": { "hostname": {"widget": "text", "label": "Server Hostname"}, "max_players": {"widget": "number", "label": "Max Players", "min": 1, "max": 1000}, "password": {"widget": "password", "label": "Player Password"}, "password_admin": {"widget": "password", "label": "Admin Password"}, "motd_lines": {"widget": "textarea", "label": "Message of the Day (one line per row)"}, "forced_difficulty": {"widget": "select", "label": "Difficulty Preset", "options": ["Recruit", "Regular", "Veteran", "Custom"]}, "battleye": {"widget": "toggle", "label": "BattleEye Anti-Cheat"}, "disable_von": {"widget": "toggle", "label": "Disable Voice over Net (VoN)"}, "verify_signatures": {"widget": "number", "label": "Verify Signatures (0=off, 1=on, 2=strict)", "min": 0, "max": 2}, "persistent": {"widget": "toggle", "label": "Persistent (keep running when empty)"}, "admin_uids": {"widget": "tag-list", "label": "Admin Steam UIDs", "placeholder": "76561198000000000"}, }, "basic": { "max_custom_file_size": {"widget": "number", "label": "Max Custom File Size (bytes)"}, }, "launch": { "extra_params": {"widget": "tag-list", "label": "Additional Startup Parameters", "placeholder": "-limitFPS=100"}, }, "rcon": { "rcon_password": {"widget": "password", "label": "RCon Password"}, "max_ping": {"widget": "number", "label": "RCon Port"}, }, } def preview_config( self, server_id: int, server_dir: Path, config_sections: dict[str, dict], ) -> dict[str, str]: server_cfg = ServerConfig(**config_sections.get("server", {})) basic_cfg = BasicConfig(**config_sections.get("basic", {})) profile_cfg = ProfileConfig(**config_sections.get("profile", {})) rcon_cfg = RConConfig(**config_sections.get("rcon", {})) return { "server.cfg": self._render_server_cfg(server_cfg), "basic.cfg": self._render_basic_cfg(basic_cfg), "server/server.Arma3Profile": self._render_arma3profile(profile_cfg), "battleye/beserver.cfg": self._render_beserver_cfg(rcon_cfg), }