edward/paddington-eurostar
1
0
Fork 0
Code Issues 2 Pull requests Projects Releases Packages Wiki Activity Actions

Add return and inbound journey support

Browse source
This commit is contained in:
Edward Betts 2026-05-21 08:46:35 +01:00
parent 6ba71447ef
commit 9691632f65
12 changed files with 1687 additions and 486 deletions
Download patch file Download diff file Expand all files Collapse all files

75
circle_line.py
View file

@ -1,5 +1,5 @@
"""
Circle Line timetable: Paddington (H&C Line) → King's Cross St Pancras.
Circle Line timetable between Paddington (H&C Line) and King's Cross St Pancras.
Parses the TransXChange XML file on first use and caches the result in memory.
"""
@ -14,9 +14,9 @@ _KXP_STOP = '9400ZZLUKSX3' # King's Cross St Pancras
from config.default import CIRCLE_LINE_XML as _TXC_XML # overridden by app config after import
_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
# Populated on first call to next_service(); maps direction -> day-type -> sorted
# list of (origin_depart_seconds, destination_arrive_seconds) measured from midnight.
_timetable: dict[str, dict[str, list[tuple[int, int]]]] | None = None
def _parse_duration(s: str | None) -> int:
@ -26,7 +26,7 @@ def _parse_duration(s: str | None) -> int:
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]]]:
def _load_timetable() -> dict[str, dict[str, list[tuple[int, int]]]]:
tree = ET.parse(_TXC_XML)
root = tree.getroot()
@ -66,8 +66,8 @@ def _load_timetable() -> dict[str, list[tuple[int, int]]]:
return elapsed
return None
# Map JP id -> (pad_offset_secs, kxp_arrive_offset_secs)
jp_offsets: dict[str, tuple[int, int]] = {}
# Map JP id -> [(direction, origin_depart_offset_secs, destination_arrive_offset_secs)].
jp_offsets: dict[str, list[tuple[str, 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)
@ -75,6 +75,7 @@ def _load_timetable() -> dict[str, list[tuple[int, int]]]:
continue
links = jps_map.get(jps_ref.text, [])
stops = [l[0] for l in links] + ([links[-1][1]] if links else [])
offsets = []
if (
_PAD_STOP in stops
and _KXP_STOP in stops
@ -83,12 +84,30 @@ def _load_timetable() -> dict[str, list[tuple[int, int]]]:
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)
offsets.append(('pad_to_kx', pad_off, kxp_off))
if (
_PAD_STOP in stops
and _KXP_STOP in stops
and stops.index(_KXP_STOP) < stops.index(_PAD_STOP)
):
kxp_off = _seconds_to_depart(links, _KXP_STOP)
pad_off = _seconds_to_arrive(links, _PAD_STOP)
if kxp_off is not None and pad_off is not None:
offsets.append(('kx_to_pad', kxp_off, pad_off))
if offsets:
jp_offsets[jp.get('id')] = offsets
result: dict[str, list[tuple[int, int]]] = {
'MondayToFriday': [],
'Saturday': [],
'Sunday': [],
result: dict[str, dict[str, list[tuple[int, int]]]] = {
'pad_to_kx': {
'MondayToFriday': [],
'Saturday': [],
'Sunday': [],
},
'kx_to_pad': {
'MondayToFriday': [],
'Saturday': [],
'Sunday': [],
},
}
for vj in root.find('t:VehicleJourneys', _NS):
@ -97,7 +116,6 @@ def _load_timetable() -> dict[str, list[tuple[int, int]]]:
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
@ -105,15 +123,20 @@ def _load_timetable() -> dict[str, list[tuple[int, int]]]:
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 direction, origin_off, dest_off in jp_offsets[jp_ref.text]:
if day_type in result[direction]:
result[direction][day_type].append((
dep_secs + origin_off,
dep_secs + dest_off,
))
for key in result:
result[key].sort()
for direction in result:
for key in result[direction]:
result[direction][key].sort()
return result
def _get_timetable() -> dict[str, list[tuple[int, int]]]:
def _get_timetable() -> dict[str, dict[str, list[tuple[int, int]]]]:
global _timetable
if _timetable is None:
_timetable = _load_timetable()
@ -126,7 +149,9 @@ def _day_type(weekday: int) -> str:
return 'Saturday' if weekday == 5 else 'Sunday'
def next_service(earliest_board: datetime) -> tuple[datetime, datetime] | None:
def next_service(
earliest_board: datetime, direction: str = 'pad_to_kx'
) -> 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
@ -135,20 +160,20 @@ def next_service(earliest_board: datetime) -> tuple[datetime, datetime] | None:
The caller is responsible for adding any walk time from the GWR platform
before passing *earliest_board*.
"""
services = upcoming_services(earliest_board, count=1)
services = upcoming_services(earliest_board, count=1, direction=direction)
return services[0] if services else None
def upcoming_services(
earliest_board: datetime, count: int = 2
earliest_board: datetime, count: int = 2, direction: str = 'pad_to_kx'
) -> list[tuple[datetime, datetime]]:
"""
Return up to *count* Circle line services from Paddington (H&C Line) to
King's Cross St Pancras, starting from *earliest_board*.
Return up to *count* Circle line services for *direction*, starting from
*earliest_board*.
Each element is (depart_paddington, arrive_kings_cross) as datetimes.
Each element is (depart_origin, arrive_destination) as datetimes.
"""
timetable = _get_timetable()[_day_type(earliest_board.weekday())]
timetable = _get_timetable().get(direction, {})[_day_type(earliest_board.weekday())]
board_secs = (
earliest_board.hour * 3600
+ earliest_board.minute * 60