313 lines
9.6 KiB
Python
313 lines
9.6 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
|
|
|
|
|
|
class UnknownStation(Exception):
|
|
"""Unknown station."""
|
|
|
|
pass
|
|
|
|
|
|
def load_travel(travel_type: str, data_dir: str) -> list[StrDict]:
|
|
"""Read flight and train journeys."""
|
|
items = travel.parse_yaml(travel_type + "s", data_dir)
|
|
for item in items:
|
|
item["type"] = travel_type
|
|
return items
|
|
|
|
|
|
def load_trains(
|
|
data_dir: str, route_distances: travel.RouteDistances | None = None
|
|
) -> list[StrDict]:
|
|
"""Load trains."""
|
|
trains = load_travel("train", data_dir)
|
|
stations = travel.parse_yaml("stations", data_dir)
|
|
by_name = {station["name"]: station for station in stations}
|
|
|
|
for train in trains:
|
|
if train["from"] not in by_name:
|
|
raise UnknownStation(train["from"])
|
|
if train["to"] not in by_name:
|
|
raise UnknownStation(train["to"])
|
|
train["from_station"] = by_name[train["from"]]
|
|
train["to_station"] = by_name[train["to"]]
|
|
|
|
for leg in train["legs"]:
|
|
assert leg["from"] in by_name
|
|
assert leg["to"] in by_name
|
|
leg["from_station"] = by_name[leg["from"]]
|
|
leg["to_station"] = by_name[leg["to"]]
|
|
|
|
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 depart_datetime(item: StrDict) -> datetime:
|
|
depart = item["depart"]
|
|
if isinstance(depart, datetime):
|
|
return depart
|
|
return datetime.combine(depart, time.min).replace(tzinfo=ZoneInfo("UTC"))
|
|
|
|
|
|
def load_flights(data_dir: str) -> list[StrDict]:
|
|
"""Load flights."""
|
|
flights = load_travel("flight", data_dir)
|
|
airlines = yaml.safe_load(open(os.path.join(data_dir, "airlines.yaml")))
|
|
airports = travel.parse_yaml("airports", data_dir)
|
|
for flight in flights:
|
|
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"] = airlines.get(flight["airline"], "[unknown]")
|
|
|
|
flight["distance"] = travel.flight_distance(flight)
|
|
return flights
|
|
|
|
|
|
def build_trip_list(
|
|
data_dir: str | None = None,
|
|
route_distances: travel.RouteDistances | None = None,
|
|
) -> list[Trip]:
|
|
"""Generate list of trips."""
|
|
trips: dict[date, Trip] = {}
|
|
|
|
if data_dir is None:
|
|
data_dir = flask.current_app.config["PERSONAL_DATA"]
|
|
|
|
yaml_trip_list = travel.parse_yaml("trips", data_dir)
|
|
|
|
yaml_trip_lookup = {item["trip"]: item for item in yaml_trip_list}
|
|
|
|
travel_items = sorted(
|
|
load_flights(data_dir) + load_trains(data_dir, route_distances=route_distances),
|
|
key=depart_datetime,
|
|
)
|
|
|
|
data = {
|
|
"travel": travel_items,
|
|
"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)
|
|
|
|
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 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 collect_trip_coordinates(trip: Trip) -> list[StrDict]:
|
|
"""Extract and deduplicate airport and station coordinates from trip."""
|
|
stations = {}
|
|
station_list = []
|
|
airports = {}
|
|
for t in trip.travel:
|
|
if t["type"] == "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"])
|
|
else:
|
|
assert t["type"] == "flight"
|
|
for field in "from_airport", "to_airport":
|
|
if field in t:
|
|
airports[t[field]["iata"]] = t[field]
|
|
|
|
for s in station_list:
|
|
if s["uic"] in stations:
|
|
continue
|
|
stations[s["uic"]] = s
|
|
|
|
coords = []
|
|
|
|
src = [
|
|
("accommodation", trip.accommodation),
|
|
("conference", trip.conferences),
|
|
("event", trip.events),
|
|
]
|
|
for coord_type, item_list in src:
|
|
coords += [
|
|
{
|
|
"name": item["name"],
|
|
"type": coord_type,
|
|
"latitude": item["latitude"],
|
|
"longitude": item["longitude"],
|
|
}
|
|
for item in item_list
|
|
if "latitude" in item and "longitude" in item
|
|
]
|
|
|
|
for coord_type, coord_dict in ("station", stations), ("airport", airports):
|
|
coords += [
|
|
{
|
|
"name": s["name"],
|
|
"type": coord_type,
|
|
"latitude": s["latitude"],
|
|
"longitude": s["longitude"],
|
|
}
|
|
for s in coord_dict.values()
|
|
]
|
|
|
|
return coords
|
|
|
|
|
|
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, "train_routes", filename + ".geojson")).read()
|
|
|
|
|
|
def get_trip_routes(trip: Trip) -> list[StrDict]:
|
|
"""Get routes for given trip to show on map."""
|
|
routes = []
|
|
seen_geojson = set()
|
|
for t in trip.travel:
|
|
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["uic"])
|
|
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": 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(item),
|
|
}
|
|
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]]:
|
|
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):
|
|
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)
|