diff --git a/circle_line.py b/circle_line.py new file mode 100644 index 0000000..1fee867 --- /dev/null +++ b/circle_line.py @@ -0,0 +1,148 @@ +""" +Circle Line timetable: Paddington (H&C Line) → King's Cross St Pancras. + +Parses the TransXChange XML file on first use and caches the result in memory. +""" +import os +import re +import xml.etree.ElementTree as ET +from datetime import datetime, timedelta + +_PAD_STOP = '9400ZZLUPAH1' # Paddington (H&C Line) +_KXP_STOP = '9400ZZLUKSX3' # King's Cross St Pancras + +_TXC_XML = os.path.join(os.path.dirname(__file__), 'output_txc_01CIR_.xml') +_NS = {'t': 'http://www.transxchange.org.uk/'} + +# Populated on first call to next_service(); maps day-type -> sorted list of +# (pad_depart_seconds, kxp_arrive_seconds) measured from midnight. +_timetable: dict[str, list[tuple[int, int]]] | None = None + + +def _parse_duration(s: str | None) -> int: + if not s: + return 0 + m = re.match(r'PT(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?', s) + return int(m.group(1) or 0) * 3600 + int(m.group(2) or 0) * 60 + int(m.group(3) or 0) + + +def _load_timetable() -> dict[str, list[tuple[int, int]]]: + tree = ET.parse(_TXC_XML) + root = tree.getroot() + + # Build JPS id -> [(from_stop, to_stop, runtime_secs, wait_secs)] + jps_map: dict[str, list[tuple]] = {} + for jps_el in root.find('t:JourneyPatternSections', _NS): + links = [] + for link in jps_el.findall('t:JourneyPatternTimingLink', _NS): + fr = link.find('t:From/t:StopPointRef', _NS) + to = link.find('t:To/t:StopPointRef', _NS) + rt = link.find('t:RunTime', _NS) + wait = link.find('t:From/t:WaitTime', _NS) + links.append(( + fr.text if fr is not None else None, + to.text if to is not None else None, + _parse_duration(rt.text if rt is not None else None), + _parse_duration(wait.text if wait is not None else None), + )) + jps_map[jps_el.get('id')] = links + + def _seconds_to_depart(links, stop): + """Seconds from journey start until departure from *stop*.""" + elapsed = 0 + for fr, to, rt, wait in links: + elapsed += wait + if fr == stop: + return elapsed + elapsed += rt + return None + + def _seconds_to_arrive(links, stop): + """Seconds from journey start until arrival at *stop*.""" + elapsed = 0 + for fr, to, rt, wait in links: + elapsed += wait + rt + if to == stop: + return elapsed + return None + + # Map JP id -> (pad_offset_secs, kxp_arrive_offset_secs) + jp_offsets: dict[str, tuple[int, int]] = {} + for svc in root.find('t:Services', _NS): + for jp in svc.findall('.//t:JourneyPattern', _NS): + jps_ref = jp.find('t:JourneyPatternSectionRefs', _NS) + if jps_ref is None: + continue + links = jps_map.get(jps_ref.text, []) + stops = [l[0] for l in links] + ([links[-1][1]] if links else []) + if ( + _PAD_STOP in stops + and _KXP_STOP in stops + and stops.index(_PAD_STOP) < stops.index(_KXP_STOP) + ): + pad_off = _seconds_to_depart(links, _PAD_STOP) + kxp_off = _seconds_to_arrive(links, _KXP_STOP) + if pad_off is not None and kxp_off is not None: + jp_offsets[jp.get('id')] = (pad_off, kxp_off) + + result: dict[str, list[tuple[int, int]]] = { + 'MondayToFriday': [], + 'Saturday': [], + 'Sunday': [], + } + + for vj in root.find('t:VehicleJourneys', _NS): + jp_ref = vj.find('t:JourneyPatternRef', _NS) + dep_time = vj.find('t:DepartureTime', _NS) + op = vj.find('t:OperatingProfile', _NS) + if jp_ref is None or dep_time is None or jp_ref.text not in jp_offsets: + continue + pad_off, kxp_off = jp_offsets[jp_ref.text] + h, m, s = map(int, dep_time.text.split(':')) + dep_secs = h * 3600 + m * 60 + s + rdt = op.find('.//t:DaysOfWeek', _NS) if op is not None else None + if rdt is None: + continue + for day_el in rdt: + day_type = day_el.tag.split('}')[-1] + if day_type in result: + result[day_type].append((dep_secs + pad_off, dep_secs + kxp_off)) + + for key in result: + result[key].sort() + return result + + +def _get_timetable() -> dict[str, list[tuple[int, int]]]: + global _timetable + if _timetable is None: + _timetable = _load_timetable() + return _timetable + + +def _day_type(weekday: int) -> str: + if weekday < 5: + return 'MondayToFriday' + return 'Saturday' if weekday == 5 else 'Sunday' + + +def next_service(earliest_board: datetime) -> tuple[datetime, datetime] | None: + """ + Given the earliest time a passenger can board at Paddington (H&C Line), + return (circle_line_depart, arrive_kings_cross) as datetimes, or None if + no service is found before midnight. + + The caller is responsible for adding any walk time from the GWR platform + before passing *earliest_board*. + """ + timetable = _get_timetable()[_day_type(earliest_board.weekday())] + board_secs = ( + earliest_board.hour * 3600 + + earliest_board.minute * 60 + + earliest_board.second + ) + midnight = earliest_board.replace(hour=0, minute=0, second=0, microsecond=0) + for pad_secs, kxp_secs in timetable: + if pad_secs >= board_secs: + return midnight + timedelta(seconds=pad_secs), midnight + timedelta(seconds=kxp_secs) + return None diff --git a/templates/results.html b/templates/results.html index 5426995..403d01d 100644 --- a/templates/results.html +++ b/templates/results.html @@ -106,7 +106,7 @@ Paddington GWR Fare Transfer - Depart St Pancras + Depart STP {{ destination }} ES Std Total @@ -159,6 +159,7 @@ {{ row.arrive_destination }} (CET) + {% if row.eurostar_duration %}
({{ row.eurostar_duration }}){% endif %} {% if row.eurostar_price is not none %} @@ -191,6 +192,7 @@ {{ row.arrive_destination }} (CET) + {% if row.eurostar_duration %}
({{ row.eurostar_duration }}){% endif %} {% if row.eurostar_price is not none %} diff --git a/trip_planner.py b/trip_planner.py index e732b93..5a0b718 100644 --- a/trip_planner.py +++ b/trip_planner.py @@ -3,17 +3,16 @@ Combine GWR Bristol→Paddington trains with Eurostar St Pancras→destination t """ 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' -# 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 +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. @@ -51,26 +50,22 @@ 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]: +def _circle_line_times(arrive_paddington: datetime) -> tuple[str, str] | tuple[None, None]: """ - Given GWR arrival at Paddington, return (circle_line_depart, arrive_checkin). + Given GWR arrival at Paddington, return (circle_line_depart_str, arrive_checkin_str). - 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. + 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) - 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 + 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: @@ -149,8 +144,10 @@ def combine_trips( 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 = _next_circle_line(arr_pad) + circle_depart, arrive_checkin = _circle_line_times(arr_pad) trips.append({ 'depart_bristol': gwr['depart_bristol'], 'arrive_paddington': gwr['arrive_paddington'], @@ -158,10 +155,11 @@ def combine_trips( '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), + '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, @@ -198,6 +196,11 @@ def find_unreachable_morning_eurostars( ): continue - unreachable.append(es) + 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'])