From b7d95f26ecd542e942180fb6ac4a7fc65dfb84a0 Mon Sep 17 00:00:00 2001 From: Edward Betts Date: Thu, 9 Jul 2026 10:11:47 +0100 Subject: [PATCH] Add car journey route generator --- agenda/car_journey_yaml.py | 756 ++++++++++++++++++++++++++++++ scripts/add-car-journeys-for-trip | 15 + tests/test_car_journey_yaml.py | 344 ++++++++++++++ 3 files changed, 1115 insertions(+) create mode 100644 agenda/car_journey_yaml.py create mode 100755 scripts/add-car-journeys-for-trip create mode 100644 tests/test_car_journey_yaml.py diff --git a/agenda/car_journey_yaml.py b/agenda/car_journey_yaml.py new file mode 100644 index 0000000..cbf360a --- /dev/null +++ b/agenda/car_journey_yaml.py @@ -0,0 +1,756 @@ +"""Generate car journey YAML and routes for trips.""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import typing +import unicodedata +from dataclasses import dataclass +from datetime import date +from pathlib import Path + +import openrouteservice # type: ignore[import-untyped] +import yaml + +from . import trip as trip_module +from . import utils + +PERSONAL_DATA_DIR = Path("~/src/personal-data").expanduser() +DEFAULT_HOME_LABEL = "PCH" +DEFAULT_PROFILE = "driving-car" +DEFAULT_SNAP_RADIUS_METRES = 5000 + +LatLon = tuple[float, float] +LonLat = tuple[float, float] +GeoJSON = dict[str, typing.Any] +StrDict = dict[str, typing.Any] +RouteFetcher = typing.Callable[[LonLat, LonLat], GeoJSON] + + +class NoAliasDumper(yaml.SafeDumper): + """YAML dumper that avoids anchors for repeated scalar values.""" + + def ignore_aliases(self, data: typing.Any) -> bool: + """Disable aliases for all values.""" + return True + + +class TripLike(typing.Protocol): + """Trip fields needed for car journey generation.""" + + start: date + accommodation: list[StrDict] + travel: list[StrDict] + + +@dataclass(frozen=True) +class CarJourney: + """A car journey ready to write to YAML and GeoJSON.""" + + trip: date + depart: date + arrive: date + route: str + start: LonLat + end: LonLat + + def as_yaml_item(self) -> StrDict: + """Return the YAML mapping for this journey.""" + return { + "trip": self.trip, + "depart": self.depart, + "arrive": self.arrive, + "route": self.route, + } + + +def latlon_to_lonlat(coord: LatLon) -> LonLat: + """Convert a lat/lon pair to lon/lat for routing APIs.""" + return (coord[1], coord[0]) + + +def route_label(value: str) -> str: + """Return a filesystem-safe route label.""" + ascii_value = ( + unicodedata.normalize( + "NFKD", + value.strip() + .replace("æ", "ae") + .replace("Æ", "AE") + .replace("ð", "d") + .replace("Ð", "D") + .replace("þ", "th") + .replace("Þ", "Th"), + ) + .encode("ascii", "ignore") + .decode("ascii") + ) + label = re.sub(r"[^A-Za-z0-9]+", "_", ascii_value).strip("_") + if not label: + raise ValueError(f"empty route label from {value!r}") + return label + + +def load_yaml_list(path: Path) -> list[StrDict]: + """Load a YAML list, returning an empty list when the file does not exist.""" + if not path.exists(): + return [] + loaded = yaml.safe_load(path.read_text()) + if loaded is None: + return [] + if isinstance(loaded, list) and all(isinstance(item, dict) for item in loaded): + return typing.cast(list[StrDict], loaded) + raise ValueError(f"{path} must contain a YAML list") + + +def dump_yaml_list_with_blank_lines(items: list[StrDict]) -> str: + """Dump a top-level YAML list using the personal-data spacing style.""" + if not items: + return "---\n" + text = yaml.dump( + items, sort_keys=False, allow_unicode=True, Dumper=NoAliasDumper + ).lstrip() + return "---\n" + text.replace("\n- ", "\n\n- ") + + +def car_route_path(data_dir: Path, route: str) -> Path: + """Return the GeoJSON path for a car route name.""" + route_filename = trip_module.route_filename_without_extension(route) + return data_dir / "car_routes" / f"{route_filename}.geojson" + + +def read_geojson(path: Path) -> GeoJSON: + """Read a GeoJSON file.""" + return typing.cast(GeoJSON, json.loads(path.read_text())) + + +def home_latlon_from_existing_routes(data_dir: Path, home_label: str) -> LatLon: + """Find the home coordinate from existing car routes.""" + route_dir = data_dir / "car_routes" + if not route_dir.exists(): + raise ValueError(f"missing car route directory: {route_dir}") + + home_route_prefix = f"{home_label}_to_" + home_route_suffix = f"_to_{home_label}" + + for path in sorted(route_dir.glob("*.geojson")): + geojson_data = read_geojson(path) + endpoints = trip_module.geojson_route_endpoints(geojson_data) + if endpoints is None: + continue + route_name = path.stem + if route_name.startswith(home_route_prefix): + return endpoints[0] + if route_name.endswith(home_route_suffix): + return endpoints[1] + + raise ValueError( + f"could not find home coordinate for {home_label!r} in {route_dir}" + ) + + +def trip_for_date(data_dir: Path, trip_date: date) -> TripLike: + """Return the trip with the given start date.""" + trips = typing.cast( + list[TripLike], trip_module.build_trip_list(data_dir=str(data_dir)) + ) + for item in trips: + if item.start == trip_date: + return item + raise ValueError(f"could not find trip starting {trip_date}") + + +def primary_accommodation(trip: TripLike) -> StrDict: + """Return the first accommodation for a trip.""" + if not trip.accommodation: + raise ValueError(f"trip {trip.start} has no accommodation") + return min(trip.accommodation, key=lambda item: utils.as_datetime(item["from"])) + + +def accommodation_label(accommodation: StrDict) -> str: + """Return the destination label for an accommodation.""" + label = accommodation.get("location") or accommodation.get("name") + if not isinstance(label, str): + raise ValueError("accommodation has no location or name") + return route_label(label) + + +def accommodation_latlon(accommodation: StrDict) -> LatLon: + """Return accommodation coordinates.""" + latitude = accommodation.get("latitude") + longitude = accommodation.get("longitude") + if not isinstance(latitude, (int, float)) or not isinstance( + longitude, (int, float) + ): + raise ValueError("accommodation is missing latitude/longitude") + return (float(latitude), float(longitude)) + + +def sorted_accommodation(trip: TripLike) -> list[StrDict]: + """Return trip accommodation sorted by check-in time.""" + return sorted(trip.accommodation, key=lambda item: utils.as_datetime(item["from"])) + + +def airport_latlon(airport: StrDict) -> LatLon: + """Return airport coordinates.""" + latitude = airport.get("latitude") + longitude = airport.get("longitude") + if not isinstance(latitude, (int, float)) or not isinstance( + longitude, (int, float) + ): + raise ValueError("airport is missing latitude/longitude") + return (float(latitude), float(longitude)) + + +def flight_departure_datetime(item: StrDict) -> typing.Any: + """Return flight departure value for sorting.""" + return item["depart"] + + +def trip_flights(trip: TripLike) -> list[StrDict]: + """Return trip flights sorted by departure time.""" + flights = [item for item in trip.travel if item.get("type") == "flight"] + if not flights: + raise ValueError(f"trip {trip.start} has no flights") + return sorted( + flights, key=lambda item: utils.as_datetime(flight_departure_datetime(item)) + ) + + +def trip_ferries(trip: TripLike) -> list[StrDict]: + """Return trip ferries sorted by departure time.""" + ferries = [item for item in trip.travel if item.get("type") == "ferry"] + if not ferries: + raise ValueError(f"trip {trip.start} has no ferries") + return sorted(ferries, key=lambda item: utils.as_datetime(item["depart"])) + + +def location_latlon(location: StrDict, label: str) -> LatLon: + """Return coordinates from a loaded travel location.""" + latitude = location.get("latitude") + longitude = location.get("longitude") + if not isinstance(latitude, (int, float)) or not isinstance( + longitude, (int, float) + ): + raise ValueError(f"{label} is missing latitude/longitude") + return (float(latitude), float(longitude)) + + +def location_name(location: StrDict, fallback: str) -> str: + """Return a route label source for a loaded travel location.""" + name = location.get("name") or fallback + if not isinstance(name, str): + raise ValueError(f"location name must be a string: {location!r}") + return name + + +def car_journeys_for_accommodation( + trip: TripLike, + data_dir: Path, + home_label: str = DEFAULT_HOME_LABEL, + destination_label: str | None = None, +) -> list[CarJourney]: + """Build outbound and return car journeys for a trip's primary accommodation.""" + accommodation = primary_accommodation(trip) + home = latlon_to_lonlat(home_latlon_from_existing_routes(data_dir, home_label)) + destination = latlon_to_lonlat(accommodation_latlon(accommodation)) + destination_route_label = destination_label or accommodation_label(accommodation) + outbound_route = f"{route_label(home_label)}_to_{destination_route_label}" + return_route = f"{destination_route_label}_to_{route_label(home_label)}" + + check_in = utils.as_date(accommodation["from"]) + check_out = utils.as_date(accommodation["to"]) + + return [ + CarJourney( + trip=trip.start, + depart=check_in, + arrive=check_in, + route=outbound_route, + start=home, + end=destination, + ), + CarJourney( + trip=trip.start, + depart=check_out, + arrive=check_out, + route=return_route, + start=destination, + end=home, + ), + ] + + +def car_journeys_for_ferry( + trip: TripLike, + data_dir: Path, + home_label: str = DEFAULT_HOME_LABEL, + destination_label: str | None = None, +) -> list[CarJourney]: + """Build car journeys around an outbound and return car ferry crossing.""" + ferries = trip_ferries(trip) + outbound_ferry = ferries[0] + return_ferry = ferries[-1] + accommodation = primary_accommodation(trip) + + home = latlon_to_lonlat(home_latlon_from_existing_routes(data_dir, home_label)) + accommodation_coord = latlon_to_lonlat(accommodation_latlon(accommodation)) + accommodation_route_label = destination_label or accommodation_label(accommodation) + + outbound_from_terminal = outbound_ferry.get("from_terminal") + outbound_to_terminal = outbound_ferry.get("to_terminal") + return_from_terminal = return_ferry.get("from_terminal") + return_to_terminal = return_ferry.get("to_terminal") + terminals = ( + outbound_from_terminal, + outbound_to_terminal, + return_from_terminal, + return_to_terminal, + ) + if not all(isinstance(terminal, dict) for terminal in terminals): + raise ValueError("ferry is missing terminal details") + + outbound_from = typing.cast(StrDict, outbound_from_terminal) + outbound_to = typing.cast(StrDict, outbound_to_terminal) + return_from = typing.cast(StrDict, return_from_terminal) + return_to = typing.cast(StrDict, return_to_terminal) + + home_route_label = route_label(home_label) + outbound_from_label = route_label( + location_name(outbound_from, typing.cast(str, outbound_ferry.get("from", ""))) + ) + outbound_to_label = route_label( + location_name(outbound_to, typing.cast(str, outbound_ferry.get("to", ""))) + ) + return_from_label = route_label( + location_name(return_from, typing.cast(str, return_ferry.get("from", ""))) + ) + return_to_label = route_label( + location_name(return_to, typing.cast(str, return_ferry.get("to", ""))) + ) + + outbound_from_coord = latlon_to_lonlat( + location_latlon(outbound_from, outbound_from_label) + ) + outbound_to_coord = latlon_to_lonlat( + location_latlon(outbound_to, outbound_to_label) + ) + return_from_coord = latlon_to_lonlat( + location_latlon(return_from, return_from_label) + ) + return_to_coord = latlon_to_lonlat(location_latlon(return_to, return_to_label)) + + return [ + CarJourney( + trip=trip.start, + depart=utils.as_date(outbound_ferry["depart"]), + arrive=utils.as_date(outbound_ferry["depart"]), + route=f"{home_route_label}_to_{outbound_from_label}", + start=home, + end=outbound_from_coord, + ), + CarJourney( + trip=trip.start, + depart=utils.as_date(outbound_ferry["arrive"]), + arrive=utils.as_date(outbound_ferry["arrive"]), + route=f"{outbound_to_label}_to_{accommodation_route_label}", + start=outbound_to_coord, + end=accommodation_coord, + ), + CarJourney( + trip=trip.start, + depart=utils.as_date(return_ferry["depart"]), + arrive=utils.as_date(return_ferry["depart"]), + route=f"{accommodation_route_label}_to_{return_from_label}", + start=accommodation_coord, + end=return_from_coord, + ), + CarJourney( + trip=trip.start, + depart=utils.as_date(return_ferry["arrive"]), + arrive=utils.as_date(return_ferry["arrive"]), + route=f"{return_to_label}_to_{home_route_label}", + start=return_to_coord, + end=home, + ), + ] + + +def car_journeys_for_airport( + trip: TripLike, + data_dir: Path, + home_label: str = DEFAULT_HOME_LABEL, +) -> list[CarJourney]: + """Build outbound and return car journeys for the trip's UK airport.""" + flights = trip_flights(trip) + outbound_flight = flights[0] + return_flight = flights[-1] + outbound_airport = outbound_flight.get("from_airport") + return_airport = return_flight.get("to_airport") + if not isinstance(outbound_airport, dict) or not isinstance(return_airport, dict): + raise ValueError("flight is missing airport details") + + outbound_iata = outbound_airport.get("iata") + return_iata = return_airport.get("iata") + if not isinstance(outbound_iata, str) or not isinstance(return_iata, str): + raise ValueError("flight airport is missing IATA code") + + home = latlon_to_lonlat(home_latlon_from_existing_routes(data_dir, home_label)) + outbound_destination = latlon_to_lonlat(airport_latlon(outbound_airport)) + return_start = latlon_to_lonlat(airport_latlon(return_airport)) + home_route_label = route_label(home_label) + outbound_route_label = route_label(outbound_iata) + return_route_label = route_label(return_iata) + + return [ + CarJourney( + trip=trip.start, + depart=utils.as_date(outbound_flight["depart"]), + arrive=utils.as_date(outbound_flight["depart"]), + route=f"{home_route_label}_to_{outbound_route_label}", + start=home, + end=outbound_destination, + ), + CarJourney( + trip=trip.start, + depart=utils.as_date(return_flight["arrive"]), + arrive=utils.as_date(return_flight["arrive"]), + route=f"{return_route_label}_to_{home_route_label}", + start=return_start, + end=home, + ), + ] + + +def country_code(item: StrDict) -> str | None: + """Return a lower-case country code from a YAML item.""" + country = item.get("country") + return country.casefold() if isinstance(country, str) else None + + +def accommodation_route_label(accommodation: StrDict) -> str: + """Return the route label for an accommodation.""" + return accommodation_label(accommodation) + + +def airport_iata(airport: StrDict) -> str: + """Return an airport IATA code.""" + iata = airport.get("iata") + if not isinstance(iata, str): + raise ValueError(f"airport missing IATA code: {airport!r}") + return iata + + +def flight_airport(flight: StrDict, field: str) -> StrDict: + """Return airport detail for a flight field.""" + airport = flight.get(field) + if not isinstance(airport, dict): + raise ValueError(f"flight missing {field}") + return typing.cast(StrDict, airport) + + +def car_journeys_for_fly_drive( + trip: TripLike, + data_dir: Path, + home_label: str = DEFAULT_HOME_LABEL, +) -> list[CarJourney]: + """Build car journeys for home-airport-hotel plus destination rental car legs.""" + flights = trip_flights(trip) + accommodations = sorted_accommodation(trip) + if not accommodations: + raise ValueError(f"trip {trip.start} has no accommodation") + + outbound_flight = flights[0] + return_flight = flights[-1] + outbound_destination_airport = flight_airport(outbound_flight, "to_airport") + return_origin_airport = flight_airport(return_flight, "from_airport") + return_destination_airport = flight_airport(return_flight, "to_airport") + home = latlon_to_lonlat(home_latlon_from_existing_routes(data_dir, home_label)) + home_route_label = route_label(home_label) + + preflight_accommodation = next( + ( + accommodation + for accommodation in accommodations + if country_code(accommodation) == "gb" + and utils.as_date(accommodation["from"]) + <= utils.as_date(outbound_flight["depart"]) + ), + None, + ) + destination_country = country_code(outbound_destination_airport) + rental_accommodations = [ + accommodation + for accommodation in accommodations + if country_code(accommodation) == destination_country + ] + if not rental_accommodations: + raise ValueError("trip has no destination-country accommodation") + + journeys: list[CarJourney] = [] + if preflight_accommodation is not None: + preflight_label = accommodation_route_label(preflight_accommodation) + preflight_coord = latlon_to_lonlat( + accommodation_latlon(preflight_accommodation) + ) + journeys.append( + CarJourney( + trip=trip.start, + depart=utils.as_date(preflight_accommodation["from"]), + arrive=utils.as_date(preflight_accommodation["from"]), + route=f"{home_route_label}_to_{preflight_label}", + start=home, + end=preflight_coord, + ) + ) + + destination_airport_label = route_label(airport_iata(outbound_destination_airport)) + destination_airport_coord = latlon_to_lonlat( + airport_latlon(outbound_destination_airport) + ) + first_accommodation = rental_accommodations[0] + first_label = accommodation_route_label(first_accommodation) + first_coord = latlon_to_lonlat(accommodation_latlon(first_accommodation)) + journeys.append( + CarJourney( + trip=trip.start, + depart=utils.as_date(outbound_flight["arrive"]), + arrive=utils.as_date(outbound_flight["arrive"]), + route=f"{destination_airport_label}_to_{first_label}", + start=destination_airport_coord, + end=first_coord, + ) + ) + + for previous, current in zip(rental_accommodations, rental_accommodations[1:]): + previous_label = accommodation_route_label(previous) + current_label = accommodation_route_label(current) + journeys.append( + CarJourney( + trip=trip.start, + depart=utils.as_date(previous["to"]), + arrive=utils.as_date(previous["to"]), + route=f"{previous_label}_to_{current_label}", + start=latlon_to_lonlat(accommodation_latlon(previous)), + end=latlon_to_lonlat(accommodation_latlon(current)), + ) + ) + + last_accommodation = rental_accommodations[-1] + return_origin_airport_label = route_label(airport_iata(return_origin_airport)) + journeys.append( + CarJourney( + trip=trip.start, + depart=utils.as_date(return_flight["depart"]), + arrive=utils.as_date(return_flight["depart"]), + route=f"{accommodation_route_label(last_accommodation)}_to_{return_origin_airport_label}", + start=latlon_to_lonlat(accommodation_latlon(last_accommodation)), + end=latlon_to_lonlat(airport_latlon(return_origin_airport)), + ) + ) + + return_destination_airport_label = route_label( + airport_iata(return_destination_airport) + ) + journeys.append( + CarJourney( + trip=trip.start, + depart=utils.as_date(return_flight["arrive"]), + arrive=utils.as_date(return_flight["arrive"]), + route=f"{return_destination_airport_label}_to_{home_route_label}", + start=latlon_to_lonlat(airport_latlon(return_destination_airport)), + end=home, + ) + ) + + return journeys + + +def car_journeys_for_trip( + trip: TripLike, + data_dir: Path, + destination: str, + home_label: str = DEFAULT_HOME_LABEL, + destination_label: str | None = None, +) -> list[CarJourney]: + """Build outbound and return car journeys for a trip.""" + if destination == "accommodation": + return car_journeys_for_accommodation( + trip, + data_dir, + home_label=home_label, + destination_label=destination_label, + ) + if destination == "airport": + return car_journeys_for_airport(trip, data_dir, home_label=home_label) + if destination == "ferry": + return car_journeys_for_ferry( + trip, + data_dir, + home_label=home_label, + destination_label=destination_label, + ) + if destination == "fly-drive": + return car_journeys_for_fly_drive(trip, data_dir, home_label=home_label) + raise ValueError(f"unknown car journey destination {destination!r}") + + +def openrouteservice_fetcher(api_key: str) -> RouteFetcher: + """Return a route fetcher backed by openrouteservice.""" + client = openrouteservice.Client(key=api_key) + + def fetch(start: LonLat, end: LonLat) -> GeoJSON: + return typing.cast( + GeoJSON, + client.directions( + [start, end], + profile=DEFAULT_PROFILE, + format="geojson", + instructions=False, + radiuses=[DEFAULT_SNAP_RADIUS_METRES, DEFAULT_SNAP_RADIUS_METRES], + ), + ) + + return fetch + + +def write_route_files( + data_dir: Path, + journeys: list[CarJourney], + fetch_route: RouteFetcher, + overwrite_routes: bool = False, +) -> int: + """Write missing route GeoJSON files and return count written.""" + route_dir = data_dir / "car_routes" + route_dir.mkdir(parents=True, exist_ok=True) + written = 0 + + for journey in journeys: + path = car_route_path(data_dir, journey.route) + if path.exists() and not overwrite_routes: + continue + route_geojson = fetch_route(journey.start, journey.end) + path.write_text(json.dumps(route_geojson, separators=(",", ":")) + "\n") + written += 1 + + return written + + +def journey_key(item: StrDict) -> tuple[date, date, date, str]: + """Return a stable key for duplicate detection and sorting.""" + route = item.get("route") + if not isinstance(route, str): + raise ValueError(f"car journey route must be a string: {item!r}") + return ( + utils.as_date(item["trip"]), + utils.as_date(item["depart"]), + utils.as_date(item["arrive"]), + trip_module.route_filename_without_extension(route), + ) + + +def import_car_journeys(data_dir: Path, journeys: list[CarJourney]) -> int: + """Insert car journeys into car_journeys.yaml and return count added.""" + path = data_dir / "car_journeys.yaml" + existing = load_yaml_list(path) + existing_keys = {journey_key(item) for item in existing} + new_items = [ + journey.as_yaml_item() + for journey in journeys + if journey_key(journey.as_yaml_item()) not in existing_keys + ] + if not new_items: + return 0 + + combined = existing + new_items + combined.sort(key=journey_key) + path.write_text(dump_yaml_list_with_blank_lines(combined)) + return len(new_items) + + +def add_car_journeys_for_trip( + trip_date: date, + data_dir: Path = PERSONAL_DATA_DIR, + destination: str = "accommodation", + home_label: str = DEFAULT_HOME_LABEL, + destination_label: str | None = None, + fetch_route: RouteFetcher | None = None, + overwrite_routes: bool = False, + dry_run: bool = False, +) -> tuple[list[CarJourney], int, int]: + """Add car journeys and routes for a trip.""" + trip = trip_for_date(data_dir, trip_date) + journeys = car_journeys_for_trip( + trip, + data_dir, + destination=destination, + home_label=home_label, + destination_label=destination_label, + ) + + if dry_run: + return (journeys, 0, 0) + + if fetch_route is None: + api_key = os.environ.get("ORS_API_KEY") + if not api_key: + raise ValueError("ORS_API_KEY must be set") + fetch_route = openrouteservice_fetcher(api_key) + + routes_written = write_route_files( + data_dir, journeys, fetch_route, overwrite_routes=overwrite_routes + ) + journeys_added = import_car_journeys(data_dir, journeys) + return (journeys, routes_written, journeys_added) + + +def parse_date(value: str) -> date: + """Parse an ISO date.""" + return date.fromisoformat(value) + + +def main(argv: list[str] | None = None) -> int: + """CLI entrypoint.""" + parser = argparse.ArgumentParser( + description=( + "Add car journeys from home to trip accommodation and back, " + "using openrouteservice for GeoJSON routes." + ) + ) + parser.add_argument("trip_date", type=parse_date) + parser.add_argument( + "--destination", + choices=("accommodation", "airport", "ferry", "fly-drive"), + default="accommodation", + help=( + "Route to accommodation check-in/out, the trip airport, " + "around an outbound/return car ferry, or a fly-drive chain." + ), + ) + parser.add_argument("--data-dir", default=str(PERSONAL_DATA_DIR)) + parser.add_argument("--home-label", default=DEFAULT_HOME_LABEL) + parser.add_argument("--destination-label") + parser.add_argument("--overwrite-routes", action="store_true") + parser.add_argument("--dry-run", action="store_true") + args = parser.parse_args(argv) + + journeys, routes_written, journeys_added = add_car_journeys_for_trip( + args.trip_date, + data_dir=Path(args.data_dir).expanduser(), + destination=args.destination, + home_label=args.home_label, + destination_label=args.destination_label, + overwrite_routes=args.overwrite_routes, + dry_run=args.dry_run, + ) + + for journey in journeys: + print(f"{journey.depart}: {journey.route}") + if args.dry_run: + return 0 + print(f"Wrote {routes_written} route file(s)") + print(f"Added {journeys_added} car journey YAML item(s)") + return 0 diff --git a/scripts/add-car-journeys-for-trip b/scripts/add-car-journeys-for-trip new file mode 100755 index 0000000..9117068 --- /dev/null +++ b/scripts/add-car-journeys-for-trip @@ -0,0 +1,15 @@ +#!/usr/bin/python3 + +import os +import sys + +SCRIPT_PATH = os.path.realpath(__file__) +SCRIPT_DIR = os.path.dirname(SCRIPT_PATH) +REPO_ROOT = os.path.dirname(SCRIPT_DIR) +if REPO_ROOT not in sys.path: + sys.path.insert(0, REPO_ROOT) + +from agenda.car_journey_yaml import main + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_car_journey_yaml.py b/tests/test_car_journey_yaml.py new file mode 100644 index 0000000..e746cab --- /dev/null +++ b/tests/test_car_journey_yaml.py @@ -0,0 +1,344 @@ +"""Tests for car journey YAML generation.""" + +import json +from datetime import date, datetime +from pathlib import Path + +import agenda.car_journey_yaml +from agenda.types import Trip + + +def write_home_route(data_dir: Path) -> None: + """Write a route that identifies PCH as the home coordinate.""" + route_dir = data_dir / "car_routes" + route_dir.mkdir() + (route_dir / "PCH_to_EMF.geojson").write_text( + json.dumps( + { + "type": "Feature", + "properties": {}, + "geometry": { + "type": "LineString", + "coordinates": [[-2.60283, 51.44083], [-2.37789, 52.038]], + }, + } + ) + ) + + +def test_car_journeys_for_trip_uses_accommodation_dates_and_locations( + tmp_path: Path, +) -> None: + """Build outbound and return journeys for the trip accommodation.""" + write_home_route(tmp_path) + trip = Trip( + start=date(2025, 4, 22), + accommodation=[ + { + "location": "Callestick", + "from": datetime(2025, 4, 22, 15, 0), + "to": datetime(2025, 4, 25, 10, 0), + "latitude": 50.307093, + "longitude": -5.13466, + } + ], + ) + + journeys = agenda.car_journey_yaml.car_journeys_for_trip( + trip, tmp_path, destination="accommodation" + ) + + assert [journey.route for journey in journeys] == [ + "PCH_to_Callestick", + "Callestick_to_PCH", + ] + assert [journey.depart for journey in journeys] == [ + date(2025, 4, 22), + date(2025, 4, 25), + ] + assert journeys[0].start == (-2.60283, 51.44083) + assert journeys[0].end == (-5.13466, 50.307093) + assert journeys[1].start == (-5.13466, 50.307093) + assert journeys[1].end == (-2.60283, 51.44083) + + +def test_car_journeys_for_trip_can_route_to_airport(tmp_path: Path) -> None: + """Build outbound and return journeys for airport access.""" + write_home_route(tmp_path) + trip = Trip( + start=date(2025, 11, 17), + travel=[ + { + "type": "flight", + "depart": datetime(2025, 11, 17, 17, 55), + "arrive": datetime(2025, 11, 18, 14, 50), + "from_airport": { + "iata": "LHR", + "latitude": 51.4775, + "longitude": -0.461389, + }, + "to_airport": { + "iata": "HKG", + "latitude": 22.308889, + "longitude": 113.914444, + }, + }, + { + "type": "flight", + "depart": datetime(2025, 11, 30, 22, 45), + "arrive": datetime(2025, 12, 1, 5, 35), + "from_airport": { + "iata": "HKG", + "latitude": 22.308889, + "longitude": 113.914444, + }, + "to_airport": { + "iata": "LHR", + "latitude": 51.4775, + "longitude": -0.461389, + }, + }, + ], + ) + + journeys = agenda.car_journey_yaml.car_journeys_for_trip( + trip, tmp_path, destination="airport" + ) + + assert [journey.route for journey in journeys] == ["PCH_to_LHR", "LHR_to_PCH"] + assert [journey.depart for journey in journeys] == [ + date(2025, 11, 17), + date(2025, 12, 1), + ] + assert journeys[0].start == (-2.60283, 51.44083) + assert journeys[0].end == (-0.461389, 51.4775) + assert journeys[1].start == (-0.461389, 51.4775) + assert journeys[1].end == (-2.60283, 51.44083) + + +def test_car_journeys_for_trip_can_route_around_ferry(tmp_path: Path) -> None: + """Build home, ferry terminal and accommodation driving legs.""" + write_home_route(tmp_path) + trip = Trip( + start=date(2025, 7, 4), + travel=[ + { + "type": "ferry", + "depart": datetime(2025, 7, 4, 22, 0), + "arrive": datetime(2025, 7, 5, 8, 0), + "from": "Plymouth", + "to": "Roscoff", + "from_terminal": { + "name": "Plymouth", + "latitude": 50.3651254, + "longitude": -4.1578164, + }, + "to_terminal": { + "name": "Roscoff", + "latitude": 48.721672, + "longitude": -3.966925, + }, + }, + { + "type": "ferry", + "depart": datetime(2025, 7, 21, 15, 0), + "arrive": datetime(2025, 7, 21, 20, 10), + "from": "Roscoff", + "to": "Plymouth", + "from_terminal": { + "name": "Roscoff", + "latitude": 48.721672, + "longitude": -3.966925, + }, + "to_terminal": { + "name": "Plymouth", + "latitude": 50.3651254, + "longitude": -4.1578164, + }, + }, + ], + accommodation=[ + { + "location": "Brest", + "from": datetime(2025, 7, 5, 16, 0), + "to": datetime(2025, 7, 21, 10, 0), + "latitude": 48.3645307, + "longitude": -4.5534685, + } + ], + ) + + journeys = agenda.car_journey_yaml.car_journeys_for_trip( + trip, tmp_path, destination="ferry" + ) + + assert [journey.route for journey in journeys] == [ + "PCH_to_Plymouth", + "Roscoff_to_Brest", + "Brest_to_Roscoff", + "Plymouth_to_PCH", + ] + assert [journey.depart for journey in journeys] == [ + date(2025, 7, 4), + date(2025, 7, 5), + date(2025, 7, 21), + date(2025, 7, 21), + ] + assert journeys[0].start == (-2.60283, 51.44083) + assert journeys[0].end == (-4.1578164, 50.3651254) + assert journeys[1].start == (-3.966925, 48.721672) + assert journeys[1].end == (-4.5534685, 48.3645307) + assert journeys[2].start == (-4.5534685, 48.3645307) + assert journeys[2].end == (-3.966925, 48.721672) + assert journeys[3].start == (-4.1578164, 50.3651254) + assert journeys[3].end == (-2.60283, 51.44083) + + +def test_car_journeys_for_trip_can_route_fly_drive_chain(tmp_path: Path) -> None: + """Build home, destination rental and return airport driving legs.""" + write_home_route(tmp_path) + trip = Trip( + start=date(2025, 9, 13), + travel=[ + { + "type": "flight", + "depart": datetime(2025, 9, 14, 6, 50), + "arrive": datetime(2025, 9, 14, 8, 55), + "from_airport": { + "iata": "LTN", + "latitude": 51.8747, + "longitude": -0.3683, + "country": "gb", + }, + "to_airport": { + "iata": "KEF", + "latitude": 63.985, + "longitude": -22.605556, + "country": "is", + }, + }, + { + "type": "flight", + "depart": datetime(2025, 10, 1, 9, 0), + "arrive": datetime(2025, 10, 1, 13, 0), + "from_airport": { + "iata": "KEF", + "latitude": 63.985, + "longitude": -22.605556, + "country": "is", + }, + "to_airport": { + "iata": "LTN", + "latitude": 51.8747, + "longitude": -0.3683, + "country": "gb", + }, + }, + ], + accommodation=[ + { + "location": "Luton Airport", + "country": "gb", + "from": datetime(2025, 9, 13, 15, 0), + "to": datetime(2025, 9, 14, 12, 0), + "latitude": 51.8745678, + "longitude": -0.3844256, + }, + { + "location": "Akranes", + "country": "is", + "from": datetime(2025, 9, 14, 16, 0), + "to": datetime(2025, 9, 16, 11, 0), + "latitude": 64.4344723, + "longitude": -21.5517961, + }, + { + "location": "Ísafjörður", + "country": "is", + "from": datetime(2025, 9, 16, 15, 0), + "to": datetime(2025, 10, 1, 11, 0), + "latitude": 66.072445, + "longitude": -23.120111, + }, + ], + ) + + journeys = agenda.car_journey_yaml.car_journeys_for_trip( + trip, tmp_path, destination="fly-drive" + ) + + assert [journey.route for journey in journeys] == [ + "PCH_to_Luton_Airport", + "KEF_to_Akranes", + "Akranes_to_Isafjordur", + "Isafjordur_to_KEF", + "LTN_to_PCH", + ] + assert [journey.depart for journey in journeys] == [ + date(2025, 9, 13), + date(2025, 9, 14), + date(2025, 9, 16), + date(2025, 10, 1), + date(2025, 10, 1), + ] + + +def test_write_route_files_skips_existing_routes(tmp_path: Path) -> None: + """Existing route files should not call the route API.""" + write_home_route(tmp_path) + journey = agenda.car_journey_yaml.CarJourney( + trip=date(2025, 4, 22), + depart=date(2025, 4, 22), + arrive=date(2025, 4, 22), + route="PCH_to_EMF", + start=(-2.60283, 51.44083), + end=(-2.37789, 52.038), + ) + + def fail_fetch( + _start: agenda.car_journey_yaml.LonLat, + _end: agenda.car_journey_yaml.LonLat, + ) -> agenda.car_journey_yaml.GeoJSON: + raise AssertionError("route API should not be called") + + written = agenda.car_journey_yaml.write_route_files(tmp_path, [journey], fail_fetch) + + assert written == 0 + + +def test_import_car_journeys_inserts_chronologically_and_is_idempotent( + tmp_path: Path, +) -> None: + """Generated journeys should be inserted without duplicates.""" + (tmp_path / "car_journeys.yaml").write_text("""--- +- trip: 2025-05-01 + depart: 2025-05-01 + arrive: 2025-05-01 + route: PCH_to_Later +""") + journeys = [ + agenda.car_journey_yaml.CarJourney( + trip=date(2025, 4, 22), + depart=date(2025, 4, 22), + arrive=date(2025, 4, 22), + route="PCH_to_Callestick", + start=(-2.60283, 51.44083), + end=(-5.13466, 50.307093), + ), + agenda.car_journey_yaml.CarJourney( + trip=date(2025, 4, 22), + depart=date(2025, 4, 25), + arrive=date(2025, 4, 25), + route="Callestick_to_PCH", + start=(-5.13466, 50.307093), + end=(-2.60283, 51.44083), + ), + ] + + added = agenda.car_journey_yaml.import_car_journeys(tmp_path, journeys) + added_again = agenda.car_journey_yaml.import_car_journeys(tmp_path, journeys) + + assert added == 2 + assert added_again == 0 + text = (tmp_path / "car_journeys.yaml").read_text() + assert text.index("PCH_to_Callestick") < text.index("PCH_to_Later")