1198 lines
40 KiB
Python
1198 lines
40 KiB
Python
"""Trips."""
|
|
|
|
import decimal
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import typing
|
|
import unicodedata
|
|
from datetime import date, datetime, timedelta, timezone
|
|
|
|
import flask
|
|
import pycountry
|
|
import yaml
|
|
from geopy.distance import geodesic # type: ignore
|
|
|
|
from agenda import conference, ical, travel, trip_schengen
|
|
from agenda.types import StrDict, Trip, TripElement
|
|
from agenda.utils import as_date, as_datetime, depart_datetime
|
|
|
|
|
|
class Airline(typing.TypedDict, total=False):
|
|
"""Airline."""
|
|
|
|
iata: str
|
|
icao: str
|
|
name: str
|
|
|
|
|
|
def load_flight_destination_rules(
|
|
data_dir: str,
|
|
) -> list[tuple[str, set[str]]]:
|
|
"""Load flight destination rules from personal data.
|
|
|
|
YAML schema:
|
|
- origin: BRS
|
|
airline: U2
|
|
destinations: [AGP, ALC]
|
|
"""
|
|
filename = os.path.join(data_dir, "flight_destinations.yaml")
|
|
if not os.path.exists(filename):
|
|
return []
|
|
|
|
raw = yaml.safe_load(open(filename))
|
|
if not isinstance(raw, list):
|
|
return []
|
|
|
|
rules: list[tuple[str, set[str]]] = []
|
|
for item in raw:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
from_iata = item.get("origin")
|
|
airline = item.get("airline")
|
|
destinations = item.get("destinations")
|
|
if (
|
|
not isinstance(from_iata, str)
|
|
or not isinstance(airline, str)
|
|
or not isinstance(destinations, list)
|
|
):
|
|
continue
|
|
destination_set = {d.upper() for d in destinations if isinstance(d, str)}
|
|
if destination_set:
|
|
rules.append((from_iata.upper(), destination_set))
|
|
return rules
|
|
|
|
|
|
def normalize_place_name(value: str) -> str:
|
|
"""Normalize place names for case/diacritics-insensitive matching."""
|
|
normalized = unicodedata.normalize("NFKD", value)
|
|
no_marks = "".join(ch for ch in normalized if not unicodedata.combining(ch))
|
|
return " ".join(no_marks.casefold().split())
|
|
|
|
|
|
def preferred_airport_iata(stop: StrDict, airports: dict[str, StrDict]) -> str | None:
|
|
"""Return preferred airport IATA for a stop from airport data."""
|
|
location = stop.get("location")
|
|
country = stop.get("country")
|
|
if not isinstance(location, str) or not isinstance(country, str):
|
|
return None
|
|
|
|
location_normalized = normalize_place_name(location)
|
|
country_lower = country.casefold()
|
|
match_scores: list[tuple[int, str]] = []
|
|
for iata, airport in airports.items():
|
|
airport_country = airport.get("country")
|
|
if (
|
|
not isinstance(airport_country, str)
|
|
or airport_country.casefold() != country_lower
|
|
):
|
|
continue
|
|
|
|
score = 0
|
|
city = airport.get("city")
|
|
if isinstance(city, str) and normalize_place_name(city) == location_normalized:
|
|
score = max(score, 3)
|
|
alt_name = airport.get("alt_name")
|
|
if (
|
|
isinstance(alt_name, str)
|
|
and normalize_place_name(alt_name) == location_normalized
|
|
):
|
|
score = max(score, 2)
|
|
name = airport.get("name")
|
|
if isinstance(name, str) and normalize_place_name(name) == location_normalized:
|
|
score = max(score, 1)
|
|
if score:
|
|
match_scores.append((score, iata))
|
|
|
|
if not match_scores:
|
|
return None
|
|
match_scores.sort(key=lambda item: (-item[0], item[1]))
|
|
return match_scores[0][1]
|
|
|
|
|
|
def get_unbooked_flight_origin_iata(
|
|
destination_iata: str | None, origin_rules: list[tuple[str, set[str]]]
|
|
) -> str:
|
|
"""Choose origin airport for an unbooked flight."""
|
|
if destination_iata:
|
|
for from_iata, destinations in origin_rules:
|
|
if destination_iata in destinations:
|
|
return from_iata
|
|
return "LHR"
|
|
|
|
|
|
UNBOOKED_RAIL_DESTINATIONS: dict[tuple[str, str], tuple[str, str]] = {
|
|
("fr", "paris"): ("London St Pancras", "Paris Gare du Nord"),
|
|
}
|
|
|
|
|
|
def load_station_lookup(data_dir: str) -> dict[str, StrDict]:
|
|
"""Load stations keyed by station name."""
|
|
stations = travel.parse_yaml("stations", data_dir)
|
|
return {station["name"]: station for station in stations}
|
|
|
|
|
|
def get_unbooked_rail_route(item: StrDict, data_dir: str) -> StrDict | None:
|
|
"""Return an assumed rail route for a conference without booked travel."""
|
|
location = item.get("location")
|
|
country = item.get("country")
|
|
if not isinstance(location, str) or not isinstance(country, str):
|
|
return None
|
|
|
|
station_names = UNBOOKED_RAIL_DESTINATIONS.get(
|
|
(country.casefold(), normalize_place_name(location))
|
|
)
|
|
if station_names is None:
|
|
return None
|
|
|
|
stations = load_station_lookup(data_dir)
|
|
from_station = stations.get(station_names[0])
|
|
to_station = stations.get(station_names[1])
|
|
if from_station is None or to_station is None:
|
|
return None
|
|
|
|
key = "_".join(["train"] + sorted([from_station["name"], to_station["name"]]))
|
|
route: StrDict = {"type": "train", "key": key}
|
|
geojson_filename = from_station.get("routes", {}).get(to_station["name"])
|
|
if geojson_filename:
|
|
route["geojson_filename"] = os.path.join("train_routes", geojson_filename)
|
|
else:
|
|
route["from"] = latlon_tuple(from_station)
|
|
route["to"] = latlon_tuple(to_station)
|
|
return route
|
|
|
|
|
|
def load_travel(travel_type: str, plural: str, data_dir: str) -> list[StrDict]:
|
|
"""Read flight and train journeys."""
|
|
items: list[StrDict] = travel.parse_yaml(plural, data_dir)
|
|
for item in items:
|
|
item["type"] = travel_type
|
|
return items
|
|
|
|
|
|
def add_station_objects(item: StrDict, by_name: dict[str, StrDict]) -> None:
|
|
"""Lookup stations and add to train or leg."""
|
|
item["from_station"] = by_name[item["from"]]
|
|
item["to_station"] = by_name[item["to"]]
|
|
|
|
|
|
def load_trains(
|
|
data_dir: str, route_distances: travel.RouteDistances | None = None
|
|
) -> list[StrDict]:
|
|
"""Load trains."""
|
|
trains = load_travel("train", "trains", data_dir)
|
|
stations = travel.parse_yaml("stations", data_dir)
|
|
by_name = {station["name"]: station for station in stations}
|
|
|
|
for train in trains:
|
|
add_station_objects(train, by_name)
|
|
for leg in train["legs"]:
|
|
add_station_objects(leg, by_name)
|
|
if route_distances:
|
|
travel.add_leg_route_distance(leg, route_distances)
|
|
|
|
# Calculate CO2 emissions for train leg (0.037 kg CO2e per passenger per km)
|
|
if "distance" in leg:
|
|
leg["co2_kg"] = leg["distance"] * 0.037
|
|
|
|
if all("distance" in leg for leg in train["legs"]):
|
|
train["distance"] = sum(leg["distance"] for leg in train["legs"])
|
|
# Calculate total CO2 for entire train journey
|
|
train["co2_kg"] = sum(leg["co2_kg"] for leg in train["legs"])
|
|
|
|
return trains
|
|
|
|
|
|
def load_ferries(
|
|
data_dir: str, route_distances: travel.RouteDistances | None = None
|
|
) -> list[StrDict]:
|
|
"""Load ferries."""
|
|
ferries = load_travel("ferry", "ferries", data_dir)
|
|
terminals = travel.parse_yaml("ferry_terminals", data_dir)
|
|
by_name = {terminal["name"]: terminal for terminal in terminals}
|
|
|
|
for item in ferries:
|
|
assert item["from"] in by_name and item["to"] in by_name
|
|
from_terminal, to_terminal = by_name[item["from"]], by_name[item["to"]]
|
|
item["from_terminal"] = from_terminal
|
|
item["to_terminal"] = to_terminal
|
|
|
|
if route_distances:
|
|
travel.add_leg_route_distance(item, route_distances)
|
|
|
|
# Calculate CO2 emissions for ferry (0.02254 kg CO2e per passenger per km)
|
|
if "distance" in item:
|
|
item["co2_kg"] = item["distance"] * 0.02254
|
|
|
|
geojson = from_terminal["routes"].get(item["to"])
|
|
if geojson:
|
|
item["geojson_filename"] = geojson
|
|
|
|
return ferries
|
|
|
|
|
|
def load_road_transport(
|
|
travel_type: str,
|
|
plural: str,
|
|
stops_yaml: str,
|
|
data_dir: str,
|
|
co2_factor: float,
|
|
route_distances: travel.RouteDistances | None = None,
|
|
) -> list[StrDict]:
|
|
"""Load road transport (bus or coach)."""
|
|
items = load_travel(travel_type, plural, data_dir)
|
|
stops = travel.parse_yaml(stops_yaml, data_dir)
|
|
by_name = {stop["name"]: stop for stop in stops}
|
|
|
|
for item in items:
|
|
add_station_objects(item, by_name)
|
|
if route_distances:
|
|
travel.add_leg_route_distance(item, route_distances)
|
|
if "distance" in item:
|
|
item["co2_kg"] = item["distance"] * co2_factor
|
|
from_station = item.get("from_station")
|
|
to_station = item.get("to_station")
|
|
if from_station and to_station:
|
|
# Support scalar or mapping routes: string or dict of stop name -> geojson filename
|
|
routes_val = from_station.get("routes", {})
|
|
if isinstance(routes_val, str):
|
|
geo: str | None = routes_val
|
|
else:
|
|
geo = routes_val.get(to_station.get("name"))
|
|
if geo:
|
|
item["geojson_filename"] = geo
|
|
|
|
return items
|
|
|
|
|
|
def load_coaches(
|
|
data_dir: str, route_distances: travel.RouteDistances | None = None
|
|
) -> list[StrDict]:
|
|
"""Load coaches."""
|
|
return load_road_transport(
|
|
"coach", "coaches", "coach_stations", data_dir, 0.027, route_distances
|
|
)
|
|
|
|
|
|
def load_buses(
|
|
data_dir: str, route_distances: travel.RouteDistances | None = None
|
|
) -> list[StrDict]:
|
|
"""Load buses."""
|
|
return load_road_transport(
|
|
"bus", "buses", "bus_stops", data_dir, 0.1, route_distances
|
|
)
|
|
|
|
|
|
def route_filename_without_extension(route: str) -> str:
|
|
"""Return route filename without a GeoJSON extension."""
|
|
return route.removesuffix(".geojson")
|
|
|
|
|
|
def line_strings_from_geojson(geojson_data: StrDict) -> list[list[list[float]]]:
|
|
"""Extract LineString coordinate arrays from GeoJSON."""
|
|
if geojson_data["type"] == "FeatureCollection":
|
|
features = typing.cast(list[StrDict], geojson_data["features"])
|
|
return [
|
|
line for feature in features for line in line_strings_from_geojson(feature)
|
|
]
|
|
|
|
if geojson_data["type"] == "Feature":
|
|
return line_strings_from_geojson(typing.cast(StrDict, geojson_data["geometry"]))
|
|
|
|
if geojson_data["type"] == "LineString":
|
|
return [typing.cast(list[list[float]], geojson_data["coordinates"])]
|
|
|
|
if geojson_data["type"] == "MultiLineString":
|
|
return typing.cast(list[list[list[float]]], geojson_data["coordinates"])
|
|
|
|
return []
|
|
|
|
|
|
def geojson_lonlat_to_latlon(coord: list[float]) -> tuple[float, float]:
|
|
"""Convert a GeoJSON lon/lat coordinate to a Leaflet lat/lon tuple."""
|
|
return (coord[1], coord[0])
|
|
|
|
|
|
def geojson_route_endpoints(
|
|
geojson_data: StrDict,
|
|
) -> tuple[tuple[float, float], tuple[float, float]] | None:
|
|
"""Return the first and last points of a GeoJSON route."""
|
|
lines = line_strings_from_geojson(geojson_data)
|
|
populated_lines = [line for line in lines if line]
|
|
if not populated_lines:
|
|
return None
|
|
return (
|
|
geojson_lonlat_to_latlon(populated_lines[0][0]),
|
|
geojson_lonlat_to_latlon(populated_lines[-1][-1]),
|
|
)
|
|
|
|
|
|
def geojson_route_distance_km(geojson_data: StrDict) -> float:
|
|
"""Calculate the length of a GeoJSON route in kilometres."""
|
|
total = 0.0
|
|
for line in line_strings_from_geojson(geojson_data):
|
|
for i in range(len(line) - 1):
|
|
total += float(
|
|
geodesic(
|
|
geojson_lonlat_to_latlon(line[i]),
|
|
geojson_lonlat_to_latlon(line[i + 1]),
|
|
).km
|
|
)
|
|
return total
|
|
|
|
|
|
def car_endpoint_type(name: str) -> str:
|
|
"""Choose a marker type for a car route endpoint."""
|
|
return (
|
|
"home"
|
|
if name.casefold() in {"home", "pch", "picture house court"}
|
|
else "car_stop"
|
|
)
|
|
|
|
|
|
def car_route_labels(item: StrDict) -> tuple[str | None, str | None]:
|
|
"""Return car route from/to labels from YAML or the route filename."""
|
|
from_label = item.get("from")
|
|
to_label = item.get("to")
|
|
if isinstance(from_label, str) and isinstance(to_label, str):
|
|
return (from_label, to_label)
|
|
|
|
route = item.get("route")
|
|
if not isinstance(route, str):
|
|
return (
|
|
from_label if isinstance(from_label, str) else None,
|
|
to_label if isinstance(to_label, str) else None,
|
|
)
|
|
|
|
route_stem = os.path.basename(route_filename_without_extension(route))
|
|
if "_to_" not in route_stem:
|
|
return (
|
|
from_label if isinstance(from_label, str) else None,
|
|
to_label if isinstance(to_label, str) else None,
|
|
)
|
|
|
|
start, end = route_stem.split("_to_", 1)
|
|
return (
|
|
from_label if isinstance(from_label, str) else start.replace("_", " "),
|
|
to_label if isinstance(to_label, str) else end.replace("_", " "),
|
|
)
|
|
|
|
|
|
def car_endpoint_marker_type(item: StrDict, direction: str, label: str) -> str:
|
|
"""Return the marker type for a car route endpoint."""
|
|
marker_type = item.get(direction + "_type")
|
|
return marker_type if isinstance(marker_type, str) else car_endpoint_type(label)
|
|
|
|
|
|
def show_car_endpoint_marker(item: StrDict, direction: str, label: str) -> bool:
|
|
"""Return whether to show a car endpoint marker on the map."""
|
|
direction_marker = item.get(direction + "_show_marker")
|
|
if isinstance(direction_marker, bool):
|
|
return direction_marker
|
|
|
|
show_markers = item.get("show_markers")
|
|
if isinstance(show_markers, bool):
|
|
return show_markers
|
|
|
|
return car_endpoint_type(label) == "home"
|
|
|
|
|
|
def load_cars(data_dir: str) -> list[StrDict]:
|
|
"""Load car journeys."""
|
|
filename = os.path.join(data_dir, "car_journeys.yaml")
|
|
if not os.path.exists(filename):
|
|
return []
|
|
|
|
items = load_travel("car", "car_journeys", data_dir)
|
|
for item in items:
|
|
route = item.get("route")
|
|
if not isinstance(route, str):
|
|
continue
|
|
|
|
route_filename = route_filename_without_extension(route)
|
|
item["geojson_filename"] = route_filename
|
|
|
|
with open(
|
|
os.path.join(data_dir, "car_routes", route_filename + ".geojson")
|
|
) as f:
|
|
geojson_data = typing.cast(StrDict, json.load(f))
|
|
|
|
if "distance" not in item:
|
|
item["distance"] = geojson_route_distance_km(geojson_data)
|
|
|
|
endpoints = geojson_route_endpoints(geojson_data)
|
|
from_label, to_label = car_route_labels(item)
|
|
if endpoints is None:
|
|
continue
|
|
|
|
for direction, field, label, endpoint in (
|
|
("from", "from_location", from_label, endpoints[0]),
|
|
("to", "to_location", to_label, endpoints[1]),
|
|
):
|
|
if not label or not show_car_endpoint_marker(item, direction, label):
|
|
continue
|
|
item[field] = {
|
|
"name": label,
|
|
"type": car_endpoint_marker_type(item, direction, label),
|
|
"latitude": endpoint[0],
|
|
"longitude": endpoint[1],
|
|
}
|
|
|
|
if from_label:
|
|
item["from"] = from_label
|
|
if to_label:
|
|
item["to"] = to_label
|
|
|
|
return items
|
|
|
|
|
|
def process_flight(
|
|
flight: StrDict, by_iata: dict[str, Airline], airports: list[StrDict]
|
|
) -> None:
|
|
"""Add airport detail, airline name and distance to flight."""
|
|
if flight["from"] in airports:
|
|
flight["from_airport"] = airports[flight["from"]]
|
|
if flight["to"] in airports:
|
|
flight["to_airport"] = airports[flight["to"]]
|
|
if "airline" in flight:
|
|
airline = by_iata[flight["airline"]]
|
|
flight["airline_detail"] = airline
|
|
flight["airline_code"] = airline[
|
|
"iata" if not airline.get("flight_number_prefer_icao") else "icao"
|
|
]
|
|
|
|
flight["distance"] = travel.flight_distance(flight)
|
|
|
|
|
|
def load_flight_bookings(data_dir: str) -> list[StrDict]:
|
|
"""Load flight bookings."""
|
|
bookings = load_travel("flight", "flights", data_dir)
|
|
airlines = yaml.safe_load(open(os.path.join(data_dir, "airlines.yaml")))
|
|
by_iata = {a["iata"]: a for a in airlines}
|
|
airports = travel.parse_yaml("airports", data_dir)
|
|
for booking in bookings:
|
|
for flight in booking["flights"]:
|
|
process_flight(flight, by_iata, airports)
|
|
return bookings
|
|
|
|
|
|
def load_flights(flight_bookings: list[StrDict]) -> list[StrDict]:
|
|
"""Load flights."""
|
|
flights = []
|
|
for booking in flight_bookings:
|
|
for flight in booking["flights"]:
|
|
for f in "type", "trip", "booking_reference", "price", "currency":
|
|
if f in booking:
|
|
flight[f] = booking[f]
|
|
flights.append(flight)
|
|
return flights
|
|
|
|
|
|
def collect_travel_items(
|
|
flight_bookings: list[StrDict],
|
|
data_dir: str | None = None,
|
|
route_distances: travel.RouteDistances | None = None,
|
|
) -> list[StrDict]:
|
|
"""Generate list of trips."""
|
|
if data_dir is None:
|
|
data_dir = flask.current_app.config["PERSONAL_DATA"]
|
|
|
|
return sorted(
|
|
load_flights(load_flight_bookings(data_dir))
|
|
+ load_trains(data_dir, route_distances=route_distances)
|
|
+ load_ferries(data_dir, route_distances=route_distances)
|
|
+ load_coaches(data_dir, route_distances=route_distances)
|
|
+ load_buses(data_dir, route_distances=route_distances)
|
|
+ load_cars(data_dir),
|
|
key=depart_datetime,
|
|
)
|
|
|
|
|
|
def group_travel_items_into_trips(
|
|
data: StrDict, yaml_trip_list: list[StrDict]
|
|
) -> list[Trip]:
|
|
"""Group travel items into trips."""
|
|
trips: dict[date, Trip] = {}
|
|
yaml_trip_lookup = {item["trip"]: item for item in yaml_trip_list}
|
|
for key, item_list in data.items():
|
|
assert isinstance(item_list, list)
|
|
for item in item_list:
|
|
if not (start := item.get("trip")):
|
|
continue
|
|
if start not in trips:
|
|
from_yaml = yaml_trip_lookup.get(start, {})
|
|
trips[start] = Trip(
|
|
start=start, **{k: v for k, v in from_yaml.items() if k != "trip"}
|
|
)
|
|
getattr(trips[start], key).append(item)
|
|
|
|
return [trip for _, trip in sorted(trips.items())]
|
|
|
|
|
|
def build_trip_list(
|
|
data_dir: str | None = None,
|
|
route_distances: travel.RouteDistances | None = None,
|
|
) -> list[Trip]:
|
|
"""Generate list of trips."""
|
|
if data_dir is None:
|
|
data_dir = flask.current_app.config["PERSONAL_DATA"]
|
|
|
|
yaml_trip_list = travel.parse_yaml("trips", data_dir)
|
|
|
|
flight_bookings = load_flight_bookings(data_dir)
|
|
conferences = travel.parse_yaml("conferences", data_dir)
|
|
for conf in conferences:
|
|
date_fields = conference.conference_date_fields(conf)
|
|
if "start" in date_fields:
|
|
conf["start"] = date_fields["start"]
|
|
conf["end"] = date_fields["end"]
|
|
|
|
data = {
|
|
"flight_bookings": flight_bookings,
|
|
"travel": collect_travel_items(flight_bookings, data_dir, route_distances),
|
|
"accommodation": travel.parse_yaml("accommodation", data_dir),
|
|
"conferences": conferences,
|
|
"events": travel.parse_yaml("events", data_dir),
|
|
}
|
|
|
|
for item in data["accommodation"]:
|
|
price = item.get("price")
|
|
if price:
|
|
item["price"] = decimal.Decimal(price)
|
|
|
|
return group_travel_items_into_trips(data, yaml_trip_list)
|
|
|
|
|
|
def add_coordinates_for_unbooked_flights(
|
|
routes: list[StrDict], coordinates: list[StrDict], data_dir: str
|
|
) -> None:
|
|
"""Add coordinates for flights that haven't been booked yet."""
|
|
if not any(route["type"] == "unbooked_flight" for route in routes):
|
|
return
|
|
|
|
airports = typing.cast(dict[str, StrDict], travel.parse_yaml("airports", data_dir))
|
|
existing_airport_names = {
|
|
typing.cast(str, pin["name"])
|
|
for pin in coordinates
|
|
if pin.get("type") == "airport" and isinstance(pin.get("name"), str)
|
|
}
|
|
iata_codes: set[str] = set()
|
|
for route in routes:
|
|
if route["type"] != "unbooked_flight":
|
|
continue
|
|
iata_codes.add(typing.cast(str, route.get("from_iata", "LHR")))
|
|
to_iata = route.get("to_iata")
|
|
if isinstance(to_iata, str):
|
|
iata_codes.add(to_iata)
|
|
|
|
for iata in sorted(iata_codes):
|
|
airport = airports.get(iata)
|
|
if not airport:
|
|
continue
|
|
airport_name = typing.cast(str, airport["name"])
|
|
if airport_name in existing_airport_names:
|
|
continue
|
|
coordinates.append(
|
|
{
|
|
"name": airport_name,
|
|
"type": "airport",
|
|
"latitude": airport["latitude"],
|
|
"longitude": airport["longitude"],
|
|
}
|
|
)
|
|
existing_airport_names.add(airport_name)
|
|
|
|
|
|
def stations_from_travel(t: StrDict) -> list[StrDict]:
|
|
"""Stations from train journey."""
|
|
station_list = [t["from_station"], t["to_station"]]
|
|
for leg in t["legs"]:
|
|
station_list.append(leg["from_station"])
|
|
station_list.append(leg["to_station"])
|
|
|
|
return station_list
|
|
|
|
|
|
def process_station_list(station_list: list[StrDict]) -> StrDict:
|
|
"""Proess sation list."""
|
|
stations = {}
|
|
for s in station_list:
|
|
if s["name"] in stations:
|
|
continue
|
|
stations[s["name"]] = s
|
|
return stations
|
|
|
|
|
|
def get_locations(trip: Trip) -> dict[str, StrDict]:
|
|
"""Collect locations of all travel locations in trip."""
|
|
locations: dict[str, StrDict] = {
|
|
"station": {},
|
|
"airport": {},
|
|
"ferry_terminal": {},
|
|
"coach_station": {},
|
|
"bus_stop": {},
|
|
"home": {},
|
|
"car_stop": {},
|
|
}
|
|
|
|
station_list = []
|
|
for t in trip.travel:
|
|
match t["type"]:
|
|
case "train":
|
|
station_list += stations_from_travel(t)
|
|
case "coach":
|
|
for field in ("from_station", "to_station"):
|
|
s = t[field]
|
|
locations["coach_station"][s["name"]] = s
|
|
case "bus":
|
|
for field in ("from_station", "to_station"):
|
|
s = t[field]
|
|
locations["bus_stop"][s["name"]] = s
|
|
case "flight":
|
|
for field in "from_airport", "to_airport":
|
|
if field in t:
|
|
locations["airport"][t[field]["iata"]] = t[field]
|
|
case "ferry":
|
|
for field in "from_terminal", "to_terminal":
|
|
terminal = t[field]
|
|
locations["ferry_terminal"][terminal["name"]] = terminal
|
|
case "car":
|
|
for field in ("from_location", "to_location"):
|
|
location = t.get(field)
|
|
if not location:
|
|
continue
|
|
location_type = location.get("type", "car_stop")
|
|
locations[location_type][location["name"]] = location
|
|
|
|
locations["station"] = process_station_list(station_list)
|
|
return locations
|
|
|
|
|
|
def coordinate_dict(item: StrDict, coord_type: str) -> StrDict:
|
|
"""Build coordinate dict for item."""
|
|
return {
|
|
"name": item["name"],
|
|
"type": coord_type,
|
|
"latitude": item["latitude"],
|
|
"longitude": item["longitude"],
|
|
}
|
|
|
|
|
|
def conference_free_days(trip: Trip) -> dict[str, tuple[int, int]]:
|
|
"""Return (days_before, days_after) exploration days for each conference.
|
|
|
|
Keyed by the conference start date as an ISO string. Days are relative to
|
|
the trip boundary or the adjacent conference's end/start for multi-conference
|
|
trips.
|
|
"""
|
|
if not trip.conferences or not trip.end:
|
|
return {}
|
|
|
|
def conf_attend_start(c: StrDict) -> date:
|
|
return as_date(c.get("attend_start") or c["start"])
|
|
|
|
def conf_attend_end(c: StrDict) -> date:
|
|
return as_date(c.get("attend_end") or c["end"])
|
|
|
|
sorted_confs = sorted(trip.conferences, key=conf_attend_start)
|
|
result: dict[str, tuple[int, int]] = {}
|
|
|
|
for i, conf in enumerate(sorted_confs):
|
|
before_boundary = conf_attend_end(sorted_confs[i - 1]) if i > 0 else trip.start
|
|
after_boundary = (
|
|
conf_attend_start(sorted_confs[i + 1])
|
|
if i < len(sorted_confs) - 1
|
|
else trip.end
|
|
)
|
|
days_before = (conf_attend_start(conf) - before_boundary).days
|
|
days_after = (after_boundary - conf_attend_end(conf)).days
|
|
result[str(conf["start"])] = (days_before, days_after)
|
|
|
|
return result
|
|
|
|
|
|
def collect_trip_coordinates(trip: Trip) -> list[StrDict]:
|
|
"""Extract and de-duplicate travel location coordinates from trip."""
|
|
coords = []
|
|
|
|
src = [
|
|
("accommodation", trip.accommodation),
|
|
("conference", trip.conferences),
|
|
("event", trip.events),
|
|
]
|
|
for coord_type, item_list in src:
|
|
coords += [
|
|
coordinate_dict(item, coord_type)
|
|
for item in item_list
|
|
if "latitude" in item and "longitude" in item
|
|
]
|
|
|
|
locations = get_locations(trip)
|
|
for coord_type, coord_dict in locations.items():
|
|
coords += [coordinate_dict(s, coord_type) for s in coord_dict.values()]
|
|
|
|
return coords
|
|
|
|
|
|
def destination_latlon(
|
|
stop: StrDict, airports: dict[str, StrDict]
|
|
) -> tuple[float, float] | None:
|
|
"""Resolve destination coordinates from airport lookup or explicit lat/lon."""
|
|
iata = preferred_airport_iata(stop, airports)
|
|
if iata:
|
|
airport = airports.get(iata)
|
|
if airport:
|
|
return latlon_tuple(airport)
|
|
if "latitude" in stop and "longitude" in stop:
|
|
return latlon_tuple(stop)
|
|
return None
|
|
|
|
|
|
def latlon_tuple(stop: StrDict) -> tuple[float, float]:
|
|
"""Given a transport stop return the lat/lon as a tuple."""
|
|
return (stop["latitude"], stop["longitude"])
|
|
|
|
|
|
def read_geojson(data_dir: str, filename: str) -> str:
|
|
"""Read GeoJSON from file."""
|
|
return open(os.path.join(data_dir, filename + ".geojson")).read()
|
|
|
|
|
|
def get_trip_routes(trip: Trip, data_dir: str) -> list[StrDict]:
|
|
"""Get routes for given trip to show on map."""
|
|
routes: list[StrDict] = []
|
|
seen_geojson = set()
|
|
for t in trip.travel:
|
|
if t["type"] == "ferry":
|
|
ferry_from, ferry_to = t["from_terminal"], t["to_terminal"]
|
|
|
|
key = "_".join(["ferry"] + sorted([ferry_from["name"], ferry_to["name"]]))
|
|
filename = os.path.join("ferry_routes", t["geojson_filename"])
|
|
|
|
routes.append(
|
|
{
|
|
"type": "train",
|
|
"key": key,
|
|
"geojson_filename": filename,
|
|
}
|
|
)
|
|
continue
|
|
if t["type"] == "flight":
|
|
if "from_airport" not in t or "to_airport" not in t:
|
|
continue
|
|
fly_from, fly_to = t["from_airport"], t["to_airport"]
|
|
key = "_".join(["flight"] + sorted([fly_from["iata"], fly_to["iata"]]))
|
|
routes.append(
|
|
{
|
|
"type": "flight",
|
|
"key": key,
|
|
"from": latlon_tuple(fly_from),
|
|
"to": latlon_tuple(fly_to),
|
|
}
|
|
)
|
|
continue
|
|
if t["type"] in ("coach", "bus"):
|
|
route_type = t["type"]
|
|
stop_from, stop_to = t["from_station"], t["to_station"]
|
|
key = "_".join([route_type] + sorted([stop_from["name"], stop_to["name"]]))
|
|
if t.get("geojson_filename"):
|
|
filename = os.path.join(f"{route_type}_routes", t["geojson_filename"])
|
|
routes.append(
|
|
{"type": route_type, "key": key, "geojson_filename": filename}
|
|
)
|
|
else:
|
|
routes.append(
|
|
{
|
|
"type": route_type,
|
|
"key": key,
|
|
"from": latlon_tuple(stop_from),
|
|
"to": latlon_tuple(stop_to),
|
|
}
|
|
)
|
|
continue
|
|
if t["type"] == "car":
|
|
route = t.get("geojson_filename")
|
|
if not isinstance(route, str):
|
|
continue
|
|
route_filename = route_filename_without_extension(route)
|
|
key = "_".join(
|
|
[
|
|
"car",
|
|
typing.cast(str, t.get("from", "")),
|
|
typing.cast(str, t.get("to", "")),
|
|
route_filename,
|
|
]
|
|
)
|
|
routes.append(
|
|
{
|
|
"type": "car",
|
|
"key": key,
|
|
"geojson_filename": os.path.join("car_routes", route_filename),
|
|
}
|
|
)
|
|
continue
|
|
if t["type"] == "train":
|
|
for leg in t["legs"]:
|
|
train_from, train_to = leg["from_station"], leg["to_station"]
|
|
geojson_filename = train_from.get("routes", {}).get(train_to["name"])
|
|
key = "_".join(
|
|
["train"] + sorted([train_from["name"], train_to["name"]])
|
|
)
|
|
if not geojson_filename:
|
|
routes.append(
|
|
{
|
|
"type": "train",
|
|
"key": key,
|
|
"from": latlon_tuple(train_from),
|
|
"to": latlon_tuple(train_to),
|
|
}
|
|
)
|
|
continue
|
|
|
|
if geojson_filename in seen_geojson:
|
|
continue
|
|
seen_geojson.add(geojson_filename)
|
|
|
|
routes.append(
|
|
{
|
|
"type": "train",
|
|
"key": key,
|
|
"geojson_filename": os.path.join(
|
|
"train_routes", geojson_filename
|
|
),
|
|
}
|
|
)
|
|
|
|
if routes:
|
|
return routes
|
|
|
|
airports = typing.cast(dict[str, StrDict], travel.parse_yaml("airports", data_dir))
|
|
origin_rules = load_flight_destination_rules(data_dir)
|
|
|
|
unbooked_routes = []
|
|
for item in trip.conferences:
|
|
unbooked_rail_route = get_unbooked_rail_route(item, data_dir)
|
|
if unbooked_rail_route is not None:
|
|
unbooked_routes.append(unbooked_rail_route)
|
|
continue
|
|
|
|
if item["country"] in {"gb", "be"}: # not flying to Belgium
|
|
continue
|
|
destination_iata = preferred_airport_iata(item, airports)
|
|
destination = destination_latlon(item, airports)
|
|
if not destination:
|
|
continue
|
|
from_iata = get_unbooked_flight_origin_iata(destination_iata, origin_rules)
|
|
from_airport = airports.get(from_iata)
|
|
if not from_airport:
|
|
from_airport = airports["LHR"]
|
|
from_iata = "LHR"
|
|
|
|
unbooked_routes.append(
|
|
{
|
|
"type": "unbooked_flight",
|
|
"key": f'{from_iata}_{item["location"]}_{item["country"]}',
|
|
"from_iata": from_iata,
|
|
"to_iata": destination_iata,
|
|
"from": latlon_tuple(from_airport),
|
|
"to": destination,
|
|
}
|
|
)
|
|
|
|
return unbooked_routes
|
|
|
|
|
|
def get_coordinates_and_routes(
|
|
trip_list: list[Trip], data_dir: str | None = None
|
|
) -> tuple[list[StrDict], list[StrDict]]:
|
|
"""Given a list of trips return the associated coordinates and routes."""
|
|
if data_dir is None:
|
|
data_dir = flask.current_app.config["PERSONAL_DATA"]
|
|
coordinates = []
|
|
seen_coordinates: set[tuple[str, str]] = set()
|
|
routes = []
|
|
seen_routes: set[str] = set()
|
|
for trip in trip_list:
|
|
for stop in collect_trip_coordinates(trip):
|
|
key = (stop["type"], stop["name"])
|
|
if key in seen_coordinates:
|
|
continue
|
|
coordinates.append(stop)
|
|
seen_coordinates.add(key)
|
|
|
|
for route in get_trip_routes(trip, data_dir):
|
|
if route["key"] in seen_routes:
|
|
continue
|
|
routes.append(route)
|
|
seen_routes.add(route["key"])
|
|
|
|
add_coordinates_for_unbooked_flights(routes, coordinates, data_dir)
|
|
|
|
for route in routes:
|
|
if "geojson_filename" in route:
|
|
route["geojson"] = read_geojson(data_dir, route.pop("geojson_filename"))
|
|
|
|
return (coordinates, routes)
|
|
|
|
|
|
def get_trip_list(
|
|
route_distances: travel.RouteDistances | None = None,
|
|
) -> list[Trip]:
|
|
"""Get list of trips respecting current authentication status."""
|
|
trips = [
|
|
trip
|
|
for trip in build_trip_list(route_distances=route_distances)
|
|
if flask.g.user.is_authenticated or not trip.private
|
|
]
|
|
|
|
# Add Schengen compliance information to each trip
|
|
for trip in trips:
|
|
trip_schengen.add_schengen_compliance_to_trip(trip)
|
|
|
|
return trips
|
|
|
|
|
|
DEFAULT_EVENT_DURATION = timedelta(hours=1)
|
|
|
|
|
|
def _ensure_utc(dt_value: datetime) -> datetime:
|
|
"""Ensure datetimes are timezone-aware in UTC."""
|
|
if dt_value.tzinfo is None:
|
|
return dt_value.replace(tzinfo=timezone.utc)
|
|
return dt_value.astimezone(timezone.utc)
|
|
|
|
|
|
def _event_datetimes(element: TripElement) -> tuple[datetime, datetime]:
|
|
"""Return start and end datetimes for a trip element."""
|
|
start_dt = as_datetime(element.start_time)
|
|
end_dt = as_datetime(element.end_time) if element.end_time else None
|
|
if end_dt is None or end_dt <= start_dt:
|
|
end_dt = start_dt + DEFAULT_EVENT_DURATION
|
|
return _ensure_utc(start_dt), _ensure_utc(end_dt)
|
|
|
|
|
|
def _event_all_day_dates(element: TripElement) -> tuple[date, date]:
|
|
"""Return start and exclusive end dates for all-day events."""
|
|
start_date = as_date(element.start_time)
|
|
end_source = element.end_time or element.start_time
|
|
end_date = as_date(end_source) + timedelta(days=1)
|
|
return start_date, end_date
|
|
|
|
|
|
def _trip_element_location(element: TripElement) -> str | None:
|
|
"""Derive a location string for the element."""
|
|
start_loc: str | None = element.start_loc
|
|
end_loc: str | None = element.end_loc
|
|
if start_loc and end_loc:
|
|
return f"{start_loc} → {end_loc}"
|
|
if start_loc:
|
|
return start_loc
|
|
if end_loc:
|
|
return end_loc
|
|
return None
|
|
|
|
|
|
def _trip_element_description(trip: Trip, element: TripElement) -> str:
|
|
"""Build a textual description for the element."""
|
|
lines = [
|
|
f"Trip: {trip.title}",
|
|
f"Type: {element.element_type}",
|
|
]
|
|
if element.element_type == "conference":
|
|
location_label = element.detail.get("location")
|
|
venue_label = element.detail.get("venue") or location_label
|
|
if place := _format_place(
|
|
location_label,
|
|
element.start_country,
|
|
element.element_type,
|
|
label_prefix="Location",
|
|
skip_country_if_in_label=True,
|
|
):
|
|
lines.append(place)
|
|
if place := _format_place(
|
|
venue_label,
|
|
element.end_country,
|
|
element.element_type,
|
|
label_prefix="Venue",
|
|
):
|
|
lines.append(place)
|
|
else:
|
|
if place := _format_place(
|
|
element.start_loc,
|
|
element.start_country,
|
|
element.element_type,
|
|
label_prefix="From",
|
|
):
|
|
lines.append(place)
|
|
if place := _format_place(
|
|
element.end_loc,
|
|
element.end_country,
|
|
element.element_type,
|
|
label_prefix="To",
|
|
):
|
|
lines.append(place)
|
|
lines.append(f"Trip start date: {_format_trip_start_date(trip)}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _format_place(
|
|
label: str | None,
|
|
country: pycountry.db.Country | None,
|
|
element_type: str,
|
|
label_prefix: str | None = None,
|
|
skip_country_if_in_label: bool = False,
|
|
) -> str | None:
|
|
"""Combine location label and country into a single string."""
|
|
parts: list[str] = []
|
|
formatted_label = label
|
|
if element_type == "train" and label:
|
|
stripped = label.strip()
|
|
if not stripped.lower().endswith("station"):
|
|
formatted_label = f"{stripped} station"
|
|
else:
|
|
formatted_label = stripped
|
|
if formatted_label:
|
|
parts.append(formatted_label)
|
|
if country:
|
|
include_country = True
|
|
if (element_type == "train" or skip_country_if_in_label) and formatted_label:
|
|
include_country = country.name.lower() not in formatted_label.lower()
|
|
if include_country:
|
|
parts.append(country.name)
|
|
if not parts:
|
|
return None
|
|
if label_prefix:
|
|
return f"{label_prefix}: {', '.join(parts)}"
|
|
return ", ".join(parts)
|
|
|
|
|
|
def _format_trip_start_date(trip: Trip) -> str:
|
|
"""Return trip start date in UK human-readable form with day name."""
|
|
day_name = trip.start.strftime("%A")
|
|
month_year = trip.start.strftime("%B %Y")
|
|
return f"{day_name} {trip.start.day} {month_year}"
|
|
|
|
|
|
def _flight_iata(detail: StrDict, key: str) -> str | None:
|
|
"""Extract an IATA code from flight detail."""
|
|
airport = typing.cast(StrDict | None, detail.get(key))
|
|
if not airport:
|
|
return None
|
|
return typing.cast(str | None, airport.get("iata"))
|
|
|
|
|
|
def _flight_label(detail: StrDict) -> str | None:
|
|
"""Return a compact label for flight elements."""
|
|
from_code = _flight_iata(detail, "from_airport")
|
|
to_code = _flight_iata(detail, "to_airport")
|
|
if from_code and to_code:
|
|
return f"{from_code}→{to_code}"
|
|
if from_code:
|
|
return from_code
|
|
if to_code:
|
|
return to_code
|
|
return None
|
|
|
|
|
|
def _trip_element_label(element: TripElement) -> str:
|
|
"""Return a succinct label describing the element."""
|
|
if element.element_type == "flight":
|
|
if label := _flight_label(element.detail):
|
|
return label
|
|
if element.element_type == "conference":
|
|
conference_name = element.detail.get("name")
|
|
if isinstance(conference_name, str):
|
|
return conference_name
|
|
|
|
start_loc = element.start_loc
|
|
end_loc = element.end_loc
|
|
if isinstance(start_loc, str) and isinstance(end_loc, str):
|
|
return f"{start_loc} → {end_loc}"
|
|
if isinstance(start_loc, str):
|
|
return start_loc
|
|
if isinstance(end_loc, str):
|
|
return end_loc
|
|
return element.title
|
|
|
|
|
|
def _trip_element_summary(trip: Trip, element: TripElement) -> str:
|
|
"""Build the calendar summary text."""
|
|
label = _trip_element_label(element)
|
|
if element.element_type == "conference":
|
|
return f"{element.element_type}: {label}"
|
|
return f"{element.element_type}: {label} [{trip.title}]"
|
|
|
|
|
|
def _trip_element_uid(trip: Trip, element: TripElement, index: int) -> str:
|
|
"""Generate a deterministic UID for calendar clients."""
|
|
raw = "|".join(
|
|
[
|
|
trip.start.isoformat(),
|
|
trip.title,
|
|
element.element_type,
|
|
element.title,
|
|
str(index),
|
|
]
|
|
)
|
|
digest = hashlib.sha1(raw.encode("utf-8")).hexdigest()
|
|
return f"trip-{digest}@agenda-codex"
|
|
|
|
|
|
def build_trip_ical(trips: list[Trip]) -> bytes:
|
|
"""Return an iCal feed containing all trip elements."""
|
|
lines = [
|
|
"BEGIN:VCALENDAR",
|
|
"VERSION:2.0",
|
|
"PRODID:-//Agenda Codex//Trips//EN",
|
|
"CALSCALE:GREGORIAN",
|
|
"METHOD:PUBLISH",
|
|
"X-WR-CALNAME:Trips",
|
|
]
|
|
generated = datetime.now(tz=timezone.utc)
|
|
|
|
for trip in trips:
|
|
for index, element in enumerate(trip.elements()):
|
|
lines.append("BEGIN:VEVENT")
|
|
ical.append_property(lines, "UID", _trip_element_uid(trip, element, index))
|
|
ical.append_property(lines, "DTSTAMP", ical.format_datetime_utc(generated))
|
|
if element.all_day:
|
|
start_date, end_date = _event_all_day_dates(element)
|
|
ical.append_property(
|
|
lines, "DTSTART;VALUE=DATE", ical.format_date(start_date)
|
|
)
|
|
ical.append_property(
|
|
lines, "DTEND;VALUE=DATE", ical.format_date(end_date)
|
|
)
|
|
else:
|
|
start_dt, end_dt = _event_datetimes(element)
|
|
ical.append_property(
|
|
lines, "DTSTART", ical.format_datetime_utc(start_dt)
|
|
)
|
|
ical.append_property(lines, "DTEND", ical.format_datetime_utc(end_dt))
|
|
summary = ical.escape_text(_trip_element_summary(trip, element))
|
|
ical.append_property(lines, "SUMMARY", summary)
|
|
|
|
description = ical.escape_text(_trip_element_description(trip, element))
|
|
ical.append_property(lines, "DESCRIPTION", description)
|
|
|
|
if location := _trip_element_location(element):
|
|
ical.append_property(lines, "LOCATION", ical.escape_text(location))
|
|
|
|
lines.append("END:VEVENT")
|
|
|
|
lines.append("END:VCALENDAR")
|
|
ical_text = "\r\n".join(lines) + "\r\n"
|
|
return ical_text.encode("utf-8")
|
|
|
|
|
|
def get_current_trip(today: date) -> Trip | None:
|
|
"""Get current trip."""
|
|
trip_list = get_trip_list(route_distances=None)
|
|
|
|
current = [
|
|
item
|
|
for item in trip_list
|
|
if item.start <= today and (item.end or item.start) >= today
|
|
]
|
|
assert len(current) < 2
|
|
return current[0] if current else None
|