412 lines
13 KiB
Python
412 lines
13 KiB
Python
"""Trips."""
|
|
|
|
import decimal
|
|
import os
|
|
import typing
|
|
from datetime import date, datetime, time
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import flask
|
|
import yaml
|
|
|
|
from agenda import travel
|
|
from agenda.types import StrDict, Trip
|
|
|
|
|
|
def load_travel(travel_type: str, plural: str, data_dir: str) -> list[StrDict]:
|
|
"""Read flight and train journeys."""
|
|
items = 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)
|
|
|
|
if all("distance" in leg for leg in train["legs"]):
|
|
train["distance"] = sum(leg["distance"] 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)
|
|
|
|
geojson = from_terminal["routes"].get(item["to"])
|
|
if geojson:
|
|
item["geojson_filename"] = geojson
|
|
|
|
return ferries
|
|
|
|
|
|
def depart_datetime(item: StrDict) -> datetime:
|
|
"""Return a datetime for this travel item.
|
|
|
|
If the travel item already has a datetime return that, otherwise if the
|
|
departure time is just a date return midnight UTC for that date.
|
|
"""
|
|
depart = item["depart"]
|
|
if isinstance(depart, datetime):
|
|
return depart
|
|
return datetime.combine(depart, time.min).replace(tzinfo=ZoneInfo("UTC"))
|
|
|
|
|
|
def process_flight(
|
|
flight: StrDict, iata: dict[str, str], 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:
|
|
flight["airline_name"] = iata.get(flight["airline"], "[unknown]")
|
|
|
|
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")))
|
|
iata = {a["iata"]: a["name"] for a in airlines}
|
|
airports = travel.parse_yaml("airports", data_dir)
|
|
for booking in bookings:
|
|
for flight in booking["flights"]:
|
|
process_flight(flight, 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),
|
|
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]
|
|
) -> None:
|
|
"""Add coordinates for flights that haven't been booked yet."""
|
|
if not (
|
|
any(route["type"] == "unbooked_flight" for route in routes)
|
|
and not any(pin["type"] == "airport" for pin in coordinates)
|
|
):
|
|
return
|
|
|
|
data_dir = flask.current_app.config["PERSONAL_DATA"]
|
|
airports = typing.cast(dict[str, StrDict], travel.parse_yaml("airports", data_dir))
|
|
lhr = airports["LHR"]
|
|
coordinates.append(
|
|
{
|
|
"name": lhr["name"],
|
|
"type": "airport",
|
|
"latitude": lhr["latitude"],
|
|
"longitude": lhr["longitude"],
|
|
}
|
|
)
|
|
|
|
|
|
def get_locations(trip: Trip) -> dict[str, StrDict]:
|
|
"""Collect locations of all travel locations in trip."""
|
|
locations: dict[str, StrDict] = {
|
|
"station": {},
|
|
"airport": {},
|
|
"ferry_terminal": {},
|
|
}
|
|
|
|
station_list = []
|
|
for t in trip.travel:
|
|
match t["type"]:
|
|
case "train":
|
|
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"])
|
|
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
|
|
|
|
for s in station_list:
|
|
if s["name"] in locations["station"]:
|
|
continue
|
|
locations["station"][s["name"]] = s
|
|
|
|
return locations
|
|
|
|
|
|
def coordinate_dict(item: StrDict, coord_type: str) -> StrDict:
|
|
"""Build coodinate 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 latlon_tuple_prefer_airport(stop: StrDict, data_dir: str) -> tuple[float, float]:
|
|
airport_lookup = {
|
|
("Berlin", "de"): "BER",
|
|
("Hamburg", "de"): "HAM",
|
|
}
|
|
iata = airport_lookup.get((stop["location"], stop["country"]))
|
|
if not iata:
|
|
return latlon_tuple(stop)
|
|
|
|
airports = typing.cast(dict[str, StrDict], travel.parse_yaml("airports", data_dir))
|
|
return latlon_tuple(airports[iata])
|
|
|
|
|
|
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
|
|
assert 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
|
|
|
|
lhr = (51.4775, -0.461389)
|
|
|
|
return [
|
|
{
|
|
"type": "unbooked_flight",
|
|
"key": f'LHR_{item["location"]}_{item["country"]}',
|
|
"from": lhr,
|
|
"to": latlon_tuple_prefer_airport(item, data_dir),
|
|
}
|
|
for item in trip.conferences
|
|
if "latitude" in item
|
|
and "longitude" in item
|
|
and item["country"] not in ("gb", "be") # not flying to Belgium
|
|
]
|
|
|
|
|
|
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"])
|
|
|
|
for route in routes:
|
|
if "geojson_filename" in route:
|
|
route["geojson"] = read_geojson(data_dir, route.pop("geojson_filename"))
|
|
|
|
return (coordinates, routes)
|