94 lines
3.2 KiB
Python
94 lines
3.2 KiB
Python
"""
|
|
Combine GWR Bristol→Paddington trains with Eurostar St Pancras→destination trains.
|
|
"""
|
|
from datetime import datetime, timedelta
|
|
|
|
MIN_CONNECTION_MINUTES = 50
|
|
MAX_CONNECTION_MINUTES = 110
|
|
MAX_GWR_MINUTES = 110
|
|
DATE_FMT = '%Y-%m-%d'
|
|
TIME_FMT = '%H:%M'
|
|
|
|
|
|
def _parse_dt(date: str, time: str) -> datetime:
|
|
return datetime.strptime(f"{date} {time}", f"{DATE_FMT} {TIME_FMT}")
|
|
|
|
|
|
def _fmt_duration(minutes: int) -> str:
|
|
h, m = divmod(minutes, 60)
|
|
if h and m:
|
|
return f"{h}h {m}m"
|
|
if h:
|
|
return f"{h}h"
|
|
return f"{m}m"
|
|
|
|
|
|
def combine_trips(
|
|
gwr_trains: list[dict],
|
|
eurostar_trains: list[dict],
|
|
travel_date: str,
|
|
min_connection_minutes: int = MIN_CONNECTION_MINUTES,
|
|
max_connection_minutes: int = MAX_CONNECTION_MINUTES,
|
|
) -> list[dict]:
|
|
"""
|
|
Return a list of valid combined trips, sorted by Bristol departure time.
|
|
|
|
Each trip dict:
|
|
depart_bristol HH:MM
|
|
arrive_paddington HH:MM
|
|
gwr_duration str (e.g. "1h 45m")
|
|
connection_duration str
|
|
depart_st_pancras HH:MM
|
|
arrive_destination HH:MM
|
|
total_duration str (e.g. "5h 30m")
|
|
destination str
|
|
"""
|
|
trips = []
|
|
|
|
for gwr in gwr_trains:
|
|
try:
|
|
arr_pad = _parse_dt(travel_date, gwr['arrive_paddington'])
|
|
dep_bri = _parse_dt(travel_date, gwr['depart_bristol'])
|
|
except (ValueError, KeyError):
|
|
continue
|
|
|
|
if int((arr_pad - dep_bri).total_seconds() / 60) > MAX_GWR_MINUTES:
|
|
continue
|
|
|
|
earliest_eurostar = arr_pad + timedelta(minutes=min_connection_minutes)
|
|
|
|
# Find only the earliest viable Eurostar for this GWR departure
|
|
for es in eurostar_trains:
|
|
try:
|
|
dep_stp = _parse_dt(travel_date, es['depart_st_pancras'])
|
|
arr_dest = _parse_dt(travel_date, es['arrive_destination'])
|
|
except (ValueError, KeyError):
|
|
continue
|
|
|
|
# Eurostar arrives next day? (e.g. night service — unlikely but handle it)
|
|
if arr_dest < dep_stp:
|
|
arr_dest += timedelta(days=1)
|
|
|
|
if dep_stp < earliest_eurostar:
|
|
continue
|
|
if (dep_stp - arr_pad).total_seconds() / 60 > max_connection_minutes:
|
|
continue
|
|
|
|
total_mins = int((arr_dest - dep_bri).total_seconds() / 60)
|
|
trips.append({
|
|
'depart_bristol': gwr['depart_bristol'],
|
|
'arrive_paddington': gwr['arrive_paddington'],
|
|
'headcode': gwr.get('headcode', ''),
|
|
'gwr_duration': _fmt_duration(int((arr_pad - dep_bri).total_seconds() / 60)),
|
|
'connection_duration': _fmt_duration(int((dep_stp - arr_pad).total_seconds() / 60)),
|
|
'depart_st_pancras': es['depart_st_pancras'],
|
|
'arrive_destination': es['arrive_destination'],
|
|
'train_number': es.get('train_number', ''),
|
|
'total_duration': _fmt_duration(total_mins),
|
|
'total_minutes': total_mins,
|
|
'destination': es['destination'],
|
|
})
|
|
break # Only the earliest valid Eurostar per GWR departure
|
|
|
|
trips.sort(key=lambda t: (t['depart_bristol'], t['depart_st_pancras']))
|
|
return trips
|