"""Mission Control cockpit telemetry v0.1 (docs/cockpit_v0_1_design.md). Real Open-Meteo hourly values for a saved (non-reference) Journey's route, through the same LiveWeatherPointResolver (1 h TTL cache) the camps' night conditions use. Two windows, design section 3: * PLANNED_DAY_WINDOW -- the Journey has a plan: the selected day(s) of derive_day_point_times() (the planned passage time, next-morning re-anchoring included, exactly as the camps' days). The route range is sampled evenly by distance; every sample is a route point at its own planned time, read from the provider hour nearest to it. The x axis is route distance, shared with the elevation profile above the bands. * NEXT_72_HOURS -- no plan (no dates): the next 72 provider hours from now along the whole route, sampled by distance; the UI shows one chosen hour at a time. The x axis is route distance here too. Only provider values are returned -- no interpolation, no derived index, no verdict. A value that does not exist is null with a reason next to it. Planned time and provider valid time stay separate fields (Mission Control v0.2 section 7). """ from __future__ import annotations import bisect import math from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timedelta, timezone from typing import Any, Sequence from zoneinfo import ZoneInfo from mountain_twin.exposure import RoutePoint from mountain_twin.journey.activities import pace_factor_for_activity from mountain_twin.journey.camps_service import FREEZING_LEVEL_REGION_EXPLANATION from mountain_twin.journey.contracts import JourneyPlan from mountain_twin.journey.day_derivation import derive_day_point_times from mountain_twin.pace.pauses import generate_relative_pauses from mountain_twin.route_analysis import prepare_route from mountain_twin.weather.live import ( FREEZING_LEVEL_NOT_PROVIDED_FOR_REGION, FREEZING_LEVEL_VARIABLE, LIVE_WEATHER_TTL_SECONDS, LiveWeatherPointResolver, ProviderRateLimited, ) from mountain_twin.weather.temporal import TemporalWeatherSample TELEMETRY_VERSION = "cockpit_telemetry_v0_1" # Bands in display order (design section 2), each with its real Open-Meteo # fields. The Mission Control UI renders exactly these, in this order. COCKPIT_BANDS: tuple[dict[str, Any], ...] = ( {"band_id": "temperature", "variables": ("temperature_2m", "apparent_temperature")}, {"band_id": "wind", "variables": ("wind_speed_10m", "wind_gusts_10m")}, # The chance of any precipitation -- there is no storm band (Open-Meteo # has no usable storm-probability field; design section 2). {"band_id": "precipitation", "variables": ("precipitation_probability",)}, {"band_id": "humidity", "variables": ("relative_humidity_2m",)}, {"band_id": "freezing_level", "variables": ("freezing_level_height",)}, {"band_id": "soil_temperature", "variables": ("soil_temperature_0cm",)}, ) # Open-Meteo's forecast API serves today plus 15 days; a planned time later # than that has no forecast yet (not an error, not a zero). # Storms tab (docs/profile_storms_v0_1_design.md section 4): the model's # convective variables and its weather code, as facts -- never a "risk". STORM_BANDS: tuple[dict[str, Any], ...] = ( {"band_id": "cape", "variables": ("cape",)}, {"band_id": "lifted_index", "variables": ("lifted_index",)}, {"band_id": "convective_inhibition", "variables": ("convective_inhibition",)}, {"band_id": "thunderstorm_code", "variables": ("weather_code",)}, ) # WMO weather codes 95-99: thunderstorm (Open-Meteo uses 95, 96, 99). THUNDERSTORM_CODES = range(95, 100) # Descriptive reference lines on the CAPE chart (J/kg), not thresholds of # anything: the design's "progi opisowe". CAPE_REFERENCE_LINES_J_PER_KG = (300, 1000) STORM_NOTE = ( "CAPE i LI opisują niestabilność atmosfery, nie prawdopodobieństwo burzy; " "CIN może zablokować konwekcję." ) FORECAST_HORIZON_DAYS = 16 NEXT_HOURS = 72 # Route points sampled for the no-plan (next 72 h) view: one request each. NEXT_HOURS_ROUTE_SAMPLES = 16 # Samples along a day: one per planned hour of walking plus the end point -- # the provider series is hourly, so denser points would add requests (one # per point, each cached for the TTL) but no new hour of data. Clamped, and # thinned further for long multi-day ranges. MIN_SAMPLES_PER_DAY = 2 MAX_SAMPLES_PER_DAY = 12 MAX_PLANNED_SAMPLES = 36 # The provider hour nearest to a planned time is used only if it is within # half an hour of it (hourly data) -- never a value from another time. MAX_VALID_TIME_OFFSET = timedelta(minutes=30) # Parallel requests per telemetry view: 4 keeps a first load's burst inside # Open-Meteo's per-minute limit (8 drew 429s on the Lofoten route). FETCH_WORKERS = 4 def cockpit_variables() -> tuple[str, ...]: return band_variables(COCKPIT_BANDS) def band_variables(bands) -> tuple[str, ...]: return tuple(variable for band in bands for variable in band["variables"]) def planned_telemetry_document( *, journey_id: str, route_points: Sequence[RoutePoint], plan: JourneyPlan, pace_profile_id: str, live_resolver: LiveWeatherPointResolver, first_day: int | None = None, last_day: int | None = None, now: datetime | None = None, bands: Sequence[dict[str, Any]] = COCKPIT_BANDS, activity_id: str | None = None, ) -> dict[str, Any]: """Telemetry along the planned days [first_day, last_day] (0-based, inclusive; default: the whole Journey), for ``bands`` (the Weather tab's by default, the Storms tab's with STORM_BANDS).""" variables = band_variables(bands) prepared = prepare_route(route_points).points days = derive_day_point_times( route_points, pace_factor=pace_factor_for_activity(activity_id, pace_profile_id), pauses=generate_relative_pauses(prepared[-1].cumulative_distance_m), start_datetime=datetime.fromisoformat(plan.planned_start_local), camp_markers=plan.camp_markers, timezone_name=plan.journey_timezone, rest_days_before_start=plan.rest_days_before_start, ) first = 0 if first_day is None else first_day last = len(days) - 1 if last_day is None else last_day if not (0 <= first <= last < len(days)): raise ValueError("telemetry day range is outside the plan's days") selected = days[first : last + 1] zone = ZoneInfo(plan.journey_timezone) today = (now or datetime.now(timezone.utc)).astimezone(zone).date() # Keyed by route point index: prepare_route may drop points, so list # positions are not indexes. distance_by_index = {point.point_index: point.cumulative_distance_m for point in prepared} day_cap = max(MIN_SAMPLES_PER_DAY, MAX_PLANNED_SAMPLES // len(selected)) requests = [] for segment, times in selected: hours = (segment.arrival_time - segment.departure_time).total_seconds() / 3600 count = max(MIN_SAMPLES_PER_DAY, min(MAX_SAMPLES_PER_DAY, day_cap, math.ceil(hours) + 1)) for index in _even_indices(sorted(times), distance_by_index, count): requests.append((segment.day_number, index, times[index])) def sample_for(request): day_number, index, planned_time = request point = route_points[index] base = { "route_point_index": index, "route_distance_m": distance_by_index[index], "elevation_m": point.elevation_m, "day_number": day_number, "planned_time": planned_time.isoformat(), } planned_date = planned_time.astimezone(zone).date() if planned_date > today + timedelta(days=FORECAST_HORIZON_DAYS - 1): return {**base, **_unavailable("BEYOND_FORECAST_HORIZON", variables)}, None try: series = live_resolver.resolve( latitude=point.latitude, longitude=point.longitude, elevation_m=point.elevation_m, start_date=planned_date, end_date=planned_date, timezone_name=plan.journey_timezone, ) except ProviderRateLimited as limited: return {**base, **_unavailable(limited.reason_code, variables)}, None except RuntimeError: # One point's failed request is that point's missing data, not # a failed instrument: the other samples still show. return {**base, **_unavailable("PROVIDER_NETWORK_FAILURE", variables)}, None return {**base, **_values_at(series, planned_time, variables)}, series with ThreadPoolExecutor(max_workers=FETCH_WORKERS) as pool: results = list(pool.map(sample_for, requests)) samples = [sample for sample, _ in results] range_start, range_end = selected[0][0].start_point_index, selected[-1][0].end_point_index elevations = [ route_points[index].elevation_m for index in distance_by_index if range_start <= index <= range_end and route_points[index].elevation_m is not None ] highest_point_m = max(elevations) if elevations else None return { **_envelope( journey_id, "PLANNED_DAY_WINDOW", "route_distance", plan.journey_timezone, bands ), "first_day": first, "last_day": last, "window_start": selected[0][0].departure_time.isoformat(), "window_end": selected[-1][0].arrival_time.isoformat(), "range_start_distance_m": distance_by_index[range_start], "range_end_distance_m": distance_by_index[range_end], "highest_point_m": highest_point_m, "freezing_level": _freezing_level_summary(samples, highest_point_m), "units": _units((series for _, series in results), variables), "daily": _daily_summaries(selected, samples, variables), "samples": samples, } def next_hours_telemetry_document( *, journey_id: str, route_points: Sequence[RoutePoint], timezone_name: str, live_resolver: LiveWeatherPointResolver, now: datetime | None = None, bands: Sequence[dict[str, Any]] = COCKPIT_BANDS, point_range: tuple[int, int] | None = None, day_range: tuple[int, int] | None = None, ) -> dict[str, Any]: """No plan (no dates): the next NEXT_HOURS provider hours along the whole route -- or, for a plan without a start (step T3), along ``point_range`` (first and last route point index of its days ``day_range``, 0-based). The route is sampled evenly by distance -- the same x axis as the elevation profile, so the cockpit's cursor stands at the selected point's own distance -- and every sample carries its full hourly series (``values_by_hour``, aligned with ``hours``); the UI shows one chosen hour at a time along the route. Only provider values, no interpolation.""" variables = band_variables(bands) zone = ZoneInfo(timezone_name) start = ( (now or datetime.now(timezone.utc)) .astimezone(zone) .replace(minute=0, second=0, microsecond=0) ) end = start + timedelta(hours=NEXT_HOURS) distance_by_index = { prepared.point_index: prepared.cumulative_distance_m for prepared in prepare_route(route_points).points } in_range = [ index for index in sorted(distance_by_index) if point_range is None or point_range[0] <= index <= point_range[1] ] indexes = _even_indices(in_range, distance_by_index, NEXT_HOURS_ROUTE_SAMPLES) def sample_for(index): point = route_points[index] base = { "route_point_index": index, "route_distance_m": distance_by_index[index], "elevation_m": point.elevation_m, "day_number": None, "planned_time": None, } try: series = live_resolver.resolve( latitude=point.latitude, longitude=point.longitude, elevation_m=point.elevation_m, start_date=start.date(), end_date=end.date(), timezone_name=timezone_name, ) except ProviderRateLimited as limited: return {**base, **_unavailable_hours(limited.reason_code, variables)}, None except RuntimeError: return {**base, **_unavailable_hours("PROVIDER_NETWORK_FAILURE", variables)}, None if series is None: return {**base, **_unavailable_hours("PROVIDER_NO_DATA", variables)}, None positions = [i for i, valid_time in enumerate(series.times) if start <= valid_time < end] return { **base, "state": "AVAILABLE" if positions else "UNAVAILABLE", "unavailable_reason": None if positions else "PROVIDER_NO_DATA", "hours": [series.times[i].isoformat() for i in positions], "values_by_hour": { variable: [_series_value(series, variable, i) for i in positions] for variable in variables }, "missing_reasons": _missing_reasons(series, positions[0], variables) if positions else {}, "variable_models": dict(series.variable_models), }, series with ThreadPoolExecutor(max_workers=FETCH_WORKERS) as pool: results = list(pool.map(sample_for, indexes)) samples = [sample for sample, _ in results] # One hour axis for all: every point is asked for the same dates in the # same timezone, so the provider hours coincide. hours = next((sample["hours"] for sample in samples if sample.get("hours")), []) for sample in samples: sample.pop("hours", None) elevations = [ route_points[i].elevation_m for i in in_range if route_points[i].elevation_m is not None ] highest_point_m = max(elevations) if elevations else None # The freezing summary covers every point and every hour of the window. flattened = [ { "values": {FREEZING_LEVEL_VARIABLE: value}, "variable_models": sample["variable_models"], "missing_reasons": sample["missing_reasons"], } for sample in samples for value in (sample["values_by_hour"].get(FREEZING_LEVEL_VARIABLE) or [None]) ] first, last = in_range[0], in_range[-1] return { **_envelope(journey_id, "NEXT_72_HOURS", "route_distance", timezone_name, bands), **({} if day_range is None else {"first_day": day_range[0], "last_day": day_range[1]}), "window_start": start.isoformat(), "window_end": end.isoformat(), "hours": hours, "range_start_distance_m": distance_by_index[first], "range_end_distance_m": distance_by_index[last], "highest_point_m": highest_point_m, "freezing_level": _freezing_level_summary(flattened, highest_point_m), "units": _units((series for _, series in results), variables), "samples": samples, } def _unavailable_hours(reason: str, variables: Sequence[str]) -> dict[str, Any]: return { "state": "UNAVAILABLE", "unavailable_reason": reason, "values_by_hour": {variable: [] for variable in variables}, "missing_reasons": {}, "variable_models": {}, } def _envelope( journey_id: str, mode: str, axis: str, timezone_name: str, bands=COCKPIT_BANDS ) -> dict[str, Any]: return { "telemetry_version": TELEMETRY_VERSION, "journey_id": journey_id, "mode": mode, "axis": axis, "timezone": timezone_name, "bands": [ {"band_id": band["band_id"], "variables": list(band["variables"])} for band in bands ], "source": { "provider": "Open-Meteo", "model_selection": "best_match", "cache_ttl_seconds": LIVE_WEATHER_TTL_SECONDS, }, } def _even_indices(indexes: list[int], distance_by_index: dict[int, float], count: int) -> list[int]: """Up to ``count`` of the day's (sorted) route point indexes, evenly spaced by distance, always including both ends, without repeats.""" if len(indexes) <= 1: return list(indexes) distances = [distance_by_index[index] for index in indexes] chosen: list[int] = [] for k in range(count): target = distances[0] + (distances[-1] - distances[0]) * k / (count - 1) position = bisect.bisect_left(distances, target) candidates = [i for i in (position - 1, position) if 0 <= i < len(indexes)] best = indexes[min(candidates, key=lambda i: abs(distances[i] - target))] if not chosen or best != chosen[-1]: chosen.append(best) return chosen def _daily_summaries( selected, samples: list[dict[str, Any]], variables: Sequence[str] ) -> list[dict[str, Any]]: """docs/profile_storms_v0_1_design.md section 1: for the whole Journey or a group, one summary per planned day -- each variable's min and max over that day's samples (provider values at the planned hours). A day with no provider value is an explicit UNAVAILABLE with its reason (e.g. beyond the forecast horizon), never numbers from another date.""" days = [] for segment, _ in selected: own = [sample for sample in samples if sample["day_number"] == segment.day_number] stats = {} for variable in variables: values = [ sample["values"][variable] for sample in own if sample["values"].get(variable) is not None ] stats[variable] = {"min": min(values), "max": max(values)} if values else None available = [sample for sample in own if sample["state"] != "UNAVAILABLE"] reasons = sorted( {sample["unavailable_reason"] for sample in own if sample.get("unavailable_reason")} ) days.append( { "day_number": segment.day_number, "departure_time": segment.departure_time.isoformat(), "arrival_time": segment.arrival_time.isoformat(), "sample_count": len(own), "state": "AVAILABLE" if len(available) == len(own) and own else "PARTIAL" if available else "UNAVAILABLE", "unavailable_reasons": reasons, "stats": stats, } ) return days def _freezing_level_summary( samples: list[dict[str, Any]], highest_point_m: float | None ) -> dict[str, Any]: """Design section 4: the freezing level as a line *and* as text -- its lowest value in the window and how far that is above (negative: below) the highest route point in the same range. Plain differences of real values; no verdict, no colour scale (Mission Control v0.2 section 6.7). Where no model provides it, an explicit UNAVAILABLE with the reason.""" values = [ sample["values"][FREEZING_LEVEL_VARIABLE] for sample in samples if sample["values"].get(FREEZING_LEVEL_VARIABLE) is not None ] models = sorted( { sample["variable_models"].get(FREEZING_LEVEL_VARIABLE, "best_match") for sample in samples if sample["values"].get(FREEZING_LEVEL_VARIABLE) is not None } ) if values: lowest = min(values) return { "state": "AVAILABLE", "models": models, "lowest_m": lowest, "highest_point_m": highest_point_m, "margin_above_highest_point_m": None if highest_point_m is None else lowest - highest_point_m, "explanation": None, } region_gap = any( sample["missing_reasons"].get(FREEZING_LEVEL_VARIABLE) == FREEZING_LEVEL_NOT_PROVIDED_FOR_REGION for sample in samples ) return { "state": "UNAVAILABLE", "models": [], "lowest_m": None, "highest_point_m": highest_point_m, "margin_above_highest_point_m": None, "unavailable_reason": FREEZING_LEVEL_NOT_PROVIDED_FOR_REGION if region_gap else None, "explanation": FREEZING_LEVEL_REGION_EXPLANATION if region_gap else None, } def _units(all_series, variables: Sequence[str]) -> dict[str, str | None]: """The provider's own unit per variable (e.g. wind in km/h), from the first series that reports one -- the UI never assumes a unit.""" units: dict[str, str | None] = {variable: None for variable in variables} for series in all_series: if series is None: continue for variable in units: unit = series.base.units.get(variable) # Open-Meteo names the unit of a variable it does not provide # "undefined" -- that is not a unit. if units[variable] is None and unit and unit != "undefined": units[variable] = unit return units def _unavailable(reason: str, variables: Sequence[str]) -> dict[str, Any]: return { "valid_time": None, "state": "UNAVAILABLE", "unavailable_reason": reason, "values": {variable: None for variable in variables}, "missing_reasons": {}, "variable_models": {}, } def _values_at( series: TemporalWeatherSample | None, planned_time: datetime, variables: Sequence[str] ) -> dict[str, Any]: if series is None or not series.times: return _unavailable("PROVIDER_NO_DATA", variables) target = planned_time.astimezone(timezone.utc) index = min( range(len(series.times)), key=lambda i: abs(series.times[i].astimezone(timezone.utc) - target), ) valid_time = series.times[index] if abs(valid_time.astimezone(timezone.utc) - target) > MAX_VALID_TIME_OFFSET: return _unavailable("PLANNED_TIME_OUTSIDE_PROVIDER_HOURS", variables) values = {variable: _series_value(series, variable, index) for variable in variables} return { "valid_time": valid_time.isoformat(), "state": "AVAILABLE" if all(value is not None for value in values.values()) else "PARTIAL", "unavailable_reason": None, "values": values, "missing_reasons": _missing_reasons(series, index, variables), "variable_models": dict(series.variable_models), } def _series_value(series: TemporalWeatherSample, variable: str, index: int): values = series.variables_by_time.get(variable) return None if values is None or index >= len(values) else values[index] def _missing_reasons( series: TemporalWeatherSample, index: int, variables: Sequence[str] ) -> dict[str, str]: reasons = {} for variable in variables: if _series_value(series, variable, index) is None: reasons[variable] = series.base.missing_variable_reasons.get( variable, "WEATHER_PROVIDER_VALUE_MISSING" ) return reasons def storms_document(*, planned: bool, **kwargs) -> dict[str, Any]: """docs/profile_storms_v0_1_design.md section 4: the Storms tab's data -- the same windows and samples as the Weather tab (planned days, or the next 72 hours without a plan) for CAPE, LI, CIN and the weather code. Per day: the day's highest CAPE, lowest LI (the ``daily`` min/max) and the planned hours whose forecast code is a thunderstorm (WMO 95-99). A variable no model provides here (e.g. LI in Scandinavia) is an explicit UNAVAILABLE -- never 0, never a value from another model.""" if planned: document = planned_telemetry_document(bands=STORM_BANDS, **kwargs) else: document = next_hours_telemetry_document(bands=STORM_BANDS, **kwargs) document.pop("freezing_level", None) document["tab"] = "STORMS" document["reference_lines"] = {"cape": list(CAPE_REFERENCE_LINES_J_PER_KG)} document["thunderstorm_codes"] = [THUNDERSTORM_CODES.start, THUNDERSTORM_CODES.stop - 1] document["note"] = STORM_NOTE samples = document["samples"] document["variable_availability"] = { variable: _variable_availability(samples, variable, planned) for variable in band_variables(STORM_BANDS) } for day in document.get("daily", []): own = [ sample for sample in samples if sample["day_number"] == day["day_number"] and sample["values"].get("weather_code") is not None ] stormy = [ sample for sample in own if int(sample["values"]["weather_code"]) in THUNDERSTORM_CODES ] day["thunderstorm"] = { "coded_sample_count": len(own), "hours": [sample["valid_time"] for sample in stormy], } return document def _variable_availability(samples, variable: str, planned: bool) -> dict[str, Any]: if planned: values = [sample["values"].get(variable) for sample in samples] else: values = [ value for sample in samples for value in sample["values_by_hour"].get(variable, []) ] present = sum(value is not None for value in values) reasons = sorted( { sample["missing_reasons"][variable] for sample in samples if variable in sample.get("missing_reasons", {}) } | {sample["unavailable_reason"] for sample in samples if sample.get("unavailable_reason")} ) return { "state": "AVAILABLE" if values and present == len(values) else "PARTIAL" if present else "UNAVAILABLE", "reasons": reasons, }