"""Per-point astronomical daylight snapshot; mountain occlusion is not calculated.""" from __future__ import annotations import argparse import csv import math from dataclasses import asdict, dataclass from datetime import datetime, timezone from pathlib import Path from typing import Sequence from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from mountain_twin.solar.sun_engine import solar_position @dataclass(frozen=True) class RoutePoint: route_id: str point_index: int track_index: int segment_index: int latitude: float longitude: float elevation_m: float | None @dataclass(frozen=True) class ExposurePoint: route_id: str point_index: int track_index: int segment_index: int latitude: float longitude: float elevation_m: float | None analysis_time_local: str analysis_time_utc: str timezone: str solar_azimuth_deg: float solar_elevation_deg: float sun_above_astronomical_horizon: bool astronomical_status: str terrain_shadow_status: str def resolve_local_time(value: str, zone_name: str, fold: int | None = None) -> datetime: """Reject DST gaps and require an explicit fold for repeated local times.""" local = datetime.fromisoformat(value) if local.tzinfo is not None: raise ValueError("supply a local date/time without an offset and an IANA timezone") zone = ZoneInfo(zone_name) candidates = [] for choice in (0, 1): aware = local.replace(tzinfo=zone, fold=choice) back = aware.astimezone(timezone.utc).astimezone(zone) if back.replace(tzinfo=None) == local and back.fold == choice: candidates.append(aware) if not candidates: raise ValueError("nonexistent local time during DST transition") if len(candidates) == 2 and fold is None: raise ValueError("ambiguous local time: supply --fold 0 or --fold 1") if fold is not None and fold not in (0, 1): raise ValueError("fold must be 0 or 1") return candidates[fold or 0] if len(candidates) == 2 else candidates[0] def load_day1(path: Path) -> list[RoutePoint]: """Read generated geometry only; source GPX timestamps are never hiking times.""" points = [] with path.open(newline="", encoding="utf-8") as handle: for row in csv.DictReader(handle): if row["route_id"] != "tmb_day_01": continue if row["assigned_hike_date"] != "2026-07-02": raise ValueError("TMB Day 1 assigned hiking date must be 2026-07-02") point = RoutePoint( row["route_id"], int(row["point_index"]), int(row["track_index"]), int(row["segment_index"]), float(row["lat"]), float(row["lon"]), float(row["elevation_m_raw"]) if row["elevation_m_raw"] else None, ) if not (-90 <= point.latitude <= 90 and -180 <= point.longitude <= 180): raise ValueError("invalid route coordinates") if point.elevation_m is not None and not math.isfinite(point.elevation_m): raise ValueError("invalid route elevation") if point.point_index != len(points): raise ValueError("route point indices must be contiguous and ordered from zero") points.append(point) if not points: raise ValueError("generated input contains no TMB Day 1 points") return points def analyze(points: Sequence[RoutePoint], instant: datetime) -> list[ExposurePoint]: """Evaluate every point at one instant, using geometric solar center > 0 degrees.""" if instant.utcoffset() is None: raise ValueError("analysis time must be timezone-aware") results = [] for point in points: elevation, azimuth = solar_position(instant, point.latitude, point.longitude) above = elevation > 0 results.append( ExposurePoint( **asdict(point), analysis_time_local=instant.isoformat(), analysis_time_utc=instant.astimezone(timezone.utc).isoformat(), timezone=getattr(instant.tzinfo, "key", str(instant.tzinfo)), solar_azimuth_deg=azimuth, solar_elevation_deg=elevation, sun_above_astronomical_horizon=above, astronomical_status="astronomical_daylight" if above else "astronomical_night", terrain_shadow_status="terrain_shadow_unknown", ) ) return results def write_csv(results: Sequence[ExposurePoint], output: Path) -> None: if not results: raise ValueError("no analysis results") output.parent.mkdir(parents=True, exist_ok=True) # Exclusive creation protects generated route artifacts and previous analyses. with output.open("x", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=list(asdict(results[0])), lineterminator="\n") writer.writeheader() for result in results: writer.writerow(asdict(result)) def main(argv: Sequence[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--input", type=Path, default=Path("data/generated/trail_points_master.csv") ) parser.add_argument("--date", default="2026-07-02", help="Local analysis date") parser.add_argument("--time", required=True, help="Local time, e.g. 12:00:00") parser.add_argument("--timezone", default="Europe/Paris", help="IANA timezone") parser.add_argument("--fold", type=int, choices=(0, 1)) parser.add_argument("--output", type=Path) args = parser.parse_args(argv) try: instant = resolve_local_time(f"{args.date}T{args.time}", args.timezone, args.fold) output = args.output or Path("data/generated/exposure") / ( f"tmb_day_01_{instant.astimezone(timezone.utc):%Y%m%dT%H%M%S%fZ}.csv" ) root = Path("data/generated").resolve() if root not in output.resolve().parents: raise ValueError("analysis output must be under data/generated/") results = analyze(load_day1(args.input), instant) write_csv(results, output) print(f"Wrote {len(results)} points to {output}; terrain shadow unknown.") return 0 except (OSError, ValueError, KeyError, ZoneInfoNotFoundError) as exc: print(f"ERROR: {exc}") return 2 if __name__ == "__main__": raise SystemExit(main())