Add car journey route generator

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

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())

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")