"""Compact product projection over existing Explorer/producer-owned results.""" from __future__ import annotations import json from dataclasses import asdict, dataclass from typing import Any, Mapping from .contracts import AnalysisRun, AnalysisScenario, Journey, JourneyPlan, ProductAvailability, RouteRevision PROJECTION_VERSION = "journey_intelligence_projection_v0_1" @dataclass(frozen=True) class JourneyIntelligenceProjection: identity: Mapping[str, str] journey_summary: Mapping[str, Any] route_summary: Mapping[str, Any] planning_context: Mapping[str, Any] timeline_summary: Mapping[str, Any] availability: Mapping[str, Mapping[str, Any]] domains: Mapping[str, Mapping[str, Any]] selection_index: Mapping[str, Any] why_index: Mapping[str, Mapping[str, Any]] limitations: tuple[str, ...] contract_version: str = PROJECTION_VERSION def to_dict(self) -> dict[str, Any]: return asdict(self) def serialize_journey_intelligence_projection(result: JourneyIntelligenceProjection) -> str: return json.dumps(result.to_dict(), sort_keys=True, separators=(",", ":"), allow_nan=False) def product_availability(native_state: str | None, *, reason_codes=(), native_state_override: str | None = None) -> dict[str, Any]: """Map only faithful producer coverage/status vocabulary; retain native state.""" state = str(native_state or "UNKNOWN").upper() mapping = { "FULL": ProductAvailability.AVAILABLE.value, "SUCCESS": ProductAvailability.AVAILABLE.value, "AVAILABLE": ProductAvailability.AVAILABLE.value, "COMPLETE": ProductAvailability.AVAILABLE.value, "PARTIAL": ProductAvailability.PARTIAL.value, "SAMPLED": ProductAvailability.PARTIAL.value, "UNAVAILABLE": ProductAvailability.UNAVAILABLE.value, "UNSUPPORTED": ProductAvailability.UNAVAILABLE.value, "UNKNOWN": ProductAvailability.UNKNOWN.value, "UNRESOLVED": ProductAvailability.UNKNOWN.value, "NOT_APPLICABLE": ProductAvailability.NOT_APPLICABLE.value, "NOT_EVALUATED": ProductAvailability.UNKNOWN.value, } return { "state": mapping.get(state, ProductAvailability.UNKNOWN.value), "native_state": native_state_override or state, "reason_codes": list(reason_codes), } def build_journey_intelligence_projection( *, journey: Journey, route_revision: RouteRevision, plan: JourneyPlan, run: AnalysisRun, scenario: AnalysisScenario, explorer_document: Mapping[str, Any], ) -> JourneyIntelligenceProjection: """Project one selected Explorer scenario without recalculating any science.""" if route_revision.route_id != explorer_document.get("route", {}).get("route_id"): raise ValueError("producer route does not match route revision") selected = next( (item for item in explorer_document.get("scenarios", ()) if item.get("name") == scenario.name), None, ) if selected is None: raise ValueError("producer result does not contain selected scenario") points = selected.get("points", ()) if len(points) != route_revision.point_count: raise ValueError("producer route point count does not match route revision") coverage = selected.get("coverage", {}) weather = coverage.get("weather", {}) solar = coverage.get("solar", {}) rci = selected.get("rci", {}) photographer = selected.get("photographer", {}) snow = selected.get("snow", {}) history = selected.get("history", {}) trail = explorer_document.get("trail_terrain_character", {}) terrain_availability = product_availability("AVAILABLE") not_evaluated_reasons = _not_evaluated_reasons(points) availability = { "terrain": terrain_availability, "light": { **product_availability(solar.get("mode")), "not_evaluated_reason_codes": not_evaluated_reasons, "not_evaluated_state": "NOT_EVALUATED" if not_evaluated_reasons else None, }, "weather": product_availability(weather.get("mode")), "conditions": product_availability((rci.get("quality") or {}).get("coverage")), "photography": product_availability((photographer.get("quality") or {}).get("coverage")), "snow": product_availability((snow.get("quality") or {}).get("coverage")), "history": product_availability(history.get("availability"), reason_codes=history.get("reason_codes", ())), "trail_character": product_availability(trail.get("availability")), } producer_refs = { "unified": f"unified:{run.analysis_run_id}:{scenario.analysis_scenario_id}", "rci": f"rci:{run.analysis_run_id}:{scenario.analysis_scenario_id}", "photographer": f"photographer:{run.analysis_run_id}:{scenario.analysis_scenario_id}", "snow": f"snow:{run.analysis_run_id}:{scenario.analysis_scenario_id}", "history": f"history:{run.analysis_run_id}:{scenario.analysis_scenario_id}", "trail_character": f"trail_character:{route_revision.route_revision_id}", } route = explorer_document["route"] timeline = selected.get("scenario", {}) domains = { "terrain": { "minimum_elevation_m": route.get("minimum_route_elevation_m"), "maximum_elevation_m": route.get("maximum_route_elevation_m"), "producer_result_reference": producer_refs["unified"], }, "light": { "summary": selected.get("summary", {}).get("solar_counts", {}), "transition_count": len(selected.get("events", ())), "producer_result_reference": producer_refs["unified"], }, "weather": { "summary": selected.get("summary", {}).get("weather", selected.get("summary", {})), "source_type": (selected.get("provenance", {}).get("weather", {}).get("source_type")), "producer_result_reference": producer_refs["unified"], }, "conditions": _compact_conditions(rci, producer_refs["rci"]), "photography": _compact_photography(photographer, producer_refs["photographer"]), "snow": _compact_snow(snow, producer_refs["snow"]), "history": _compact_history(history, producer_refs["history"]), "trail_character": _compact_trail(trail, producer_refs["trail_character"]), } why_index = { key: { "producer": producer, "producer_result_reference": producer_refs[key], "producer_contract_version": _contract_version_for(key, explorer_document, selected), "availability": availability[domain], "source_snapshot_ids": list(run.source_snapshot_ids), } for key, producer, domain in ( ("unified", "Mountain Twin Unified Route Analysis", "light"), ("rci", "Mountain Twin Route Conditions Intelligence", "conditions"), ("photographer", "Mountain Twin Photographer Intelligence", "photography"), ("snow", "Mountain Twin Snow Intelligence", "snow"), ("history", "Mountain Twin History / Time Machine", "history"), ("trail_character", "Mountain Twin Trail / Terrain Character", "trail_character"), ) } return JourneyIntelligenceProjection( identity={ "projection_id": f"journey-projection:{run.analysis_run_id}:{scenario.analysis_scenario_id}", "journey_id": journey.journey_id, "journey_plan_id": plan.journey_plan_id, "route_revision_id": route_revision.route_revision_id, "analysis_run_id": run.analysis_run_id, "analysis_scenario_id": scenario.analysis_scenario_id, }, journey_summary={"title": journey.title, "activity_type": journey.activity_type}, route_summary={ "route_name": route.get("route_name"), "distance_m": route.get("distance_m"), "point_count": route.get("point_count"), "elevation_unit": "m", }, planning_context={ "journey_timezone": plan.journey_timezone, "planned_start_local": scenario.planned_start_local, "planned_start_utc": scenario.planned_start_utc, "pace_policy_id": scenario.pace_policy_id, "pause_policy_id": scenario.pause_policy_id, "scenario_name": scenario.name, }, timeline_summary={ "planned_finish": timeline.get("finish_datetime"), "total_planned_time_s": timeline.get("total_planned_time_s"), "total_moving_time_s": timeline.get("moving_time_s"), "total_pause_time_s": timeline.get("pause_time_s"), "route_timeline_reference": producer_refs["unified"], }, availability=availability, domains=domains, selection_index={ "route_revision_id": route_revision.route_revision_id, "analysis_scenario_id": scenario.analysis_scenario_id, "route_point_index_range": [0, route_revision.point_count - 1], "route_distance_m_range": [0, route.get("distance_m")], "route_series_reference": producer_refs["unified"], "native_segmentation": "producer-owned intervals and segments are not merged", }, why_index=why_index, limitations=tuple(explorer_document.get("limitations", ())), ) def _not_evaluated_reasons(points): return sorted({code for point in points if point.get("solar", {}).get("status") == "NOT_EVALUATED" for code in point.get("solar", {}).get("reason_codes", ())}) def _compact_conditions(value, reference): return {"signal_count": len(value.get("signals", ())), "cluster_count": len(value.get("clusters", ())), "event_count": len(value.get("events", ())), "producer_result_reference": reference} def _compact_photography(value, reference): terrain = value.get("terrain_light", {}) return {"golden_window_count": len(value.get("golden_hour", {}).get("windows", ())), "blue_window_count": len(value.get("blue_hour", {}).get("windows", ())), "terrain_event_count": len(terrain.get("events", ())), "producer_result_reference": reference} def _compact_snow(value, reference): facts = value.get("snow_facts", ()) return {"fact_count": len(facts), "fact_types": sorted({fact.get("semantics", {}).get("fact_type") for fact in facts if fact.get("semantics", {}).get("fact_type")}), "producer_result_reference": reference} def _compact_history(value, reference): return {"mode": value.get("mode"), "engine_count": len(value.get("engines", ())), "producer_result_reference": reference} def _compact_trail(value, reference): mapped = value.get("mapped_character", {}) return {"section": value.get("route_reference", {}), "coverage": mapped.get("coverage"), "characteristic_count": len(mapped.get("characteristics", ())), "producer_result_reference": reference, "limitations": value.get("limitations", ())} def _contract_version_for(key, document, scenario): if key == "unified": return "0.4" if key == "trail_character": return document.get("trail_terrain_character", {}).get("contract_version") return scenario.get(key, {}).get("contract_version")