- 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
432 lines
19 KiB
Python
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),
|
|
} |