Punctuality

# pip install polars requests
import io
import requests
import polars as pl

from datetime import datetime, timedelta, timezone

yesterday_noon = (datetime.now(timezone.utc) - timedelta(days=1)).replace(
    hour=12, minute=0, second=0, microsecond=0
)
params = {"timestamp": yesterday_noon.isoformat()}

url = "https://api.mobilitytwin.brussels/sncb/punctuality"
data = requests.get(url, headers={
    "Authorization": "Bearer [MY_API_KEY]"
}, params=params).content

df = pl.read_parquet(io.BytesIO(data))
print(f"rows: {df.height:,}")

# Note: SNCB's GTFS-RT feed does not populate stop_sequence (always null).
# Join to static GTFS on stop_id when needed.
not_cancelled = df.filter(~pl.col("cancelled") & pl.col("arrival_delay").is_not_null())

# Trains with the largest arrival delays, and where they were observed.
print("\nTop 20 worst delays of the day:")
print(not_cancelled.sort("arrival_delay", descending=True)
      .select(["trip_id", "stop_id", "arrival_time", "arrival_delay"])
      .head(20))

# Cancellation rate per route (one row per cancelled trip).
cancelled = df.filter(pl.col("cancelled"))
print(f"\ncancelled trip-instances: "
      f"{cancelled.select(['trip_id', 'start_date', 'start_time']).unique().height}")
print("\nMost-cancelled routes:")
print(cancelled.group_by("route_id").len()
      .sort("len", descending=True).head(10))