Sncb

import requests
import geopandas as gpd

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

data = requests.get(url, headers={
        'Authorization': 'Bearer [MY_API_KEY]'
}).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

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

data = requests.get(url, headers={
        'Authorization': 'Bearer [MY_API_KEY]'
}).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

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

data = requests.get(url, headers={
        'Authorization': 'Bearer [MY_API_KEY]'
}).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)}")

import requests
import tempfile

from gtfs_parquet import read_parquet

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

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

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

feed = read_parquet(f.name)

# List all routes (Polars DataFrame — zero-copy, extremely low RAM usage)
print(feed.routes.select("route_short_name", "route_long_name"))

# Count trips per route
print(feed.trips.group_by("route_id").len().sort("len", descending=True))

import requests
from google.transit import gtfs_realtime_pb2

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

data = requests.get(url, headers={
        'Authorization': 'Bearer [MY_API_KEY]'
}).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

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

data = requests.get(url, headers={
        'Authorization': 'Bearer [MY_API_KEY]'
}).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

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

data = requests.get(url, headers={
        'Authorization': 'Bearer [MY_API_KEY]'
}).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}")