from __future__ import annotations import json from pathlib import Path from typing import Any from .analysis import analyze_session_dir, compare_laps as compare_lap_objects, load_session, segment_laps from .ac_app import install_ac_python_app from .bridge_app import install_codex_bridge_app from .content import build_content_library from .live_coach import LiveCoachConfig, load_live_coach_status, start_live_coach_background, stop_live_coach_background from .launching import discover_launch_surfaces, prepare_direct_training_launch, prepare_stage_training_launch from .moza import moza_state from .moza_install import install_moza_ac_plugin, verify_moza_ac_plugin from .paths import EnvironmentPaths, discover_environment, read_race_context from .recorder import SessionRecorder from .shared_memory import AssettoCorsaReader from .training import materialize_training_setup, recommend_training_curriculum, save_training_plan class AssettoCoachService: def __init__(self, paths: EnvironmentPaths | None = None) -> None: self.paths = paths or discover_environment() self.reader = AssettoCorsaReader() def _latest_session_dir(self) -> Path: candidates = [ path for path in self.paths.sessions_root.iterdir() if path.is_dir() and (path / "meta.json").exists() ] if not candidates: raise RuntimeError("No captured sessions are available yet") return max(candidates, key=lambda path: path.stat().st_mtime) def _resolve_session_dir(self, session_id: str | None) -> Path: if session_id: session_dir = self.paths.sessions_root / session_id if not session_dir.exists(): raise RuntimeError(f"Session '{session_id}' was not found") return session_dir return self._latest_session_dir() def health(self) -> dict[str, Any]: snapshot = self.reader.read_snapshot() moza_pending_count = 0 if self.paths.moza_pending_root and self.paths.moza_pending_root.exists(): moza_pending_count = len(list(self.paths.moza_pending_root.iterdir())) return { "paths": self.paths.to_dict(), "race_context": read_race_context(self.paths), "assetto_shared_memory_available": snapshot is not None, "current_snapshot": snapshot, "existing_exports": { "aim_dump_exists": (self.paths.aim_root / "telemetry_dump.act").exists(), "race_out_exists": (self.paths.out_root / "race_out.json").exists(), "laps_ini_exists": (self.paths.out_root / "laps.ini").exists(), }, "moza": { **moza_state(self.paths), "pit_house_present": self.paths.moza_pit_house_root is not None, "telemetry_root": str(self.paths.moza_telemetry_root) if self.paths.moza_telemetry_root else None, "pending_file_count": moza_pending_count, }, "live_coach": load_live_coach_status(self.paths), } def snapshot(self) -> dict[str, Any]: snapshot = self.reader.read_snapshot() if snapshot is None: raise RuntimeError("Assetto Corsa shared memory is not available") snapshot["race_context"] = read_race_context(self.paths) return snapshot def capture_run( self, *, duration_s: float = 120.0, sample_hz: float = 50.0, label: str | None = None, ) -> dict[str, Any]: recorder = SessionRecorder(self.paths, self.reader) result = recorder.capture(duration_s=duration_s, sample_hz=sample_hz, label=label) return result.to_dict() def capture_laps( self, *, lap_count: int = 3, sample_hz: float = 50.0, max_duration_s: float = 1800.0, label: str | None = None, ) -> dict[str, Any]: recorder = SessionRecorder(self.paths, self.reader) result = recorder.capture( lap_count=lap_count, sample_hz=sample_hz, max_duration_s=max_duration_s, label=label, ) return result.to_dict() def list_sessions(self, limit: int = 10) -> list[dict[str, Any]]: if not self.paths.sessions_root.exists(): return [] sessions = [] for session_dir in self.paths.sessions_root.iterdir(): meta_path = session_dir / "meta.json" if not session_dir.is_dir() or not meta_path.exists(): continue meta = json.loads(meta_path.read_text(encoding="utf-8")) sessions.append( { "session_id": meta["session_id"], "created_at_unix_s": meta.get("created_at_unix_s"), "sample_count": meta.get("sample_count"), "capture_mode": meta.get("capture_mode"), "car_model": meta.get("initial_snapshot", {}).get("car_model"), "track": meta.get("initial_snapshot", {}).get("track"), "session_dir": str(session_dir), } ) sessions.sort(key=lambda item: item.get("created_at_unix_s", 0), reverse=True) return sessions[:limit] def library_summary(self) -> dict[str, Any]: library = build_content_library(self.paths) return { "summary": library.summary(), "launch_surfaces": discover_launch_surfaces(self.paths).to_dict(), "sample_cars": [ { "car_model": car.folder_name, "name": car.name, "drive_layout": car.drive_layout, "bhp": car.bhp, "setup_count": car.setup_count, "official": car.official, } for car in list(library.cars.values())[:20] ], "sample_tracks": [ { "layout_id": track.layout_id, "name": track.name, "length_m": track.length_m, "training_suitable": track.training_suitable, } for track in list(library.tracks.values())[:20] ], } def recommend_training( self, *, materialize_setups: bool = True, ) -> dict[str, Any]: library = build_content_library(self.paths) race_context = read_race_context(self.paths) recommendation = recommend_training_curriculum( library, current_car_model=race_context.get("car_model"), current_track=race_context.get("track"), current_track_config=race_context.get("track_config"), ) generated_setups = [] if materialize_setups: for item in recommendation["curriculum"]: try: generated = materialize_training_setup( self.paths, car_model=item["car_model"], track=item["track"], track_layout=item["track_layout"], profile_id=item["setup_profile"], setup_name=f"codex_{item['stage']}.ini", ) except Exception as exc: generated = {"error": str(exc), "car_model": item["car_model"], "stage": item["stage"]} generated_setups.append(generated) recommendation["generated_setups"] = generated_setups recommendation["plan_path"] = str(save_training_plan(self.paths, recommendation)) return recommendation def materialize_training_setup( self, *, car_model: str, track: str | None, track_layout: str | None, profile_id: str, setup_name: str | None = None, ) -> dict[str, Any]: return materialize_training_setup( self.paths, car_model=car_model, track=track, track_layout=track_layout, profile_id=profile_id, setup_name=setup_name, ) def start_live_coach( self, *, mode: str = "consistency", reference_session_id: str | None = None, car_model: str | None = None, track: str | None = None, track_layout: str | None = None, tts_enabled: bool = True, muted: bool = False, max_callouts_per_lap: int = 3, sample_hz: float = 20.0, zone_count: int = 4, setup_profile: str | None = None, plan_label: str | None = None, ) -> dict[str, Any]: race_context = read_race_context(self.paths) config = LiveCoachConfig( car_model=car_model or race_context.get("car_model"), track=track or race_context.get("track"), track_layout=track_layout or race_context.get("track_config"), mode=mode, reference_session_id=reference_session_id, sample_hz=sample_hz, tts_enabled=tts_enabled, muted=muted, max_callouts_per_lap=max_callouts_per_lap, zone_count=zone_count, setup_profile=setup_profile, plan_label=plan_label, ) return start_live_coach_background(self.paths, config) def live_coach_status(self) -> dict[str, Any]: return load_live_coach_status(self.paths) def stop_live_coach(self) -> dict[str, Any]: return stop_live_coach_background(self.paths) def install_ac_app(self, *, enable: bool = True) -> dict[str, Any]: return install_ac_python_app(self.paths, enable=enable) def install_bridge_app(self, *, enable: bool = True) -> dict[str, Any]: return install_codex_bridge_app(self.paths, enable=enable) def verify_moza_plugin(self) -> dict[str, Any]: return verify_moza_ac_plugin(self.paths) def install_moza_plugin(self, *, enable: bool = True) -> dict[str, Any]: return install_moza_ac_plugin(self.paths, enable=enable) def prepare_training_launch( self, *, stage: str = "smooth_inputs", apply_setup_as_last: bool = True, launch_surface: str = "assettocorsa", dry_run: bool = True, ) -> dict[str, Any]: return prepare_stage_training_launch( self.paths, stage=stage, apply_setup_as_last=apply_setup_as_last, launch_surface=launch_surface, dry_run=dry_run, ) def prepare_direct_training_launch( self, *, 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 = "assettocorsa", dry_run: bool = True, ) -> dict[str, Any]: return prepare_direct_training_launch( self.paths, car_model=car_model, track=track, track_layout=track_layout, profile_id=profile_id, live_mode=live_mode, title=title, apply_setup_as_last=apply_setup_as_last, launch_surface=launch_surface, dry_run=dry_run, ) def analyze_session(self, session_id: str | None = None) -> dict[str, Any]: session_dir = self._resolve_session_dir(session_id) return analyze_session_dir(session_dir) def compare_laps( self, *, session_id: str | None = None, slower_lap_number: int | None = None, reference_lap_number: int | None = None, ) -> dict[str, Any]: session_dir = self._resolve_session_dir(session_id) meta, samples = load_session(session_dir) laps = segment_laps(samples, started_mid_lap=bool(meta.get("started_mid_lap"))) if bool(meta.get("started_mid_lap")) and len(laps) < 2: laps = segment_laps(samples, started_mid_lap=False) if len(laps) < 2: raise RuntimeError("Need at least two complete laps in the session") reference = None slower = None if reference_lap_number is not None: reference = next((lap for lap in laps if lap.lap_number == reference_lap_number), None) if slower_lap_number is not None: slower = next((lap for lap in laps if lap.lap_number == slower_lap_number), None) if reference is None: reference = min(laps, key=lambda lap: lap.lap_time_ms) if slower is None: candidates = [lap for lap in laps if lap.lap_number != reference.lap_number] slower = max(candidates, key=lambda lap: lap.lap_time_ms) if slower.lap_number == reference.lap_number: raise RuntimeError("Reference and slower lap must be different") return compare_lap_objects(reference, slower)