"""Derive day boundaries from camp markers, for a route with no predefined stage file. Parallel path, not a replacement: mountain_twin/journey/stages.py (the TMB fixture's static per-day CSV+GeoJSON reader) is untouched and stays authoritative for the TMB fixture's 9-day list. This module is for routes that have camp markers on their JourneyPlan (docs/camps_staging_v0_1_design.md section 4) instead of a predefined stage file. Adds no new terrain/pace/timing mathematics: it is a thin "cut a continuous route into segments at these positions" layer over the existing, unmodified mountain_twin.pace.analyze_route_pace (moving time) and mountain_twin.weather.temporal.assign_timeline (planned arrival times, already correctly accounting for cumulative pauses). See docs/design_reference/camps_staging_spike_v0_1.md section 3.2 for why stages.py itself cannot be generalised in place. Nights at camp: the whole route's pace timeline is computed once (unchanged moving time and pauses), then every day after the first is re-anchored to the plan's own start time of day on the calendar day after the previous day's arrival -- a camp means sleeping there, so day N+1 starts the next morning, not at the moment day N arrived. Each day keeps its own duration. Rest days (docs/design_reference/rest_days_spike_v0_1.md): the same gap, made longer. A camp's ``rest_days`` are whole extra days spent there, so the next day starts that many calendar days later; ``rest_days_before_start`` holds the first day back the same way from the plan's start (the start is when the trip begins, not necessarily when day 1 walks). Day numbers count walking days only and every day keeps its own duration and clock times -- only dates move. With no rest days anywhere nothing changes at all. """ from __future__ import annotations import threading from collections import OrderedDict from dataclasses import dataclass from datetime import datetime, timedelta, timezone, tzinfo from typing import Any, Sequence from zoneinfo import ZoneInfo from mountain_twin.journey.contracts import CampMarker from mountain_twin.journey.route_segments import elevation_gaps, fill_short_gaps from mountain_twin.pace.analysis import analyze_route_pace from mountain_twin.route_analysis import prepare_route_cached, route_fingerprint from mountain_twin.weather.temporal import PlanningScenario, assign_timeline DAY_DERIVATION_VERSION = "camp_day_derivation_v0_2" # The reason code for a plan saved without a start (migration 004, docs/ # design_reference/weather_start_models_spike_v0_1.md step T1): its route # has no calendar days, clock times or camp nights until a start is set. NO_PLANNED_START = "NO_PLANNED_START" # Anchors the one pace timeline of a plan without a start (steps T2/T3): # its days are known by where they start and end and how long they take, # which does not depend on the start -- this instant is never shown or # returned (relative_day_document() drops every clock time). RELATIVE_TIMING_REFERENCE_START = datetime(2000, 1, 3, 6, 0, tzinfo=timezone.utc) class DayDerivationUnavailable(ValueError): """The route cannot be timed (pace unresolved), so no days, planned times or camp arrivals exist -- an explicit missing-data state, not bad input. ``reason_codes`` are the pace engine's (e.g. ``PACE_ELEVATION_UNRESOLVED`` for route points without elevation, docs/design_reference/ elevation_gaps_spike_v0_1.md); ``unavailable_elevation_points`` counts them.""" def __init__( self, reason_codes: Sequence[str], unavailable_elevation_points: int, elevation_gaps: Sequence[dict[str, Any]] = (), ): super().__init__(f"day derivation pace unresolved: {tuple(reason_codes)}") self.reason_codes = tuple(reason_codes) self.unavailable_elevation_points = unavailable_elevation_points # AV-064: the gaps too long to fill for timing, where they are. self.elevation_gaps = tuple(elevation_gaps) def gaps_document(self) -> list[dict[str, Any]]: return [ { "from_m": round(gap["from_m"]), "to_m": round(gap["to_m"]), "points": len(gap["point_indexes"]), } for gap in self.elevation_gaps ] @dataclass(frozen=True) class DaySegment: day_number: int start_point_index: int end_point_index: int start_route_distance_m: float end_route_distance_m: float distance_m: float departure_time: datetime arrival_time: datetime # The camp that closes this day; None for the final day (the route's own # end has no camp marker there). camp_label: str | None # AV-064: points of this day whose height was filled in for timing only # (a short gap) -- its time is an estimate. estimated_elevation_points: int = 0 def to_dict(self) -> dict[str, Any]: return { "day_number": self.day_number, "start_point_index": self.start_point_index, "end_point_index": self.end_point_index, "start_route_distance_m": self.start_route_distance_m, "end_route_distance_m": self.end_route_distance_m, "distance_m": self.distance_m, "departure_time": self.departure_time.isoformat(), "arrival_time": self.arrival_time.isoformat(), "camp_label": self.camp_label, "estimated_elevation_points": self.estimated_elevation_points, } def derive_day_segments( route_points: Sequence[Any], *, pace_factor: float, pauses: Sequence[Any] = (), start_datetime: datetime, camp_markers: Sequence[CampMarker] = (), timezone_name: str | None = None, rest_days_before_start: int = 0, ) -> tuple[DaySegment, ...]: """route_points is any mountain_twin.exposure.RoutePoint-shaped sequence -- this works for a drawn/imported route exactly as it does for a frozen fixture; nothing here is TMB-specific. timezone_name (IANA) decides which calendar day "the next morning" is and keeps the start time of day right across DST changes on a long trip; without it, start_datetime's own offset is used.""" segments, _ = _derive( route_points, pace_factor=pace_factor, pauses=pauses, start_datetime=start_datetime, camp_markers=camp_markers, timezone_name=timezone_name, rest_days_before_start=rest_days_before_start, ) return segments def derive_day_point_times( route_points: Sequence[Any], *, pace_factor: float, pauses: Sequence[Any] = (), start_datetime: datetime, camp_markers: Sequence[CampMarker] = (), timezone_name: str | None = None, rest_days_before_start: int = 0, ) -> tuple[tuple[DaySegment, dict[int, datetime]], ...]: """Each derived day with the planned time at every one of its route points: the same single route timeline derive_day_segments() uses, shifted by that day's own re-anchoring offset (a camp night moves the whole next day, not just its departure). Nothing new is computed.""" segments, timeline_by_index = _derive( route_points, pace_factor=pace_factor, pauses=pauses, start_datetime=start_datetime, camp_markers=camp_markers, timezone_name=timezone_name, rest_days_before_start=rest_days_before_start, ) days = [] for segment in segments: offset = ( segment.departure_time - timeline_by_index[segment.start_point_index].planned_arrival ) days.append( ( segment, { index: point.planned_arrival + offset for index, point in timeline_by_index.items() if segment.start_point_index <= index <= segment.end_point_index }, ) ) return tuple(days) def _derive( route_points: Sequence[Any], *, pace_factor: float, pauses: Sequence[Any], start_datetime: datetime, camp_markers: Sequence[CampMarker], timezone_name: str | None, rest_days_before_start: int = 0, ) -> tuple[tuple[DaySegment, ...], dict[int, Any]]: # AV-048: the same derivation is asked for several times per Weather # opening (the plan, the walking times); its result depends only on these # inputs, so the last few are kept (callers only read them). try: key = ( route_fingerprint(route_points) if route_points else None, pace_factor, tuple(pauses), # Not the datetime itself: equal instants in other zones compare # (and hash) equal, and the result carries the start's own zone. start_datetime.isoformat(), str(start_datetime.tzinfo), tuple(camp_markers), timezone_name, rest_days_before_start, ) hash(key) except TypeError: key = None if key is not None: with _derive_lock: cached = _derive_cache.get(key) if cached is not None: _derive_cache.move_to_end(key) return cached result = _derive_uncached( route_points, pace_factor=pace_factor, pauses=pauses, start_datetime=start_datetime, camp_markers=camp_markers, timezone_name=timezone_name, rest_days_before_start=rest_days_before_start, ) if key is not None: with _derive_lock: _derive_cache[key] = result while len(_derive_cache) > _PACE_CACHE_SIZE: _derive_cache.popitem(last=False) return result def _derive_uncached( route_points: Sequence[Any], *, pace_factor: float, pauses: Sequence[Any], start_datetime: datetime, camp_markers: Sequence[CampMarker], timezone_name: str | None, rest_days_before_start: int = 0, ) -> tuple[tuple[DaySegment, ...], dict[int, Any]]: if not route_points: raise ValueError("day derivation requires route points") if rest_days_before_start < 0: raise ValueError("rest days before the start must not be negative") if start_datetime.tzinfo is None or start_datetime.utcoffset() is None: raise ValueError("day derivation start must be timezone-aware") prepared = prepare_route_cached(route_points).points valid_indices = {point.point_index for point in prepared} # Sorted by route_point_index (the geometrically authoritative order # along the prepared route), not by the caller-supplied route_distance_m # -- a route_point_index strictly orders points along the route by # construction (mountain_twin.route_analysis.prepare_route), while a # distance value is just cached/informational and must never be the # thing that decides day order. sorted_markers = tuple(sorted(camp_markers, key=lambda marker: marker.route_point_index)) seen_indices: set[int] = set() for marker in sorted_markers: if marker.route_point_index not in valid_indices: raise ValueError( f"camp marker route_point_index {marker.route_point_index} is not a real route point" ) if marker.route_point_index in seen_indices: raise ValueError("camp markers must not repeat the same route point") seen_indices.add(marker.route_point_index) scenario = PlanningScenario( "camps", start_datetime, moving_speed_mps=1.0, pauses=tuple(pauses), pace_factor=pace_factor ) # AV-064: short elevation gaps filled for the pace model only (in # memory, route_segments.fill_short_gaps); the timeline below keeps the # route's own points. timed, filled = fill_short_gaps(prepared) pace = _route_pace(route_points, timed, pace_factor, tuple(pauses)) if pace.state.value != "COMPLETE": raise DayDerivationUnavailable( pace.reason_codes, sum(1 for point in timed if point.gpx_elevation_m is None), [gap for gap in elevation_gaps(prepared) if not gap["fillable"]], ) timeline = assign_timeline(prepared, scenario, pace) timeline_by_index = {point.point_index: point for point in timeline} last_index = prepared[-1].point_index boundary_indices = [marker.route_point_index for marker in sorted_markers] if not boundary_indices or boundary_indices[-1] != last_index: boundary_indices.append(last_index) zone: tzinfo = ZoneInfo(timezone_name) if timezone_name else start_datetime.tzinfo start_time_of_day = start_datetime.astimezone(zone).time().replace(tzinfo=None) segments: list[DaySegment] = [] previous_index = prepared[0].point_index for day_number, boundary_index in enumerate(boundary_indices, start=1): start_point = timeline_by_index[previous_index] end_point = timeline_by_index[boundary_index] camp_label = ( sorted_markers[day_number - 1].label if day_number <= len(sorted_markers) else None ) # Moving time plus this day's own pauses, from the one route timeline. duration = end_point.planned_arrival - start_point.planned_arrival if segments: # The camp that closed the previous day: its rest days are spent # there, before this day sets out. rest_days = sorted_markers[day_number - 2].rest_days departure = _next_morning(segments[-1].arrival_time, start_time_of_day, zone, rest_days) elif rest_days_before_start: trip_start = start_point.planned_arrival.astimezone(zone) departure = datetime.combine( trip_start.date() + timedelta(days=rest_days_before_start), start_time_of_day, tzinfo=zone, ) else: departure = start_point.planned_arrival segments.append( DaySegment( day_number=day_number, start_point_index=previous_index, end_point_index=boundary_index, start_route_distance_m=start_point.route_distance_m, end_route_distance_m=end_point.route_distance_m, distance_m=end_point.route_distance_m - start_point.route_distance_m, departure_time=departure, arrival_time=departure + duration, camp_label=camp_label, estimated_elevation_points=sum( 1 for index in filled if previous_index < index <= boundary_index ), ) ) previous_index = boundary_index return tuple(segments), timeline_by_index def _next_morning( arrival: datetime, start_time_of_day, zone: tzinfo, rest_days: int = 0 ) -> datetime: """The plan's start time of day on the calendar day after ``arrival`` (in ``zone``) -- a camp night between two days -- and ``rest_days`` whole days later still when the camp is a longer stop.""" next_day = arrival.astimezone(zone).date() + timedelta(days=1 + rest_days) return datetime.combine(next_day, start_time_of_day, tzinfo=zone) def derive_relative_day_segments( route_points: Sequence[Any], *, pace_factor: float, pauses: Sequence[Any] = (), camp_markers: Sequence[CampMarker] = (), ) -> tuple[DaySegment, ...]: """The days of a route with no start: the same split and durations as derive_day_segments() gives for any start (camps close days, the next day restarts the next morning), anchored on RELATIVE_TIMING_REFERENCE_START. Only their places, distances and durations mean anything -- publish them through relative_day_document(), never their clock times.""" return derive_day_segments( route_points, pace_factor=pace_factor, pauses=pauses, start_datetime=RELATIVE_TIMING_REFERENCE_START, camp_markers=camp_markers, ) def relative_day_document(segment: DaySegment) -> dict[str, Any]: """A day without clock times: where it runs, how far, and how long it takes (moving time plus its pauses) -- departure/arrival are null.""" return { **segment.to_dict(), "departure_time": None, "arrival_time": None, "duration_s": (segment.arrival_time - segment.departure_time).total_seconds(), } # --- Days without a pace model (AV-032) ------------------------------------ # A route drawn for an activity with no pace model yet (bikes, AV-031) still # has days: the user puts the camps, a camp closes a day and the next one # starts the next morning. Where each day runs and how far is geometry; its # calendar date is calendar arithmetic (the plan's start date, one date per # day, rest days in between) -- neither needs a pace. Clock times, durations # and arrivals do, and stay null: never a walking time, never a guess. @dataclass(frozen=True) class UntimedDay: day_number: int start_point_index: int end_point_index: int start_route_distance_m: float end_route_distance_m: float distance_m: float camp_label: str | None def derive_untimed_day_segments( route_points: Sequence[Any], *, camp_markers: Sequence[CampMarker] = () ) -> tuple[UntimedDay, ...]: """The route cut at its camps, in route order -- no pace involved.""" if not route_points: raise ValueError("day derivation requires route points") prepared = prepare_route_cached(route_points).points distance = {point.point_index: point.cumulative_distance_m for point in prepared} markers = tuple(sorted(camp_markers, key=lambda marker: marker.route_point_index)) seen: set[int] = set() for marker in markers: if marker.route_point_index not in distance: raise ValueError( f"camp marker route_point_index {marker.route_point_index} is not a real route point" ) if marker.route_point_index in seen: raise ValueError("camp markers must not repeat the same route point") seen.add(marker.route_point_index) last = prepared[-1].point_index ends = [ (marker.route_point_index, marker.label) for marker in markers if marker.route_point_index < last ] ends.append((last, None)) days, start = [], prepared[0].point_index for number, (end, label) in enumerate(ends, start=1): days.append( UntimedDay( day_number=number, start_point_index=start, end_point_index=end, start_route_distance_m=distance[start], end_route_distance_m=distance[end], distance_m=distance[end] - distance[start], camp_label=label, ) ) start = end return tuple(days) def untimed_day_dates( start: Any, rest_days_before_start: int, rest_days_after_days: Sequence[int], count: int ) -> list[Any]: """Each day's calendar date: the trip starts on ``start`` (a date, or None for a plan without a start -> every date None), rest days before day 1 hold it back, a camp's rest days push every later day.""" if start is None: return [None] * count dates, current = [], start + timedelta(days=rest_days_before_start) for index in range(count): dates.append(current) rest = rest_days_after_days[index] if index < len(rest_days_after_days) else 0 current = current + timedelta(days=1 + rest) return dates def untimed_day_document(day: UntimedDay, date_value: Any) -> dict[str, Any]: """A day with its place, distance and (when the plan has a start) its date; departure, arrival and duration are null -- no pace model.""" return { "day_number": day.day_number, "start_point_index": day.start_point_index, "end_point_index": day.end_point_index, "start_route_distance_m": day.start_route_distance_m, "end_route_distance_m": day.end_route_distance_m, "distance_m": day.distance_m, "camp_label": day.camp_label, "date": None if date_value is None else date_value.isoformat(), "departure_time": None, "arrival_time": None, "duration_s": None, } # AV-048: the pace analysis of a route depends only on its content, the # pace factor and the pauses (not on the start), and the Weather tab with the # profile, storms and camps requests beside it asked for the same one up to # five times per opening. The last few are kept (PaceResult is frozen). _PACE_CACHE_SIZE = 16 _pace_cache: "OrderedDict[tuple, Any]" = OrderedDict() _pace_lock = threading.Lock() _derive_cache: "OrderedDict[tuple, Any]" = OrderedDict() _derive_lock = threading.Lock() def _route_pace(route_points, prepared, pace_factor, pauses): key = (route_fingerprint(route_points), pace_factor, pauses) with _pace_lock: cached = _pace_cache.get(key) if cached is not None: _pace_cache.move_to_end(key) return cached pace = analyze_route_pace( prepared, scenario_name="camps", scenario_factor=pace_factor, pauses=pauses ) with _pace_lock: _pace_cache[key] = pace while len(_pace_cache) > _PACE_CACHE_SIZE: _pace_cache.popitem(last=False) return pace