engine/model.py

"""Compiled network: the in-memory layout RAPTOR scans.

The key idea (RAPTOR's speed trick): group trips into *patterns* — sets of
trips that visit the exact same ordered sequence of stops. Each pattern is a
"route" here. Within a route, trips are sorted by departure time, so boarding
the earliest catchable trip is a binary search, and a single linear sweep down
the stop sequence relaxes every downstream arrival.
"""
from __future__ import annotations

import math
from dataclasses import dataclass, field


@dataclass
class Route:
    """A unique stop-pattern and all trips that run it (sorted by departure)."""
    stops: list[int]                       # ordered internal stop indices
    arr: list[list[int]]                   # arr[trip][pos] -> arrival sec
    dep: list[list[int]]                   # dep[trip][pos] -> departure sec
    route_id: str = ""                     # source GTFS route_id (for label)
    route_type: int = 3                    # GTFS route_type (1=metro, 3=bus, 2=rail)
    route_color: str = ""                  # GTFS route_color hex (no '#'), e.g. "00B300"
    route_name: str = ""                   # short name for labels, e.g. "1" or "165"
    shape_id: str = ""                     # GTFS shape this pattern follows (key into Network.shapes)
    stop_shape_idx: list[float] = field(default_factory=list)  # per stop: float pos (seg+t) on the shape

    # Transposed departure columns: dep_cols[pos] = packed array of every trip's
    # departure at that stop position, ascending (trips are sorted, no overtaking).
    # Lets the "earliest boardable trip" lookup be a C-level bisect over a flat
    # array instead of a Python binary search over dep[mid][pos]. Built lazily.
    dep_cols: object = None

    @property
    def n_trips(self) -> int:
        return len(self.dep)


@dataclass
class Network:
    # --- stops (index == internal stop id) ---
    stop_ids: list[str] = field(default_factory=list)
    stop_name: list[str] = field(default_factory=list)
    stop_lat: list[float] = field(default_factory=list)
    stop_lon: list[float] = field(default_factory=list)
    stop_index: dict[str, int] = field(default_factory=dict)

    # --- routes / patterns ---
    routes: list[Route] = field(default_factory=list)
    # per stop: list of (route_idx, position-of-stop-within-route)
    stop_routes: list[list[tuple[int, int]]] = field(default_factory=list)
    # per stop: list of (to_stop_idx, walk_seconds) footpath transfers
    transfers: list[list[tuple[int, float]]] = field(default_factory=list)
    # transfer leg geometry (street path), keyed by (from_stop, to_stop). Filled
    # once the OSM walk graph exists; absent => draw a straight line.
    transfer_geom: dict = field(default_factory=dict)

    # shape_id -> polyline [[lon, lat], ...] for tracing transit legs along the route
    shapes: dict = field(default_factory=dict)

    # metadata
    service_date: str = ""
    feeds: list[str] = field(default_factory=list)

    @property
    def n_stops(self) -> int:
        return len(self.stop_ids)


def haversine_m(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
    """Great-circle distance in metres."""
    r = 6_371_000.0
    p1, p2 = math.radians(lat1), math.radians(lat2)
    dphi = math.radians(lat2 - lat1)
    dlmb = math.radians(lon2 - lon1)
    a = math.sin(dphi / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dlmb / 2) ** 2
    return 2 * r * math.asin(math.sqrt(a))


def parse_gtfs_time(t: str) -> int:
    """'HH:MM:SS' (hours may exceed 24) -> seconds since midnight."""
    h, m, s = t.split(":")
    return int(h) * 3600 + int(m) * 60 + int(s)


def project_stops_to_shape(stop_lonlat: list[tuple[float, float]],
                           shape: list[list[float]]) -> list[float]:
    """For each stop, its float position `segment_index + t` (t in [0,1)) of the
    NEAREST POINT on the shape polyline, scanning forward (monotonic) and made
    strictly increasing.

    No shape_dist_traveled in the feed, so we map each stop to the closest point
    on the polyline. Nearest *point on a segment* (not nearest vertex): a stop in
    the middle of a long, sparsely-sampled express segment maps onto that segment
    instead of jumping to a far vertex — the bug that crammed the remaining stops
    and collapsed legs into straight cross-town cuts. Strict increase keeps every
    leg's slice non-empty.
    """
    n = len(shape)
    out: list[float] = []
    if n < 2:
        return [0.0] * len(stop_lonlat)
    seg_start = 0
    for lon, lat in stop_lonlat:
        best_d2, best_pos, best_seg = float("inf"), float(seg_start), seg_start
        for s in range(seg_start, n - 1):
            ax, ay = shape[s]
            dx, dy = shape[s + 1][0] - ax, shape[s + 1][1] - ay
            seg2 = dx * dx + dy * dy
            t = 0.0 if seg2 == 0 else ((lon - ax) * dx + (lat - ay) * dy) / seg2
            t = 0.0 if t < 0 else 1.0 if t > 1 else t
            px, py = ax + t * dx, ay + t * dy
            d2 = (lon - px) ** 2 + (lat - py) ** 2
            if d2 < best_d2:
                best_d2, best_pos, best_seg = d2, s + t, s
        if out and best_pos <= out[-1]:
            best_pos = min(out[-1] + 1e-3, n - 1.0)
        out.append(best_pos)
        seg_start = best_seg  # monotonic progress along the shape
    return out


def seconds_to_hhmmss(sec: int) -> str:
    sec %= 24 * 3600
    return f"{sec // 3600:02d}:{(sec % 3600) // 60:02d}:{sec % 60:02d}"