diff --git a/agenda/trip.py b/agenda/trip.py new file mode 100644 index 0000000..6bc4a7d --- /dev/null +++ b/agenda/trip.py @@ -0,0 +1,180 @@ +import operator +import os +from datetime import date + +import flask + +from agenda import travel +from agenda.types import StrDict, Trip + + +def load_travel(travel_type: str) -> list[StrDict]: + """Read flight and train journeys.""" + data_dir = flask.current_app.config["PERSONAL_DATA"] + items = travel.parse_yaml(travel_type + "s", data_dir) + for item in items: + item["type"] = travel_type + return items + + +def load_trains() -> list[StrDict]: + """Load trains.""" + data_dir = flask.current_app.config["PERSONAL_DATA"] + + trains = load_travel("train") + stations = travel.parse_yaml("stations", data_dir) + by_name = {station["name"]: station for station in stations} + + for train in trains: + assert train["from"] in by_name + assert train["to"] in by_name + 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"]] + + return trains + + +def load_flights() -> list[StrDict]: + """Load flights.""" + data_dir = flask.current_app.config["PERSONAL_DATA"] + flights = load_travel("flight") + 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"]] + return flights + + +def build_trip_list() -> list[Trip]: + """Generate list of trips.""" + trips: dict[date, Trip] = {} + + data_dir = flask.current_app.config["PERSONAL_DATA"] + + travel_items = sorted( + load_flights() + load_trains(), key=operator.itemgetter("depart") + ) + + 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 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: + trips[start] = Trip(start=start) + getattr(trips[start], key).append(item) + + return [trip for _, trip in sorted(trips.items())] + + +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 + + return [ + { + "name": s["name"], + "type": "station", + "latitude": s["latitude"], + "longitude": s["longitude"], + } + for s in stations.values() + ] + [ + { + "name": s["name"], + "type": "airport", + "latitude": s["latitude"], + "longitude": s["longitude"], + } + for s in airports.values() + ] + + +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(filename: str) -> str: + """Read GeoJSON from file.""" + data_dir = flask.current_app.config["PERSONAL_DATA"] + 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"] + routes.append( + { + "type": "flight", + "from": latlon_tuple(fly_from), + "to": latlon_tuple(fly_to), + } + ) + + else: + 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"]) + if not geojson_filename: + routes.append( + { + "type": "train", + "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", + "geojson": read_geojson(geojson_filename), + } + ) + + return routes diff --git a/web_view.py b/web_view.py index c4557f3..9def860 100755 --- a/web_view.py +++ b/web_view.py @@ -17,8 +17,8 @@ import yaml import agenda.data import agenda.error_mail import agenda.thespacedevs +import agenda.trip from agenda import format_list_with_ampersand, travel -from agenda.types import StrDict, Trip app = flask.Flask(__name__) app.debug = False @@ -155,84 +155,10 @@ def accommodation_list() -> str: ) -def load_travel(travel_type: str) -> list[StrDict]: - """Read flight and train journeys.""" - data_dir = app.config["PERSONAL_DATA"] - items = travel.parse_yaml(travel_type + "s", data_dir) - for item in items: - item["type"] = travel_type - return items - - -def load_trains() -> list[StrDict]: - """Load trains.""" - data_dir = app.config["PERSONAL_DATA"] - - trains = load_travel("train") - stations = travel.parse_yaml("stations", data_dir) - by_name = {station["name"]: station for station in stations} - - for train in trains: - assert train["from"] in by_name - assert train["to"] in by_name - 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"]] - - return trains - - -def load_flights() -> list[StrDict]: - """Load flights.""" - data_dir = app.config["PERSONAL_DATA"] - flights = load_travel("flight") - 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"]] - return flights - - -def build_trip_list() -> list[Trip]: - """Generate list of trips.""" - trips: dict[date, Trip] = {} - - data_dir = app.config["PERSONAL_DATA"] - - travel_items = sorted( - load_flights() + load_trains(), key=operator.itemgetter("depart") - ) - - 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 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: - trips[start] = Trip(start=start) - getattr(trips[start], key).append(item) - - return [trip for _, trip in sorted(trips.items())] - - @app.route("/trip") def trip_list() -> str: """Page showing a list of trips.""" - trip_list = build_trip_list() + trip_list = agenda.trip.build_trip_list() today = date.today() current = [ @@ -255,106 +181,10 @@ def trip_list() -> str: ) -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 - - return [ - { - "name": s["name"], - "type": "station", - "latitude": s["latitude"], - "longitude": s["longitude"], - } - for s in stations.values() - ] + [ - { - "name": s["name"], - "type": "airport", - "latitude": s["latitude"], - "longitude": s["longitude"], - } - for s in airports.values() - ] - - -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(filename: str) -> str: - data_dir = app.config["PERSONAL_DATA"] - return open(os.path.join(data_dir, "train_routes", filename + ".geojson")).read() - - -def get_trip_routes(trip: Trip) -> list[StrDict]: - 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"] - routes.append( - { - "type": "flight", - "from": latlon_tuple(fly_from), - "to": latlon_tuple(fly_to), - } - ) - - else: - 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"]) - if not geojson_filename: - routes.append( - { - "type": "train", - "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", - "geojson": read_geojson(geojson_filename), - } - ) - - return routes - - @app.route("/trip/") def trip_page(start: str) -> str: """Individual trip page.""" - trip_list = build_trip_list() + trip_list = agenda.trip.build_trip_list() today = date.today() trip = next((trip for trip in trip_list if trip.start.isoformat() == start), None) @@ -362,8 +192,8 @@ def trip_page(start: str) -> str: if not trip: flask.abort(404) - coordinates = collect_trip_coordinates(trip) - routes = get_trip_routes(trip) + coordinates = agenda.trip.collect_trip_coordinates(trip) + routes = agenda.trip.get_trip_routes(trip) return flask.render_template( "trip_page.html",