Tec

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/tec/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 TEC schedule
print(f"Routes: {len(feed.routes)}, Stops: {len(feed.stops)}, Trips: {len(feed.trips)}")
print("\nSample routes:")
print(feed.routes[['route_short_name', 'route_long_name']].head(10))

# 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/tec/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/tec/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
for entity in feed.entity:
    v = entity.vehicle
    pos = v.position
    print(f"Route {v.trip.route_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/tec/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/tec/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 trips (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/tec/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: TEC's GTFS-RT feed publishes delays only — no absolute arrival_time /
# departure_time. Join to static GTFS on (trip_id, stop_id) to recover times.
not_cancelled = df.filter(~pl.col("cancelled") & pl.col("arrival_delay").is_not_null())

# Distribution of arrival delays.
print("\nArrival delay distribution (seconds):")
print(not_cancelled.select([
    pl.col("arrival_delay").quantile(0.05).alias("p05"),
    pl.col("arrival_delay").median().alias("median"),
    pl.col("arrival_delay").mean().round(1).alias("mean"),
    pl.col("arrival_delay").quantile(0.95).alias("p95"),
    pl.col("arrival_delay").max().alias("max"),
    (pl.col("arrival_delay") <= 60).mean().round(3).alias("share_within_60s"),
]))

# Where do the worst delays accumulate?
print("\nMean arrival delay per route (top 10):")
print(not_cancelled.group_by("route_id")
      .agg([pl.col("arrival_delay").mean().round(1).alias("mean_delay_s"),
            pl.len().alias("n")])
      .filter(pl.col("n") >= 50)
      .sort("mean_delay_s", descending=True)
      .head(10))

# Cancellations.
cancelled = df.filter(pl.col("cancelled"))
print(f"\ncancelled rows: {cancelled.height:,}, unique trip-instances: "
      f"{cancelled.select(['trip_id', 'start_date', 'start_time']).unique().height}")