Sncb

import requests
import geopandas as gpd

from datetime import datetime, timedelta, timezone

end = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0)
start = end - timedelta(hours=1)
params = {
    "start_timestamp": start.isoformat(),      # ISO accepted; epoch also works
    "end_timestamp": end.isoformat(),
}

url = "https://api.mobilitytwin.brussels/sncb/vehicle-position"

data = requests.get(url, headers={
        'Authorization': 'Bearer [MY_API_KEY]'
}, params=params).json()

gdf = gpd.GeoDataFrame.from_features(data["features"])

# Show all currently running trains with their route
for _, train in gdf.iterrows():
    print(f"{train['name_start']} → {train['name_end']} ({train['trip_headsign']})")

# Plot train positions on a map
gdf.plot(figsize=(10, 8), color="#006ab3", markersize=10)

import requests
import movingpandas as mpd

from datetime import datetime, timedelta, timezone

end = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0)
start = end - timedelta(hours=1)
params = {
    "start_timestamp": start.isoformat(),      # ISO accepted; epoch also works
    "end_timestamp": end.isoformat(),
}

url = "https://api.mobilitytwin.brussels/sncb/trips"

data = requests.get(url, headers={
        'Authorization': 'Bearer [MY_API_KEY]'
}, params=params).json()

# Load as a MovingPandas TrajectoryCollection
tc = mpd.io.read_mf_dict(data, traj_id_property="trip_id")

# Print summary of each trajectory
for traj in tc.trajectories:
    print(f"Trip {traj.id}: {traj.get_length():.0f}m, duration: {traj.get_duration()}")

# Plot all train trajectories on a map
tc.plot(figsize=(12, 8), linewidth=1)

import gtfs_kit as gk
import requests
import tempfile

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(),   # ISO accepted; epoch also works
}

url = "https://api.mobilitytwin.brussels/sncb/gtfs"

data = requests.get(url, headers={
        'Authorization': 'Bearer [MY_API_KEY]'
}, params=params).content

with tempfile.NamedTemporaryFile(delete=False, suffix=".zip") as f:
    f.write(data)

feed = gk.read_feed(f.name, 'm')

# List all routes
print(feed.routes[['route_short_name', 'route_long_name']])

# Count trips per route
trips_per_route = feed.trips.groupby('route_id').size()
print(f"\nTotal routes: {len(feed.routes)}, Total trips: {len(feed.trips)}")

# pip install gtfs-parquet>=0.4.0
import requests
import tempfile

from gtfs_parquet import read_parquet
from gtfs_parquet.ops.network import describe
from gtfs_parquet.ops.calendar import get_first_week, compute_busiest_date
from gtfs_parquet.ops.routes import compute_route_stats

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(),   # ISO accepted; epoch also works
}

url = "https://api.mobilitytwin.brussels/sncb/gtfs-parquet"

data = requests.get(url, headers={
        'Authorization': 'Bearer [MY_API_KEY]'
}, params=params).content

with tempfile.NamedTemporaryFile(delete=False, suffix=".zip") as f:
    f.write(data)

feed = read_parquet(f.name)

# Quick summary of the feed
print(describe(feed))

# Find the busiest date in the first week
week = get_first_week(feed)
busiest = compute_busiest_date(feed, week)
print(f"Busiest date: {busiest}")

# Compute per-route statistics for that week
route_stats = compute_route_stats(feed, week)
print(route_stats.sort("num_trips", descending=True))

import requests
from google.transit import gtfs_realtime_pb2

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(),   # ISO accepted; epoch also works
}

url = "https://api.mobilitytwin.brussels/sncb/gtfs-realtime"

data = requests.get(url, headers={
        'Authorization': 'Bearer [MY_API_KEY]'
}, params=params).content

feed = gtfs_realtime_pb2.FeedMessage()
feed.ParseFromString(data)

# List all vehicle positions with their current status
for entity in feed.entity:
    v = entity.vehicle
    pos = v.position
    print(f"Trip {v.trip.trip_id} at ({pos.latitude:.4f}, {pos.longitude:.4f})")

import requests
from google.transit import gtfs_realtime_pb2

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(),   # ISO accepted; epoch also works
}

url = "https://api.mobilitytwin.brussels/sncb/gtfs-rt-alert"

data = requests.get(url, headers={
        'Authorization': 'Bearer [MY_API_KEY]'
}, params=params).content

feed = gtfs_realtime_pb2.FeedMessage()
feed.ParseFromString(data)

# Display all active service alerts
for entity in feed.entity:
    alert = entity.alert
    header = alert.header_text.translation[0].text if alert.header_text.translation else "N/A"
    routes = [s.route_id for s in alert.informed_entity]
    print(f"[{alert.cause}] {header} — Routes: {', '.join(routes)}")

import requests
from google.transit import gtfs_realtime_pb2

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(),   # ISO accepted; epoch also works
}

url = "https://api.mobilitytwin.brussels/sncb/gtfs-rt-trip-update"

data = requests.get(url, headers={
        'Authorization': 'Bearer [MY_API_KEY]'
}, params=params).content

feed = gtfs_realtime_pb2.FeedMessage()
feed.ParseFromString(data)

# Show delayed trains (delay > 5 minutes)
for entity in feed.entity:
    tu = entity.trip_update
    for stu in tu.stop_time_update:
        delay = stu.arrival.delay if stu.HasField('arrival') else 0
        if delay > 300:
            print(f"Trip {tu.trip.trip_id} — {delay // 60}min late at stop {stu.stop_id}")

# 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))