"""Weather tab v0.1: the route's weather series along the elevation profile (docs/pogoda_v0_1_design.md; sampling decided in docs/design_reference/pogoda_spike_v0_1.md). One document serves both slider modes, so switching mode never refetches and the browser never computes anything: * ``hours`` -- the slider's axis: the next 72 provider hours, or the planned dates (clamped to the forecast horizon) when the plan can be timed -- AV-043: cut to its first 24 h / 72 h when that horizon is chosen (``plan_horizon``); ``hour_states`` is the astronomical state of each hour at the middle of the range (deterministic calculation, mountain_twin.solar) for the night shading. * ``samples`` -- route points every SAMPLE_SPACING_M (at most MAX_SAMPLES per view), each with its provider series on that axis (``values_by_hour``) and, when the plan is timed, ``planned_hour_index``: the axis hour nearest its planned passage (within 30 min), or null with the reason. * ``plan_mode`` -- whether "Według planu" is possible: UNAVAILABLE with NO_PLAN, or with PLAN_TIMING_UNAVAILABLE and the same fields the rest of Mission Control uses (docs/design_reference/elevation_gaps_spike_v0_1.md). A route point without elevation is not queried (without ``elevation=`` the provider would answer for its grid cell's height, not the route's): it is an explicit ROUTE_ELEVATION_UNAVAILABLE gap. Only provider values -- no interpolation, no derived index, no verdict. """ from __future__ import annotations import math 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 activity_for, pace_factor_for_activity from mountain_twin.journey.camps_service import day_boundaries_document from mountain_twin.journey.contracts import JourneyPlan from mountain_twin.journey.day_derivation import ( NO_PLANNED_START, DayDerivationUnavailable, derive_day_point_times, derive_relative_day_segments, derive_untimed_day_segments, untimed_day_dates, ) from mountain_twin.journey.telemetry import FORECAST_HORIZON_DAYS, MAX_VALID_TIME_OFFSET, NEXT_HOURS from mountain_twin.journey.weather_anchor import anchor_document, anchored_plan, next_start from mountain_twin.pace.pauses import generate_relative_pauses from mountain_twin.route_analysis import prepare_route_cached from mountain_twin.solar.states import classify_astronomical_state from mountain_twin.solar.sun_engine import solar_position from mountain_twin.weather.forecast_cache import call_source from mountain_twin.weather.live import LIVE_WEATHER_TTL_SECONDS, LiveWeatherPointResolver from mountain_twin.weather.sources import budget_document, sources_document WEATHER_SERIES_CONTRACT = "route_weather_series_v0_1" # The Weather tab's fields. Open-Meteo weighs a request by its variables in # tens (fractionally): AV-024's twelve fields weighed 1.2 calls per location. # AV-051: the basic request is ten -- exactly one call per location, what # the prefetch and every opening ask for -- and the ground temperature and # the modelled snow depth come only when asked for ("Więcej wykresów"), in # a request of their own (a call per location: worth it only on request). WEATHER_SERIES_VARIABLES: tuple[str, ...] = ( "temperature_2m", "apparent_temperature", "wind_speed_10m", "wind_gusts_10m", "wind_direction_10m", "precipitation", "precipitation_probability", "relative_humidity_2m", "freezing_level_height", "snowfall", ) WEATHER_SERIES_EXTRA_VARIABLES: tuple[str, ...] = ("soil_temperature_0cm", "snow_depth") WEATHER_SERIES_ALL_VARIABLES = WEATHER_SERIES_VARIABLES + WEATHER_SERIES_EXTRA_VARIABLES # Spike section a: wind, humidity and freezing level change every ~1 km along # a route (a ~2 km model grid); temperature follows the elevation at every # point. 250 m keeps four samples per real change at ~1 request per 100 points. SAMPLE_SPACING_M = 250.0 MAX_SAMPLES = 240 # Planning Workspace's pace default (aventurro/app.js, "normal_hiker"). DEFAULT_WALK_PACE_PROFILE_ID = "normal_hiker" # Step M2 (docs/design_reference/weather_start_models_spike_v0_1.md section # 2.4): named models compared on request, never by default. The comparison # models' grids are coarser than best_match's (ECMWF IFS and GFS ~25 km, # ICON 2-7 km in Europe), so 1 km keeps every real change at a quarter of # the samples; 60 samples x 3 models <= 180 provider calls per opening # (each model weighs one call per location, spike section 2.2). COMPARISON_MODELS: tuple[str, ...] = ("icon_seamless", "ecmwf_ifs025", "gfs_seamless") COMPARISON_SPACING_M = 1000.0 # Open-Meteo weighs a request by its variables in tens (fractionally, at # least one): the basic ten = 1 call per location and model. PROVIDER_CALLS_PER_LOCATION = max(1.0, len(WEATHER_SERIES_VARIABLES) / 10) COMPARISON_MAX_SAMPLES = 60 def weather_series_document( *, journey_id: str, route_points: Sequence[RoutePoint], plan: JourneyPlan | None, timezone_name: str, live_resolver: LiveWeatherPointResolver, first_day: int | None = None, last_day: int | None = None, now: datetime | None = None, walk_start_local: str | None = None, walk_pace_profile_id: str | None = None, compare_models: bool = False, activity_id: str | None = None, horizon: str | None = None, prefer_saved_dates: bool = False, extras: bool = False, ) -> dict[str, Any]: """``horizon``: "24", "72" or "TRIP"; None is the axis's own default -- 72 h on the fixed-hour axis, the whole plan on the plan's dates. ``prefer_saved_dates``: AV-048's "Pokaż według zapisanych dat". ``extras``: AV-051's further fields (ground, snow depth), asked for.""" if horizon is not None and horizon not in HORIZONS: raise ValueError("horizon must be 24, 72 or TRIP") # AV-048: the weather follows the plan's start time and pace -- simulated # from the next start unless the whole trip is ahead within the # forecast, or the saved dates are asked for (weather_anchor). plan, anchor = anchored_plan( plan, route_points, activity_id, now, prefer_saved=prefer_saved_dates ) scope = _scope(route_points, plan, timezone_name, first_day, last_day, now, activity_id) distance_by_index = scope["distance_by_index"] zone_name, zone, now_local = scope["zone_name"], scope["zone"], scope["now_local"] plan_mode, selected, first, last = ( scope["plan_mode"], scope["selected"], scope["first"], scope["last"], ) start_m, end_m, sample_indexes = scope["start_m"], scope["end_m"], scope["sample_indexes"] planned_times: dict[int, datetime] = {} for _, times in selected or (): planned_times.update(times) # The slider's axis: the planned dates (a timed plan), else the next 72 h. horizon_end = now_local.date() + timedelta(days=FORECAST_HORIZON_DAYS - 1) first_hour = now_local.replace(minute=0, second=0, microsecond=0) plan_window = None if selected: plan_first = selected[0][0].departure_time.astimezone(zone).date() plan_last = selected[-1][0].arrival_time.astimezone(zone).date() start_date, end_date = max(plan_first, now_local.date()), min(plan_last, horizon_end) axis_kind = "PLANNED_DATES" # AV-043: 24 h / 72 h on the plan's dates -- the first hours from the # departure (from now, when the trip is under way). if horizon in ("24", "72"): departure = selected[0][0].departure_time.astimezone(zone) window_start = max(departure.replace(minute=0, second=0, microsecond=0), first_hour) plan_window = (window_start, window_start + timedelta(hours=int(horizon))) end_date = min(end_date, plan_window[1].date()) else: # AV-036: the fixed-hour axis spans 24 h, 72 h or the whole trip (to # the end of its last day, cut at the forecast's reach). span = _axis_span(horizon or "72", scope, plan, first_hour, now_local) start_date, end_date = ( first_hour.date(), (first_hour + timedelta(hours=span["hours"])).date(), ) axis_kind = "NEXT_72_HOURS" axis_available = start_date <= end_date axis_reason = None if not axis_available: axis_reason = ( "PLANNED_DATES_PASSED" if end_date < now_local.date() else "BEYOND_FORECAST_HORIZON" ) queried = [index for index in sample_indexes if route_points[index].elevation_m is not None] # AV-048: when the answers used were fetched, and whether stale ones are # being refreshed in the background (the shared cell cache fills it). freshness: dict[str, Any] = {} locations = [ ( route_points[index].latitude, route_points[index].longitude, route_points[index].elevation_m, ) for index in queried ] def ask(variables): return dict( zip( queried, live_resolver.resolve_many( locations, start_date=start_date, end_date=end_date, timezone_name=zone_name, variables=variables, **({"freshness": freshness} if _takes_freshness(live_resolver) else {}), ), ) ) resolved = ask(WEATHER_SERIES_VARIABLES) if axis_available and queried else {} extra_resolved = ( ask(WEATHER_SERIES_EXTRA_VARIABLES) if extras and axis_available and queried else {} ) variables = WEATHER_SERIES_ALL_VARIABLES if extras else WEATHER_SERIES_VARIABLES hours = _axis_hours(resolved, first_hour, axis_kind, span["hours"] if not selected else None) if plan_window: hours = [hour for hour in hours if plan_window[0] <= hour < plan_window[1]] if not hours and resolved and not axis_reason: # No hour axis because no point got data: say why (a spent provider # quota, a network failure), not just "no data". failures = {failure for _, failure in resolved.values() if failure} axis_reason = next( ( code for code in ( "PROVIDER_DAILY_LIMIT", "PROVIDER_HOURLY_LIMIT", "PROVIDER_RATE_LIMITED", "PROVIDER_NETWORK_FAILURE", ) if code in failures ), None, ) # AV-048: why a planned passage has no hour -- passed, or beyond the # forecast (and from when it will be in it) -- never a bare dash. timing = { "now": now_local, "until": hours[-1] if hours else first_hour + timedelta(days=FORECAST_HORIZON_DAYS), } samples = [ _sample( index, route_points[index], distance_by_index[index], resolved.get(index), hours, planned_times.get(index), axis_reason, plan_window, timing, variables=variables, extra=extra_resolved.get(index), ) for index in sample_indexes ] middle = route_points[sample_indexes[len(sample_indexes) // 2]] comparison = ( { "model_comparison": _model_comparison( scope, route_points, live_resolver, hours, planned_times, axis_reason, start_date, end_date, plan_window, freshness, timing, ) } if compare_models else {} ) return { "contract": WEATHER_SERIES_CONTRACT, "journey_id": journey_id, "timezone": zone_name, "axis": axis_kind, "axis_state": "AVAILABLE" if hours else "UNAVAILABLE", "axis_unavailable_reason": None if hours else axis_reason or "PROVIDER_NO_DATA", "hours": [hour.isoformat() for hour in hours], "hour_states": [ classify_astronomical_state( solar_position(hour, middle.latitude, middle.longitude)[0] ).value for hour in hours ], "plan_mode": plan_mode, "plan_anchor": anchor_document(anchor), # AV-048: where the forecast ends (the axis's last hour) -- the page # marks it; passages after it say when they come into the forecast. "forecast_until": hours[-1].isoformat() if hours else None, "axis_span": None if selected else {**span, "data_until": hours[-1].isoformat() if hours else None}, # AV-043: the horizon on the plan's dates (None on the fixed-hour axis). "plan_horizon": None if not selected else { "choice": horizon if plan_window else "TRIP", "hours": int(horizon) if plan_window else None, "window_start": plan_window[0].isoformat() if plan_window else None, "window_end": plan_window[1].isoformat() if plan_window else None, }, "first_day": None if first_day is None and last_day is None else first, "last_day": None if first_day is None and last_day is None else last, "range_start_distance_m": start_m, "range_end_distance_m": end_m, "sample_spacing_m": round((end_m - start_m) / max(1, len(sample_indexes) - 1), 1), "variables": list(variables), "units": _units(list(resolved.values()) + list(extra_resolved.values()), variables), # AV-051: the further charts -- included, or what asking costs. "extras": { "included": extras, "variables": list(WEATHER_SERIES_EXTRA_VARIABLES), "provider_call_weight": float(len(queried)) if axis_available else 0.0, }, "unavailable_elevation_samples": sum( 1 for index in sample_indexes if route_points[index].elevation_m is None ), "source": { "provider": "Open-Meteo", "model_selection": "best_match", "cache_ttl_seconds": LIVE_WEATHER_TTL_SECONDS, }, "walking": _walking( scope, route_points, plan, walk_start_local, walk_pace_profile_id, activity_id ), "samples": samples, "model_comparison_offer": _comparison_offer(scope, route_points, hours, axis_reason), "forecast_freshness": _freshness_document(freshness, zone), # AV-051: a limit in force -- which one and when it comes back. "provider_limit": _limit_document(live_resolver), # AV-051: which source answered how many points (MET Norway when # Open-Meteo refused), and the attributions their licences ask for. **sources_document(resolved.values()), **budget_document(), "source_primary": getattr(live_resolver, "_primary", "open_meteo"), **comparison, **_rest_dates(scope, plan), } def _limit_document(resolver) -> dict[str, Any] | None: document = getattr(resolver, "rate_limit_document", None) return document() if callable(document) else None def _takes_freshness(resolver) -> bool: """The live resolver reports freshness; test doubles may not.""" import inspect try: return "freshness" in inspect.signature(resolver.resolve_many).parameters except (TypeError, ValueError): return False def _freshness_document(freshness: dict[str, Any], zone) -> dict[str, Any] | None: """AV-048: when the forecast shown was fetched (the oldest answer used), whether some of it is past its model's TTL and being fetched again in the background (stale-while-revalidate: the page asks again shortly), and how many grid cells were used and fetched now.""" if not freshness: return None oldest = freshness.get("oldest_fetched_at") return { "fetched_at": datetime.fromtimestamp(oldest, timezone.utc).astimezone(zone).isoformat() if oldest is not None else None, "stale": freshness.get("stale_cells", 0) > 0, "refreshing": bool(freshness.get("refreshing")), "cells": freshness.get("cells", 0), "fetched_cells": freshness.get("fetched_cells", 0), } def _rest_dates(scope, plan) -> dict[str, Any]: """The plan's rest days as calendar dates (docs/design_reference/ rest_days_spike_v0_1.md): days on the slider's axis with no walking, so the Weather tab can say which they are. The same list the Kalendarz tab gets (camps_service.day_boundaries_document). The key is there only for a timed plan that has rest days -- a document without it has none.""" days = scope["days"] if plan is None or not days: return {} markers = tuple(sorted(plan.camp_markers, key=lambda marker: marker.route_point_index)) boundaries = day_boundaries_document( plan, markers, scope["prepared"][-1].point_index, [segment for segment, _ in days] ) dates = [rest_date for boundary in boundaries for rest_date in boundary["rest_dates"]] return {"rest_dates": dates} if dates else {} def _model_comparison( scope, route_points, live_resolver, hours, planned_times, axis_reason, start_date, end_date, plan_window=None, freshness=None, timing=None, ) -> dict[str, Any]: """Step M2: the same range and hour axis as the best_match series, every COMPARISON_SPACING_M, each sample with one series per named model -- side by side, never averaged or merged (ADR-003: model outputs stay apart). A model's gap is its own explicit reason, a failed request each sample's (PROVIDER_DAILY_LIMIT etc.), as in the main series.""" distance_by_index = scope["distance_by_index"] start_m, end_m = scope["start_m"], scope["end_m"] indexes, queried = _comparison_indexes(scope, route_points) resolved: dict[int, tuple[dict | None, str | None]] = {} # No hour axis (a spent quota, no data): nothing to compare against, # and no further quota spent on it. if hours and queried: with call_source("COMPARE"): answers = live_resolver.resolve_many_models( [ ( route_points[index].latitude, route_points[index].longitude, route_points[index].elevation_m, ) for index in queried ], models=COMPARISON_MODELS, start_date=start_date, end_date=end_date, timezone_name=scope["zone_name"], variables=WEATHER_SERIES_VARIABLES, **( {"freshness": freshness} if freshness is not None and _takes_freshness(live_resolver) else {} ), ) resolved = dict(zip(queried, answers)) reason = None if hours else axis_reason or "PROVIDER_NO_DATA" samples = [] for index in indexes: by_model, failure = resolved.get(index, (None, None)) per_model = { model: _sample( index, route_points[index], distance_by_index[index], ((by_model or {}).get(model), failure), hours, planned_times.get(index), reason, plan_window, timing, ) for model in COMPARISON_MODELS } first = per_model[COMPARISON_MODELS[0]] samples.append( { **{key: first[key] for key in _SAMPLE_PLACE_KEYS}, "models": { model: { key: value for key, value in sample.items() if key not in _SAMPLE_PLACE_KEYS } for model, sample in per_model.items() }, } ) available = any( entry["state"] != "UNAVAILABLE" for sample in samples for entry in sample["models"].values() ) failures = [failure for _, failure in resolved.values() if failure] return { "state": "AVAILABLE" if available else "UNAVAILABLE", "unavailable_reason": None if available else reason or (failures[0] if failures else "PROVIDER_NO_DATA"), "models": list(COMPARISON_MODELS), "sample_spacing_m": round((end_m - start_m) / max(1, len(indexes) - 1), 1), # The provider's weight for this comparison (spike section 2.2): per # location and model, the Weather tab's variables in tens (1.2). "provider_call_weight": _call_weight(len(queried)) if resolved else 0, "units": { model: _units( ((by_model or {}).get(model), failure) for by_model, failure in resolved.values() ) for model in COMPARISON_MODELS }, "source": { "provider": "Open-Meteo", "model_selection": "named", "cache_ttl_seconds": LIVE_WEATHER_TTL_SECONDS, }, "samples": samples, } def _comparison_indexes(scope, route_points) -> tuple[list[int], list[int]]: """The comparison's samples over the tab's range (1 km apart, at most COMPARISON_MAX_SAMPLES) and those of them the provider is asked for (points with elevation).""" distance_by_index, in_range = scope["distance_by_index"], scope["in_range"] count = min( COMPARISON_MAX_SAMPLES, max(2, math.ceil((scope["end_m"] - scope["start_m"]) / COMPARISON_SPACING_M) + 1), ) indexes = _even_indexes(in_range, distance_by_index, count) return indexes, [index for index in indexes if route_points[index].elevation_m is not None] def _comparison_offer(scope, route_points, hours, axis_reason) -> dict[str, Any]: """What turning the comparison on would ask for (step M3): its models, samples and provider call weight -- known before any request is made, so the switch can say what it costs. Nothing is fetched for it.""" indexes, queried = _comparison_indexes(scope, route_points) return { "state": "AVAILABLE" if hours and queried else "UNAVAILABLE", "unavailable_reason": None if hours and queried else axis_reason or ("PROVIDER_NO_DATA" if not hours else "ROUTE_ELEVATION_UNAVAILABLE"), "models": list(COMPARISON_MODELS), "sample_count": len(indexes), "sample_spacing_m": round( (scope["end_m"] - scope["start_m"]) / max(1, len(indexes) - 1), 1 ), "provider_call_weight": _call_weight(len(queried)), } def _call_weight(locations: int) -> float: return round(locations * len(COMPARISON_MODELS) * PROVIDER_CALLS_PER_LOCATION, 1) # A sample's place on the route and its planned hour -- the same for every # model at that point. _SAMPLE_PLACE_KEYS = ( "route_point_index", "route_distance_m", "elevation_m", "planned_time", "planned_hour_index", "planned_hour_reason", "planned_hour_available_from", ) def walking_times_document( *, journey_id: str, route_points: Sequence[RoutePoint], plan: JourneyPlan | None, timezone_name: str, first_day: int | None = None, last_day: int | None = None, start_local: str | None = None, pace_profile_id: str | None = None, now: datetime | None = None, activity_id: str | None = None, prefer_saved_dates: bool = False, ) -> dict[str, Any]: """docs/pogoda_v0_1_design.md section 7.3: the walking-time row for the same samples as the weather series (same range, same spacing), from an unsaved start/pace preview -- nothing is written, the plan is untouched.""" plan, anchor = anchored_plan( plan, route_points, activity_id, now, prefer_saved=prefer_saved_dates ) scope = _scope(route_points, plan, timezone_name, first_day, last_day, now, activity_id) return { "journey_id": journey_id, "plan_anchor": anchor_document(anchor), "first_day": scope["first"], "last_day": scope["last"], **_walking(scope, route_points, plan, start_local, pace_profile_id, activity_id), } def _scope( route_points, plan, timezone_name, first_day, last_day, now, activity_id=None ) -> dict[str, Any]: """The tab's range (whole route, or the tree's planned days) and its sample points -- shared by the series and the walking-time preview.""" prepared = prepare_route_cached(route_points).points distance_by_index = {point.point_index: point.cumulative_distance_m for point in prepared} ordered = sorted(distance_by_index) zone_name = plan.journey_timezone if plan is not None else timezone_name zone = ZoneInfo(zone_name) now_local = (now or datetime.now(timezone.utc)).astimezone(zone) days, plan_mode = None, {"state": "UNAVAILABLE", "reason": "NO_PLAN"} # A plan without a start (migration 004): no planned hours to follow, # but its days are real places on the route (step T3) -- a day range is # that stretch of route on the next-72-hours axis. relative_days = None if plan is not None and plan.planned_start_local is None: plan_mode = {"state": "UNAVAILABLE", "reason": NO_PLANNED_START} try: relative_days = derive_relative_day_segments( route_points, pace_factor=pace_factor_for_activity(activity_id, plan.pace_policy_id), pauses=generate_relative_pauses(prepared[-1].cumulative_distance_m), camp_markers=plan.camp_markers, ) except DayDerivationUnavailable: relative_days = None elif plan is not None: try: days = derive_day_point_times( route_points, pace_factor=pace_factor_for_activity(activity_id, plan.pace_policy_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, ) plan_mode = {"state": "AVAILABLE", "reason": None} except DayDerivationUnavailable as unavailable: plan_mode = _timing_unavailable(unavailable) # AV-032: no pace model (a bike) -- its days are still places on the # route, so a day range is that stretch, as for a plan without a start. if ( plan is not None and days is None and relative_days is None and not activity_for(activity_id).pace_model_available ): relative_days = derive_untimed_day_segments(route_points, camp_markers=plan.camp_markers) # The range (tree selection): planned days need a timed plan. first = last = None if (first_day is not None or last_day is not None) and relative_days is not None: first = 0 if first_day is None else first_day last = len(relative_days) - 1 if last_day is None else last_day if not (0 <= first <= last < len(relative_days)): raise ValueError("weather day range is outside the plan's days") selected = None range_start = relative_days[first].start_point_index range_end = relative_days[last].end_point_index elif first_day is not None or last_day is not None: if days is None: raise ValueError("a day range needs a plan that can be timed") 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("weather day range is outside the plan's days") selected = days[first : last + 1] range_start, range_end = selected[0][0].start_point_index, selected[-1][0].end_point_index else: selected = days range_start, range_end = ordered[0], ordered[-1] in_range = [index for index in ordered if range_start <= index <= range_end] start_m, end_m = distance_by_index[in_range[0]], distance_by_index[in_range[-1]] count = min(MAX_SAMPLES, max(2, math.ceil((end_m - start_m) / SAMPLE_SPACING_M) + 1)) return { "prepared": prepared, "route_points": route_points, "distance_by_index": distance_by_index, "zone_name": zone_name, "zone": zone, "now_local": now_local, "plan_mode": plan_mode, "days": days, "selected": selected, "first": first, "last": last, "start_m": start_m, "end_m": end_m, "in_range": in_range, "sample_indexes": _even_indexes(in_range, distance_by_index, count), } def _timing_unavailable(unavailable: DayDerivationUnavailable) -> dict[str, Any]: return { "state": "UNAVAILABLE", "reason": "PLAN_TIMING_UNAVAILABLE", "reason_codes": list(unavailable.reason_codes), "unavailable_elevation_points": unavailable.unavailable_elevation_points, "elevation_gaps": unavailable.gaps_document(), # AV-064: where the long gaps are } def _walking( scope, route_points, plan, start_local, pace_profile_id, activity_id=None ) -> dict[str, Any]: """Section 7.3: the clock time of reaching each sample -- the same pace model as the plan's days (derive_day_point_times: pace profile, pauses, camps, next-morning restart), from the saved plan by default, or from an unsaved preview (start and/or pace given). Without a plan the default is the next 06:00 at the normal pace (Planning Workspace's defaults); a plan without a start (migration 004) takes that default start with its own pace and camps (source PLAN_WITHOUT_START) -- a preview, never saved.""" zone, prepared = scope["zone"], scope["prepared"] preview = start_local is not None or pace_profile_id is not None # AV-048: the next 06:00 (today while it is still ahead, else tomorrow), # as for a plan whose date is not used. default_start = next_start("06:00", scope["zone_name"], scope["now_local"]) default_pace, source = DEFAULT_WALK_PACE_PROFILE_ID, "DEFAULT" if plan is not None: default_pace = plan.pace_policy_id if plan.planned_start_local is None: source = "PLAN_WITHOUT_START" else: default_start = datetime.fromisoformat(plan.planned_start_local).astimezone(zone) source = "PLAN" start = default_start if start_local is None else _local(start_local, zone) pace = pace_profile_id or default_pace base = { "source": "PREVIEW" if preview else source, "start_local": start.strftime("%Y-%m-%dT%H:%M"), "pace_profile_id": pace, "timezone": scope["zone_name"], # The start is when the trip begins; with rest days before day 1 the # walking begins that many days later (only said when there are any). **( {"rest_days_before_start": plan.rest_days_before_start} if plan is not None and plan.rest_days_before_start else {} ), } try: days = derive_day_point_times( route_points, pace_factor=pace_factor_for_activity(activity_id, pace), pauses=generate_relative_pauses(prepared[-1].cumulative_distance_m), start_datetime=start, camp_markers=plan.camp_markers if plan is not None else (), timezone_name=scope["zone_name"], rest_days_before_start=plan.rest_days_before_start if plan is not None else 0, ) except DayDerivationUnavailable as unavailable: return {**base, **_timing_unavailable(unavailable), "times": {}} times: dict[int, datetime] = {} for _, day_times in days: times.update(day_times) return { **base, "state": "AVAILABLE", "reason": None, "times": { str(index): times[index].isoformat() for index in scope["sample_indexes"] if index in times }, } def _local(value: str, zone: ZoneInfo) -> datetime: parsed = datetime.fromisoformat(value) return parsed.replace(tzinfo=zone) if parsed.tzinfo is None else parsed.astimezone(zone) def _even_indexes(indexes: list[int], distance_by_index: dict[int, float], count: int) -> list[int]: """``count`` route point indexes evenly spaced by distance, ends included.""" if len(indexes) <= count: return list(indexes) distances = [distance_by_index[index] for index in indexes] chosen, position = [], 0 for k in range(count): target = distances[0] + (distances[-1] - distances[0]) * k / (count - 1) while position + 1 < len(indexes) and abs(distances[position + 1] - target) <= abs( distances[position] - target ): position += 1 if not chosen or indexes[position] != chosen[-1]: chosen.append(indexes[position]) return chosen HORIZONS = ("24", "72", "TRIP") def _axis_span(horizon, scope, plan, first_hour, now_local) -> dict[str, Any]: """AV-036: how far the fixed-hour axis reaches. TRIP: to the end of the trip's last day -- its planned arrival (a timed plan), else the end of its last dated day (a ride with camps, AV-032) -- cut at the forecast's reach; a trip with no end in time (no dates) or of one day falls back to 72 h and says so.""" reach = first_hour + timedelta(days=FORECAST_HORIZON_DAYS) if horizon != "TRIP": return { "choice": horizon, "hours": int(horizon), "trip_end": None, "forecast_until": None, "fallback_reason": None, } trip_end, days = None, scope.get("days") if days and len(days) > 1: trip_end = days[-1][0].arrival_time elif plan is not None and plan.planned_start_local is not None and len(plan.camp_markers) > 0: untimed = derive_untimed_day_segments(scope["route_points"], camp_markers=plan.camp_markers) first_day = ( datetime.fromisoformat(plan.planned_start_local).astimezone(scope["zone"]).date() ) dates = untimed_day_dates( first_day, plan.rest_days_before_start, [m.rest_days for m in sorted(plan.camp_markers, key=lambda m: m.route_point_index)], len(untimed), ) trip_end = datetime.combine( dates[-1] + timedelta(days=1), datetime.min.time(), tzinfo=scope["zone"] ) if trip_end is None: return { "choice": "TRIP", "hours": NEXT_HOURS, "trip_end": None, "forecast_until": reach.isoformat(), "fallback_reason": "NO_MULTI_DAY_TRIP_IN_TIME", } end = min(trip_end, reach) hours = max(1, math.ceil((end - first_hour).total_seconds() / 3600)) if trip_end <= first_hour: return { "choice": "TRIP", "hours": NEXT_HOURS, "trip_end": trip_end.isoformat(), "forecast_until": reach.isoformat(), "fallback_reason": "TRIP_ALREADY_OVER", } return { "choice": "TRIP", "hours": hours, "trip_end": trip_end.isoformat(), "forecast_until": reach.isoformat(), "fallback_reason": None, "beyond_forecast": trip_end > reach, } def _axis_hours( resolved, first_hour: datetime, axis_kind: str, span_hours: int | None = None ) -> list[datetime]: """The provider's own hours (every point is asked for the same dates in the same timezone, so they coincide); the fixed-hour axis starts at the current hour and runs ``span_hours`` (AV-036: 24, 72 or the trip).""" series = next((sample.times for sample, _ in resolved.values() if sample and sample.times), ()) if axis_kind == "NEXT_72_HOURS": end = first_hour + timedelta(hours=span_hours or NEXT_HOURS) return [time for time in series if first_hour <= time < end] return list(series) def _sample( index, point, distance_m, resolved, hours, planned_time, axis_reason, window=None, timing=None, variables=WEATHER_SERIES_VARIABLES, extra=None, ): base = { "route_point_index": index, "route_distance_m": distance_m, "elevation_m": point.elevation_m, "planned_time": planned_time.isoformat() if planned_time else None, } base.update(_planned_hour(planned_time, hours, window, timing)) if point.elevation_m is None: return {**base, **_unavailable("ROUTE_ELEVATION_UNAVAILABLE", variables)} if axis_reason: return {**base, **_unavailable(axis_reason, variables)} sample, failure = resolved if resolved else (None, None) if sample is None: return {**base, **_unavailable(failure or "PROVIDER_NO_DATA", variables)} position = {time: i for i, time in enumerate(sample.times)} # AV-048: each hour's position looked up once, not once per variable. positions = [position.get(hour) for hour in hours] values = { variable: [_value(sample, variable, at) for at in positions] for variable in WEATHER_SERIES_VARIABLES } reasons = dict(sample.base.missing_variable_reasons) models = dict(sample.variable_models) if variables != WEATHER_SERIES_VARIABLES: # AV-051: the further fields from their own answer (asked for). extra_sample, extra_failure = extra if extra else (None, None) extra_positions = ( [{t: i for i, t in enumerate(extra_sample.times)}.get(hour) for hour in hours] if extra_sample is not None else [] ) for variable in WEATHER_SERIES_EXTRA_VARIABLES: values[variable] = ( [_value(extra_sample, variable, at) for at in extra_positions] if extra_sample is not None else [None] * len(hours) ) if extra_sample is None: reasons[variable] = extra_failure or "PROVIDER_NO_DATA" else: reasons.update( { k: v for k, v in extra_sample.base.missing_variable_reasons.items() if k == variable } ) models.update( {k: v for k, v in extra_sample.variable_models.items() if k == variable} ) missing = { variable: reasons.get(variable, "WEATHER_PROVIDER_VALUE_MISSING") for variable, series in values.items() if not any(value is not None for value in series) } present = sum(1 for series in values.values() for value in series if value is not None) total = sum(len(series) for series in values.values()) return { **base, "state": "AVAILABLE" if present == total and total else "PARTIAL" if present else "UNAVAILABLE", "unavailable_reason": None if present else "PROVIDER_NO_DATA", "values_by_hour": values, "missing_reasons": missing, "variable_models": models, } def _planned_hour(planned_time, hours, window=None, timing=None): if planned_time is None: return { "planned_hour_index": None, "planned_hour_reason": None, "planned_hour_available_from": None, } target = planned_time.astimezone(timezone.utc) best = min( range(len(hours)), key=lambda i: abs(hours[i].astimezone(timezone.utc) - target), default=None, ) if best is None or abs(hours[best].astimezone(timezone.utc) - target) > MAX_VALID_TIME_OFFSET: # AV-043: a passage beyond the chosen 24 h / 72 h is that, not # "outside the forecast" -- the whole trip would show it. outside_window = window is not None and ( planned_time < window[0] or planned_time >= window[1] - MAX_VALID_TIME_OFFSET ) available_from = None if outside_window: reason = "PLANNED_TIME_OUTSIDE_HORIZON" elif timing is not None and planned_time < timing["now"] - MAX_VALID_TIME_OFFSET: # AV-048: the saved dates on request -- that passage is behind us. reason = "PLANNED_TIME_PASSED" elif timing is not None and planned_time > timing["until"] + MAX_VALID_TIME_OFFSET: # AV-048: beyond the forecast's reach; it comes into the forecast # FORECAST_HORIZON_DAYS - 1 days before (today's date included). reason = "PLANNED_TIME_BEYOND_FORECAST" available_from = ( planned_time.date() - timedelta(days=FORECAST_HORIZON_DAYS - 1) ).isoformat() else: reason = "PLANNED_TIME_OUTSIDE_FORECAST" return { "planned_hour_index": None, "planned_hour_reason": reason, "planned_hour_available_from": available_from, } return { "planned_hour_index": best, "planned_hour_reason": None, "planned_hour_available_from": None, } def _value(sample, variable, position): series = sample.variables_by_time.get(variable) return ( None if series is None or position is None or position >= len(series) else series[position] ) def _unavailable(reason: str, variables=WEATHER_SERIES_VARIABLES) -> dict[str, Any]: return { "state": "UNAVAILABLE", "unavailable_reason": reason, "values_by_hour": {variable: [] for variable in variables}, "missing_reasons": {}, "variable_models": {}, } def _units(resolved, variables=WEATHER_SERIES_VARIABLES) -> dict[str, str | None]: units: dict[str, str | None] = {variable: None for variable in variables} for sample, _ in resolved: if sample is None: continue for variable in units: unit = sample.base.units.get(variable) if units[variable] is None and unit and unit != "undefined": units[variable] = unit return units