"""Project the accepted Golden Hour convention from existing Unified facts.""" from __future__ import annotations import math from dataclasses import asdict from typing import Any, Mapping, Sequence from mountain_twin.analysis_contract import AnalysisIdentity from mountain_twin.photographer.atmospheric import project_atmospheric_context from mountain_twin.photographer.contracts import ( BoundarySemantics, ConventionMembership, ConventionWindowResult, PhotographerIntelligenceResult, PhotographicWindow, RouteBoundary, RouteSampleReference, RouteTimelineReference, derive_photographic_window_id, ) from mountain_twin.photographer.policy import ( PHOTOGRAPHER_POLICY_V0_1, ConventionSupportState, PhotographerPolicy, ) from mountain_twin.photographer.terrain_light import project_terrain_light from mountain_twin.rci.contracts import CoverageState, EvidenceQuality def project_photographer_route( unified_analysis: Any, *, policy: PhotographerPolicy = PHOTOGRAPHER_POLICY_V0_1, ) -> PhotographerIntelligenceResult: """Create compact convention windows from one already-composed route analysis. This function deliberately consumes only upstream planned arrival and solar elevation facts. It neither calculates a route pace nor calls the solar engine, so FAST/NOMINAL/SLOW remain the existing scenario timelines. """ route = _mapping(_value(unified_analysis, "route", {})) scenario = _mapping(_value(unified_analysis, "scenario", {})) identity = _mapping(_value(unified_analysis, "identity", {})) route_id = route.get("route_id") scenario_name = scenario.get("name") if not isinstance(route_id, str) or not route_id: raise ValueError("Unified analysis route ID is required") if not isinstance(scenario_name, str) or not scenario_name: raise ValueError("Unified analysis scenario name is required") upstream_reference = _analysis_reference(unified_analysis) records = _records(_value(unified_analysis, "points", ())) coverage, coverage_reasons = _coverage(unified_analysis, records) golden = _convention_window_result( records, coverage, coverage_reasons, policy, policy.golden_hour, upstream_reference, ) blue = _convention_window_result( records, coverage, coverage_reasons, policy, policy.blue_hour, upstream_reference, ) terrain_light = project_terrain_light( unified_analysis, upstream_analysis_reference=upstream_reference, policy=policy, ) atmospheric_context = project_atmospheric_context( unified_analysis, upstream_analysis_reference=upstream_reference, golden_hour=golden, blue_hour=blue, ) pace_reference = _pace_reference(_value(unified_analysis, "points", ())) timeline_reference = f"{upstream_reference}:timeline" reason_codes = tuple(sorted(set((*coverage_reasons, *golden.reason_codes)))) result = PhotographerIntelligenceResult( identity=AnalysisIdentity( analysis_type="photographer_intelligence", semantic_type="route_photographic_light_context", version="0.1", subject_id=route_id, scenario_datetime=scenario.get("start_datetime"), timezone=route.get("timezone") or identity.get("timezone"), ), analysis_metadata={ "upstream_analysis_reference": upstream_reference, "upstream_analysis_type": identity.get("analysis_type"), "upstream_analysis_version": identity.get("version"), "solar_computation": "REUSED_FROM_UNIFIED_ROUTE_ANALYSIS", "timeline_computation": "REUSED_FROM_UNIFIED_ROUTE_ANALYSIS", "state_series_retained_publicly": False, }, route_timeline_reference=RouteTimelineReference( route_id=route_id, scenario_name=scenario_name, upstream_analysis_reference=upstream_reference, timeline_reference=timeline_reference, pace_reference=pace_reference, ), policy=policy, golden_hour=golden, blue_hour=blue, terrain_light=terrain_light, atmospheric_context=atmospheric_context, quality=EvidenceQuality( coverage=coverage, source="unified_route_analysis.solar", derived_status="GEOMETRIC_SOLAR_ELEVATION_CONVENTION", reason_codes=reason_codes, limitations=( "PHOTOGRAPHIC_CONVENTIONS_ARE_NOT_ASTRONOMICAL_STANDARDS", "NO_ATMOSPHERIC_REFRACTION_OR_APPARENT_SOLAR_LIMB_MODEL", "NO_EXACT_ROUTE_THRESHOLD_CROSSING_INFERRED", ), ), provenance={ "policy": policy.to_dict(), "upstream_analysis_reference": upstream_reference, "upstream_solar_reference": "unified_route_analysis.points[].solar", "upstream_timeline_reference": timeline_reference, }, diagnostics=( "PHOTOGRAPHIC_CONVENTIONS_ARE_NOT_PHOTOGRAPHIC_QUALITY", "WINDOWS_ARE_SAMPLED_OR_BOUNDED_NOT_EXACT_CROSSINGS", ), ) return result def project_golden_hour_route( unified_analysis: Any, *, policy: PhotographerPolicy = PHOTOGRAPHER_POLICY_V0_1, ) -> PhotographerIntelligenceResult: """Backward-compatible entry point returning all v0.1 convention projections.""" return project_photographer_route(unified_analysis, policy=policy) def project_golden_hour_scenarios( unified_analyses: Sequence[Any], *, policy: PhotographerPolicy = PHOTOGRAPHER_POLICY_V0_1, ) -> tuple[PhotographerIntelligenceResult, ...]: """Project independently composed scenarios without recalculating their pace.""" results = tuple(project_photographer_route(analysis, policy=policy) for analysis in unified_analyses) names = tuple(result.route_timeline_reference.scenario_name for result in results) if len(set(names)) != len(names): raise ValueError("Photographer scenario names must be unique") return tuple(sorted(results, key=lambda result: result.route_timeline_reference.scenario_name)) def _records(points: Sequence[Any]): records = [] previous_index = None for point in points: sample = _sample_reference(point) if previous_index is not None and sample.route_point_index <= previous_index: raise ValueError("Unified route points must be strictly ordered") previous_index = sample.route_point_index solar = _mapping(_value(point, "solar", {})) records.append((sample, _finite_elevation(solar.get("solar_elevation_deg")))) return tuple(records) def _finite_elevation(value: Any) -> float | None: if isinstance(value, bool): return None try: elevation = float(value) except (TypeError, ValueError): return None if not math.isfinite(elevation): return None return elevation def _coverage(unified_analysis: Any, records) -> tuple[CoverageState, tuple[str, ...]]: declared = _value(_value(_value(unified_analysis, "coverage", {}), "solar", {}), "mode") declared = getattr(declared, "value", declared) declared_coverage = { "FULL": CoverageState.FULL, "SAMPLED": CoverageState.PARTIAL, "PARTIAL": CoverageState.PARTIAL, "UNAVAILABLE": CoverageState.UNAVAILABLE, }.get(declared, CoverageState.UNKNOWN) route_point_count = _value(_value(unified_analysis, "route", {}), "point_count") known_count = sum(elevation is not None for _, elevation in records) if declared_coverage is CoverageState.UNAVAILABLE: return CoverageState.UNAVAILABLE, ("SOLAR_COVERAGE_UNAVAILABLE",) if not records: return CoverageState.UNAVAILABLE, ("SOLAR_ROUTE_POINTS_UNAVAILABLE",) if known_count == 0: return CoverageState.UNKNOWN, ("SOLAR_ELEVATION_UNAVAILABLE",) if ( declared_coverage is CoverageState.FULL and isinstance(route_point_count, int) and route_point_count == len(records) and known_count == len(records) ): return CoverageState.FULL, () reasons = [] if declared_coverage is CoverageState.PARTIAL or route_point_count != len(records): reasons.append("SOLAR_COVERAGE_PARTIAL") if known_count != len(records): reasons.append("SOLAR_ELEVATION_UNAVAILABLE") if declared_coverage is CoverageState.UNKNOWN: reasons.append("SOLAR_COVERAGE_UNKNOWN") return CoverageState.PARTIAL if reasons else CoverageState.UNKNOWN, tuple(sorted(set(reasons))) def _convention_window_result( records, coverage, coverage_reasons, policy, convention_policy, upstream_reference ): counts = {state.value: 0 for state in ConventionMembership} memberships = tuple( (sample, _membership(elevation, convention_policy)) for sample, elevation in records ) for _, membership in memberships: counts[membership.value] += 1 if coverage is CoverageState.UNAVAILABLE: return ConventionWindowResult( convention=convention_policy.convention, support_state=ConventionSupportState.SUPPORTED, coverage=coverage, intersects_route=None, threshold_provenance=convention_policy.threshold_provenance, reason_codes=coverage_reasons, sampled_membership_counts=counts, ) windows = _windows(memberships, policy, convention_policy.convention, upstream_reference) unknown_present = counts[ConventionMembership.UNKNOWN.value] > 0 if windows: intersects_route = True elif coverage is CoverageState.FULL and not unknown_present: intersects_route = False else: intersects_route = None return ConventionWindowResult( convention=convention_policy.convention, support_state=ConventionSupportState.SUPPORTED, coverage=coverage, intersects_route=intersects_route, windows=windows, threshold_provenance=convention_policy.threshold_provenance, reason_codes=coverage_reasons, sampled_membership_counts=counts, ) def _membership(value: float | None, convention_policy) -> ConventionMembership: if value is None: return ConventionMembership.UNKNOWN return ( ConventionMembership.IN_WINDOW if convention_policy.contains(value) else ConventionMembership.OUTSIDE_WINDOW ) def _windows(records, policy, convention, upstream_reference): windows = [] index = 0 while index < len(records): if records[index][1] is not ConventionMembership.IN_WINDOW: index += 1 continue start = index while index + 1 < len(records) and records[index + 1][1] is ConventionMembership.IN_WINDOW: index += 1 end = index first, last = records[start][0], records[end][0] windows.append( PhotographicWindow( window_id=derive_photographic_window_id( upstream_analysis_reference=upstream_reference, policy_id=policy.policy_id, policy_version=policy.version, convention=convention, first_route_point_index=first.route_point_index, last_route_point_index=last.route_point_index, ), convention=convention, first_in_window=first, last_in_window=last, start_boundary=_start_boundary(records, start), end_boundary=_end_boundary(records, end), ) ) index += 1 return tuple(windows) def _start_boundary(records, start): current = records[start][0] if start == 0: return RouteBoundary(BoundarySemantics.ANALYSIS_WINDOW_START, (current,)) previous, membership = records[start - 1] semantics = ( BoundarySemantics.TRANSITION_BRACKET if membership is ConventionMembership.OUTSIDE_WINDOW else BoundarySemantics.DATA_GAP_BRACKET ) return RouteBoundary(semantics, (previous, current)) def _end_boundary(records, end): current = records[end][0] if end == len(records) - 1: return RouteBoundary(BoundarySemantics.ANALYSIS_WINDOW_END, (current,)) following, membership = records[end + 1] semantics = ( BoundarySemantics.TRANSITION_BRACKET if membership is ConventionMembership.OUTSIDE_WINDOW else BoundarySemantics.DATA_GAP_BRACKET ) return RouteBoundary(semantics, (current, following)) def _sample_reference(point: Any) -> RouteSampleReference: return RouteSampleReference( route_point_index=_value(point, "point_index"), route_distance_m=_value(point, "route_distance_m"), planned_time=_value(point, "planned_arrival_time"), elapsed_planned_seconds=_value(point, "elapsed_route_seconds"), ) def _pace_reference(points: Sequence[Any]) -> str | None: for point in points: reference = _value(_value(point, "pace", {}), "model_reference") if reference is not None: return str(reference) return None def _analysis_reference(analysis: Any) -> str: identity = _value(analysis, "identity", {}) route = _value(analysis, "route", {}) scenario = _value(analysis, "scenario", {}) return ":".join( str(value) for value in ( _value(identity, "analysis_type", "unified_route_analysis"), _value(identity, "version", "unknown"), _value(identity, "subject_id", _value(route, "route_id", "unknown")), _value(scenario, "name", "unknown"), ) ) def _mapping(value: Any) -> Mapping[str, Any]: if isinstance(value, Mapping): return value if hasattr(value, "__dataclass_fields__"): return asdict(value) return {} def _value(item: Any, key: str, default: Any = None) -> Any: return item.get(key, default) if isinstance(item, Mapping) else getattr(item, key, default)