"""Immutable saved-Journey contracts, deliberately separate from engine results.""" from __future__ import annotations import json from dataclasses import asdict, dataclass, field from datetime import date, datetime, timezone from enum import Enum from typing import Any, Mapping from zoneinfo import ZoneInfo class ProductAvailability(str, Enum): AVAILABLE = "AVAILABLE" PARTIAL = "PARTIAL" UNAVAILABLE = "UNAVAILABLE" UNKNOWN = "UNKNOWN" NOT_APPLICABLE = "NOT_APPLICABLE" class JourneyDateIntentKind(str, Enum): """Honest precision for user-entered Journey dates.""" UNSPECIFIED = "UNSPECIFIED" MONTH = "MONTH" DATE_RANGE = "DATE_RANGE" class JourneyActivityIntent(str, Enum): """Provider-neutral planning intent; it is not an Activity classification.""" HIKING = "HIKING" TREKKING = "TREKKING" TRAIL_RUNNING = "TRAIL_RUNNING" CYCLING = "CYCLING" MOUNTAINEERING = "MOUNTAINEERING" OTHER = "OTHER" def _is_time_of_day(value: object) -> bool: if not isinstance(value, str) or len(value) != 5 or value[2] != ":": return False hours, minutes = value[:2], value[3:] return hours.isdigit() and minutes.isdigit() and int(hours) < 24 and int(minutes) < 60 def _aware(value: str, label: str) -> datetime: result = datetime.fromisoformat(value) if result.tzinfo is None or result.utcoffset() is None: raise ValueError(f"{label} must be timezone-aware") return result @dataclass(frozen=True) class Journey: journey_id: str title: str current_plan_id: str | None = None current_analysis_run_id: str | None = None # Legacy analyzed-Journey display context retained for the TMB fixture. activity_type: str | None = None owner_id: str | None = None destination_label: str | None = None country_label: str | None = None date_intent: JourneyDateIntent = field(default_factory=lambda: JourneyDateIntent.unspecified()) activity_intent: JourneyActivityIntent | None = None created_at: str | None = None updated_at: str | None = None # Route persistence v0.1 (docs/route_persistence_v0_1_design.md), added # at the end to preserve every existing positional construction (in # particular mountain_twin.journey.repository.tmb_day_1_repository(), # which is untouched and leaves this None -- TMB resolves its route via # JourneyPlan.route_revision_id, not this pointer). current_route_revision_id: str | None = None # AV-027: when the owner marked the Journey as done (migration 007); # None while it is upcoming. Only that explicit action sets it. completed_at: str | None = None def __post_init__(self) -> None: if not self.journey_id: raise ValueError("journey requires stable identity") if not self.title or self.title != self.title.strip(): raise ValueError("journey title must be non-empty and trimmed") if not isinstance(self.date_intent, JourneyDateIntent): raise ValueError("journey date intent is invalid") if self.activity_intent is not None and not isinstance( self.activity_intent, JourneyActivityIntent ): raise ValueError("journey activity intent is unsupported") for label, value in ( ("owner", self.owner_id), ("destination", self.destination_label), ("country", self.country_label), ): if value is not None and (not value or value != value.strip()): raise ValueError(f"journey {label} must be non-empty and trimmed when supplied") for label, value in ( ("journey creation", self.created_at), ("journey update", self.updated_at), ): if value is not None: _aware(value, label) @dataclass(frozen=True) class JourneyDateIntent: """A date-precision value that never fabricates an instant from a month.""" kind: JourneyDateIntentKind year: int | None = None month: int | None = None start_date: date | None = None end_date: date | None = None def __post_init__(self) -> None: if self.kind is JourneyDateIntentKind.UNSPECIFIED: if any( value is not None for value in (self.year, self.month, self.start_date, self.end_date) ): raise ValueError("unspecified Journey date intent cannot contain dates") return if self.kind is JourneyDateIntentKind.MONTH: if ( self.year is None or self.month is None or self.start_date is not None or self.end_date is not None ): raise ValueError("month Journey date intent requires only year and month") date(self.year, self.month, 1) return if self.kind is JourneyDateIntentKind.DATE_RANGE: if ( self.start_date is None or self.end_date is None or self.year is not None or self.month is not None ): raise ValueError("date-range Journey date intent requires only start and end dates") if self.start_date > self.end_date: raise ValueError("Journey date range start must not follow end") return raise ValueError("unsupported Journey date intent") @classmethod def unspecified(cls) -> JourneyDateIntent: return cls(JourneyDateIntentKind.UNSPECIFIED) @classmethod def month_intent(cls, year: int, month: int) -> JourneyDateIntent: return cls(JourneyDateIntentKind.MONTH, year=year, month=month) @classmethod def date_range(cls, start_date: date, end_date: date) -> JourneyDateIntent: return cls(JourneyDateIntentKind.DATE_RANGE, start_date=start_date, end_date=end_date) @dataclass(frozen=True) class RouteRevision: route_revision_id: str # Route persistence v0.1 (docs/route_persistence_v0_1_design.md section # 3; docs/journey_creator_v0_1_design.md line 103): explicit ownership. # route_id remains the logical route lineage within this Journey, while # route_revision_id identifies one immutable geometry/source snapshot. journey_id: str route_id: str route_name: str geometry_fingerprint: str point_count: int source_reference: str # AV-031 (migration 008): what the route was drawn for. None for the TMB # fixture and for durable routes saved before AV-031 -- both read as # trekking (mountain_twin.journey.activities.activity_for). activity_id: str | None = None routing_profile: str | None = None def __post_init__(self) -> None: if not all( ( self.route_revision_id, self.journey_id, self.route_id, self.geometry_fingerprint, self.source_reference, ) ): raise ValueError("route revision requires stable source identity") if self.point_count <= 0: raise ValueError("route revision requires route points") @dataclass(frozen=True) class JourneyPlan: journey_plan_id: str journey_id: str route_revision_id: str version: int journey_timezone: str # None (both) for a plan saved without a start (migration 004, docs/ # design_reference/weather_start_models_spike_v0_1.md step T1): the # route is timed relative to its start until a real start is set as a # new plan version -- never a guessed or default time. planned_start_local: str | None planned_start_utc: str | None pace_policy_id: str pause_policy_id: str activity_type: str | None = None # Additive, camps/day-staging v0.1 (docs/camps_staging_v0_1_design.md # section 4): overnight camp markers on this plan version. A new/edited # marker is a new JourneyPlan version via the existing # LocalJourneyRepository.create_plan versioning -- this field is never # mutated in place. Independent of mountain_twin/journey/stages.py's # static per-TMB-day CSV+GeoJSON list, which stays authoritative for the # TMB fixture. camp_markers: tuple["CampMarker", ...] = () # Rest days at the two ends of the trip, which have no camp to carry them # (migration 006, docs/design_reference/rest_days_spike_v0_1.md): whole # days between the plan's start and day 1 setting out, and after the last # day's arrival when the route does not end at a camp. Rest days between # two walking days live on the camp between them (CampMarker.rest_days). rest_days_before_start: int = 0 rest_days_after_finish: int = 0 # AV-035 (migration 010): the start time of day, "HH:MM" in the Journey's # timezone -- with or without a date. None with a dated start means "the # time of planned_start_local" (start_time_of_day() reads either). planned_start_time: str | None = None def start_time_of_day(self) -> str | None: """The plan's start time "HH:MM", whether or not it has a date.""" if self.planned_start_time is not None: return self.planned_start_time return None if self.planned_start_local is None else self.planned_start_local[11:16] def __post_init__(self) -> None: if self.planned_start_time is not None and not _is_time_of_day(self.planned_start_time): raise ValueError("planned_start_time must be HH:MM") for name in ("rest_days_before_start", "rest_days_after_finish"): _validate_rest_days(getattr(self, name), name) if (self.planned_start_local is None) != (self.planned_start_utc is None): raise ValueError("a plan's start is both its local and UTC start, or neither") zone = ZoneInfo(self.journey_timezone) if self.planned_start_local is not None: local = _aware(self.planned_start_local, "planned local start") utc = _aware(self.planned_start_utc, "planned UTC start") if local.astimezone(timezone.utc) != utc.astimezone(timezone.utc): raise ValueError("planned local and UTC starts must identify the same instant") resolved = local.astimezone(zone) if (resolved.replace(tzinfo=None), resolved.utcoffset()) != ( local.replace(tzinfo=None), local.utcoffset(), ): raise ValueError( "planned local start is not a valid instant in journey IANA timezone" ) if self.version < 1: raise ValueError("journey plan version must be positive") if not all(isinstance(marker, CampMarker) for marker in self.camp_markers): raise ValueError("camp_markers must contain CampMarker values") distances = [marker.route_distance_m for marker in self.camp_markers] if distances != sorted(distances): raise ValueError("camp_markers must be sorted by route_distance_m") if len(set(distances)) != len(distances): raise ValueError("camp_markers must not repeat the same route_distance_m") CAMP_ACCOMMODATION_TYPES = ("WILD_CAMP", "CAMPSITE", "HUT", "CABIN", "OTHER") # A bound against typing mistakes, not a product rule about how long one may # rest: a single stop longer than this is refused rather than silently stored. MAX_REST_DAYS = 60 def _validate_rest_days(value: object, name: str) -> None: if isinstance(value, bool) or not isinstance(value, int) or not 0 <= value <= MAX_REST_DAYS: raise ValueError(f"{name} must be a whole number of days from 0 to {MAX_REST_DAYS}") @dataclass(frozen=True) class CampMarker: """One overnight camp on a JourneyPlan: a position on the route plus a user label. Deliberately minimal -- no POI/name suggestion (Phase 2, out of scope), no per-camp identity beyond the plan version it lives on. accommodation_type/accommodation_type_other/cost_amount/cost_currency (AV-009, task 2) are the Kalendarz tab's single source of truth for a camp's accommodation edit -- read and written through this same dataclass and the camp_markers_json column (never a second, parallel field), so an edit from Kalendarz is visible in Noclegi immediately. accommodation_type is None until a user actively picks one -- never defaulted to a category that might not fit. rest_days: whole extra days spent at this camp beyond its one night (the day that follows sets out that many calendar days later). Stored with the camp, not with a day number, so it stays with its place when camps are added or removed before it.""" route_point_index: int route_distance_m: float label: str accommodation_type: str | None = None accommodation_type_other: str | None = None cost_amount: float | None = None cost_currency: str = "PLN" rest_days: int = 0 def __post_init__(self) -> None: _validate_rest_days(self.rest_days, "camp marker rest_days") if self.route_point_index < 0: raise ValueError("camp marker route point index must be non-negative") if self.route_distance_m < 0: raise ValueError("camp marker route distance must be non-negative") if not self.label or self.label != self.label.strip(): raise ValueError("camp marker label must be non-empty and trimmed") if ( self.accommodation_type is not None and self.accommodation_type not in CAMP_ACCOMMODATION_TYPES ): raise ValueError( f"accommodation_type must be one of {CAMP_ACCOMMODATION_TYPES} or None" ) if self.accommodation_type_other and self.accommodation_type != "OTHER": raise ValueError( "accommodation_type_other is only meaningful when accommodation_type is OTHER" ) if self.cost_amount is not None and self.cost_amount < 0: raise ValueError("camp marker cost_amount must be non-negative") if not self.cost_currency or self.cost_currency != self.cost_currency.strip(): raise ValueError("camp marker cost_currency must be non-empty and trimmed") def to_dict(self) -> dict[str, Any]: return { "route_point_index": self.route_point_index, "route_distance_m": self.route_distance_m, "label": self.label, "accommodation_type": self.accommodation_type, "accommodation_type_other": self.accommodation_type_other, "cost_amount": self.cost_amount, "cost_currency": self.cost_currency, "rest_days": self.rest_days, } @dataclass(frozen=True) class AnalysisScenario: analysis_scenario_id: str analysis_run_id: str name: str planned_start_local: str planned_start_utc: str pace_policy_id: str pause_policy_id: str semantic_version: str = "v0_1" def __post_init__(self) -> None: local = _aware(self.planned_start_local, "scenario local start") utc = _aware(self.planned_start_utc, "scenario UTC start") if local.astimezone(timezone.utc) != utc.astimezone(timezone.utc): raise ValueError("scenario local and UTC starts must identify the same instant") @dataclass(frozen=True) class SourceSnapshot: source_snapshot_id: str source_type: str source_reference: str provider: str | None = None provider_run_at: str | None = None retrieved_at: str | None = None def __post_init__(self) -> None: if not all((self.source_snapshot_id, self.source_type, self.source_reference)): raise ValueError("source snapshot requires stable identity") for label, value in ( ("provider run", self.provider_run_at), ("retrieved", self.retrieved_at), ): if value is not None: _aware(value, label) @dataclass(frozen=True) class AnalysisRun: analysis_run_id: str journey_plan_id: str route_revision_id: str executed_at: str input_fingerprint: str projection_version: str source_snapshot_ids: tuple[str, ...] producer_result_references: Mapping[str, str] = field(default_factory=dict) status: str = "AVAILABLE" def __post_init__(self) -> None: _aware(self.executed_at, "analysis execution") if not all( ( self.analysis_run_id, self.journey_plan_id, self.route_revision_id, self.input_fingerprint, ) ): raise ValueError("analysis run requires stable identity") @dataclass(frozen=True) class RouteSelectionAnchor: route_revision_id: str analysis_scenario_id: str route_point_index: int | None = None route_distance_m: float | None = None planned_arrival: str | None = None def __post_init__(self) -> None: if (self.route_point_index is None) == (self.route_distance_m is None): raise ValueError( "selection requires exactly one of route point index or route distance" ) if self.route_point_index is not None and self.route_point_index < 0: raise ValueError("route point index must be non-negative") if self.route_distance_m is not None and self.route_distance_m < 0: raise ValueError("route distance must be non-negative") if self.planned_arrival is not None: _aware(self.planned_arrival, "selection planned arrival") def serialize_contract(value: Any) -> str: """Canonical serialization for local deterministic contracts.""" return json.dumps( asdict(value), sort_keys=True, separators=(",", ":"), allow_nan=False, default=lambda item: item.value if isinstance(item, Enum) else item.isoformat(), )