Show cheapest GWR fare per journey and flag unreachable morning Eurostars

Add cheapest_gwr_ticket() to trip_planner.py encoding the SSS/SVS/SDS
walk-on single restrictions for Bristol Temple Meads → Paddington: on
weekdays, Super Off-Peak (£45) is valid before 05:05 or from 09:58,
Off-Peak (£63.60) from 08:26, and Anytime (£138.70) covers the gap.
Weekends have no restrictions. The fare is included in each trip dict
and displayed in a new GWR Fare column on the results page.

Also wire up find_unreachable_morning_eurostars() into the results view
so early Eurostar services unreachable from Bristol appear in the table,
with tests covering both features.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Edward Betts 2026-04-04 10:22:47 +01:00
parent b88d23a270
commit 804fcedfad
5 changed files with 428 additions and 44 deletions

16
app.py
View file

@ -7,7 +7,7 @@ from datetime import date, timedelta
from cache import get_cached, set_cached from cache import get_cached, set_cached
import scraper.eurostar as eurostar_scraper import scraper.eurostar as eurostar_scraper
import scraper.realtime_trains as rtt_scraper import scraper.realtime_trains as rtt_scraper
from trip_planner import combine_trips from trip_planner import combine_trips, find_unreachable_morning_eurostars
RTT_PADDINGTON_URL = ( RTT_PADDINGTON_URL = (
"https://www.realtimetrains.co.uk/search/detailed/" "https://www.realtimetrains.co.uk/search/detailed/"
@ -109,6 +109,18 @@ def results(slug, travel_date):
error = f"{error}; {msg}" if error else msg error = f"{error}; {msg}" if error else msg
trips = combine_trips(gwr_trains, eurostar_trains, travel_date, min_connection, max_connection) trips = combine_trips(gwr_trains, eurostar_trains, travel_date, min_connection, max_connection)
unreachable_morning_services = find_unreachable_morning_eurostars(
gwr_trains,
eurostar_trains,
travel_date,
min_connection,
max_connection,
)
result_rows = sorted(
[{'row_type': 'trip', **trip} for trip in trips]
+ [{'row_type': 'unreachable', **service} for service in unreachable_morning_services],
key=lambda row: row['depart_st_pancras'],
)
dt = date.fromisoformat(travel_date) dt = date.fromisoformat(travel_date)
prev_date = (dt - timedelta(days=1)).isoformat() prev_date = (dt - timedelta(days=1)).isoformat()
@ -121,6 +133,8 @@ def results(slug, travel_date):
return render_template( return render_template(
'results.html', 'results.html',
trips=trips, trips=trips,
result_rows=result_rows,
unreachable_morning_services=unreachable_morning_services,
destinations=DESTINATIONS, destinations=DESTINATIONS,
destination=destination, destination=destination,
travel_date=travel_date, travel_date=travel_date,

View file

@ -75,6 +75,12 @@
{{ gwr_count }} GWR service{{ 's' if gwr_count != 1 }} {{ gwr_count }} GWR service{{ 's' if gwr_count != 1 }}
&nbsp;&middot;&nbsp; &nbsp;&middot;&nbsp;
{{ eurostar_count }} Eurostar service{{ 's' if eurostar_count != 1 }} {{ eurostar_count }} Eurostar service{{ 's' if eurostar_count != 1 }}
{% if unreachable_morning_services %}
&nbsp;&middot;&nbsp;
<span style="color:#718096">
{{ unreachable_morning_services | length }} morning service{{ 's' if unreachable_morning_services | length != 1 }} unavailable from Bristol
</span>
{% endif %}
{% if from_cache %} {% if from_cache %}
&nbsp;&middot;&nbsp; <span style="color:#718096;font-size:0.85rem">(cached)</span> &nbsp;&middot;&nbsp; <span style="color:#718096;font-size:0.85rem">(cached)</span>
{% endif %} {% endif %}
@ -86,13 +92,14 @@
{% endif %} {% endif %}
</div> </div>
{% if trips %} {% if trips or unreachable_morning_services %}
<div class="card" style="overflow-x:auto"> <div class="card" style="overflow-x:auto">
<table style="width:100%;border-collapse:collapse;font-size:0.95rem"> <table style="width:100%;border-collapse:collapse;font-size:0.95rem">
<thead> <thead>
<tr style="border-bottom:2px solid #e2e8f0;text-align:left"> <tr style="border-bottom:2px solid #e2e8f0;text-align:left">
<th style="padding:0.6rem 0.8rem;white-space:nowrap">Bristol</th> <th style="padding:0.6rem 0.8rem;white-space:nowrap">Bristol</th>
<th style="padding:0.6rem 0.8rem;white-space:nowrap">Paddington</th> <th style="padding:0.6rem 0.8rem;white-space:nowrap">Paddington</th>
<th style="padding:0.6rem 0.8rem;white-space:nowrap">GWR Fare</th>
<th style="padding:0.6rem 0.8rem;white-space:nowrap">Transfer</th> <th style="padding:0.6rem 0.8rem;white-space:nowrap">Transfer</th>
<th style="padding:0.6rem 0.8rem;white-space:nowrap">Depart St&nbsp;Pancras</th> <th style="padding:0.6rem 0.8rem;white-space:nowrap">Depart St&nbsp;Pancras</th>
<th style="padding:0.6rem 0.8rem">{{ destination }} <th style="padding:0.6rem 0.8rem">{{ destination }}
@ -101,47 +108,73 @@
</tr> </tr>
</thead> </thead>
<tbody> <tbody>
{% if trips %}
{% set best_mins = trips | map(attribute='total_minutes') | min %} {% set best_mins = trips | map(attribute='total_minutes') | min %}
{% set worst_mins = trips | map(attribute='total_minutes') | max %} {% set worst_mins = trips | map(attribute='total_minutes') | max %}
{% for trip in trips %} {% endif %}
{% if trip.total_minutes == best_mins and trips | length > 1 %} {% for row in result_rows %}
{% if row.row_type == 'trip' and row.total_minutes <= best_mins + 5 and trips | length > 1 %}
{% set row_bg = 'background:#f0fff4' %} {% set row_bg = 'background:#f0fff4' %}
{% elif trip.total_minutes == worst_mins and trips | length > 1 %} {% elif row.row_type == 'trip' and row.total_minutes >= worst_mins - 5 and trips | length > 1 %}
{% set row_bg = 'background:#fff5f5' %} {% set row_bg = 'background:#fff5f5' %}
{% elif row.row_type == 'unreachable' %}
{% set row_bg = 'background:#f7fafc;color:#a0aec0' %}
{% elif loop.index is odd %} {% elif loop.index is odd %}
{% set row_bg = 'background:#f7fafc' %} {% set row_bg = 'background:#f7fafc' %}
{% else %} {% else %}
{% set row_bg = '' %} {% set row_bg = '' %}
{% endif %} {% endif %}
<tr style="border-bottom:1px solid #e2e8f0;{{ row_bg }}"> <tr style="border-bottom:1px solid #e2e8f0;{{ row_bg }}">
{% if row.row_type == 'trip' %}
<td style="padding:0.6rem 0.8rem;font-weight:600"> <td style="padding:0.6rem 0.8rem;font-weight:600">
{{ trip.depart_bristol }} {{ row.depart_bristol }}
{% if trip.headcode %}<br><span style="font-size:0.75rem;font-weight:400;color:#718096">{{ trip.headcode }}</span>{% endif %} {% if row.headcode %}<br><span style="font-size:0.75rem;font-weight:400;color:#718096">{{ row.headcode }}</span>{% endif %}
</td> </td>
<td style="padding:0.6rem 0.8rem"> <td style="padding:0.6rem 0.8rem">
{{ trip.arrive_paddington }} {{ row.arrive_paddington }}
<span style="font-size:0.8rem;color:#718096">({{ trip.gwr_duration }})</span> <span style="font-size:0.8rem;color:#718096">({{ row.gwr_duration }})</span>
</td>
<td style="padding:0.6rem 0.8rem;white-space:nowrap">
£{{ "%.2f"|format(row.ticket_price) }}
<br><span style="font-size:0.75rem;color:#718096">{{ row.ticket_name }}</span>
</td> </td>
<td style="padding:0.6rem 0.8rem;color:#4a5568"> <td style="padding:0.6rem 0.8rem;color:#4a5568">
{{ trip.connection_duration }} {{ row.connection_duration }}
</td> </td>
<td style="padding:0.6rem 0.8rem;font-weight:600"> <td style="padding:0.6rem 0.8rem;font-weight:600">
{{ trip.depart_st_pancras }} {{ row.depart_st_pancras }}
{% if trip.train_number %}<br><span style="font-size:0.75rem;font-weight:400;color:#718096">{{ trip.train_number }}</span>{% endif %} {% if row.train_number %}<br><span style="font-size:0.75rem;font-weight:400;color:#718096">{{ row.train_number }}</span>{% endif %}
</td> </td>
<td style="padding:0.6rem 0.8rem"> <td style="padding:0.6rem 0.8rem">
{{ trip.arrive_destination }} {{ row.arrive_destination }}
<span style="font-weight:400;color:#718096;font-size:0.85em">(CET)</span> <span style="font-weight:400;color:#718096;font-size:0.85em">(CET)</span>
</td> </td>
<td style="padding:0.6rem 0.8rem;font-weight:600"> <td style="padding:0.6rem 0.8rem;font-weight:600">
{% if trip.total_minutes == best_mins and trips | length > 1 %} {% if row.total_minutes <= best_mins + 5 and trips | length > 1 %}
<span style="color:#276749" title="Fastest option">{{ trip.total_duration }} ⚡</span> <span style="color:#276749" title="Fastest option">{{ row.total_duration }} ⚡</span>
{% elif trip.total_minutes == worst_mins and trips | length > 1 %} {% elif row.total_minutes >= worst_mins - 5 and trips | length > 1 %}
<span style="color:#c53030" title="Slowest option">{{ trip.total_duration }} 🐢</span> <span style="color:#c53030" title="Slowest option">{{ row.total_duration }} 🐢</span>
{% else %} {% else %}
<span style="color:#00539f">{{ trip.total_duration }}</span> <span style="color:#00539f">{{ row.total_duration }}</span>
{% endif %} {% endif %}
</td> </td>
{% else %}
<td style="padding:0.6rem 0.8rem;font-weight:600">&mdash;</td>
<td style="padding:0.6rem 0.8rem">&mdash;</td>
<td style="padding:0.6rem 0.8rem">&mdash;</td>
<td style="padding:0.6rem 0.8rem">Unavailable</td>
<td style="padding:0.6rem 0.8rem;font-weight:600">
{{ row.depart_st_pancras }}
{% if row.train_number %}<br><span style="font-size:0.75rem;font-weight:400;color:#a0aec0">{{ row.train_number }}</span>{% endif %}
</td>
<td style="padding:0.6rem 0.8rem">
{{ row.arrive_destination }}
<span style="font-weight:400;color:#a0aec0;font-size:0.85em">(CET)</span>
</td>
<td style="padding:0.6rem 0.8rem;font-weight:600">
<span title="No same-day Bristol connection">Unavailable from Bristol</span>
</td>
{% endif %}
</tr> </tr>
{% endfor %} {% endfor %}
</tbody> </tbody>
@ -150,6 +183,9 @@
<p style="margin-top:1rem;font-size:0.82rem;color:#718096"> <p style="margin-top:1rem;font-size:0.82rem;color:#718096">
Paddington &rarr; St&nbsp;Pancras connection: {{ min_connection }}&ndash;{{ max_connection }}&nbsp;min. Paddington &rarr; St&nbsp;Pancras connection: {{ min_connection }}&ndash;{{ max_connection }}&nbsp;min.
{% if unreachable_morning_services %}
Morning means Eurostar departures before 12:00 from St&nbsp;Pancras.
{% endif %}
Eurostar times are from the general timetable and may vary; always check Eurostar times are from the general timetable and may vary; always check
<a href="{{ eurostar_url }}" target="_blank" rel="noopener">eurostar.com</a> to book. <a href="{{ eurostar_url }}" target="_blank" rel="noopener">eurostar.com</a> to book.
&nbsp;&middot;&nbsp; &nbsp;&middot;&nbsp;

View file

@ -89,3 +89,166 @@ def test_results_title_and_social_meta_include_destination(monkeypatch):
'to Lille Europe on Friday 10 April 2026 via Paddington, St Pancras, and Eurostar.">' 'to Lille Europe on Friday 10 April 2026 via Paddington, St Pancras, and Eurostar.">'
) in html ) in html
assert '<meta property="og:url" content="http://localhost/results/lille/2026-04-10?min_connection=60&amp;max_connection=120">' in html assert '<meta property="og:url" content="http://localhost/results/lille/2026-04-10?min_connection=60&amp;max_connection=120">' in html
def test_results_marks_trips_within_five_minutes_of_fastest_and_slowest(monkeypatch):
monkeypatch.setattr(app_module, 'get_cached', lambda key: None)
monkeypatch.setattr(app_module, 'set_cached', lambda key, data: None)
monkeypatch.setattr(
app_module.rtt_scraper,
'fetch',
lambda travel_date, user_agent: [
{'depart_bristol': '07:00', 'arrive_paddington': '08:30', 'headcode': '1A01'},
{'depart_bristol': '07:05', 'arrive_paddington': '08:36', 'headcode': '1A02'},
{'depart_bristol': '07:10', 'arrive_paddington': '08:46', 'headcode': '1A03'},
{'depart_bristol': '07:15', 'arrive_paddington': '08:56', 'headcode': '1A04'},
{'depart_bristol': '07:20', 'arrive_paddington': '09:06', 'headcode': '1A05'},
],
)
monkeypatch.setattr(
app_module.eurostar_scraper,
'fetch',
lambda destination, travel_date, user_agent: [
{
'depart_st_pancras': '09:30',
'arrive_destination': '11:50',
'destination': destination,
'train_number': 'ES 1001',
},
{
'depart_st_pancras': '09:40',
'arrive_destination': '12:00',
'destination': destination,
'train_number': 'ES 1002',
},
{
'depart_st_pancras': '09:50',
'arrive_destination': '12:20',
'destination': destination,
'train_number': 'ES 1003',
},
{
'depart_st_pancras': '10:00',
'arrive_destination': '12:35',
'destination': destination,
'train_number': 'ES 1004',
},
{
'depart_st_pancras': '10:10',
'arrive_destination': '12:45',
'destination': destination,
'train_number': 'ES 1005',
},
],
)
monkeypatch.setattr(
app_module.eurostar_scraper,
'timetable_url',
lambda destination: f'https://example.test/{destination.lower().replace(" ", "-")}',
)
client = _client()
resp = client.get('/results/paris/2026-04-10?min_connection=60&max_connection=120')
html = resp.get_data(as_text=True)
assert resp.status_code == 200
assert html.count('title="Fastest option"') == 2
assert html.count('title="Slowest option"') == 2
assert '4h 50m ⚡' in html
assert '4h 55m ⚡' in html
assert '5h 20m 🐢' in html
assert '5h 25m 🐢' in html
assert '5h 10m ⚡' not in html
assert '5h 10m 🐢' not in html
def test_results_shows_unreachable_morning_eurostar_services(monkeypatch):
monkeypatch.setattr(app_module, 'get_cached', lambda key: None)
monkeypatch.setattr(app_module, 'set_cached', lambda key, data: None)
monkeypatch.setattr(
app_module.rtt_scraper,
'fetch',
lambda travel_date, user_agent: [
{'depart_bristol': '07:00', 'arrive_paddington': '08:45', 'headcode': '1A23'},
],
)
monkeypatch.setattr(
app_module.eurostar_scraper,
'fetch',
lambda destination, travel_date, user_agent: [
{
'depart_st_pancras': '09:30',
'arrive_destination': '12:00',
'destination': destination,
'train_number': 'ES 9001',
},
{
'depart_st_pancras': '10:15',
'arrive_destination': '13:40',
'destination': destination,
'train_number': 'ES 9002',
},
{
'depart_st_pancras': '12:30',
'arrive_destination': '15:55',
'destination': destination,
'train_number': 'ES 9003',
},
],
)
monkeypatch.setattr(
app_module.eurostar_scraper,
'timetable_url',
lambda destination: f'https://example.test/{destination.lower().replace(" ", "-")}',
)
client = _client()
resp = client.get('/results/paris/2026-04-10?min_connection=60&max_connection=120')
html = resp.get_data(as_text=True)
assert resp.status_code == 200
assert '1 morning service unavailable from Bristol' in html
assert '09:30' in html
assert 'ES 9001' in html
assert 'Unavailable from Bristol' in html
assert 'Morning means Eurostar departures before 12:00 from St&nbsp;Pancras.' in html
assert html.index('09:30') < html.index('10:15')
def test_results_can_show_only_unreachable_morning_services(monkeypatch):
monkeypatch.setattr(app_module, 'get_cached', lambda key: None)
monkeypatch.setattr(app_module, 'set_cached', lambda key, data: None)
monkeypatch.setattr(
app_module.rtt_scraper,
'fetch',
lambda travel_date, user_agent: [
{'depart_bristol': '07:00', 'arrive_paddington': '08:45', 'headcode': '1A23'},
],
)
monkeypatch.setattr(
app_module.eurostar_scraper,
'fetch',
lambda destination, travel_date, user_agent: [
{
'depart_st_pancras': '09:30',
'arrive_destination': '12:00',
'destination': destination,
'train_number': 'ES 9001',
},
],
)
monkeypatch.setattr(
app_module.eurostar_scraper,
'timetable_url',
lambda destination: f'https://example.test/{destination.lower().replace(" ", "-")}',
)
client = _client()
resp = client.get('/results/paris/2026-04-10?min_connection=60&max_connection=120')
html = resp.get_data(as_text=True)
assert resp.status_code == 200
assert 'No valid journeys found.' not in html
assert '1 morning service unavailable from Bristol' in html
assert '09:30' in html
assert 'Unavailable from Bristol' in html

View file

@ -1,5 +1,5 @@
import pytest import pytest
from trip_planner import combine_trips, _fmt_duration from trip_planner import combine_trips, find_unreachable_morning_eurostars, _fmt_duration, cheapest_gwr_ticket
DATE = '2026-03-30' DATE = '2026-03-30'
@ -129,3 +129,88 @@ def test_connection_duration_in_trip():
# arrive Paddington 08:45, depart St Pancras 10:01 → 1h 16m # arrive Paddington 08:45, depart St Pancras 10:01 → 1h 16m
trips = combine_trips([GWR_FAST], [ES_PARIS], DATE) trips = combine_trips([GWR_FAST], [ES_PARIS], DATE)
assert trips[0]['connection_duration'] == '1h 16m' assert trips[0]['connection_duration'] == '1h 16m'
def test_unreachable_morning_eurostars_lists_only_unreachable_morning_services():
gwr = [
{'depart_bristol': '07:00', 'arrive_paddington': '08:45'},
]
eurostar = [
{'depart_st_pancras': '09:30', 'arrive_destination': '12:00', 'destination': 'Paris Gare du Nord', 'train_number': 'ES 9001'},
{'depart_st_pancras': '10:15', 'arrive_destination': '13:40', 'destination': 'Paris Gare du Nord', 'train_number': 'ES 9002'},
{'depart_st_pancras': '12:30', 'arrive_destination': '15:55', 'destination': 'Paris Gare du Nord', 'train_number': 'ES 9003'},
]
unreachable = find_unreachable_morning_eurostars(gwr, eurostar, DATE)
assert [service['depart_st_pancras'] for service in unreachable] == ['09:30']
# ---------------------------------------------------------------------------
# cheapest_gwr_ticket — Bristol Temple Meads → Paddington
# ---------------------------------------------------------------------------
# 2026-03-30 is a Monday; 2026-03-28 is a Saturday
def test_cheapest_ticket_weekday_super_off_peak_morning():
# 05:00 on Monday: dep ≤ 05:04 → Super Off-Peak
t = cheapest_gwr_ticket('05:00', '2026-03-30')
assert t['ticket'] == 'Super Off-Peak'
assert t['price'] == 45.00
def test_cheapest_ticket_weekday_anytime_window():
# 07:00 on Monday: 05:0508:25 → Anytime only
t = cheapest_gwr_ticket('07:00', '2026-03-30')
assert t['ticket'] == 'Anytime'
assert t['price'] == 138.70
def test_cheapest_ticket_weekday_off_peak():
# 08:30 on Monday: dep ≥ 08:26 but < 09:58 → Off-Peak
t = cheapest_gwr_ticket('08:30', '2026-03-30')
assert t['ticket'] == 'Off-Peak'
assert t['price'] == 63.60
def test_cheapest_ticket_weekday_super_off_peak_late():
# 10:00 on Monday: dep ≥ 09:58 → Super Off-Peak
t = cheapest_gwr_ticket('10:00', '2026-03-30')
assert t['ticket'] == 'Super Off-Peak'
assert t['price'] == 45.00
def test_cheapest_ticket_boundary_super_off_peak_cutoff():
# 05:04 is last valid minute for early Super Off-Peak
assert cheapest_gwr_ticket('05:04', '2026-03-30')['ticket'] == 'Super Off-Peak'
# 05:05 falls into the Anytime window (off-peak starts at 08:26)
assert cheapest_gwr_ticket('05:05', '2026-03-30')['ticket'] == 'Anytime'
def test_cheapest_ticket_boundary_off_peak_start():
assert cheapest_gwr_ticket('08:25', '2026-03-30')['ticket'] == 'Anytime'
assert cheapest_gwr_ticket('08:26', '2026-03-30')['ticket'] == 'Off-Peak'
def test_cheapest_ticket_boundary_super_off_peak_resumes():
assert cheapest_gwr_ticket('09:57', '2026-03-30')['ticket'] == 'Off-Peak'
assert cheapest_gwr_ticket('09:58', '2026-03-30')['ticket'] == 'Super Off-Peak'
def test_cheapest_ticket_weekend_always_super_off_peak():
# Saturday — no restrictions
t = cheapest_gwr_ticket('07:00', '2026-03-28')
assert t['ticket'] == 'Super Off-Peak'
assert t['price'] == 45.00
def test_combine_trips_includes_ticket_fields():
trips = combine_trips([GWR_FAST], [ES_PARIS], DATE)
assert len(trips) == 1
t = trips[0]
assert 'ticket_name' in t
assert 'ticket_price' in t
assert 'ticket_code' in t
def test_unreachable_morning_eurostars_returns_empty_when_morning_service_is_reachable():
gwr = [
{'depart_bristol': '07:00', 'arrive_paddington': '08:45'},
]
eurostar = [
{'depart_st_pancras': '10:15', 'arrive_destination': '13:40', 'destination': 'Paris Gare du Nord', 'train_number': 'ES 9002'},
]
assert find_unreachable_morning_eurostars(gwr, eurostar, DATE) == []

View file

@ -1,15 +1,47 @@
""" """
Combine GWR BristolPaddington trains with Eurostar St Pancrasdestination trains. Combine GWR BristolPaddington trains with Eurostar St Pancrasdestination trains.
""" """
from datetime import datetime, timedelta from datetime import datetime, timedelta, time as _time
MIN_CONNECTION_MINUTES = 50 MIN_CONNECTION_MINUTES = 50
MAX_CONNECTION_MINUTES = 110 MAX_CONNECTION_MINUTES = 110
MAX_GWR_MINUTES = 110 MAX_GWR_MINUTES = 110
MORNING_CUTOFF_HOUR = 12
DATE_FMT = '%Y-%m-%d' DATE_FMT = '%Y-%m-%d'
TIME_FMT = '%H:%M' TIME_FMT = '%H:%M'
# Bristol Temple Meads → London Paddington walk-on single fares.
# Weekday restrictions (MonFri 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:0005: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 (MonFri) 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: def _parse_dt(date: str, time: str) -> datetime:
return datetime.strptime(f"{date} {time}", f"{DATE_FMT} {TIME_FMT}") return datetime.strptime(f"{date} {time}", f"{DATE_FMT} {TIME_FMT}")
@ -23,6 +55,36 @@ def _fmt_duration(minutes: int) -> str:
return f"{m}m" 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( def combine_trips(
gwr_trains: list[dict], gwr_trains: list[dict],
eurostar_trains: list[dict], eurostar_trains: list[dict],
@ -46,35 +108,21 @@ def combine_trips(
trips = [] trips = []
for gwr in gwr_trains: 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 # Find only the earliest viable Eurostar for this GWR departure
for es in eurostar_trains: for es in eurostar_trains:
try: connection = _is_viable_connection(
dep_stp = _parse_dt(travel_date, es['depart_st_pancras']) gwr,
arr_dest = _parse_dt(travel_date, es['arrive_destination']) es,
except (ValueError, KeyError): travel_date,
continue min_connection_minutes,
max_connection_minutes,
# Eurostar arrives next day? (e.g. night service — unlikely but handle it) )
if arr_dest < dep_stp: if not connection:
arr_dest += timedelta(days=1)
if dep_stp < earliest_eurostar:
continue
if (dep_stp - arr_pad).total_seconds() / 60 > max_connection_minutes:
continue continue
dep_bri, arr_pad, dep_stp, arr_dest = connection
total_mins = int((arr_dest - dep_bri).total_seconds() / 60) total_mins = int((arr_dest - dep_bri).total_seconds() / 60)
ticket = cheapest_gwr_ticket(gwr['depart_bristol'], travel_date)
trips.append({ trips.append({
'depart_bristol': gwr['depart_bristol'], 'depart_bristol': gwr['depart_bristol'],
'arrive_paddington': gwr['arrive_paddington'], 'arrive_paddington': gwr['arrive_paddington'],
@ -87,8 +135,46 @@ def combine_trips(
'total_duration': _fmt_duration(total_mins), 'total_duration': _fmt_duration(total_mins),
'total_minutes': total_mins, 'total_minutes': total_mins,
'destination': es['destination'], 'destination': es['destination'],
'ticket_name': ticket['ticket'],
'ticket_price': ticket['price'],
'ticket_code': ticket['code'],
}) })
break # Only the earliest valid Eurostar per GWR departure break # Only the earliest valid Eurostar per GWR departure
trips.sort(key=lambda t: (t['depart_bristol'], t['depart_st_pancras'])) trips.sort(key=lambda t: (t['depart_bristol'], t['depart_st_pancras']))
return trips 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:
try:
dep_stp = _parse_dt(travel_date, es['depart_st_pancras'])
except (ValueError, KeyError):
continue
if dep_stp.hour >= MORNING_CUTOFF_HOUR:
continue
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'])