Move trip code into separate file
This commit is contained in:
parent
e0735b4185
commit
4a990a9fe5
180
agenda/trip.py
Normal file
180
agenda/trip.py
Normal file
|
|
@ -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
|
||||
180
web_view.py
180
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/<start>")
|
||||
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",
|
||||
|
|
|
|||
Loading…
Reference in a new issue