Stib

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/stib/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')

# Explore the STIB schedule
print(f"Routes: {len(feed.routes)}")
print(f"Stops: {len(feed.stops)}")
print(f"Trips: {len(feed.trips)}")
print("\nRoutes:")
print(feed.routes[['route_short_name', 'route_long_name', 'route_type']])

# 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/stib/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
import geopandas as gpd

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/stib/segments"

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

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

# One stop-to-stop segment per feature; `distance` orders them along the line
line = gdf[(gdf['line_id'] == '71') & (gdf['direction'] == 1)].sort_values('distance')
line['length_m'] = line.to_crs(31370).geometry.length   # Belgian Lambert 72, metres
print(line[['start', 'end', 'length_m']].round(0).to_string(index=False))

# Plot the whole network in the lines' own colours
gdf.plot(figsize=(12, 10), color=gdf['color'], linewidth=1)

import requests
import geopandas as gpd

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/stib/stops"

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

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

# List stops for a specific line
line = "1"
line_stops = gdf[gdf['route_short_name'] == line].sort_values('stop_sequence')
print(f"Stops on line {line}:")
for _, stop in line_stops.iterrows():
    print(f"  {stop['stop_sequence']}. {stop['stop_name']}")

# Plot all stops on a map
gdf.plot(figsize=(10, 8), markersize=3, color="red")

import requests
import pandas as pd

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/stib/vehicle-distance"

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

# Convert to DataFrame for analysis
df = pd.DataFrame(data)

# Group by line and compute average distance from stop
avg_distance = df.groupby('lineId')['distanceFromPoint'].mean()
print("Average distance from stop per line:")
print(avg_distance.sort_values(ascending=False))

import requests
import geopandas as gpd

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/stib/vehicle-position"

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

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

# Count vehicles per line
vehicles_per_line = gdf.groupby('lineId').size().sort_values(ascending=False)
print("Active vehicles per line:")
print(vehicles_per_line)

# Plot vehicles colored by line
gdf.plot(figsize=(10, 8), column='color', legend=True, markersize=8)

import requests

url = "https://api.mobilitytwin.brussels/stib/vehicle-trip"

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

# Every vehicle on the road, with its GTFS trip and delay where established
for feature in data["features"]:
    p = feature["properties"]
    if p["tripId"]:
        print(f"Line {p['lineId']} trip {p['tripId']} ({p['startTime']}): "
              f"{p['delay']:+d} s, {p['status']} stop {p['stopId']}")
    else:
        print(f"Line {p['lineId']} vehicle {p['uuid'][:8]}: no trip yet")

import requests
from google.transit import gtfs_realtime_pb2

url = "https://api.mobilitytwin.brussels/stib/gtfs-rt-vehicle-position"

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

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

# List all vehicle positions (trip_id is empty for a vehicle not yet matched)
for entity in feed.entity:
    v = entity.vehicle
    print(f"Route {v.trip.route_id} trip {v.trip.trip_id or '-'}: "
          f"({v.position.latitude:.4f}, {v.position.longitude:.4f}), "
          f"stop {v.stop_id or '-'} seq {v.current_stop_sequence}")

import requests
from google.transit import gtfs_realtime_pb2

from datetime import datetime

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

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

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

# Predicted arrival at the next stops of every trip with a vehicle on it
for entity in feed.entity:
    update = entity.trip_update
    upcoming = [(s.stop_id, datetime.fromtimestamp(s.arrival.time).strftime("%H:%M"))
                for s in update.stop_time_update[:3]]
    print(f"Trip {update.trip.trip_id} ({update.delay:+d} s): {upcoming}")

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

from datetime import datetime, timedelta, timezone

# Yesterday's Brussels-day file (the harvester anchors windows to local midnight,
# so any timestamp inside that day returns the same parquet).
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/stib/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:,}, trips: {df['trip_id'].n_unique():,}")

# STIB publishes no realtime feed: every time is reconstructed from vehicle
# positions. Keep the measured ones (observed) of timetabled trips.
measured = df.filter(pl.col("observed") & (pl.col("trip_schedule_relationship") == 0))

delays = (
    measured.group_by("route_id")          # route_id is the line number
      .agg([
          pl.col("arrival_delay").median().alias("median_delay_s"),
          (pl.col("arrival_delay") > 300).mean().round(3).alias("share_over_5_min"),
          pl.len().alias("stop_calls"),
      ])
      .sort("median_delay_s", descending=True)
)
print("\nLines by median arrival delay:")
print(delays.head(10))

import requests
import pandas as pd

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/stib/speed"

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

df = pd.DataFrame(data)

# Average speed per line (km/h)
speed_per_line = df.groupby('lineId')['speed'].mean().sort_values()
print("Average speed per line (km/h):")
print(speed_per_line)

# Find the slowest segments
slowest = df.nsmallest(5, 'speed')[['lineId', 'pointId', 'speed']]
print("\nSlowest segments:")
print(slowest)

import requests
import pandas as pd

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/stib/aggregated-speed"

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

df = pd.DataFrame(data)

# Compare average speed across lines over 10-minute intervals
speed_by_line = df.groupby('lineId')['speed'].agg(['mean', 'min', 'max'])
print("Speed statistics per line (km/h):")
print(speed_by_line.sort_values('mean'))

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/stib/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="uuid")

# Compute speed and distance for each vehicle trajectory
for traj in tc.trajectories[:5]:
    print(f"Vehicle {traj.id}: {traj.get_length():.0f}m, duration: {traj.get_duration()}")

# Plot all vehicle trajectories
tc.plot(figsize=(12, 8), linewidth=0.5)

import requests
import geopandas as gpd

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/stib/shapefile"

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

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

# Plot the full STIB network colored by line
gdf.plot(figsize=(12, 10), color=gdf['color_hex'], linewidth=1)

# List all unique lines
print("STIB lines:", sorted(gdf['ligne'].unique()))

import requests
import json

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/stib/stop-details"

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

# Parse and display stop details
for stop in data['results'][:10]:
    name = json.loads(stop['name'])
    coords = json.loads(stop['gpscoordinates'])
    print(f"Stop {stop['id']}: {name['fr']} / {name['nl']} ({coords['latitude']}, {coords['longitude']})")

import requests
import json

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/stib/stops-by-line"

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

# Display the route of a specific line
for route in data['results']:
    if route['lineid'] == '1':
        dest = json.loads(route['destination'])
        stops = json.loads(route['points'])
        print(f"Line {route['lineid']} → {dest['fr']} ({route['direction']})")
        print(f"  {len(stops)} stops: {stops[0]['id']} → {stops[-1]['id']}")

import requests
import json

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/stib/waiting-times"

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

# Show next arrivals at each stop
for entry in data['results'][:10]:
    times = json.loads(entry['passingtimes'])
    for t in times:
        dest = t.get('destination', {}).get('fr', '?')   # absent on a "last departure" message
        arrival = datetime.fromisoformat(t['expectedArrivalTime'])
        print(f"Stop {entry['pointid']} — Line {entry['lineid']} → {dest} at {arrival:%H:%M}")

import requests
import json

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/stib/travellers-information"

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

# Display active service alerts sorted by priority
for alert in sorted(data['results'], key=lambda x: x['priority'], reverse=True):
    content = json.loads(alert['content'])
    lines = json.loads(alert['lines'])
    text = content[0]['text'][0].get('en') or content[0]['text'][0]['fr']   # a few alerts are French/Dutch only
    line_ids = ', '.join(l['id'] for l in lines)
    print(f"[Priority {alert['priority']}] Lines {line_ids}: {text}")