""" Combine GWR Bristol→Paddington trains with Eurostar St Pancras→destination trains. """ from datetime import datetime, timedelta, time as _time MIN_CONNECTION_MINUTES = 50 MAX_CONNECTION_MINUTES = 110 MAX_GWR_MINUTES = 110 DATE_FMT = '%Y-%m-%d' TIME_FMT = '%H:%M' # Circle Line: Paddington (H&C) → Kings Cross St Pancras CIRCLE_LINE_MINUTES_PAST = [8, 18, 28, 38, 48, 58] # departures past each hour CIRCLE_LINE_JOURNEY_MINUTES = 11 PAD_WALK_TO_UNDERGROUND_MINUTES = 8 # GWR platform → Paddington Underground KX_WALK_TO_CHECKIN_MINUTES = 10 # Kings Cross St Pancras platform → St Pancras check-in # Bristol Temple Meads → London Paddington walk-on single fares. # Weekday restrictions (Mon–Fri only): # Super Off-Peak (£45.00, SSS): not valid if 05:05 ≤ dep ≤ 09:57 # Off-Peak (£63.60, SVS): not valid if dep < 08:26, except 02:00–05:10 # Anytime (£138.70, SDS): no restrictions _TICKET_SUPER_OFF_PEAK = {'ticket': 'Super Off-Peak', 'price': 45.00, 'code': 'SSS'} _TICKET_OFF_PEAK = {'ticket': 'Off-Peak', 'price': 63.60, 'code': 'SVS'} _TICKET_ANYTIME = {'ticket': 'Anytime', 'price': 138.70, 'code': 'SDS'} def cheapest_gwr_ticket(depart_time: str, travel_date: str) -> dict: """ Return the cheapest walk-on single for Bristol Temple Meads → Paddington. Weekday (Mon–Fri) restrictions derived from the SSS/SVS ticket conditions: Super Off-Peak valid: dep ≤ 05:04 or dep ≥ 09:58 Off-Peak valid: 02:00 ≤ dep ≤ 05:10 or dep ≥ 08:26 Weekends: no restrictions — always Super Off-Peak. """ dt = datetime.strptime(f"{travel_date} {depart_time}", f"{DATE_FMT} {TIME_FMT}") if dt.weekday() >= 5: # Saturday or Sunday return _TICKET_SUPER_OFF_PEAK dep = dt.time() if dep <= _time(5, 4) or dep >= _time(9, 58): return _TICKET_SUPER_OFF_PEAK if dep >= _time(8, 26): return _TICKET_OFF_PEAK return _TICKET_ANYTIME def _parse_dt(date: str, time: str) -> datetime: return datetime.strptime(f"{date} {time}", f"{DATE_FMT} {TIME_FMT}") def _next_circle_line(arrive_paddington: datetime) -> tuple[datetime, datetime]: """ Given GWR arrival at Paddington, return (circle_line_depart, arrive_checkin). Walk 10 min to Paddington Underground, catch next Circle Line at :08/:18/:28/:38/:48/:58, 11 min journey to Kings Cross St Pancras, 10 min walk to St Pancras check-in. """ earliest_board = arrive_paddington + timedelta(minutes=PAD_WALK_TO_UNDERGROUND_MINUTES) minute = earliest_board.minute depart_minute = next((m for m in CIRCLE_LINE_MINUTES_PAST if m >= minute), None) if depart_minute is None: circle_depart = (earliest_board + timedelta(hours=1)).replace( minute=CIRCLE_LINE_MINUTES_PAST[0], second=0, microsecond=0 ) else: circle_depart = earliest_board.replace(minute=depart_minute, second=0, microsecond=0) arrive_checkin = circle_depart + timedelta( minutes=CIRCLE_LINE_JOURNEY_MINUTES + KX_WALK_TO_CHECKIN_MINUTES ) return circle_depart, arrive_checkin 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 _is_viable_connection( gwr: dict, eurostar: dict, travel_date: str, min_connection_minutes: int, max_connection_minutes: int, ) -> tuple[datetime, datetime, datetime, datetime] | None: try: arr_pad = _parse_dt(travel_date, gwr['arrive_paddington']) dep_bri = _parse_dt(travel_date, gwr['depart_bristol']) dep_stp = _parse_dt(travel_date, eurostar['depart_st_pancras']) arr_dest = _parse_dt(travel_date, eurostar['arrive_destination']) except (ValueError, KeyError): return None if int((arr_pad - dep_bri).total_seconds() / 60) > MAX_GWR_MINUTES: return None if arr_dest < dep_stp: arr_dest += timedelta(days=1) connection_minutes = (dep_stp - arr_pad).total_seconds() / 60 if connection_minutes < min_connection_minutes: return None if connection_minutes > max_connection_minutes: return None return dep_bri, arr_pad, dep_stp, arr_dest 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: # Find only the earliest viable Eurostar for this GWR departure for es in eurostar_trains: connection = _is_viable_connection( gwr, es, travel_date, min_connection_minutes, max_connection_minutes, ) if not connection: continue dep_bri, arr_pad, dep_stp, arr_dest = connection total_mins = int((arr_dest - dep_bri).total_seconds() / 60) ticket = cheapest_gwr_ticket(gwr['depart_bristol'], travel_date) circle_depart, arrive_checkin = _next_circle_line(arr_pad) 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_minutes': int((dep_stp - arr_pad).total_seconds() / 60), 'connection_duration': _fmt_duration(int((dep_stp - arr_pad).total_seconds() / 60)), 'circle_line_depart': circle_depart.strftime(TIME_FMT), 'circle_arrive_checkin': arrive_checkin.strftime(TIME_FMT), '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'], 'ticket_name': ticket['ticket'], 'ticket_price': ticket['price'], 'ticket_code': ticket['code'], }) break # Only the earliest valid Eurostar per GWR departure trips.sort(key=lambda t: (t['depart_bristol'], t['depart_st_pancras'])) return trips def find_unreachable_morning_eurostars( 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]: unreachable = [] for es in eurostar_trains: if any( _is_viable_connection( gwr, es, travel_date, min_connection_minutes, max_connection_minutes, ) for gwr in gwr_trains ): continue unreachable.append(es) return sorted(unreachable, key=lambda s: s['depart_st_pancras'])