Refactor bus/coach loading and rendering to share code.
Extract `load_road_transport()` as a shared helper for bus and coach,
combining the near-identical route rendering blocks in `get_trip_routes`
and `Trip.elements()` into single `in ("coach", "bus")` branches.
Document transport type patterns in AGENTS.md.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
c87320415f
commit
53f4252b92
3 changed files with 59 additions and 92 deletions
110
agenda/trip.py
110
agenda/trip.py
|
|
@ -92,62 +92,32 @@ def load_ferries(
|
|||
return ferries
|
||||
|
||||
|
||||
def load_coaches(
|
||||
data_dir: str, route_distances: travel.RouteDistances | None = None
|
||||
def load_road_transport(
|
||||
travel_type: str,
|
||||
plural: str,
|
||||
stops_yaml: str,
|
||||
data_dir: str,
|
||||
co2_factor: float,
|
||||
route_distances: travel.RouteDistances | None = None,
|
||||
) -> list[StrDict]:
|
||||
"""Load coaches."""
|
||||
coaches = load_travel("coach", "coaches", data_dir)
|
||||
stations = travel.parse_yaml("coach_stations", data_dir)
|
||||
by_name = {station["name"]: station for station in stations}
|
||||
|
||||
for item in coaches:
|
||||
add_station_objects(item, by_name)
|
||||
# Add route distance and CO2 calculation if available
|
||||
if route_distances:
|
||||
travel.add_leg_route_distance(item, route_distances)
|
||||
if "distance" in item:
|
||||
# Calculate CO2 emissions for coach (0.027 kg CO2e per passenger per km)
|
||||
item["co2_kg"] = item["distance"] * 0.027
|
||||
# Include GeoJSON route if defined in stations data
|
||||
from_station = item.get("from_station")
|
||||
to_station = item.get("to_station")
|
||||
if from_station and to_station:
|
||||
# Support scalar or mapping routes: string or dict of station name -> geojson base name
|
||||
routes_val = from_station.get("routes", {})
|
||||
if isinstance(routes_val, str):
|
||||
geo = routes_val
|
||||
else:
|
||||
geo = routes_val.get(to_station.get("name"))
|
||||
if geo:
|
||||
item["geojson_filename"] = geo
|
||||
|
||||
return coaches
|
||||
|
||||
|
||||
def load_buses(
|
||||
data_dir: str, route_distances: travel.RouteDistances | None = None
|
||||
) -> list[StrDict]:
|
||||
"""Load buses."""
|
||||
items = load_travel("bus", "buses", data_dir)
|
||||
stops = travel.parse_yaml("bus_stops", data_dir)
|
||||
"""Load road transport (bus or coach)."""
|
||||
items = load_travel(travel_type, plural, data_dir)
|
||||
stops = travel.parse_yaml(stops_yaml, data_dir)
|
||||
by_name = {stop["name"]: stop for stop in stops}
|
||||
|
||||
for item in items:
|
||||
add_station_objects(item, by_name)
|
||||
# Add route distance and CO2 calculation if available
|
||||
if route_distances:
|
||||
travel.add_leg_route_distance(item, route_distances)
|
||||
if "distance" in item:
|
||||
# Calculate CO2 emissions for bus (0.027 kg CO2e per passenger per km)
|
||||
item["co2_kg"] = item["distance"] * 0.1
|
||||
# Include GeoJSON route if defined in stations data
|
||||
item["co2_kg"] = item["distance"] * co2_factor
|
||||
from_station = item.get("from_station")
|
||||
to_station = item.get("to_station")
|
||||
if from_station and to_station:
|
||||
# Support scalar or mapping routes: string or dict of station name -> geojson base name
|
||||
# Support scalar or mapping routes: string or dict of stop name -> geojson filename
|
||||
routes_val = from_station.get("routes", {})
|
||||
if isinstance(routes_val, str):
|
||||
geo = routes_val
|
||||
geo: str | None = routes_val
|
||||
else:
|
||||
geo = routes_val.get(to_station.get("name"))
|
||||
if geo:
|
||||
|
|
@ -156,6 +126,24 @@ def load_buses(
|
|||
return items
|
||||
|
||||
|
||||
def load_coaches(
|
||||
data_dir: str, route_distances: travel.RouteDistances | None = None
|
||||
) -> list[StrDict]:
|
||||
"""Load coaches."""
|
||||
return load_road_transport(
|
||||
"coach", "coaches", "coach_stations", data_dir, 0.027, route_distances
|
||||
)
|
||||
|
||||
|
||||
def load_buses(
|
||||
data_dir: str, route_distances: travel.RouteDistances | None = None
|
||||
) -> list[StrDict]:
|
||||
"""Load buses."""
|
||||
return load_road_transport(
|
||||
"bus", "buses", "bus_stops", data_dir, 0.1, route_distances
|
||||
)
|
||||
|
||||
|
||||
def process_flight(
|
||||
flight: StrDict, by_iata: dict[str, Airline], airports: list[StrDict]
|
||||
) -> None:
|
||||
|
|
@ -434,40 +422,22 @@ def get_trip_routes(trip: Trip, data_dir: str) -> list[StrDict]:
|
|||
}
|
||||
)
|
||||
continue
|
||||
if t["type"] == "coach":
|
||||
coach_from, coach_to = t["from_station"], t["to_station"]
|
||||
key = "_".join(["coach"] + sorted([coach_from["name"], coach_to["name"]]))
|
||||
# Use GeoJSON route when available, otherwise draw straight line
|
||||
if t["type"] in ("coach", "bus"):
|
||||
route_type = t["type"]
|
||||
stop_from, stop_to = t["from_station"], t["to_station"]
|
||||
key = "_".join([route_type] + sorted([stop_from["name"], stop_to["name"]]))
|
||||
if t.get("geojson_filename"):
|
||||
filename = os.path.join("coach_routes", t["geojson_filename"])
|
||||
filename = os.path.join(f"{route_type}_routes", t["geojson_filename"])
|
||||
routes.append(
|
||||
{"type": "coach", "key": key, "geojson_filename": filename}
|
||||
{"type": route_type, "key": key, "geojson_filename": filename}
|
||||
)
|
||||
else:
|
||||
routes.append(
|
||||
{
|
||||
"type": "coach",
|
||||
"type": route_type,
|
||||
"key": key,
|
||||
"from": latlon_tuple(coach_from),
|
||||
"to": latlon_tuple(coach_to),
|
||||
}
|
||||
)
|
||||
continue
|
||||
if t["type"] == "bus":
|
||||
bus_from, bus_to = t["from_station"], t["to_station"]
|
||||
key = "_".join(["bus"] + sorted([bus_from["name"], bus_to["name"]]))
|
||||
if t.get("geojson_filename"):
|
||||
filename = os.path.join("bus_routes", t["geojson_filename"])
|
||||
routes.append(
|
||||
{"type": "bus", "key": key, "geojson_filename": filename}
|
||||
)
|
||||
else:
|
||||
routes.append(
|
||||
{
|
||||
"type": "bus",
|
||||
"key": key,
|
||||
"from": latlon_tuple(bus_from),
|
||||
"to": latlon_tuple(bus_to),
|
||||
"from": latlon_tuple(stop_from),
|
||||
"to": latlon_tuple(stop_to),
|
||||
}
|
||||
)
|
||||
continue
|
||||
|
|
|
|||
|
|
@ -397,8 +397,7 @@ class Trip:
|
|||
end_country=to_country,
|
||||
)
|
||||
)
|
||||
if item["type"] == "coach":
|
||||
# Include coach journeys in trip elements
|
||||
if item["type"] in ("coach", "bus"):
|
||||
from_country = agenda.get_country(item["from_station"]["country"])
|
||||
to_country = agenda.get_country(item["to_station"]["country"])
|
||||
name = f"{item['from']} → {item['to']}"
|
||||
|
|
@ -408,26 +407,7 @@ class Trip:
|
|||
end_time=item.get("arrive"),
|
||||
title=name,
|
||||
detail=item,
|
||||
element_type="coach",
|
||||
start_loc=item["from"],
|
||||
end_loc=item["to"],
|
||||
start_country=from_country,
|
||||
end_country=to_country,
|
||||
)
|
||||
)
|
||||
|
||||
if item["type"] == "bus":
|
||||
# Include bus journeys in trip elements
|
||||
from_country = agenda.get_country(item["from_station"]["country"])
|
||||
to_country = agenda.get_country(item["to_station"]["country"])
|
||||
name = f"{item['from']} → {item['to']}"
|
||||
elements.append(
|
||||
TripElement(
|
||||
start_time=item["depart"],
|
||||
end_time=item.get("arrive"),
|
||||
title=name,
|
||||
detail=item,
|
||||
element_type="bus",
|
||||
element_type=item["type"],
|
||||
start_loc=item["from"],
|
||||
end_loc=item["to"],
|
||||
start_country=from_country,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue