feat(realtime): real-time GTFS-RT departure delay buffer

examples/config.toml

@@ -33,6 +33,20 @@ rain_buffer_minutes = 10
 snow_buffer_minutes = 20
 precip_probability_threshold = 50
 
+[realtime]
+# Pad the departure buffer when the line you board is running late, using the
+# MTA GTFS-RT trip-update feeds.  Conservative: a live delay can only move your
+# leave time *earlier* (never later) and is capped at max_buffer_minutes.
+# Disabled by default.  Feed URLs default to the current MTA endpoints — verify
+# against https://api.mta.info if matching stops working.
+enabled = false
+max_buffer_minutes = 15
+min_delay_minutes = 2
+match_window_minutes = 30
+fuzzy_threshold = 80
+# subway_tripupdate_urls / lirr_tripupdate_url / bus_tripupdate_url default to
+# the canonical MTA + OneBusAway trip-update feeds; override only if they move.
+
 [monitoring]
 # Optional dead-man's-switch.  poll_staleness_minutes controls how long the poll
 # loop may be silent before the morning digest warns the timer is dead.  Set

pyproject.toml

@@ -26,6 +26,18 @@ build-backend = "hatchling.build"
 
 [tool.hatch.build.targets.wheel]
 packages = ["src/commutecompass"]
+# Bundled GTFS stop tables consumed by commutecompass.realtime (generated by
+# scripts/build_stops.py).  Hatchling includes non-.py files under the package
+# by default, but pin it explicitly so a stray build config can't drop them.
+artifacts = ["src/commutecompass/data/*.csv"]
+
+[tool.hatch.build.targets.sdist]
+include = [
+  "src/commutecompass",
+  "scripts/build_stops.py",
+  "tests",
+]
+artifacts = ["src/commutecompass/data/*.csv"]
 
 [tool.ruff]
 line-length = 100

scripts/build_stops.py

@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+"""Generate the bundled GTFS stop tables used by :mod:`commutecompass.realtime`.
+
+The real-time departure buffer fuzzy-matches a Google Directions boarding-stop
+*name* to a GTFS ``stop_id`` so it can look that stop up in the matching
+GTFS-RT trip-update feed.  That mapping comes from the MTA's **static** GTFS
+``stops.txt``.  Rather than depend on a multi-megabyte feed at runtime, we
+distill each system's ``stops.txt`` into a compact CSV checked into the package
+(``src/commutecompass/data/stops_{subway,lirr,bus}.csv``).
+
+Run this to refresh the bundled data when the MTA reorganizes stops:
+
+    python scripts/build_stops.py
+
+It downloads the current static GTFS zips, extracts ``stops.txt``, and writes
+the three CSVs.  Network access is required (only when refreshing).
+
+CSV schema (subway, lirr): ``stop_id,stop_name,parent_station``
+CSV schema (bus):          ``stop_id,stop_name``
+
+Subway stop ids are directional at the platform level (``R20N``/``R20S``) with a
+parent station (``R20``).  We keep the parent stations (so a name matches once)
+plus their ``parent_station`` linkage; ``realtime`` expands a parent to its
+``…N``/``…S`` children when querying the feed.  Bus stops are already
+directional, so no parent column is needed.
+"""
+
+from __future__ import annotations
+
+import csv
+import io
+import sys
+import zipfile
+from pathlib import Path
+from typing import Iterable
+
+import httpx
+
+DATA_DIR = Path(__file__).resolve().parent.parent / "src" / "commutecompass" / "data"
+
+# Static GTFS bundles (verify against https://www.mta.info/developers).
+SUBWAY_GTFS_URL = "https://rrgtfsfeeds.s3.amazonaws.com/gtfs_subway.zip"
+LIRR_GTFS_URL = "https://rrgtfsfeeds.s3.amazonaws.com/gtfslirr.zip"
+# NYC bus static GTFS is split per operator; merge them all.
+BUS_GTFS_URLS = [
+    "https://rrgtfsfeeds.s3.amazonaws.com/gtfs_b.zip",  # Brooklyn
+    "https://rrgtfsfeeds.s3.amazonaws.com/gtfs_bx.zip",  # Bronx
+    "https://rrgtfsfeeds.s3.amazonaws.com/gtfs_m.zip",  # Manhattan
+    "https://rrgtfsfeeds.s3.amazonaws.com/gtfs_q.zip",  # Queens
+    "https://rrgtfsfeeds.s3.amazonaws.com/gtfs_si.zip",  # Staten Island
+    "https://rrgtfsfeeds.s3.amazonaws.com/gtfs_busco.zip",  # MTA Bus Company
+]
+
+
+def _read_stops(zip_url: str) -> list[dict[str, str]]:
+    """Download a GTFS zip and return its ``stops.txt`` rows as dicts."""
+    print(f"  fetching {zip_url}", file=sys.stderr)
+    resp = httpx.get(zip_url, timeout=60.0, follow_redirects=True)
+    resp.raise_for_status()
+    with zipfile.ZipFile(io.BytesIO(resp.content)) as zf:
+        with zf.open("stops.txt") as fh:
+            text = io.TextIOWrapper(fh, encoding="utf-8-sig")
+            return list(csv.DictReader(text))
+
+
+def _write_csv(path: Path, rows: Iterable[tuple[str, ...]], header: tuple[str, ...]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    seen: set[str] = set()
+    count = 0
+    with path.open("w", newline="", encoding="utf-8") as fh:
+        writer = csv.writer(fh)
+        writer.writerow(header)
+        for row in rows:
+            key = row[0]
+            if key in seen:
+                continue
+            seen.add(key)
+            writer.writerow(row)
+            count += 1
+    print(f"  wrote {count} rows -> {path}", file=sys.stderr)
+
+
+def build_rail(zip_url: str, out_name: str) -> None:
+    """Subway/LIRR: keep parent stations (and their parent linkage)."""
+    rows = _read_stops(zip_url)
+    out: list[tuple[str, str, str]] = []
+    for r in rows:
+        stop_id = (r.get("stop_id") or "").strip()
+        name = (r.get("stop_name") or "").strip()
+        parent = (r.get("parent_station") or "").strip()
+        if not stop_id or not name:
+            continue
+        # Prefer parent stations (location_type==1) and any stop without a parent
+        # — those are the named stations a rider boards at.  Directional children
+        # are reconstructed at query time.
+        location_type = (r.get("location_type") or "").strip()
+        if location_type == "1" or not parent:
+            out.append((stop_id, name, parent))
+    _write_csv(DATA_DIR / out_name, out, ("stop_id", "stop_name", "parent_station"))
+
+
+def build_bus(out_name: str) -> None:
+    rows: list[tuple[str, str]] = []
+    for url in BUS_GTFS_URLS:
+        for r in _read_stops(url):
+            stop_id = (r.get("stop_id") or "").strip()
+            name = (r.get("stop_name") or "").strip()
+            if stop_id and name:
+                rows.append((stop_id, name))
+    _write_csv(DATA_DIR / out_name, rows, ("stop_id", "stop_name"))
+
+
+def main() -> int:
+    print("building subway stops...", file=sys.stderr)
+    build_rail(SUBWAY_GTFS_URL, "stops_subway.csv")
+    print("building LIRR stops...", file=sys.stderr)
+    build_rail(LIRR_GTFS_URL, "stops_lirr.csv")
+    print("building bus stops...", file=sys.stderr)
+    build_bus("stops_bus.csv")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

skills/commutecompass/SKILL.md

@@ -38,6 +38,7 @@ to stdout; relay that stdout back to the user.
 | "mute pings for event X" / "mute everything today"                       | `scripts/mute.sh <selector>` *or* `scripts/mute.sh --today` |
 | "unmute event X"                                                         | `scripts/unmute.sh <selector>`                    |
 | "what alerts are hitting my commute today?"                              | `scripts/mta-alerts.sh`                           |
+| "is my train running late?" / "any real-time delays on my commute?"      | `scripts/realtime.sh`                             |
 | "send me today's digest again" / "force-run morning"                     | `scripts/morning.sh`                              |
 | "run a poll cycle now" / "check alerts now"                              | `scripts/poll.sh`                                 |
 | "what time will my alarm be tomorrow?" / "preview tomorrow's wake time"  | `scripts/tomorrow.sh` (dry-run; no HA push)       |
@@ -82,7 +83,7 @@ purpose.
   the user before running them on demand if the cause for the user's request is
   unclear.
 - `digest-preview`, `where`, `plan-event` (without `--from`), `config-show`,
-  and `mta-alerts` are pure reads — invoke freely. `plan-event --from <addr>`
+  `mta-alerts`, and `realtime` are pure reads — invoke freely. `plan-event --from <addr>`
   is also a read (preview only; never saves).
 - `adjust` only shifts `prep_at`. The `leave_at` is governed by route+event
   start and can't be moved without a replan. If the user wants to leave

skills/commutecompass/references/config-allowlist.md

@@ -18,6 +18,8 @@ intentionally read-only via the skill and must be changed by editing
 | `home_assistant.tomorrow.enabled`         | bool        | Toggle the pull-model "tomorrow alarm" push     |
 | `home_assistant.replan_window_minutes`    | int (min)   | Window before leave_at in which to replan       |
 | `home_assistant.max_age_minutes`          | int (min)   | Max acceptable age of an HA location reading    |
+| `realtime.enabled`                        | bool        | Toggle real-time GTFS-RT departure delay buffer |
+| `realtime.max_buffer_minutes`             | int (min)   | Cap on minutes a live delay can add to leave time |
 
 ## How to revert
 

skills/commutecompass/scripts/realtime.sh

@@ -0,0 +1,3 @@
+#!/usr/bin/env bash
+set -euo pipefail
+exec commutecompass-skill realtime "$@"

src/commutecompass/cli.py

@@ -1100,6 +1100,51 @@ def mta_alerts(ctx: click.Context) -> None:
     click.echo(format_alerts_block(matched))
 
 
+@cli.command(name="realtime")
+@click.pass_context
+def realtime_cmd(ctx: click.Context) -> None:
+    """Show real-time departure delays for today's planned boarding legs.
+
+    Reads the MTA GTFS-RT trip-update feeds and reports how late (if at all) the
+    line each plan boards is running.  Honors ``realtime.enabled`` — when
+    disabled it says so rather than fetching.
+    """
+    from commutecompass.realtime import realtime_delay
+    from commutecompass.store import Store
+
+    config_path: Path = ctx.obj["config_path"]
+    cfg = _load_config(config_path)
+
+    if not cfg.realtime.enabled:
+        click.echo("Real-time departures are disabled (set realtime.enabled = true).")
+        return
+
+    store = Store(Path(cfg.paths.db_path))
+    plans = store.today_plans()
+    plans_with_routes = [p for p in plans if p.route is not None and p.leave_at is not None]
+    if not plans_with_routes:
+        click.echo("No planned routes today — nothing to check.")
+        return
+
+    any_delay = False
+    for plan in plans_with_routes:
+        assert plan.route is not None
+        delay = realtime_delay(plan.route, plan.event.start, cfg.realtime)
+        boarding = next((leg for leg in plan.route.legs if leg.mode == "TRANSIT"), None)
+        if boarding and boarding.line and boarding.departure_stop:
+            board_desc = f"{boarding.line} from {boarding.departure_stop}"
+        else:
+            board_desc = "(no transit leg)"
+        if delay.reason:
+            any_delay = True
+            click.echo(f"{plan.event.title}: {board_desc} → {delay.reason}")
+        else:
+            click.echo(f"{plan.event.title}: {board_desc} → on time")
+
+    if not any_delay:
+        click.echo("All boarding legs on time.")
+
+
 # ─────────── bot (stub) ──────────────────────────────────────────────────────
 
 

src/commutecompass/config.py

@@ -158,6 +158,67 @@ def _validate_forecast_url(cls, v: str) -> str:
         return v.rstrip("/")
 
 
+# Canonical MTA GTFS-RT *trip-update* feeds (distinct from the alert feeds in
+# mta.py).  Subway is split into line-group feeds; verify against
+# https://api.mta.info.  Bus trip-updates come from OneBusAway.
+_MTA_BASE = "https://api-endpoint.mta.info/Dataservice/mtagtfsfeeds"
+_DEFAULT_SUBWAY_TRIPUPDATE_URLS = [
+    f"{_MTA_BASE}/nyct%2Fgtfs",  # 1 2 3 4 5 6 7 S
+    f"{_MTA_BASE}/nyct%2Fgtfs-ace",
+    f"{_MTA_BASE}/nyct%2Fgtfs-bdfm",
+    f"{_MTA_BASE}/nyct%2Fgtfs-g",
+    f"{_MTA_BASE}/nyct%2Fgtfs-jz",
+    f"{_MTA_BASE}/nyct%2Fgtfs-nqrw",
+    f"{_MTA_BASE}/nyct%2Fgtfs-l",
+    f"{_MTA_BASE}/nyct%2Fgtfs-si",
+]
+_DEFAULT_LIRR_TRIPUPDATE_URL = f"{_MTA_BASE}/lirr%2Fgtfs-lirr"
+_DEFAULT_BUS_TRIPUPDATE_URL = "https://gtfsrt.prod.obanyc.com/tripUpdates"
+
+
+class RealtimeConfig(BaseModel):
+    """Real-time departure buffer: pad leave time when the boarding line runs late.
+
+    Reads the MTA GTFS-RT *trip-update* feeds, fuzzy-matches the boarding stop to
+    a GTFS ``stop_id`` (via the bundled ``data/stops_*.csv``), and folds an
+    observed delay into the buffer so the alarm fires earlier.  Conservative by
+    design: a delay can only move the leave time *earlier*, never later, and is
+    capped — see :mod:`commutecompass.realtime`.  Disabled by default.
+    """
+
+    enabled: bool = False
+    subway_tripupdate_urls: list[str] = Field(
+        default_factory=lambda: list(_DEFAULT_SUBWAY_TRIPUPDATE_URLS)
+    )
+    lirr_tripupdate_url: str = _DEFAULT_LIRR_TRIPUPDATE_URL
+    bus_tripupdate_url: str = _DEFAULT_BUS_TRIPUPDATE_URL
+    # Most an observed delay can add to the buffer (guards against bad feed data).
+    max_buffer_minutes: int = Field(default=15, ge=0, le=120)
+    # Ignore delays smaller than this (feed jitter / rounding noise).
+    min_delay_minutes: int = Field(default=2, ge=0, le=60)
+    # How far from the scheduled departure to look for the matching trip.
+    match_window_minutes: int = Field(default=30, ge=1, le=180)
+    # rapidfuzz score cutoff (0-100) for boarding-stop name matching.
+    fuzzy_threshold: int = Field(default=80, ge=0, le=100)
+
+    @field_validator("subway_tripupdate_urls")
+    @classmethod
+    def _validate_subway_urls(cls, v: list[str]) -> list[str]:
+        for url in v:
+            if url and not (url.startswith("http://") or url.startswith("https://")):
+                raise ValueError(
+                    f"realtime.subway_tripupdate_urls entries must start with http(s)://, got {url!r}"
+                )
+        return v
+
+    @field_validator("lirr_tripupdate_url", "bus_tripupdate_url")
+    @classmethod
+    def _validate_tripupdate_url(cls, v: str) -> str:
+        if v and not (v.startswith("http://") or v.startswith("https://")):
+            raise ValueError(f"realtime trip-update URL must start with http(s)://, got {v!r}")
+        return v
+
+
 class MonitoringConfig(BaseModel):
     """Dead-man's-switch / heartbeat configuration.
 
@@ -194,6 +255,7 @@ class Config(BaseModel):
     notify: NotifyConfig = NotifyConfig()
     monitoring: MonitoringConfig = MonitoringConfig()
     weather: WeatherConfig = WeatherConfig()
+    realtime: RealtimeConfig = RealtimeConfig()
     # Loaded from env, not TOML:
     google_maps_api_key: str = ""
     google_oauth_client_secret_json: str = ""
@@ -383,6 +445,9 @@ def _coerce_bool(value: str) -> bool:
     "home_assistant.tomorrow.enabled": _coerce_bool,
     "home_assistant.replan_window_minutes": _coerce_int,
     "home_assistant.max_age_minutes": _coerce_int,
+    # Real-time departure buffer — safe to toggle / tune from chat.
+    "realtime.enabled": _coerce_bool,
+    "realtime.max_buffer_minutes": _coerce_int,
 }