Decoding GTFS-Realtime Protobuf Feeds

A GTFS-Realtime endpoint does not return JSON, XML, or CSV. It returns an opaque stream of Protocol Buffer bytes that means nothing until you decode it against the published schema. Analysts who expect a text format open the response, see a wall of non-printable characters, and assume the feed is broken. It is not — it is simply serialized with protobuf, the same wire format Google designed for compact, schema-versioned message exchange. This guide, part of the GTFS-Realtime Integration overview, covers how to turn those bytes into typed Python objects with gtfs-realtime-bindings, how to read the feed header, how to iterate and classify the entities inside, and how to flatten everything into a tidy pandas DataFrame that the rest of your pipeline can consume.

Why Protobuf, and What It Costs You

Static GTFS ships as CSV files inside a zip archive because the data is published once and read many times; human-readability is worth the size penalty. Realtime data has the opposite profile. A large agency emits a fresh snapshot every 10–30 seconds, each carrying thousands of vehicle positions or arrival predictions. Encoding that as JSON would multiply payload size several-fold and force every consumer to re-parse verbose keys on every poll. Protocol Buffers solve this by defining the message shape once in a .proto schema, then transmitting only field numbers and packed values on the wire.

The cost is that you cannot inspect a GTFS-Realtime payload with curl and read it. You need the compiled schema — the gtfs_realtime_pb2 module shipped inside gtfs-realtime-bindings — to map field numbers back to names and types. The schema is versioned, backward-compatible, and shared across every GTFS-Realtime feed on earth, which is what makes one decoder work against MTA, Metlink, RATP, and every other publisher without modification.

Prerequisites

Before running the code in this guide, ensure the following are in place:

• Python 3.9+ with the official bindings, an HTTP client, and pandas installed:

pip install gtfs-realtime-bindings requests pandas protobuf

• A reachable GTFS-Realtime endpoint URL. Most agencies publish separate URLs for vehicle positions, trip updates, and service alerts; some combine them into one feed. Any of them decodes with the same code.
• Familiarity with static GTFS identifiers — trip_id, route_id, stop_id, stop_sequence — because realtime entities reference them rather than restating the schedule. If your static ingestion is not yet solid, the GTFS-Realtime Integration overview links the static-feed groundwork you need first.
• An understanding that every timestamp you decode will be POSIX seconds in UTC, which ties directly into timezone handling and schedule normalization when you later join realtime data to local scheduled times.

Concept and Spec Background

The FeedMessage envelope

Every GTFS-Realtime response, regardless of what it carries, is a single top-level FeedMessage. That message has exactly two parts: a required header and a repeated entity field. The header describes the feed as a whole; the entity list holds the payload. Nothing else lives at the top level. Once you internalize that shape, every feed on the planet decodes the same way.

The FeedHeader carries three fields worth reading on every poll:

The three entity kinds

Each FeedEntity in the list carries a mandatory id (a producer-assigned string, unique within the message) and at most one of three optional payloads. An entity is never more than one kind at once:

Because the payload fields are optional, reading entity.vehicle when no vehicle was set does not raise — it returns a default-constructed empty VehiclePosition. This is the single most common decoding bug: treating a zero-valued default as real data. The HasField guard exists precisely to prevent it, and every loop in this guide uses it.

FULL_DATASET vs DIFFERENTIAL

Treat incrementality as a contract. Under FULL_DATASET, the entity list you just decoded is the complete world state as of header.timestamp; any vehicle absent from the list is no longer reporting, and you should overwrite your store wholesale. Under the rare DIFFERENTIAL mode, entities carry an is_deleted flag and you must apply them as a patch against previously held state. Assuming the wrong mode silently accumulates stale ghosts, so read the enum rather than hard-coding an assumption.

Step-by-Step Implementation

Step 1: Fetch the Bytes and Parse a FeedMessage

Fetch the endpoint with a plain HTTP GET and hand the raw body to a fresh FeedMessage. The response is binary, so read response.content (bytes), never response.text (which would corrupt the payload by applying a character decode).

import logging
import requests
from google.transit import gtfs_realtime_pb2

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")

VEHICLE_FEED_URL = "https://example-agency.gov/gtfs-realtime/vehicle-positions.pb"


def fetch_feed_message(url: str, timeout: float = 15.0) -> gtfs_realtime_pb2.FeedMessage:
    """Fetch a GTFS-Realtime endpoint and decode the protobuf body."""
    response = requests.get(url, timeout=timeout)
    response.raise_for_status()

    feed = gtfs_realtime_pb2.FeedMessage()
    feed.ParseFromString(response.content)  # decode bytes against the compiled schema

    logging.info(
        "Decoded FeedMessage: version=%s, entities=%d",
        feed.header.gtfs_realtime_version,
        len(feed.entity),
    )
    return feed


feed = fetch_feed_message(VEHICLE_FEED_URL)

ParseFromString mutates the feed object in place and raises google.protobuf.message.DecodeError if the bytes are not valid protobuf — a truncated download or an HTML error page served with a 200 status will both trip it, so wrap the call in a try/except in production.

Step 2: Read and Trust the Header

Before touching entities, read the header. The version guards compatibility, and the timestamp anchors every downstream staleness and deduplication decision.

from datetime import datetime, timezone

header = feed.header

version = header.gtfs_realtime_version
mode = header.incrementality  # 0 = FULL_DATASET, 1 = DIFFERENTIAL
built_at = datetime.fromtimestamp(header.timestamp, tz=timezone.utc)

logging.info("Feed version %s, mode=%s, built_at=%s (UTC)", version, mode, built_at.isoformat())

if mode != gtfs_realtime_pb2.FeedHeader.FULL_DATASET:
    logging.warning("Feed is DIFFERENTIAL — entities must be applied as deltas, not a replace")

# Staleness guard: reject snapshots older than 90 seconds
age_seconds = datetime.now(tz=timezone.utc).timestamp() - header.timestamp
if age_seconds > 90:
    logging.warning("Feed snapshot is %.0f s old — producer may be stalled", age_seconds)

Reading header.timestamp as POSIX UTC and converting explicitly keeps every timestamp in a single frame of reference. Local-time conversion belongs at the display edge, in step with the timezone handling and schedule normalization rules for the agency’s own zone.

Step 3: Iterate Entities and Branch on Kind

Walk feed.entity, guard every optional payload with HasField, and extract only the fields you need. The loop below counts each kind and pulls a handful of representative fields from vehicle positions.

vehicle_rows: list[dict] = []
kind_counts = {"vehicle": 0, "trip_update": 0, "alert": 0}

for entity in feed.entity:
    if entity.HasField("vehicle"):
        kind_counts["vehicle"] += 1
        vp = entity.vehicle
        vehicle_rows.append(
            {
                "entity_id": entity.id,
                "vehicle_id": vp.vehicle.id or None,
                "trip_id": vp.trip.trip_id or None,
                "route_id": vp.trip.route_id or None,
                "latitude": vp.position.latitude if vp.HasField("position") else None,
                "longitude": vp.position.longitude if vp.HasField("position") else None,
                "bearing": vp.position.bearing if vp.HasField("position") else None,
                "current_stop_sequence": vp.current_stop_sequence,
                "stop_id": vp.stop_id or None,
                "timestamp": vp.timestamp or None,
            }
        )
    elif entity.HasField("trip_update"):
        kind_counts["trip_update"] += 1
    elif entity.HasField("alert"):
        kind_counts["alert"] += 1

logging.info("Entity kinds: %s", kind_counts)

The vp.vehicle.id or None idiom matters: protobuf returns an empty string for an unset string field, and "" or None yields None, giving you a clean nullable column instead of a mix of empty strings and real IDs. Numeric fields such as current_stop_sequence default to 0, so if 0 is a meaningful sequence value in your feed, guard it with HasField rather than truthiness.

Step 4: Flatten to a Tidy DataFrame

Collect the extracted rows into a single frame with explicit dtypes. A tidy frame — one row per vehicle, one column per attribute — is what every downstream join, aggregation, and export expects.

import pandas as pd

VEHICLE_DTYPES = {
    "entity_id": "string",
    "vehicle_id": "string",
    "trip_id": "string",
    "route_id": "string",
    "latitude": "float64",
    "longitude": "float64",
    "bearing": "float64",
    "current_stop_sequence": "Int64",
    "stop_id": "string",
    "timestamp": "Int64",
}

vehicle_positions = pd.DataFrame(vehicle_rows).astype(VEHICLE_DTYPES)

# Convert POSIX seconds to a UTC-aware datetime column for downstream joins
vehicle_positions["observed_at"] = pd.to_datetime(
    vehicle_positions["timestamp"], unit="s", utc=True
)

logging.info("Built vehicle_positions frame: %d rows", len(vehicle_positions))
print(vehicle_positions.head().to_string(index=False))

Declaring string and Int64 (both nullable) rather than object and int64 keeps identifier columns from silently coercing numeric-looking IDs and lets integer columns hold missing values. The observed_at column is derived from the per-vehicle timestamp, which may lag the feed-level header.timestamp when a producer reuses a stale fix.

Step 5: MessageToDict for Exploration Only

When you are still learning a feed’s shape, MessageToDict dumps an entire message to nested Python dicts. It is invaluable for discovery and terrible for a fixed pipeline.

from google.protobuf.json_format import MessageToDict

sample = MessageToDict(
    feed.entity[0],
    preserving_proto_field_name=True,  # keep snake_case, not camelCase
    use_integers_for_enums=True,       # keep enum ints, not names
)
import json
print(json.dumps(sample, indent=2)[:800])

Without preserving_proto_field_name=True, every key is renamed to camelCase (tripId, not trip_id), which breaks any code that indexes by the spec field name. And because MessageToDict omits unset optional fields entirely, a downstream dict["stop_id"] raises KeyError on exactly the rows where the field was absent — the opposite of the safe default you get from guarded direct access. Use it to look; use direct field access to build.

Validation and Verification

Run these assertions on every decoded frame before promoting it into a store or a join. They catch the failure modes that a silent protobuf default would otherwise smuggle past you.

def validate_vehicle_frame(vehicle_positions: pd.DataFrame, header_ts: int) -> dict:
    """Validate a decoded vehicle-position frame against structural expectations."""
    assert not vehicle_positions.empty, "Decoded frame is empty — feed may be down"

    # entity_id must be unique within a single FeedMessage
    dupes = vehicle_positions["entity_id"].duplicated().sum()
    assert dupes == 0, f"{dupes} duplicate entity_id values in one message"

    # Coordinates, where present, must fall within WGS84 bounds
    coords = vehicle_positions.dropna(subset=["latitude", "longitude"])
    assert coords["latitude"].between(-90, 90).all(), "latitude out of WGS84 bounds"
    assert coords["longitude"].between(-180, 180).all(), "longitude out of WGS84 bounds"

    # Per-vehicle timestamps should not lead the header timestamp
    ahead = (vehicle_positions["timestamp"].dropna() > header_ts + 5).sum()

    results = {
        "rows": len(vehicle_positions),
        "with_position": len(coords),
        "with_trip_id": int(vehicle_positions["trip_id"].notna().sum()),
        "timestamps_ahead_of_header": int(ahead),
    }
    return results


report = validate_vehicle_frame(vehicle_positions, feed.header.timestamp)
logging.info("Validation report: %s", report)

Failure Modes and Edge Cases

• Reading response.text instead of response.content. requests decodes .text as a character string using a guessed charset, which mangles binary protobuf. Always pass response.content to ParseFromString. A DecodeError immediately after a 200 response is almost always this bug or an HTML error page served with the wrong content type.

• Trusting zero-valued defaults. Protobuf has no concept of “null” for scalar fields; an unset string is "", an unset number is 0, an unset bool is False. A vehicle whose current_stop_sequence reads 0 may be at sequence 0 or may simply not report it. Use HasField for optional message and scalar fields rather than inferring absence from a falsy value.

• Assuming every entity is a vehicle. Some agencies serve vehicle positions, trip updates, and alerts from a single combined URL. A loop that reads entity.vehicle unconditionally will silently produce empty rows for the trip-update and alert entities. Branch with HasField and count each kind, as in step 3.

• Version drift between producer and bindings. A producer advertising gtfs_realtime_version "3.0" may include fields your installed gtfs-realtime-bindings predates. Unknown fields are preserved on the wire but invisible to your decoder. Pin and periodically bump the bindings package, and log header.gtfs_realtime_version so drift is visible.

• Empty entity lists during service gaps. Overnight or during outages a feed may legitimately return a well-formed FeedMessage with zero entities. This is not an error — do not raise on an empty list; log it and let the staleness guard on header.timestamp decide whether the producer has stalled.

• Per-vehicle timestamps that lag the header. A producer may stamp the header at build time but carry a vehicle fix that is minutes old. Store both header.timestamp and the per-entity timestamp, and prefer the entity value when reasoning about how fresh a specific vehicle’s position is.

Performance and Scale Notes

A single national feed can exceed tens of thousands of entities per snapshot, polled every 15 seconds around the clock. Two practices keep decode cost bounded. First, decode straight into lists of plain dicts and build one DataFrame per snapshot with pd.DataFrame(rows).astype(dtypes), rather than appending to a growing frame — row-wise concat is quadratic and will dominate CPU at scale. Second, if you persist history, write each snapshot to partitioned Parquet keyed by feed date and hour rather than accumulating an in-memory frame; the columnar layout compresses the highly repetitive route_id and stop_id columns dramatically.

import pyarrow  # noqa: F401  (engine for to_parquet)

snapshot_ts = pd.Timestamp(feed.header.timestamp, unit="s", tz="UTC")
partition = snapshot_ts.strftime("date=%Y-%m-%d/hour=%H")
vehicle_positions.assign(header_timestamp=feed.header.timestamp).to_parquet(
    f"warehouse/vehicle_positions/{partition}/{feed.header.timestamp}.parquet",
    index=False,
)

When you only need a subset of fields, extract exactly those in the decode loop rather than round-tripping through MessageToDict — direct field access avoids materializing the entire nested message as Python objects, which is the dominant cost on large feeds. Polling cadence itself is a scale lever covered in polling GTFS-Realtime feeds efficiently.

Frequently Asked Questions

Related

• Parsing GTFS-RT Protobuf in Python — a single copy-pasteable script that decodes a feed into a DataFrame with safe field access
• Polling GTFS-Realtime Feeds Efficiently — conditional requests and header-timestamp deduplication so you never re-decode an unchanged snapshot
• Parsing GTFS with pandas and partridge — the static-feed counterpart whose identifiers realtime entities reference
• Timezone Handling and Schedule Normalization — converting the POSIX timestamps decoded here into agency-local scheduled times
• Up: GTFS-Realtime Integration — the full realtime pipeline from decoding to schedule matching · Home