917 lines
30 KiB
Python
917 lines
30 KiB
Python
"""Trips."""
|
|
|
|
import decimal
|
|
import hashlib
|
|
import os
|
|
import typing
|
|
import unicodedata
|
|
from datetime import date, datetime, timedelta, timezone
|
|
|
|
import flask
|
|
import pycountry
|
|
import yaml
|
|
|
|
from agenda import 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"
|
|
|
|
|
|
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 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),
|
|
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)
|
|
|
|
data = {
|
|
"flight_bookings": flight_bookings,
|
|
"travel": collect_travel_items(flight_bookings, data_dir, route_distances),
|
|
"accommodation": travel.parse_yaml("accommodation", data_dir),
|
|
"conferences": travel.parse_yaml("conferences", data_dir),
|
|
"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": {},
|
|
}
|
|
|
|
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
|
|
|
|
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 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"] == "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:
|
|
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
|