"""Provider-labelled weather result construction and compact route summaries.""" from __future__ import annotations from dataclasses import asdict from typing import Any, Sequence from mountain_twin.analysis_contract import ( AnalysisIdentity, AnalysisResult, EvaluationState, MethodProvenance, ProviderProvenance, ResultLocation, ResultQuality, WeatherPayload, ) from mountain_twin.weather.provider import WeatherSample WEATHER_LIMITATIONS = ( "WEATHER_MODEL_GRID_NOT_ROUTE_POINT_PRECISION", "ROUTE_ELEVATION_AND_PROVIDER_GRID_ELEVATION_NOT_LAPSE_RATE_CORRECTED", "MODEL_OUTPUT_NOT_STATION_OBSERVATION_OR_OFFICIAL_MOUNTAIN_WARNING", "OPEN_METEO_R_AND_D_PROVIDER_NOT_PRODUCTION_PROVIDER_DECISION", ) def weather_analysis_result( sample: WeatherSample, *, scenario_datetime: str, timezone: str ) -> AnalysisResult: missing = sample.missing_variable_reasons return AnalysisResult( identity=AnalysisIdentity( analysis_type="route_weather_sampling", semantic_type="weather_context", version="0.1", subject_id=sample.location.sample_id, scenario_datetime=scenario_datetime, timezone=timezone, ), location=ResultLocation( latitude=sample.location.latitude, longitude=sample.location.longitude, point_index=sample.location.point_index, route_distance_m=sample.location.route_distance_m, route_elevation_m=sample.location.route_elevation_m, ), status="PARTIAL" if missing else "COMPLETE", quality=ResultQuality( outcome=EvaluationState.UNRESOLVED if missing else EvaluationState.SUCCESS, reason_codes=tuple(sorted(set(missing.values()))), data_support_state="partial" if missing else "complete", numerical_convergence_state="not_applicable", range_convergence_state="not_applicable", limitations=WEATHER_LIMITATIONS, ), source=ProviderProvenance( provider=sample.provider, product=sample.api_family, product_variant=sample.endpoint_type, surface_semantics=None, native_resolution=None, horizontal_crs="EPSG:4326", vertical_reference=None, acquisition_context=f"model_selection={sample.model_selection}", validity_context=sample.source_type.value, limitations=WEATHER_LIMITATIONS, ), method=MethodProvenance( method="route_weather_sample_v0_1", model_version=sample.model_identity, assumptions=( "Explicit scenario time; GPX timestamps are not hiking chronology.", "No automatic lapse-rate correction is applied.", ), ), payload=WeatherPayload( source_type=sample.source_type, provider_valid_time=sample.provider_valid_time, provider_run_time=sample.provider_run_time, data_freshness=None, provider_grid_latitude=sample.provider_grid_latitude, provider_grid_longitude=sample.provider_grid_longitude, provider_grid_elevation_m=sample.provider_grid_elevation_m, route_minus_provider_elevation_m=sample.route_minus_provider_elevation_m, variables=sample.variables, units=sample.units, missing_variable_reasons=sample.missing_variable_reasons, variable_semantics=sample.variable_semantics, ), ) def weather_document( samples: Sequence[WeatherSample], *, scenario_datetime: str, timezone: str, fetched_at: str | None, ) -> dict[str, Any]: """Build deterministic summaries, except for explicit fetch timestamp provenance.""" contracts = [ weather_analysis_result( sample, scenario_datetime=scenario_datetime, timezone=timezone ).to_dict() for sample in samples ] return { "contract_version": "weather_engine_v0_1", "analysis_type": "route_weather_sampling", "time_semantics": "explicit_scenario_timestamp_not_gpx_or_hiking_time", "scenario_datetime": scenario_datetime, "timezone": timezone, "fetched_at": fetched_at, "sample_count": len(samples), "source_type": samples[0].source_type.value if samples else None, "provider": samples[0].provider if samples else None, "api_family": samples[0].api_family if samples else None, "endpoint_type": samples[0].endpoint_type if samples else None, "model_selection": samples[0].model_selection if samples else None, "requested_variables": list(samples[0].variables) if samples else [], "samples": [ { "selection": asdict(sample.location), "contract": contract, } for sample, contract in zip(samples, contracts) ], "route_wide_ranges": _ranges(samples), "limitations": list(WEATHER_LIMITATIONS), } def weather_markdown(document: dict[str, Any]) -> str: """Render a compact, deterministic human-readable companion to JSON.""" lines = [ "# TMB Day 1 Weather Engine v0.1 — " + document["source_type"].replace("_", " "), "", "This is provider-labelled model output for an explicit scenario timestamp; it is not " "observed station truth, a hiking chronology, or a mountain warning.", "", "- Scenario: `" + str(document["scenario_datetime"]) + "` (`" + document["timezone"] + "`)", "- Provider/API: " + str(document["provider"]) + " / " + str(document["api_family"]), "- Source type: `" + document["source_type"] + "`; model selection: `" + document["model_selection"] + "`", "- Samples: " + str(document["sample_count"]), "- Fetch timestamp: `" + str(document["fetched_at"]) + "`", "", "## Route-scale ranges", "", "| Variable | Min | Max | Unit |", "| --- | ---: | ---: | --- |", ] for variable, values in document["route_wide_ranges"].items(): lines.append(f"| {variable} | {values['min']} | {values['max']} | {values['unit']} |") lines.extend( [ "", "## Samples", "", "Full provider-labelled variables, units, returned grid context and missing-value reasons " "are in the adjacent JSON artifact. Route elevation is GPX-derived; provider elevation " "is model-grid context. Their difference is reported without lapse-rate correction.", "", "| Sample | Chainage km | Route elevation m | Grid elevation m | Temperature | Wind / gust | Cloud |", "| --- | ---: | ---: | ---: | ---: | ---: | ---: |", ] ) for sample in document["samples"]: selection = sample["selection"] payload = sample["contract"]["payload"] variables = payload["variables"] lines.append( "| {sample_id} | {distance:.3f} | {route_elevation} | {grid_elevation} | " "{temperature} | {wind} / {gust} | {cloud} |".format( sample_id=selection["sample_id"], distance=selection["route_distance_m"] / 1000, route_elevation=selection["route_elevation_m"], grid_elevation=payload["provider_grid_elevation_m"], temperature=variables["temperature_2m"], wind=variables["wind_speed_10m"], gust=variables["wind_gusts_10m"], cloud=variables["cloud_cover"], ) ) lines.extend( [ "", "## Limitations", "", *[f"- `{item}`" for item in document["limitations"]], "", ] ) return "\n".join(lines) def _ranges(samples: Sequence[WeatherSample]) -> dict[str, dict[str, float | int | None]]: result = {} for variable in ( "temperature_2m", "wind_speed_10m", "wind_gusts_10m", "precipitation", "cloud_cover", "freezing_level_height", ): values = [ sample.variables[variable] for sample in samples if sample.variables[variable] is not None ] result[variable] = { "min": min(values) if values else None, "max": max(values) if values else None, "unit": samples[0].units.get(variable) if samples else None, } return result