bristol-eurostar/trip_planner.py

"""
Combine GWR station→Paddington trains with Eurostar St Pancras→destination trains.
"""

from datetime import datetime, timedelta

import circle_line

MIN_CONNECTION_MINUTES = 50
MAX_CONNECTION_MINUTES = 110
DATE_FMT = "%Y-%m-%d"
TIME_FMT = "%H:%M"

PAD_WALK_TO_UNDERGROUND_MINUTES = 8  # GWR platform → Paddington (H&C Line) platform


def _parse_dt(date: str, time: str) -> datetime:
    return datetime.strptime(f"{date} {time}", f"{DATE_FMT} {TIME_FMT}")


def _circle_line_services(arrive_paddington: datetime) -> list[dict]:
    """
    Given GWR arrival at Paddington, return up to 2 upcoming Circle line services
    as [{'depart': 'HH:MM', 'arrive_kx': 'HH:MM'}, ...].

    Each entry gives the departure from Paddington (H&C Line) and the actual
    arrival at King's Cross St Pancras underground station.
    """
    earliest_board = arrive_paddington + timedelta(
        minutes=PAD_WALK_TO_UNDERGROUND_MINUTES
    )
    services = circle_line.upcoming_services(earliest_board, count=2)
    return [
        {"depart": dep.strftime(TIME_FMT), "arrive_kx": arr.strftime(TIME_FMT)}
        for dep, arr in services
    ]


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 arr_pad < dep_bri:
        arr_pad += timedelta(days=1)

    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,
    gwr_fares: dict | None = None,
) -> 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
            fare = (gwr_fares or {}).get(gwr["depart_bristol"])
            circle_svcs = _circle_line_services(arr_pad)
            trips.append(
                {
                    "depart_bristol": gwr["depart_bristol"],
                    "arrive_paddington": gwr["arrive_paddington"],
                    "arrive_platform": gwr.get("arrive_platform", ""),
                    "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_services": circle_svcs,
                    "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": fare["ticket"] if fare else None,
                    "ticket_price": fare["price"] if fare else None,
                    "ticket_code": fare["code"] if fare else None,
                }
            )
            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"])