147 lines
4.2 KiB
Python
147 lines
4.2 KiB
Python
"""Travel."""
|
|
|
|
import decimal
|
|
import json
|
|
import os
|
|
import typing
|
|
|
|
import flask
|
|
import yaml
|
|
from geopy.distance import geodesic
|
|
|
|
from .types import Event, StrDict
|
|
|
|
Leg = dict[str, str]
|
|
|
|
TravelList = list[dict[str, typing.Any]]
|
|
|
|
RouteDistances = dict[tuple[str, str], float]
|
|
|
|
|
|
def coords(airport: StrDict) -> tuple[float, float]:
|
|
"""Longitude / Latitude as coordinate tuples."""
|
|
# return (airport["longitude"], airport["latitude"])
|
|
return (airport["latitude"], airport["longitude"])
|
|
|
|
|
|
def flight_distance(f: StrDict) -> float:
|
|
"""Distance of flight."""
|
|
return float(geodesic(coords(f["from_airport"]), coords(f["to_airport"])).km)
|
|
|
|
|
|
def route_distances_as_json(route_distances: RouteDistances) -> str:
|
|
"""Format route distances as JSON string."""
|
|
return (
|
|
"[\n"
|
|
+ ",\n".join(
|
|
" " + json.dumps([s1, s2, dist])
|
|
for (s1, s2), dist in route_distances.items()
|
|
)
|
|
+ "\n]"
|
|
)
|
|
|
|
|
|
def parse_yaml(travel_type: str, data_dir: str) -> TravelList:
|
|
"""Parse flights YAML and return list of travel."""
|
|
filepath = os.path.join(data_dir, travel_type + ".yaml")
|
|
items: TravelList = yaml.safe_load(open(filepath))
|
|
if not all(isinstance(item, dict) for item in items):
|
|
return items
|
|
|
|
for item in items:
|
|
price = item.get("price")
|
|
if price:
|
|
item["price"] = decimal.Decimal(price)
|
|
|
|
return items
|
|
|
|
|
|
def get_flights(data_dir: str) -> list[Event]:
|
|
"""Get travel events."""
|
|
bookings = parse_yaml("flights", data_dir)
|
|
events = []
|
|
for booking in bookings:
|
|
for item in booking["flights"]:
|
|
if not item["depart"].date():
|
|
continue
|
|
e = Event(
|
|
date=item["depart"],
|
|
end_date=item.get("arrive"),
|
|
name="transport",
|
|
title=f'✈️ {item["from"]} to {item["to"]} ({flight_number(item)})',
|
|
url=(item.get("url") if flask.g.user.is_authenticated else None),
|
|
)
|
|
events.append(e)
|
|
return events
|
|
|
|
|
|
def get_trains(data_dir: str) -> list[Event]:
|
|
"""Get train events."""
|
|
events: list[Event] = []
|
|
for item in parse_yaml("trains", data_dir):
|
|
events += [
|
|
Event(
|
|
date=leg["depart"],
|
|
end_date=leg["arrive"],
|
|
name="transport",
|
|
title=f'🚆 {leg["from"]} to {leg["to"]}',
|
|
url=(item.get("url") if flask.g.user.is_authenticated else None),
|
|
)
|
|
for leg in item["legs"]
|
|
]
|
|
return events
|
|
|
|
|
|
def flight_number(flight: Leg) -> str:
|
|
"""Flight number."""
|
|
airline_code = flight["airline"]
|
|
# make sure this is the airline code, not the airline name
|
|
assert " " not in airline_code and not any(c.islower() for c in airline_code)
|
|
|
|
return airline_code + flight["flight_number"]
|
|
|
|
|
|
def all_events(data_dir: str) -> list[Event]:
|
|
"""Get all flights and rail journeys."""
|
|
return get_trains(data_dir) + get_flights(data_dir)
|
|
|
|
|
|
def train_leg_distance(geojson_data: StrDict) -> float:
|
|
"""Calculate the total length of a LineString in kilometers from GeoJSON data."""
|
|
# Extract coordinates
|
|
first_object = geojson_data["features"][0]["geometry"]
|
|
assert first_object["type"] in ("LineString", "MultiLineString")
|
|
|
|
if first_object["type"] == "LineString":
|
|
coord_list = [first_object["coordinates"]]
|
|
else:
|
|
first_object["type"] == "MultiLineString"
|
|
coord_list = first_object["coordinates"]
|
|
|
|
total_length_km = 0.0
|
|
|
|
for coordinates in coord_list:
|
|
total_length_km += sum(
|
|
float(geodesic(coordinates[i], coordinates[i + 1]).km)
|
|
for i in range(len(coordinates) - 1)
|
|
)
|
|
|
|
return total_length_km
|
|
|
|
|
|
def load_route_distances(data_dir: str) -> RouteDistances:
|
|
"""Load cache of route distances."""
|
|
route_distances: RouteDistances = {}
|
|
with open(os.path.join(data_dir, "route_distances.json")) as f:
|
|
for s1, s2, dist in json.load(f):
|
|
route_distances[(s1, s2)] = dist
|
|
|
|
return route_distances
|
|
|
|
|
|
def add_leg_route_distance(leg: StrDict, route_distances: RouteDistances) -> None:
|
|
s1, s2 = sorted([leg["from"], leg["to"]])
|
|
dist = route_distances.get((s1, s2))
|
|
if dist:
|
|
leg["distance"] = dist
|