"""Small replaceable in-memory repository seeded with the local TMB Journey.""" from __future__ import annotations from dataclasses import replace from typing import Iterable from .contracts import AnalysisRun, AnalysisScenario, Journey, JourneyPlan, RouteRevision, SourceSnapshot TMB_DAY_1_JOURNEY_ID = "journey-tmb-day-01-v0_1" TMB_DAY_1_ROUTE_REVISION_ID = "route-revision-tmb-day-01-v0_1" TMB_DAY_1_PLAN_ID = "journey-plan-tmb-day-01-v0_1" TMB_DAY_1_RUN_ID = "analysis-run-tmb-day-01-v0_1" class LocalJourneyRepository: """Explicit local records; callers replace records instead of mutating them.""" def __init__( self, *, journeys: Iterable[Journey], routes: Iterable[RouteRevision], plans: Iterable[JourneyPlan], runs: Iterable[AnalysisRun], scenarios: Iterable[AnalysisScenario], source_snapshots: Iterable[SourceSnapshot], ): self._journeys = {item.journey_id: item for item in journeys} self._routes = {item.route_revision_id: item for item in routes} self._plans = {item.journey_plan_id: item for item in plans} self._runs = {item.analysis_run_id: item for item in runs} self._scenarios = {item.analysis_scenario_id: item for item in scenarios} self._snapshots = {item.source_snapshot_id: item for item in source_snapshots} def get_journey(self, journey_id: str) -> Journey: return self._journeys[journey_id] def get_route_revision(self, route_revision_id: str) -> RouteRevision: return self._routes[route_revision_id] def get_plan(self, journey_plan_id: str) -> JourneyPlan: return self._plans[journey_plan_id] def get_run(self, analysis_run_id: str) -> AnalysisRun: return self._runs[analysis_run_id] def get_scenario(self, analysis_scenario_id: str) -> AnalysisScenario: return self._scenarios[analysis_scenario_id] def scenarios_for_run(self, analysis_run_id: str) -> tuple[AnalysisScenario, ...]: return tuple(item for item in self._scenarios.values() if item.analysis_run_id == analysis_run_id) def get_source_snapshot(self, source_snapshot_id: str) -> SourceSnapshot: return self._snapshots[source_snapshot_id] def create_plan(self, plan: JourneyPlan, *, make_current: bool = False) -> JourneyPlan: """Append an immutable plan revision without erasing its predecessor.""" if plan.journey_plan_id in self._plans: raise ValueError("journey plan ID already exists") if plan.journey_id not in self._journeys or plan.route_revision_id not in self._routes: raise ValueError("journey plan has unknown journey or route revision") versions = [item.version for item in self._plans.values() if item.journey_id == plan.journey_id] if versions and plan.version <= max(versions): raise ValueError("journey plan version must advance") self._plans[plan.journey_plan_id] = plan if make_current: journey = self._journeys[plan.journey_id] self._journeys[plan.journey_id] = replace(journey, current_plan_id=plan.journey_plan_id) return plan def create_reanalysis(self, run: AnalysisRun, scenarios: Iterable[AnalysisScenario]) -> AnalysisRun: """Append a new immutable run; old run and source snapshots remain unchanged.""" if run.analysis_run_id in self._runs: raise ValueError("analysis run ID already exists") if run.journey_plan_id not in self._plans or run.route_revision_id not in self._routes: raise ValueError("analysis run has unknown plan or route revision") scenario_values = tuple(scenarios) if not scenario_values or any(item.analysis_run_id != run.analysis_run_id for item in scenario_values): raise ValueError("analysis run requires its own scenarios") if any(item.analysis_scenario_id in self._scenarios for item in scenario_values): raise ValueError("analysis scenario ID already exists") self._runs[run.analysis_run_id] = run self._scenarios.update({item.analysis_scenario_id: item for item in scenario_values}) plan = self._plans[run.journey_plan_id] journey = self._journeys[plan.journey_id] self._journeys[journey.journey_id] = replace(journey, current_analysis_run_id=run.analysis_run_id) return run def tmb_day_1_repository() -> LocalJourneyRepository: """The product fixture names the journey; it contains no copied route geometry.""" route = RouteRevision( TMB_DAY_1_ROUTE_REVISION_ID, TMB_DAY_1_JOURNEY_ID, "tmb_day_01", "Tour du Mont Blanc · Day 1", "tmb-day-01-geojson-v0_1", 1515, "data/geojson/tmb_day_01.geojson", ) plan = JourneyPlan( TMB_DAY_1_PLAN_ID, TMB_DAY_1_JOURNEY_ID, route.route_revision_id, 1, "Europe/Paris", "2026-07-02T06:00:00+02:00", "2026-07-02T04:00:00+00:00", "mountain_pace_v0_1", "tmb_planning_pauses_v0_1", "HIKING", ) snapshots = ( SourceSnapshot("source-snapshot-tmb-route-v0_1", "ROUTE_INPUT", route.source_reference), SourceSnapshot("source-snapshot-tmb-dem-v0_1", "TERRAIN", "data/raw/dem/tmb_day_01_copernicus_glo30_5km.json"), SourceSnapshot("source-snapshot-tmb-weather-v0_1", "WEATHER", "docs/tmb_day_01_weather_historical_forecast_v0_1.json", "Open-Meteo"), SourceSnapshot("source-snapshot-tmb-trail-v0_1", "MAPPED_CHARACTER", "tests/fixtures/osm/tmb_day_01_route_section_osm_260919.json", "OpenStreetMap"), ) run = AnalysisRun( TMB_DAY_1_RUN_ID, plan.journey_plan_id, route.route_revision_id, "2026-09-21T00:00:00+00:00", "tmb-day-01-plan-v0_1", "journey_intelligence_projection_v0_1", tuple(item.source_snapshot_id for item in snapshots), {"explorer": "mountain_twin_explorer_v0_7:tmb_day_01"}, ) scenarios = tuple( AnalysisScenario( f"analysis-scenario-tmb-day-01-{name.lower()}-v0_1", run.analysis_run_id, name, plan.planned_start_local, plan.planned_start_utc, "mountain_pace_v0_1", "tmb_planning_pauses_v0_1", ) for name in ("FAST", "NOMINAL", "SLOW") ) journey = Journey(TMB_DAY_1_JOURNEY_ID, "Tour du Mont Blanc · Day 1", plan.journey_plan_id, run.analysis_run_id, "HIKING") return LocalJourneyRepository(journeys=(journey,), routes=(route,), plans=(plan,), runs=(run,), scenarios=scenarios, source_snapshots=snapshots)