""" Combine GWR Bristol→Paddington trains with Eurostar St Pancras→destination trains. """ from datetime import datetime, timedelta, time as _time import circle_line MIN_CONNECTION_MINUTES = 50 MAX_CONNECTION_MINUTES = 110 MAX_GWR_MINUTES = 110 DATE_FMT = '%Y-%m-%d' TIME_FMT = '%H:%M' PAD_WALK_TO_UNDERGROUND_MINUTES = 7 # GWR platform → Paddington (H&C Line) platform KX_WALK_TO_CHECKIN_MINUTES = 8 # King's 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 _circle_line_times(arrive_paddington: datetime) -> tuple[str, str] | tuple[None, None]: """ Given GWR arrival at Paddington, return (circle_line_depart_str, arrive_checkin_str). Adds PAD_WALK_TO_UNDERGROUND_MINUTES to get the earliest boarding time, looks up the next real Circle Line service from the timetable, then adds KX_WALK_TO_CHECKIN_MINUTES to the Kings Cross arrival to give an estimated St Pancras check-in time. Returns (None, None) if no service is found. """ earliest_board = arrive_paddington + timedelta(minutes=PAD_WALK_TO_UNDERGROUND_MINUTES) result = circle_line.next_service(earliest_board) if result is None: return None, None circle_depart, arrive_kx = result arrive_checkin = arrive_kx + timedelta(minutes=KX_WALK_TO_CHECKIN_MINUTES) return circle_depart.strftime(TIME_FMT), arrive_checkin.strftime(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 _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) # Destination time is CET/CEST, departure is GMT/BST; Europe is always 1h ahead. eurostar_mins = int((arr_dest - dep_stp).total_seconds() / 60) - 60 ticket = cheapest_gwr_ticket(gwr['depart_bristol'], travel_date) circle_depart, arrive_checkin = _circle_line_times(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, 'circle_arrive_checkin': arrive_checkin, 'depart_st_pancras': es['depart_st_pancras'], 'arrive_destination': es['arrive_destination'], 'eurostar_duration': _fmt_duration(eurostar_mins), '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 dep_stp = _parse_dt(travel_date, es['depart_st_pancras']) arr_dest = _parse_dt(travel_date, es['arrive_destination']) if arr_dest < dep_stp: arr_dest += timedelta(days=1) eurostar_mins = int((arr_dest - dep_stp).total_seconds() / 60) - 60 unreachable.append({**es, 'eurostar_duration': _fmt_duration(eurostar_mins)}) return sorted(unreachable, key=lambda s: s['depart_st_pancras'])