"""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