Files
languard-servers-manager/backend/adapters/arma3/config_generator.py
Tran G. (Revernomad) Khoa 4aae08420b feat: Phase 2 — Mission rotation management + multi-file upload
- Backend: add terrain field to Arma3MissionManager.list_missions()
- Backend: add missions field to ServerConfig Pydantic model
- Backend: add GET /missions/rotation and PUT /missions/rotation endpoints
- Frontend: Mission type gains terrain field; new MissionRotationEntry type
- Frontend: useServerMissionRotation and useUpdateMissionRotation hooks
- Frontend: useUploadMission updated to accept File[] with sequential upload
- Frontend: MissionList redesigned with Available Missions + Mission Rotation sections
- Frontend: per-file upload progress tracking, terrain badges, difficulty select
- Tests: 5 new tests; fixed existing useUploadMission test for File[] API; 141 pass
2026-04-17 20:33:04 +07:00

432 lines
19 KiB
Python

"""
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)
missions: list[dict] = 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),
}