Add car journeys to trips
This commit is contained in:
parent
bc7a2e89d0
commit
c6cb06b71e
9 changed files with 430 additions and 10 deletions
|
|
@ -94,7 +94,7 @@ def _require_date_value(value: typing.Any, field_name: str) -> DateOrDateTime:
|
|||
|
||||
def _require_date_only(value: typing.Any, field_name: str) -> date:
|
||||
"""Return field value as a date or raise ValueError."""
|
||||
return typing.cast(date, utils.as_date(_require_date_value(value, field_name)))
|
||||
return utils.as_date(_require_date_value(value, field_name))
|
||||
|
||||
|
||||
def conference_date_fields(item: StrDict) -> StrDict:
|
||||
|
|
|
|||
185
agenda/trip.py
185
agenda/trip.py
|
|
@ -2,6 +2,7 @@
|
|||
|
||||
import decimal
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import typing
|
||||
import unicodedata
|
||||
|
|
@ -10,6 +11,7 @@ from datetime import date, datetime, timedelta, timezone
|
|||
import flask
|
||||
import pycountry
|
||||
import yaml
|
||||
from geopy.distance import geodesic # type: ignore
|
||||
|
||||
from agenda import conference, ical, travel, trip_schengen
|
||||
from agenda.types import StrDict, Trip, TripElement
|
||||
|
|
@ -281,6 +283,156 @@ def load_buses(
|
|||
)
|
||||
|
||||
|
||||
def route_filename_without_extension(route: str) -> str:
|
||||
"""Return route filename without a GeoJSON extension."""
|
||||
return route.removesuffix(".geojson")
|
||||
|
||||
|
||||
def line_strings_from_geojson(geojson_data: StrDict) -> list[list[list[float]]]:
|
||||
"""Extract LineString coordinate arrays from GeoJSON."""
|
||||
if geojson_data["type"] == "FeatureCollection":
|
||||
features = typing.cast(list[StrDict], geojson_data["features"])
|
||||
return [
|
||||
line for feature in features for line in line_strings_from_geojson(feature)
|
||||
]
|
||||
|
||||
if geojson_data["type"] == "Feature":
|
||||
return line_strings_from_geojson(typing.cast(StrDict, geojson_data["geometry"]))
|
||||
|
||||
if geojson_data["type"] == "LineString":
|
||||
return [typing.cast(list[list[float]], geojson_data["coordinates"])]
|
||||
|
||||
if geojson_data["type"] == "MultiLineString":
|
||||
return typing.cast(list[list[list[float]]], geojson_data["coordinates"])
|
||||
|
||||
return []
|
||||
|
||||
|
||||
def geojson_lonlat_to_latlon(coord: list[float]) -> tuple[float, float]:
|
||||
"""Convert a GeoJSON lon/lat coordinate to a Leaflet lat/lon tuple."""
|
||||
return (coord[1], coord[0])
|
||||
|
||||
|
||||
def geojson_route_endpoints(
|
||||
geojson_data: StrDict,
|
||||
) -> tuple[tuple[float, float], tuple[float, float]] | None:
|
||||
"""Return the first and last points of a GeoJSON route."""
|
||||
lines = line_strings_from_geojson(geojson_data)
|
||||
populated_lines = [line for line in lines if line]
|
||||
if not populated_lines:
|
||||
return None
|
||||
return (
|
||||
geojson_lonlat_to_latlon(populated_lines[0][0]),
|
||||
geojson_lonlat_to_latlon(populated_lines[-1][-1]),
|
||||
)
|
||||
|
||||
|
||||
def geojson_route_distance_km(geojson_data: StrDict) -> float:
|
||||
"""Calculate the length of a GeoJSON route in kilometres."""
|
||||
total = 0.0
|
||||
for line in line_strings_from_geojson(geojson_data):
|
||||
for i in range(len(line) - 1):
|
||||
total += float(
|
||||
geodesic(
|
||||
geojson_lonlat_to_latlon(line[i]),
|
||||
geojson_lonlat_to_latlon(line[i + 1]),
|
||||
).km
|
||||
)
|
||||
return total
|
||||
|
||||
|
||||
def car_endpoint_type(name: str) -> str:
|
||||
"""Choose a marker type for a car route endpoint."""
|
||||
return (
|
||||
"home"
|
||||
if name.casefold() in {"home", "pch", "picture house court"}
|
||||
else "car_stop"
|
||||
)
|
||||
|
||||
|
||||
def car_route_labels(item: StrDict) -> tuple[str | None, str | None]:
|
||||
"""Return car route from/to labels from YAML or the route filename."""
|
||||
from_label = item.get("from")
|
||||
to_label = item.get("to")
|
||||
if isinstance(from_label, str) and isinstance(to_label, str):
|
||||
return (from_label, to_label)
|
||||
|
||||
route = item.get("route")
|
||||
if not isinstance(route, str):
|
||||
return (
|
||||
from_label if isinstance(from_label, str) else None,
|
||||
to_label if isinstance(to_label, str) else None,
|
||||
)
|
||||
|
||||
route_stem = os.path.basename(route_filename_without_extension(route))
|
||||
if "_to_" not in route_stem:
|
||||
return (
|
||||
from_label if isinstance(from_label, str) else None,
|
||||
to_label if isinstance(to_label, str) else None,
|
||||
)
|
||||
|
||||
start, end = route_stem.split("_to_", 1)
|
||||
return (
|
||||
from_label if isinstance(from_label, str) else start.replace("_", " "),
|
||||
to_label if isinstance(to_label, str) else end.replace("_", " "),
|
||||
)
|
||||
|
||||
|
||||
def car_endpoint_marker_type(item: StrDict, direction: str, label: str) -> str:
|
||||
"""Return the marker type for a car route endpoint."""
|
||||
marker_type = item.get(direction + "_type")
|
||||
return marker_type if isinstance(marker_type, str) else car_endpoint_type(label)
|
||||
|
||||
|
||||
def load_cars(data_dir: str) -> list[StrDict]:
|
||||
"""Load car journeys."""
|
||||
filename = os.path.join(data_dir, "car_journeys.yaml")
|
||||
if not os.path.exists(filename):
|
||||
return []
|
||||
|
||||
items = load_travel("car", "car_journeys", data_dir)
|
||||
for item in items:
|
||||
route = item.get("route")
|
||||
if not isinstance(route, str):
|
||||
continue
|
||||
|
||||
route_filename = route_filename_without_extension(route)
|
||||
item["geojson_filename"] = route_filename
|
||||
|
||||
with open(
|
||||
os.path.join(data_dir, "car_routes", route_filename + ".geojson")
|
||||
) as f:
|
||||
geojson_data = typing.cast(StrDict, json.load(f))
|
||||
|
||||
if "distance" not in item:
|
||||
item["distance"] = geojson_route_distance_km(geojson_data)
|
||||
|
||||
endpoints = geojson_route_endpoints(geojson_data)
|
||||
from_label, to_label = car_route_labels(item)
|
||||
if endpoints is None:
|
||||
continue
|
||||
|
||||
for direction, field, label, endpoint in (
|
||||
("from", "from_location", from_label, endpoints[0]),
|
||||
("to", "to_location", to_label, endpoints[1]),
|
||||
):
|
||||
if not label:
|
||||
continue
|
||||
item[field] = {
|
||||
"name": label,
|
||||
"type": car_endpoint_marker_type(item, direction, label),
|
||||
"latitude": endpoint[0],
|
||||
"longitude": endpoint[1],
|
||||
}
|
||||
|
||||
if from_label:
|
||||
item["from"] = from_label
|
||||
if to_label:
|
||||
item["to"] = to_label
|
||||
|
||||
return items
|
||||
|
||||
|
||||
def process_flight(
|
||||
flight: StrDict, by_iata: dict[str, Airline], airports: list[StrDict]
|
||||
) -> None:
|
||||
|
|
@ -337,7 +489,8 @@ def collect_travel_items(
|
|||
+ load_trains(data_dir, route_distances=route_distances)
|
||||
+ load_ferries(data_dir, route_distances=route_distances)
|
||||
+ load_coaches(data_dir, route_distances=route_distances)
|
||||
+ load_buses(data_dir, route_distances=route_distances),
|
||||
+ load_buses(data_dir, route_distances=route_distances)
|
||||
+ load_cars(data_dir),
|
||||
key=depart_datetime,
|
||||
)
|
||||
|
||||
|
|
@ -465,6 +618,8 @@ def get_locations(trip: Trip) -> dict[str, StrDict]:
|
|||
"ferry_terminal": {},
|
||||
"coach_station": {},
|
||||
"bus_stop": {},
|
||||
"home": {},
|
||||
"car_stop": {},
|
||||
}
|
||||
|
||||
station_list = []
|
||||
|
|
@ -488,6 +643,13 @@ def get_locations(trip: Trip) -> dict[str, StrDict]:
|
|||
for field in "from_terminal", "to_terminal":
|
||||
terminal = t[field]
|
||||
locations["ferry_terminal"][terminal["name"]] = terminal
|
||||
case "car":
|
||||
for field in ("from_location", "to_location"):
|
||||
location = t.get(field)
|
||||
if not location:
|
||||
continue
|
||||
location_type = location.get("type", "car_stop")
|
||||
locations[location_type][location["name"]] = location
|
||||
|
||||
locations["station"] = process_station_list(station_list)
|
||||
return locations
|
||||
|
|
@ -635,6 +797,27 @@ def get_trip_routes(trip: Trip, data_dir: str) -> list[StrDict]:
|
|||
}
|
||||
)
|
||||
continue
|
||||
if t["type"] == "car":
|
||||
route = t.get("geojson_filename")
|
||||
if not isinstance(route, str):
|
||||
continue
|
||||
route_filename = route_filename_without_extension(route)
|
||||
key = "_".join(
|
||||
[
|
||||
"car",
|
||||
typing.cast(str, t.get("from", "")),
|
||||
typing.cast(str, t.get("to", "")),
|
||||
route_filename,
|
||||
]
|
||||
)
|
||||
routes.append(
|
||||
{
|
||||
"type": "car",
|
||||
"key": key,
|
||||
"geojson_filename": os.path.join("car_routes", route_filename),
|
||||
}
|
||||
)
|
||||
continue
|
||||
if t["type"] == "train":
|
||||
for leg in t["legs"]:
|
||||
train_from, train_to = leg["from_station"], leg["to_station"]
|
||||
|
|
|
|||
|
|
@ -16,7 +16,6 @@ from agenda import format_list_with_ampersand
|
|||
|
||||
from . import utils
|
||||
|
||||
|
||||
StrDict = dict[str, typing.Any]
|
||||
DateOrDateTime = datetime.datetime | datetime.date
|
||||
|
||||
|
|
@ -46,6 +45,7 @@ class TripElement:
|
|||
"ferry": ":ferry:",
|
||||
"coach": ":bus:",
|
||||
"bus": ":bus:",
|
||||
"car": ":automobile:",
|
||||
}
|
||||
|
||||
alias = emoji_map.get(self.element_type)
|
||||
|
|
@ -414,6 +414,25 @@ class Trip:
|
|||
end_country=to_country,
|
||||
)
|
||||
)
|
||||
if item["type"] == "car":
|
||||
start_location = item.get("from_location", {})
|
||||
end_location = item.get("to_location", {})
|
||||
from_country = agenda.get_country(start_location.get("country"))
|
||||
to_country = agenda.get_country(end_location.get("country"))
|
||||
name = f"{item.get('from', 'Car journey')} → {item.get('to', 'destination')}"
|
||||
elements.append(
|
||||
TripElement(
|
||||
start_time=item["depart"],
|
||||
end_time=item.get("arrive"),
|
||||
title=name,
|
||||
detail=item,
|
||||
element_type="car",
|
||||
start_loc=item.get("from"),
|
||||
end_loc=item.get("to"),
|
||||
start_country=from_country,
|
||||
end_country=to_country,
|
||||
)
|
||||
)
|
||||
|
||||
return sorted(elements, key=lambda e: utils.as_datetime(e.start_time))
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue