"""Factual route-condition events extracted from already composed upstream facts.""" from __future__ import annotations import hashlib import json from dataclasses import replace from typing import Any, Callable, Mapping, Sequence from mountain_twin.rci.contracts import CoverageState, EvidenceQuality, RouteConditionEvent from mountain_twin.rci.why import event_why def extract_route_condition_events(analysis: Any) -> tuple[RouteConditionEvent, ...]: """Extract only point facts supported by the supplied Unified Analysis. Weather extrema describe available route samples, never a continuous or route-wide observation. Astronomical transitions and terrain visibility transitions retain their different upstream event families. """ points = tuple(_value(analysis, "points", ())) events: list[RouteConditionEvent] = [] if points: events.extend(_route_endpoint_events(points, analysis)) events.extend(_route_high_points(points, _point_coverage(analysis))) events.extend(_weather_extrema(points, "temperature_2m", "TEMP_MIN", min, analysis)) events.extend(_weather_extrema(points, "temperature_2m", "TEMP_MAX", max, analysis)) events.extend(_weather_extrema(points, "wind_gusts_10m", "MAX_GUST", max, analysis)) events.extend(_solar_transition_events(_value(analysis, "events", ()), analysis)) return tuple(sorted(events, key=lambda event: (_event_order(event), event.event_id))) def _route_endpoint_events(points: Sequence[Any], analysis: Any) -> tuple[RouteConditionEvent, ...]: total_points = _value(_value(analysis, "route", {}), "point_count") quality = EvidenceQuality(coverage=_point_coverage(analysis)) events = [] if _value(points[0], "point_index") == 0: events.append(_event("ROUTE_START", points[0], source_reference="route", quality=quality)) if total_points is not None and _value(points[-1], "point_index") == total_points - 1: events.append(_event("ROUTE_END", points[-1], source_reference="route", quality=quality)) return tuple(events) def _route_high_points( points: Sequence[Any], coverage: CoverageState ) -> tuple[RouteConditionEvent, ...]: available = tuple( (float(_value(point, "route_elevation_m")), point) for point in points if _value(point, "route_elevation_m") is not None ) if not available: return () maximum = max(value for value, _ in available) tied = tuple( sorted(_value(point, "point_index") for value, point in available if value == maximum) ) return tuple( _event( "ROUTE_HIGH_POINT", point, value=maximum, unit="m", source_reference="route_elevation", quality=EvidenceQuality(coverage=coverage), evidence_payload={ "semantics": "highest elevation represented by analyzed route data", "tied_route_point_indices": tied, }, ) for _, point in available if _value(point, "point_index") in tied ) def _weather_extrema( points: Sequence[Any], variable: str, event_type: str, reducer: Callable[[Sequence[float]], float], analysis: Any, ) -> tuple[RouteConditionEvent, ...]: values = [] for point in points: item = _value(_value(_value(point, "weather", {}), "values", {}), variable, {}) value = _value(item, "value") if value is not None: values.append((float(value), _value(item, "unit"), point)) if not values: return () extreme = reducer(tuple(value for value, _, _ in values)) tied = tuple( sorted(_value(point, "point_index") for value, _, point in values if value == extreme) ) quality = EvidenceQuality( coverage=_combine_coverage(_domain_coverage(analysis, "weather"), _point_coverage(analysis)) ) return tuple( _event( event_type, point, value=extreme, unit=unit, source_reference=_value(point, "weather_provenance_ref"), quality=quality, evidence_payload={ "semantics": ( "extremum among available analyzed route samples, " "not a continuous observation" ), "variable": variable, "tied_route_point_indices": tied, }, ) for value, unit, point in values if value == extreme ) def _solar_transition_events( upstream_events: Sequence[Any], analysis: Any ) -> tuple[RouteConditionEvent, ...]: result = [] terrain_mappings = { "SOLAR_DIRECT_TO_SHADOW": "SHADOW_ENTRY", "SOLAR_SHADOW_TO_DIRECT": "SHADOW_EXIT", } astronomical_mappings = { "ASTRONOMICAL_CIVIL_TWILIGHT_TO_DAY": ("SUNRISE", "CIVIL_TWILIGHT_EXIT"), "ASTRONOMICAL_DAY_TO_CIVIL_TWILIGHT": ("SUNSET", "CIVIL_TWILIGHT_ENTRY"), } quality = EvidenceQuality(coverage=_domain_coverage(analysis, "solar")) for upstream in upstream_events: event_type = _value(upstream, "event_type") mapped = terrain_mappings.get(event_type) mapped_types = (mapped,) if mapped is not None else astronomical_mappings.get(event_type, ()) for mapped_type in mapped_types: result.append( _event( mapped_type, { "point_index": _value(upstream, "after_point_index"), "route_distance_m": _value(upstream, "after_route_distance_m"), "planned_arrival_time": _value(upstream, "after_planned_time"), }, source_reference=_value(upstream, "provenance_ref"), quality=quality, evidence_payload={ "upstream_event_type": event_type, "upstream_event_family": _value(upstream, "event_family"), "before_state": _value(upstream, "before_state"), "after_state": _value(upstream, "after_state"), "boundary_semantics": "sampled upstream route-point transition", }, ) ) return tuple(result) def _event( event_type: str, point: Any, *, value: float | None = None, unit: str | None = None, source_reference: str | None = None, quality: EvidenceQuality | None = None, evidence_payload: Mapping[str, Any] | None = None, ) -> RouteConditionEvent: index = _value(point, "point_index") identity = { "event_type": event_type, "point_index": index, "source_reference": source_reference, } event = RouteConditionEvent( event_id=( "rci-event-" + hashlib.sha256( json.dumps(identity, sort_keys=True, separators=(",", ":")).encode("utf-8") ).hexdigest()[:24] ), event_type=event_type, route_point_index=index, route_distance_m=_value(point, "route_distance_m"), planned_time=_value(point, "planned_arrival_time", _value(point, "planned_time")), value=value, unit=unit, source_reference=source_reference, quality=quality or EvidenceQuality(coverage=CoverageState.UNKNOWN), explanation_key=f"rci.event.{event_type.lower()}", evidence_payload=evidence_payload or {}, ) return replace(event, why=event_why(event)) def _domain_coverage(analysis: Any, domain: str) -> CoverageState: mode = _value(_value(_value(analysis, "coverage", {}), domain, {}), "mode") return { "FULL": CoverageState.FULL, "PARTIAL": CoverageState.PARTIAL, "SAMPLED": CoverageState.PARTIAL, "UNAVAILABLE": CoverageState.UNAVAILABLE, }.get(mode, CoverageState.UNKNOWN) def _point_coverage(analysis: Any) -> CoverageState: total = _value(_value(analysis, "route", {}), "point_count") evaluated = len(tuple(_value(analysis, "points", ()))) if total == 0: return CoverageState.UNAVAILABLE if total is None: return CoverageState.UNKNOWN return CoverageState.FULL if evaluated == total else CoverageState.PARTIAL def _combine_coverage(left: CoverageState, right: CoverageState) -> CoverageState: if CoverageState.UNAVAILABLE in {left, right}: return CoverageState.UNAVAILABLE if CoverageState.UNKNOWN in {left, right}: return CoverageState.UNKNOWN if CoverageState.NOT_APPLICABLE in {left, right}: return CoverageState.NOT_APPLICABLE if left is CoverageState.FULL and right is CoverageState.FULL: return CoverageState.FULL return CoverageState.PARTIAL def _event_order(event: RouteConditionEvent) -> tuple[bool, int, str]: return ( event.route_point_index is None, event.route_point_index if event.route_point_index is not None else 0, event.event_type, ) def _value(item: Any, key: str, default: Any = None) -> Any: return item.get(key, default) if isinstance(item, Mapping) else getattr(item, key, default)