from __future__ import annotations import configparser import os import shutil import time from dataclasses import dataclass from pathlib import Path from typing import Any from .paths import EnvironmentPaths, read_race_context from .training import materialize_training_setup, recommend_training_curriculum from .content import build_content_library @dataclass(slots=True) class LaunchSurfaces: acs_exe: Path | None assetto_corsa_launcher_exe: Path | None content_manager_exe: Path | None content_manager_protocol: str | None def to_dict(self) -> dict[str, str | None]: return { "acs_exe": str(self.acs_exe) if self.acs_exe else None, "assetto_corsa_launcher_exe": str(self.assetto_corsa_launcher_exe) if self.assetto_corsa_launcher_exe else None, "content_manager_exe": str(self.content_manager_exe) if self.content_manager_exe else None, "content_manager_protocol": self.content_manager_protocol, } def discover_launch_surfaces(paths: EnvironmentPaths) -> LaunchSurfaces: acs_exe = None assetto_corsa_launcher_exe = None if paths.assetto_root is not None: candidate = paths.assetto_root / "acs.exe" if candidate.exists(): acs_exe = candidate launcher_candidate = paths.assetto_root / "AssettoCorsa.exe" if launcher_candidate.exists(): assetto_corsa_launcher_exe = launcher_candidate content_manager_candidates = [ Path(os.environ["USERPROFILE"]) / "Downloads" / "content-manager" / "Content Manager.exe", Path(os.environ["USERPROFILE"]) / "Desktop" / "Content Manager.exe", ] content_manager_exe = next((candidate for candidate in content_manager_candidates if candidate.exists()), None) content_manager_protocol = "acmanager://launch" return LaunchSurfaces( acs_exe=acs_exe, assetto_corsa_launcher_exe=assetto_corsa_launcher_exe, content_manager_exe=content_manager_exe, content_manager_protocol=content_manager_protocol, ) def _load_race_ini(path: Path) -> configparser.ConfigParser: parser = configparser.ConfigParser() parser.optionxform = str if path.exists(): parser.read(path, encoding="utf-8") return parser def _ensure_section(parser: configparser.ConfigParser, section: str) -> None: if not parser.has_section(section): parser.add_section(section) def _first_skin(paths: EnvironmentPaths, car_model: str) -> str | None: if paths.assetto_root is None: return None skins_root = paths.assetto_root / "content" / "cars" / car_model / "skins" if not skins_root.exists(): return None for skin_dir in skins_root.iterdir(): if skin_dir.is_dir(): return skin_dir.name return None def prepare_training_launch( paths: EnvironmentPaths, *, plan: dict[str, Any], setup_profile: str, apply_setup_as_last: bool = True, launch_surface: str = "assettocorsa", dry_run: bool = True, ) -> dict[str, Any]: setup_result = materialize_training_setup( paths, car_model=plan["car_model"], track=plan["track"], track_layout=plan["track_layout"], profile_id=setup_profile, setup_name=f"codex_{plan['stage']}.ini", ) surfaces = discover_launch_surfaces(paths) if launch_surface == "assettocorsa" and surfaces.acs_exe is None: raise RuntimeError("acs.exe was not found") if launch_surface == "content_manager" and surfaces.content_manager_exe is None: raise RuntimeError("Content Manager executable was not found") race_ini_path = paths.cfg_root / "race.ini" parser = _load_race_ini(race_ini_path) backup_path = None setup_backup_path = None race_context = read_race_context(paths) skin = race_context.get("skin") if race_context.get("car_model") == plan["car_model"] else _first_skin(paths, plan["car_model"]) if skin is None: skin = "" _ensure_section(parser, "REMOTE") parser.set("REMOTE", "ACTIVE", "0") parser.set("REMOTE", "SERVER_IP", "") parser.set("REMOTE", "SERVER_PORT", "0") parser.set("REMOTE", "SERVER_NAME", "") _ensure_section(parser, "RACE") parser.set("RACE", "MODEL", plan["car_model"]) parser.set("RACE", "MODEL_CONFIG", "") parser.set("RACE", "SKIN", skin) parser.set("RACE", "TRACK", plan["track"]) parser.set("RACE", "CONFIG_TRACK", plan["track_layout"] or "") parser.set("RACE", "AI_LEVEL", "100") parser.set("RACE", "CARS", "1") parser.set("RACE", "DRIFT_MODE", "0") parser.set("RACE", "RACE_LAPS", "0") parser.set("RACE", "FIXED_SETUP", "0") parser.set("RACE", "PENALTIES", "1") parser.set("RACE", "JUMP_START_PENALTY", "0") _ensure_section(parser, "CAR_0") parser.set("CAR_0", "SETUP", "") parser.set("CAR_0", "SKIN", skin) parser.set("CAR_0", "MODEL", "-") parser.set("CAR_0", "MODEL_CONFIG", "") parser.set("CAR_0", "BALLAST", "0") parser.set("CAR_0", "RESTRICTOR", "0") parser.set("CAR_0", "DRIVER_NAME", race_context.get("server_name") or "Codex Training") _ensure_section(parser, "SESSION_0") parser.set("SESSION_0", "NAME", "Practice") parser.set("SESSION_0", "TYPE", "1") parser.set("SESSION_0", "DURATION_MINUTES", "0") parser.set("SESSION_0", "SPAWN_SET", "HOTLAP_START") if not dry_run: paths.runtime_root.mkdir(parents=True, exist_ok=True) if race_ini_path.exists(): backup_path = paths.runtime_root / f"race.ini.backup.{time.strftime('%Y%m%d-%H%M%S')}.ini" shutil.copy2(race_ini_path, backup_path) with race_ini_path.open("w", encoding="utf-8") as handle: parser.write(handle, space_around_delimiters=False) if apply_setup_as_last: generated_setup = Path(setup_result["output_path"]) last_setup_path = paths.documents_root / "setups" / plan["car_model"] / "generic" / "last.ini" last_setup_path.parent.mkdir(parents=True, exist_ok=True) if last_setup_path.exists(): setup_backup_path = paths.runtime_root / f"{plan['car_model']}.last.backup.{time.strftime('%Y%m%d-%H%M%S')}.ini" shutil.copy2(last_setup_path, setup_backup_path) shutil.copy2(generated_setup, last_setup_path) if launch_surface == "assettocorsa": os.startfile(str(surfaces.acs_exe)) # type: ignore[arg-type] elif launch_surface == "content_manager": launched = False if surfaces.content_manager_protocol: try: os.startfile(surfaces.content_manager_protocol) # type: ignore[arg-type] launched = True except Exception: launched = False if not launched: os.startfile(str(surfaces.content_manager_exe)) # type: ignore[arg-type] else: raise RuntimeError(f"Unknown launch surface '{launch_surface}'") return { "plan": plan, "setup": setup_result, "launch_surface": launch_surface, "launch_surfaces": surfaces.to_dict(), "race_ini_path": str(race_ini_path), "race_ini_backup": str(backup_path) if backup_path else None, "setup_backup": str(setup_backup_path) if setup_backup_path else None, "dry_run": dry_run, "apply_setup_as_last": apply_setup_as_last, } def prepare_stage_training_launch( paths: EnvironmentPaths, *, stage: str = "smooth_inputs", apply_setup_as_last: bool = True, launch_surface: str = "content_manager", dry_run: bool = True, ) -> dict[str, Any]: library = build_content_library(paths) curriculum = recommend_training_curriculum( library, current_car_model=read_race_context(paths).get("car_model"), current_track=read_race_context(paths).get("track"), current_track_config=read_race_context(paths).get("track_config"), ) plan = next((item for item in curriculum["curriculum"] if item["stage"] == stage), None) if plan is None: raise RuntimeError(f"No training recommendation found for stage '{stage}'") return prepare_training_launch( paths, plan=plan, setup_profile=plan["setup_profile"], apply_setup_as_last=apply_setup_as_last, launch_surface=launch_surface, dry_run=dry_run, ) def prepare_direct_training_launch( paths: EnvironmentPaths, *, car_model: str, track: str, track_layout: str | None, profile_id: str, live_mode: str = "consistency", title: str = "Custom Training Run", apply_setup_as_last: bool = True, launch_surface: str = "content_manager", dry_run: bool = True, ) -> dict[str, Any]: plan = { "plan_id": f"direct::{car_model}::{track}/{track_layout or ''}", "stage": live_mode, "priority": 0, "title": title, "car_model": car_model, "track": track, "track_layout": track_layout, "setup_profile": profile_id, "live_mode": live_mode, "why": ["Direct launch requested."], "score": 0.0, } return prepare_training_launch( paths, plan=plan, setup_profile=profile_id, apply_setup_as_last=apply_setup_as_last, launch_surface=launch_surface, dry_run=dry_run, )