"""Local Explorer analysis requests backed by the existing cached engines.""" from __future__ import annotations import hashlib import json import time from datetime import datetime from pathlib import Path from typing import Any from mountain_twin.analysis_contract import WeatherSourceType from mountain_twin.explorer import validate_explorer_document from mountain_twin.exposure import load_day1, resolve_local_time from mountain_twin.history import HistoricalTimeWindow, compose_reconstructed_past from mountain_twin.pace.analysis import analyze_route_pace from mountain_twin.photographer.projection import project_photographer_route from mountain_twin.rci.composition import compose_rci_route_conditions from mountain_twin.route_analysis import prepare_route from mountain_twin.route_conditions import compose_route_conditions from mountain_twin.route_solar_conditions import analyze_route_solar from mountain_twin.snow.composition import compose_snow_intelligence from mountain_twin.terrain.cache import CacheIdentity, HorizonProfileCache from mountain_twin.trail_character.explorer import ( compose_frozen_fixture_character, project_trail_terrain_character, ) from mountain_twin.trail_character.osm_fixture import load_frozen_osm_fixture from mountain_twin.unified_route_analysis import ( CoverageMode, DomainCoverage, compose_unified_route_analysis, serialize_unified_route_analysis, ) from mountain_twin.weather.provider import HOURLY_VARIABLES from mountain_twin.weather.resolver import LocalWeatherProviderResolver from mountain_twin.weather.temporal import ( assign_timeline, combined_route_representation, planning_scenarios, ) TIMEZONE = "Europe/Paris" RESOLUTION_DEG = 2 RANGE_M = 5000 RANGE_POLICY = "direct adaptive continuous-bilinear 5 km reference; range-edge/nonconverged samples remain UNKNOWN" def parse_explorer_start_datetime(payload: dict[str, Any]) -> datetime: """Parse the local planning input without interpreting GPX timestamps.""" raw = payload.get("start_datetime") if raw: return datetime.fromisoformat(raw) planning_date = payload.get("planning_date", "2026-07-02") start_time = payload.get("start_time") if not start_time: raise ValueError("request requires start_datetime or start_time") timezone_name = payload.get("timezone", TIMEZONE) return resolve_local_time( f"{planning_date}T{start_time}", timezone_name, ) def _make_identity(point, source_sha256: str) -> CacheIdentity: return CacheIdentity( provider="Copernicus", product="COP-DEM_GLO-30-DGED", source_id="data/raw/dem/tmb_day_01_copernicus_glo30_5km.tif", source_sha256=source_sha256, latitude=point.latitude, longitude=point.longitude, surface_semantics="DSM", observer_model="provider_surface_bilinear_zero_height", observer_height_m=0, angular_resolution_deg=RESOLUTION_DEG, interpolation_method="circular_linear", horizon_range_m=RANGE_M, range_policy=RANGE_POLICY, method_version="horizon_profile_v0_1", ) def _compact_weather(condition) -> dict[str, Any]: values = condition.weather["values"] resolution = condition.weather.get("resolution") or {} names = ( "temperature_2m", "apparent_temperature", "wind_speed_10m", "wind_direction_10m", "wind_gusts_10m", "precipitation", "cloud_cover", "relative_humidity_2m", ) return { "values": { name: { key: values[name][key] for key in ( "value", "unit", "representation_state", "temporal_representation_state", ) if key in values[name] } for name in names if name in values }, "quality": condition.quality["weather_state"], "provenance_ref": resolution.get("source_artifact"), } def _weather_summary(conditions): def values(name): return [ row.weather["values"][name]["value"] for row in conditions if row.weather["values"].get(name, {}).get("value") is not None ] temperatures = values("temperature_2m") gusts = values("wind_gusts_10m") precipitation = values("precipitation") return { "temperature_range_c": [min(temperatures), max(temperatures)] if temperatures else [None, None], "maximum_wind_gust": max(gusts) if gusts else None, "maximum_wind_gust_unit": conditions[0].weather["values"]["wind_gusts_10m"].get("unit"), "precipitation_maximum": max(precipitation) if precipitation else None, "precipitation_positive_point_count": sum(value > 0 for value in precipitation), } def _unavailable_weather_representation(timeline, reason_codes): """Create explicit null weather values without selecting a fake source.""" return [ { "point_index": point.point_index, "route_distance_m": point.route_distance_m, "elevation_m": point.elevation_m, "planned_arrival": point.planned_arrival.isoformat(), "elapsed_route_seconds": point.elapsed_route_seconds, "scenario_datetime_timezone": getattr( point.planned_arrival.tzinfo, "key", str(point.planned_arrival.tzinfo) ), "provider": None, "source_type": None, "values": { variable: { "value": None, "representation_state": "UNKNOWN", "temporal_representation_state": "UNKNOWN", "limitation_codes": tuple(reason_codes), } for variable in HOURLY_VARIABLES }, } for point in timeline ] def _weather_coverage(conditions, resolution, total_points): evaluated = sum(row.quality["weather_state"] == "SUCCESS" for row in conditions) if resolution["mode"] == "UNAVAILABLE": mode = CoverageMode.UNAVAILABLE elif evaluated == total_points: mode = CoverageMode.FULL else: mode = CoverageMode.PARTIAL return DomainCoverage(evaluated, total_points, mode) def _rci_document(unified) -> dict[str, Any]: """Return the inspectable RCI projection without duplicating Unified points.""" result = compose_rci_route_conditions(unified) document = result.to_dict() # RCI's internal state series is intentionally not an Explorer payload. # The UI consumes qualified facts and its existing upstream point projection. document.pop("condition_states", None) return document def _photographer_document(unified) -> dict[str, Any]: """Return an inspectable compact Photographer projection for Explorer. Photographer owns the policy classifications, terrain events, and compact weather summaries. The browser receives those facts rather than a second solar, horizon, pace, or weather-alignment model. Local cache/artifact identifiers are intentionally not exposed by the local Explorer payload. """ document = project_photographer_route(unified).to_dict() source = document.get("atmospheric_context", {}).get("source", {}) source.pop("source_artifact", None) source.pop("cache_identity", None) return document def _snow_document(unified) -> dict[str, Any]: """Return existing Snow Core facts; this boundary adds no Snow science.""" return compose_snow_intelligence(unified).to_dict() def _history_document(unified, rci, photographer, snow, resolution): """Project Python-owned reconstructed-past evidence for the browser. A forecast-resolved request is not represented as reconstructed history. Other resolver states still compose an explicit evidence ledger: solar and terrain can remain available while Weather is unavailable. """ if resolution["mode"] == "FORECAST": return { "contract_version": "history_time_machine_explorer_v0_1", "availability": "UNAVAILABLE", "mode": "RECONSTRUCTED_PAST", "reason_codes": ["HISTORY_RECONSTRUCTED_PAST_NOT_AVAILABLE_FOR_FORECAST_SOURCE"], "limitations": [ "RECONSTRUCTED_PAST requires historical-capable evidence; forecast output is not presented as history." ], "data_gaps": ["HISTORY_RECONSTRUCTED_PAST_SOURCE_UNAVAILABLE"], "engines": [], } scenario = unified.scenario result = compose_reconstructed_past( unified_analysis=unified, requested_window=HistoricalTimeWindow( requested_start=scenario["start_datetime"], requested_end=scenario["finish_datetime"], timezone=unified.identity.timezone, ), snow_result=snow, photographer_result=photographer, rci_result=rci, ) return _history_projection(result.to_dict()) def _history_projection(result: dict[str, Any]) -> dict[str, Any]: """Keep the browser projection compact without reclassifying evidence.""" return { "contract_version": "history_time_machine_explorer_v0_1", "availability": "AVAILABLE", "mode": result["mode"], "requested_window": result["requested_window"], "route_timeline_references": result["route_timeline_references"], "engines": [ { "engine_id": engine["engine_id"], "availability": engine["availability"], "result_references": engine["result_references"], "evidence": [ { **entry, "display_label": _history_evidence_label(entry["evidence_class"]), "display_note": _history_evidence_note(entry["evidence_class"]), } for entry in engine["evidence"] ], "limitations": engine["limitations"], "data_gaps": engine["data_gaps"], } for engine in result["engines"] ], "limitations": result["limitations"], "data_gaps": result["data_gaps"], "diagnostics": result["diagnostics"], } def _history_evidence_label(evidence_class: str) -> str: return { "HISTORICAL_FORECAST": "Historical forecast / model output", "FORECAST": "Forecast / model output", "DETERMINISTIC_ASTRONOMY": "Deterministic astronomy", "TERRAIN_DERIVED": "Terrain-derived light", "DERIVED": "Derived from upstream evidence", "OBSERVED": "Observed evidence", "REANALYSIS": "Reanalysis model evidence", "UNKNOWN": "Unknown evidence provenance", }.get(evidence_class, "Unknown evidence provenance") def _history_evidence_note(evidence_class: str) -> str: if evidence_class == "HISTORICAL_FORECAST": return ( "Archived historical model output for the valid time; not observation or reanalysis. " "Forecast issue/run availability is not established when absent." ) if evidence_class == "DETERMINISTIC_ASTRONOMY": return "Geometric solar calculation for the requested time under the recorded method." if evidence_class == "TERRAIN_DERIVED": return "Derived from deterministic solar geometry and the recorded terrain profile." if evidence_class == "DERIVED": return "Composed from upstream evidence; it does not upgrade that evidence to observation." if evidence_class == "UNKNOWN": return "Source provenance is unavailable or not established." return "Evidence class retained from the server-side Time Machine ledger." def _scenario_document( unified, full_conditions, route_points, timeline, rci, photographer, snow, history ): sampled = {point.point_index: point for point in unified.points} route_rows = [] for point, planned, condition in zip(route_points, timeline, full_conditions): sampled_point = sampled.get(point.point_index) if sampled_point is None: solar = { "status": "NOT_EVALUATED", "evaluated": False, "reason_codes": ["SOLAR_PROFILE_NOT_AVAILABLE_FOR_EXPLORER_POINT"], } solar_quality = {"state": "NOT_EVALUATED"} else: solar = {**sampled_point.solar, "evaluated": True} solar_quality = sampled_point.solar_quality route_rows.append( { "route_id": point.route_id, "point_index": point.point_index, "route_distance_m": planned.route_distance_m, "latitude": point.latitude, "longitude": point.longitude, "route_elevation_m": point.elevation_m, "planned_arrival_time": planned.planned_arrival.isoformat(), "elapsed_route_seconds": planned.elapsed_route_seconds, "weather": _compact_weather(condition), "weather_quality": { "state": condition.quality["weather_state"], "reason_codes": list(condition.quality["weather_reason_codes"]), }, "solar": solar, "solar_quality": solar_quality, } ) return { "name": unified.scenario["name"], "scenario": unified.scenario, "route": unified.route, "coverage": unified.coverage, "points": route_rows, "events": [event.to_dict() for event in unified.events], "summary": unified.summary, "provenance": unified.provenance, "limitations": list(unified.limitations), "rci": rci, "photographer": photographer, "snow": snow, "history": history, } def _route_document( route_points, prepared, cache, identities, profiles, resolved_weather, start_datetime, elevation_source_identity, trail_character=None, ): field = resolved_weather.field resolution = resolved_weather.resolution.to_dict() source_type = resolution["source_type"] source_type = WeatherSourceType(source_type) if source_type else None provider = resolution["provider"] fetched_at = resolution["fetched_at"] weather_units = field.samples[0].base.units if field is not None else {} scenarios = [] timing = {} for scenario in planning_scenarios(start_datetime): started = time.perf_counter() pace_started = time.perf_counter() pace = analyze_route_pace( prepared, scenario_name=scenario.name, scenario_factor=scenario.pace_factor or 1.0, pauses=scenario.pauses, source_identity=elevation_source_identity, ) pace_seconds = time.perf_counter() - pace_started if pace.state.value != "COMPLETE": raise ValueError(f"terrain pace unresolved: {pace.reason_codes}") timeline_started = time.perf_counter() timeline = assign_timeline(prepared, scenario, pace) timeline_seconds = time.perf_counter() - timeline_started weather_started = time.perf_counter() rows = ( combined_route_representation(field, timeline) if field is not None else _unavailable_weather_representation(timeline, resolution["reason_codes"]) ) rows = [ { **row, "latitude": route_points[row["point_index"]].latitude, "longitude": route_points[row["point_index"]].longitude, } for row in rows ] conditions = compose_route_conditions( route_id=route_points[0].route_id, scenario_name=scenario.name, timeline=timeline, weather_rows=rows, source_type=source_type, fetched_at=fetched_at, provider=provider, weather_units=weather_units, weather_resolution=resolution, ) weather_seconds = time.perf_counter() - weather_started solar_started = time.perf_counter() solar = analyze_route_solar( route_points, scenario, cache, identities, evaluated_indices=tuple(range(len(route_points))), loaded_profiles=profiles, pace_result=pace, ) solar_seconds = time.perf_counter() - solar_started composition_started = time.perf_counter() selected_conditions = compose_route_conditions( route_id=route_points[0].route_id, scenario_name=scenario.name, timeline=timeline, weather_rows=rows, source_type=source_type, fetched_at=fetched_at, provider=provider, weather_units=weather_units, weather_resolution=resolution, solar_by_index={point.point_index: point.to_dict() for point in solar.points}, terrain=solar.terrain_provenance, ) unified = compose_unified_route_analysis( route_points=route_points, timeline=timeline, scenario=scenario, condition_results=selected_conditions, solar_series=solar, weather_coverage=_weather_coverage(conditions, resolution, len(route_points)), provenance={ "weather": { "provider": provider, "source_type": resolution["source_type"], "model_selection": resolution["model_selection"], }, "solar": solar.terrain_provenance, }, weather_summary=_weather_summary(conditions), weather_resolution=resolution, pace_result=pace, ) composition_seconds = time.perf_counter() - composition_started rci_started = time.perf_counter() rci = _rci_document(unified) rci_seconds = time.perf_counter() - rci_started photographer_started = time.perf_counter() photographer = _photographer_document(unified) photographer_seconds = time.perf_counter() - photographer_started snow_started = time.perf_counter() snow = _snow_document(unified) snow_seconds = time.perf_counter() - snow_started history_started = time.perf_counter() history = _history_document(unified, rci, photographer, snow, resolution) history_seconds = time.perf_counter() - history_started serialization_started = time.perf_counter() encoded = serialize_unified_route_analysis(unified) serialization_seconds = time.perf_counter() - serialization_started scenarios.append( _scenario_document( unified, conditions, route_points, timeline, rci, photographer, snow, history ) ) timing[scenario.name] = { "total_seconds": time.perf_counter() - started, "timeline_seconds": timeline_seconds, "pace_seconds": pace_seconds, "pace_preprocessing_seconds": pace.runtime.get("preprocessing_seconds", 0.0), "pace_calculation_seconds": pace.runtime.get("calculation_seconds", 0.0), "weather_seconds": weather_seconds, "solar_seconds": solar_seconds, "composition_seconds": composition_seconds, "rci_composition_seconds": rci_seconds, "photographer_composition_seconds": photographer_seconds, "snow_composition_seconds": snow_seconds, "history_composition_seconds": history_seconds, "serialization_seconds": serialization_seconds, "cached_solar_query_seconds": solar.runtime.cached_query_seconds, "terrain_interpolation_queries": solar.runtime.terrain_interpolation_queries, "serialized_unified_bytes": len(encoded.encode("utf-8")), "serialized_unified_sha256": hashlib.sha256(encoded.encode("utf-8")).hexdigest(), "serialized_rci_bytes": len( json.dumps(rci, sort_keys=True, separators=(",", ":"), allow_nan=False).encode( "utf-8" ) ), "serialized_photographer_bytes": len( json.dumps( photographer, sort_keys=True, separators=(",", ":"), allow_nan=False ).encode("utf-8") ), "serialized_snow_bytes": len( json.dumps(snow, sort_keys=True, separators=(",", ":"), allow_nan=False).encode( "utf-8" ) ), "serialized_history_bytes": len( json.dumps(history, sort_keys=True, separators=(",", ":"), allow_nan=False).encode( "utf-8" ) ), } route = { "route_id": route_points[0].route_id, "route_name": "TMB Day 1", "point_count": len(route_points), "distance_m": prepared[-1].cumulative_distance_m, "minimum_route_elevation_m": min( point.elevation_m for point in route_points if point.elevation_m is not None ), "maximum_route_elevation_m": max( point.elevation_m for point in route_points if point.elevation_m is not None ), } return { "contract_version": "mountain_twin_explorer_v0_7", "dataset": { "route_name": "TMB Day 1", "route_id": route_points[0].route_id, "planning_date": start_datetime.date().isoformat(), "timezone": TIMEZONE, "start_time": start_datetime.isoformat(), "time_semantics": "explicit planned scenarios; GPX timestamps are not hiking chronology", "generated_from": "Unified Route Analysis v0.4 + Route Conditions Intelligence v0.2 + Photographer Intelligence v0.1 + Snow Intelligence v0.1 with full-route 2-degree profiles", }, "route": route, "route_points": [ { "point_index": point.point_index, "latitude": point.latitude, "longitude": point.longitude, "route_distance_m": prepared_point.cumulative_distance_m, "route_elevation_m": point.elevation_m, } for point, prepared_point in zip(route_points, prepared) ], "scenarios": scenarios, "provenance": { "weather": resolution, "solar": scenarios[0]["provenance"].get("solar", {}), }, "trail_terrain_character": trail_character, "performance": { "profile_load_seconds": None, "scenario_timings": timing, "profile_construction_performed": False, }, "limitations": [ "Solar coverage is FULL only because all route point profiles are available in the exact-coordinate cache.", "Solar transition events are bounded by adjacent evaluated route points; no sub-point physical transition is inferred.", "Weather is a locally resolved provider-model interpolation, not observation truth.", "Photographer conventions and terrain-light facts are compact projections of Unified analysis; they are not photographic-quality predictions.", "Snowfall and modelled ground snow depth are separate evidence facts; no snow truth, passability, or safety conclusion is inferred.", "GLO-30 DGED is a DSM, not bare-earth physical terrain truth.", "The Explorer is analytical context, not safety or risk advice.", ], } class ExplorerAnalysisService: """Reuse route, weather, and terrain state across local Explorer requests.""" def __init__(self, root: Path): self.root = root.resolve() self.route_points = load_day1(self.root / "data/generated/trail_points_master.csv") self.prepared = prepare_route(self.route_points).points fixture = load_frozen_osm_fixture( self.root / "tests/fixtures/osm/tmb_day_01_route_section_osm_260919.json" ) section_start, section_end = 100, 343 section_result = compose_frozen_fixture_character( "tmb_day_01", self.route_points[section_start:section_end], fixture ) self.trail_terrain_character = project_trail_terrain_character( section_result, fixture, route_distance_offset_m=self.prepared[section_start].cumulative_distance_m, ) provenance_path = self.root / "data/generated/provenance.json" provenance = json.loads(provenance_path.read_text(encoding="utf-8")) self.elevation_source_identity = next( item["source_sha256"] for item in provenance["sources"] if item["metadata"]["route_id"] == "tmb_day_01" ) self.weather_resolver = LocalWeatherProviderResolver(self.root) metadata_path = self.root / "data/raw/dem/tmb_day_01_copernicus_glo30_5km.json" metadata = json.loads(metadata_path.read_text(encoding="utf-8")) self.cache = HorizonProfileCache(self.root / "data/generated/terrain_horizon_cache_v0_2") self.identities = { point.point_index: _make_identity(point, metadata["sha256"]) for point in self.route_points } started = time.perf_counter() self.profiles = { index: self.cache.get_profile(identity) for index, identity in self.identities.items() } self.profile_load_seconds = time.perf_counter() - started if len(self.profiles) != len(self.route_points) or any( profile is None for profile in self.profiles.values() ): raise RuntimeError("full-route profile cache is incomplete") def analyze_explorer_request( self, route_id: str, start_datetime: datetime, preset: str = "NOMINAL" ) -> dict[str, Any]: """Build a planned result without constructing terrain profiles.""" if route_id != "tmb_day_01": raise ValueError("only TMB Day 1 is available in the Explorer prototype") if start_datetime.tzinfo is None or start_datetime.utcoffset() is None: raise ValueError("start_datetime must be timezone-aware") if getattr(start_datetime.tzinfo, "key", None) != TIMEZONE: raise ValueError(f"start_datetime must use {TIMEZONE}") if preset not in {"FAST", "NOMINAL", "SLOW"}: raise ValueError("preset must be FAST, NOMINAL, or SLOW") scenarios = planning_scenarios(start_datetime) timeline_ends = [ timeline[-1].planned_arrival for scenario in scenarios if ( pace := analyze_route_pace( self.prepared, scenario_name=scenario.name, scenario_factor=scenario.pace_factor or 1.0, pauses=scenario.pauses, source_identity=getattr(self, "elevation_source_identity", None), ) ).state.value == "COMPLETE" and (timeline := assign_timeline(self.prepared, scenario, pace)) ] window_end = max(timeline_ends, default=start_datetime) resolution_started = time.perf_counter() resolved_weather = self.weather_resolver.resolve(start_datetime, window_end) resolution_seconds = time.perf_counter() - resolution_started document = _route_document( self.route_points, self.prepared, self.cache, self.identities, self.profiles, resolved_weather, start_datetime, elevation_source_identity=getattr(self, "elevation_source_identity", None), trail_character=getattr(self, "trail_terrain_character", None), ) validate_explorer_document(document) document["request"] = { "route_id": route_id, "start_datetime": start_datetime.isoformat(), "timezone": TIMEZONE, "preset": preset, "terrain_profile_construction_performed": False, "terrain_profiles_reused": len(self.profiles), "weather_resolution": resolved_weather.resolution.to_dict(), } document["performance"]["profile_load_seconds"] = self.profile_load_seconds document["performance"]["weather_resolution_seconds"] = resolution_seconds return document def analyze_explorer_request( service: ExplorerAnalysisService, route_id: str, start_datetime: datetime, preset: str = "NOMINAL", ) -> dict[str, Any]: """Functional adapter for local callers and a future API layer.""" return service.analyze_explorer_request(route_id, start_datetime, preset)