//! **GTFS-Realtime** vehicle positions — feature `parsers` (no extra dep).
//!
//! GTFS-Realtime is the real-time companion to the GTFS static feed: a great many
//! EU/Nordic transit agencies publish a `VehiclePositions` feed — every bus, tram,
//! metro and commuter train reporting its live `(id, lat, lon, route, timestamp)`.
//! Unlike the static feed (a ZIP of CSV, decoded by [`super::gtfs`]) it is a
//! **Protocol-Buffers binary message** (`FeedMessage`), a wire shape NO other
//! collector in this crate decodes: DCAT/ENTSO-E/RSS/WFS are XML, PxWeb/SDMX/USGS/
//! OpenAQ/Energy-Charts/Nord Pool are JSON, GTFS-static is a CSV zip, Trafikverket
//! is a JSON pointer-map. This one walks the **protobuf varint/length-delimited
//! wire format** by hand.
//!
//! ## Why hand-rolled (no `prost` / `protobuf` crate)
//! The GTFS-RT subset we need (vehicle id / lat / lon / route / trip / timestamp /
//! stop) touches only a handful of fields across five nested messages. Pulling in
//! `prost` + `prost-build` (a `protoc`/codegen build step) or the `protobuf` crate
//! would drag a heavy codegen dependency into a leaf whose whole point is to stay
//! lean (`arrow`, `serde`, `reqwest` and little else). Instead this module carries
//! a ~60-line [`PbReader`] — a varint / fixed32 / fixed64 / length-delimited
//! cursor that reads the fields we care about and SKIPS every unknown one (the
//! protobuf forward-compat contract). Zero new deps; the leaf stays pure-Rust.
//!
//! ## Wire shape (the GTFS-RT subset decoded here)
//! ```text
//! FeedMessage { FeedHeader header = 1; repeated FeedEntity entity = 2; }
//! FeedEntity { string id = 1; VehiclePosition vehicle = 4; … }
//! VehiclePosition { TripDescriptor trip = 1; Position position = 2;
//! uint64 timestamp = 5; string stop_id = 7;
//! VehicleDescriptor vehicle = 8; … }
//! Position { float latitude = 1; float longitude = 2;
//! float bearing = 3; float speed = 5; … }
//! TripDescriptor { string trip_id = 1; string route_id = 5; … }
//! VehicleDescriptor{ string id = 1; … }
//! ```
//! Only entities carrying a `vehicle` (a `VehiclePosition`) with a locatable
//! `Position` (lat AND lon) become a row — a `trip_update`/`alert` entity, or a
//! vehicle with no position, contributes NO row (analogous to USGS dropping a
//! non-Point feature). [`parse_gtfs_rt`] is the pure parse core (bytes in, rows
//! out — no I/O); [`build_gtfs_rt`] is the fixture/round-trip ENCODER the tests +
//! the offline sample use to synthesise a feed with no committed binary blob.
use anyhow::{Result, anyhow};
use arrow_array::RecordBatch;
use arrow_schema::{DataType, Schema};
use super::arrow_util::{batch, f64_col, i64_col, schema as build_schema, str_col};
use super::{
AuthKind, BatchSink, Bounds, Checkpoint, CollectStats, Collector, CollectorCfg, CollectorMeta,
CollectorPath, Direction, GdprClass, SourceEndpoint, SourceKind, SourceTarget,
};
// ── protobuf wire types (only the four the GTFS-RT subset uses) ─────────────────
const WIRE_VARINT: u8 = 0; // int32/64, uint32/64, bool, enum
const WIRE_I64: u8 = 1; // fixed64, sfixed64, double
const WIRE_LEN: u8 = 2; // string, bytes, embedded message, packed repeated
const WIRE_I32: u8 = 5; // fixed32, sfixed32, float
// ── minimal protobuf reader (a cursor; no prost / protobuf crate) ───────────────
/// A read cursor over protobuf-encoded bytes. Every accessor is bounds-checked and
/// returns an `Err` (never panics) on a truncated / malformed stream — so a
/// corrupt feed is a RAGNARÖK error, not a crash.
struct PbReader<'a> {
buf: &'a [u8],
pos: usize,
}
impl<'a> PbReader<'a> {
fn new(buf: &'a [u8]) -> Self {
Self { buf, pos: 0 }
}
fn at_end(&self) -> bool {
self.pos >= self.buf.len()
}
/// Read a base-128 varint (LEB128, low group first). Errors on a truncated
/// stream or an over-long (>10 group) encoding.
fn varint(&mut self) -> Result<u64> {
let mut result: u64 = 0;
let mut shift: u32 = 0;
loop {
let byte = *self
.buf
.get(self.pos)
.ok_or_else(|| anyhow!("truncated varint"))?;
self.pos += 1;
if shift >= 64 {
return Err(anyhow!("varint overflow (>10 bytes)"));
}
result |= ((byte & 0x7f) as u64) << shift;
if byte & 0x80 == 0 {
break;
}
shift += 7;
}
Ok(result)
}
/// Read a field tag → `(field number, wire type)`. Field 0 is illegal.
fn tag(&mut self) -> Result<(u32, u8)> {
let key = self.varint()?;
let field = (key >> 3) as u32;
let wire = (key & 0x07) as u8;
if field == 0 {
return Err(anyhow!("zero field number in tag"));
}
Ok((field, wire))
}
/// Read a length-delimited slice (string / bytes / sub-message).
fn len_slice(&mut self) -> Result<&'a [u8]> {
let len = self.varint()? as usize;
let start = self.pos;
let end = start
.checked_add(len)
.ok_or_else(|| anyhow!("length overflow"))?;
let s = self
.buf
.get(start..end)
.ok_or_else(|| anyhow!("truncated length-delimited field"))?;
self.pos = end;
Ok(s)
}
/// Read a little-endian fixed32 (a `float` lives here).
fn fixed32(&mut self) -> Result<u32> {
let b = self
.buf
.get(self.pos..self.pos + 4)
.ok_or_else(|| anyhow!("truncated fixed32"))?;
self.pos += 4;
Ok(u32::from_le_bytes([b[0], b[1], b[2], b[3]]))
}
/// Read a little-endian fixed64 (a `double` lives here).
fn fixed64(&mut self) -> Result<u64> {
let b = self
.buf
.get(self.pos..self.pos + 8)
.ok_or_else(|| anyhow!("truncated fixed64"))?;
self.pos += 8;
Ok(u64::from_le_bytes([
b[0], b[1], b[2], b[3], b[4], b[5], b[6], b[7],
]))
}
/// Read a `float` (fixed32 → f32 → f64).
fn float(&mut self) -> Result<f64> {
Ok(f32::from_bits(self.fixed32()?) as f64)
}
/// Skip a field of the given wire type (unknown / uninteresting fields — the
/// protobuf forward-compat contract).
fn skip(&mut self, wire: u8) -> Result<()> {
match wire {
WIRE_VARINT => {
self.varint()?;
}
WIRE_I64 => {
self.fixed64()?;
}
WIRE_LEN => {
self.len_slice()?;
}
WIRE_I32 => {
self.fixed32()?;
}
other => {
return Err(anyhow!(
"unsupported wire type {other} (groups unsupported)"
));
}
}
Ok(())
}
}
fn decode_utf8(b: &[u8]) -> String {
String::from_utf8_lossy(b).into_owned()
}
// ── decoded row ─────────────────────────────────────────────────────────────────
/// One decoded vehicle position — a single transit vehicle's live location + the
/// keys that resolve it (`vehicle_id` / `route_id` / `trip_id`).
#[derive(Debug, Clone, Default, PartialEq)]
pub struct VehiclePosition {
/// `FeedEntity.id` — the feed's entity key (always present).
pub entity_id: String,
/// `VehicleDescriptor.id` — the vehicle's stable id (the natural join key for
/// korp-ontology resolution); `None` if the feed omits the descriptor.
pub vehicle_id: Option<String>,
/// `TripDescriptor.trip_id` — the GTFS trip this vehicle is running.
pub trip_id: Option<String>,
/// `TripDescriptor.route_id` — the GTFS route (joins back to `gtfs-routes`).
pub route_id: Option<String>,
/// WGS84 latitude (`Position.latitude`) — required for a row (the map Y).
pub latitude: f64,
/// WGS84 longitude (`Position.longitude`) — required for a row (the map X).
pub longitude: f64,
/// Heading in degrees (`Position.bearing`), if reported.
pub bearing: Option<f64>,
/// Speed in m/s (`Position.speed`), if reported.
pub speed: Option<f64>,
/// POSIX seconds of the fix (`VehiclePosition.timestamp`), if reported.
pub timestamp: Option<i64>,
/// The stop this vehicle is at / approaching (`VehiclePosition.stop_id`).
pub stop_id: Option<String>,
}
/// Fields accumulated while walking one `VehiclePosition` sub-message; lat/lon are
/// `Option` here because a row is emitted ONLY when both are present.
#[derive(Default)]
struct VpRaw {
vehicle_id: Option<String>,
trip_id: Option<String>,
route_id: Option<String>,
latitude: Option<f64>,
longitude: Option<f64>,
bearing: Option<f64>,
speed: Option<f64>,
timestamp: Option<i64>,
stop_id: Option<String>,
}
/// Decode a GTFS-Realtime `FeedMessage` (protobuf bytes) → one row per vehicle
/// position. The `FeedHeader` (field 1) and every non-vehicle entity are skipped;
/// a vehicle whose `Position` lacks lat OR lon contributes no row.
pub fn parse_gtfs_rt(pb: &[u8]) -> Result<Vec<VehiclePosition>> {
let mut out = Vec::new();
let mut r = PbReader::new(pb);
while !r.at_end() {
let (field, wire) = r.tag()?;
match (field, wire) {
// FeedMessage.entity = 2 (repeated message)
(2, WIRE_LEN) => {
let entity = r.len_slice()?;
if let Some(vp) = parse_feed_entity(entity)? {
out.push(vp);
}
}
// FeedMessage.header = 1 and anything else → skip.
_ => r.skip(wire)?,
}
}
Ok(out)
}
/// A `FeedEntity` → a row iff it carries a `vehicle` (field 4) with a locatable
/// `Position`; a `trip_update`/`alert` entity yields `None`.
fn parse_feed_entity(buf: &[u8]) -> Result<Option<VehiclePosition>> {
let mut r = PbReader::new(buf);
let mut entity_id = String::new();
let mut vehicle_bytes: Option<&[u8]> = None;
while !r.at_end() {
let (field, wire) = r.tag()?;
match (field, wire) {
(1, WIRE_LEN) => entity_id = decode_utf8(r.len_slice()?), // FeedEntity.id
(4, WIRE_LEN) => vehicle_bytes = Some(r.len_slice()?), // FeedEntity.vehicle
_ => r.skip(wire)?,
}
}
let Some(vb) = vehicle_bytes else {
return Ok(None); // not a vehicle entity → no row
};
let vp = parse_vehicle_position(vb)?;
let (Some(latitude), Some(longitude)) = (vp.latitude, vp.longitude) else {
return Ok(None); // a vehicle with no locatable position → no row
};
Ok(Some(VehiclePosition {
entity_id,
vehicle_id: vp.vehicle_id,
trip_id: vp.trip_id,
route_id: vp.route_id,
latitude,
longitude,
bearing: vp.bearing,
speed: vp.speed,
timestamp: vp.timestamp,
stop_id: vp.stop_id,
}))
}
/// Walk one `VehiclePosition` sub-message, recursing into `trip` / `position` /
/// `vehicle` descriptors. Unknown / mismatched-wire fields fall to `skip`.
fn parse_vehicle_position(buf: &[u8]) -> Result<VpRaw> {
let mut r = PbReader::new(buf);
let mut vp = VpRaw::default();
while !r.at_end() {
let (field, wire) = r.tag()?;
match (field, wire) {
(1, WIRE_LEN) => parse_trip(r.len_slice()?, &mut vp)?, // trip
(2, WIRE_LEN) => parse_position(r.len_slice()?, &mut vp)?, // position
(5, WIRE_VARINT) => vp.timestamp = Some(r.varint()? as i64), // timestamp
(7, WIRE_LEN) => vp.stop_id = Some(decode_utf8(r.len_slice()?)), // stop_id
(8, WIRE_LEN) => parse_vehicle_descriptor(r.len_slice()?, &mut vp)?, // vehicle
_ => r.skip(wire)?, // current_stop_sequence, current_status, occupancy, …
}
}
Ok(vp)
}
/// `TripDescriptor` → `trip_id` (1) + `route_id` (5).
fn parse_trip(buf: &[u8], vp: &mut VpRaw) -> Result<()> {
let mut r = PbReader::new(buf);
while !r.at_end() {
let (field, wire) = r.tag()?;
match (field, wire) {
(1, WIRE_LEN) => vp.trip_id = Some(decode_utf8(r.len_slice()?)),
(5, WIRE_LEN) => vp.route_id = Some(decode_utf8(r.len_slice()?)),
_ => r.skip(wire)?,
}
}
Ok(())
}
/// `Position` → `latitude` (1) + `longitude` (2) + `bearing` (3) + `speed` (5),
/// each a `float`. `odometer` (4, a double) is skipped.
fn parse_position(buf: &[u8], vp: &mut VpRaw) -> Result<()> {
let mut r = PbReader::new(buf);
while !r.at_end() {
let (field, wire) = r.tag()?;
match (field, wire) {
(1, WIRE_I32) => vp.latitude = Some(r.float()?),
(2, WIRE_I32) => vp.longitude = Some(r.float()?),
(3, WIRE_I32) => vp.bearing = Some(r.float()?),
(5, WIRE_I32) => vp.speed = Some(r.float()?),
_ => r.skip(wire)?,
}
}
Ok(())
}
/// `VehicleDescriptor` → `id` (1). `label` / `license_plate` are skipped.
fn parse_vehicle_descriptor(buf: &[u8], vp: &mut VpRaw) -> Result<()> {
let mut r = PbReader::new(buf);
while !r.at_end() {
let (field, wire) = r.tag()?;
match (field, wire) {
(1, WIRE_LEN) => vp.vehicle_id = Some(decode_utf8(r.len_slice()?)),
_ => r.skip(wire)?,
}
}
Ok(())
}
// ── Arrow ───────────────────────────────────────────────────────────────────────
/// The vehicle-position table schema (a moving-map position table).
pub fn vehicles_schema() -> Schema {
build_schema(&[
("entity_id", DataType::Utf8, false),
("vehicle_id", DataType::Utf8, true),
("trip_id", DataType::Utf8, true),
("route_id", DataType::Utf8, true),
("latitude", DataType::Float64, false),
("longitude", DataType::Float64, false),
("bearing", DataType::Float64, true),
("speed", DataType::Float64, true),
("timestamp", DataType::Int64, true),
("stop_id", DataType::Utf8, true),
])
}
/// Vehicle positions → one Arrow `RecordBatch`.
pub fn vehicles_batch(rows: &[VehiclePosition]) -> Result<RecordBatch> {
let cols = vec![
str_col(
&rows
.iter()
.map(|r| Some(r.entity_id.clone()))
.collect::<Vec<_>>(),
),
str_col(
&rows
.iter()
.map(|r| r.vehicle_id.clone())
.collect::<Vec<_>>(),
),
str_col(&rows.iter().map(|r| r.trip_id.clone()).collect::<Vec<_>>()),
str_col(&rows.iter().map(|r| r.route_id.clone()).collect::<Vec<_>>()),
f64_col(&rows.iter().map(|r| Some(r.latitude)).collect::<Vec<_>>()),
f64_col(&rows.iter().map(|r| Some(r.longitude)).collect::<Vec<_>>()),
f64_col(&rows.iter().map(|r| r.bearing).collect::<Vec<_>>()),
f64_col(&rows.iter().map(|r| r.speed).collect::<Vec<_>>()),
i64_col(&rows.iter().map(|r| r.timestamp).collect::<Vec<_>>()),
str_col(&rows.iter().map(|r| r.stop_id.clone()).collect::<Vec<_>>()),
];
batch(&vehicles_schema(), cols)
}
// ── fixture ENCODER (round-trip helper — no committed binary blob) ──────────────
/// A vehicle to encode into a synthetic GTFS-RT feed (the fixture spec).
#[derive(Debug, Clone, Default)]
pub struct VehicleFixture<'a> {
pub entity_id: &'a str,
pub vehicle_id: Option<&'a str>,
pub trip_id: Option<&'a str>,
pub route_id: Option<&'a str>,
pub latitude: f32,
pub longitude: f32,
pub bearing: Option<f32>,
pub speed: Option<f32>,
pub timestamp: Option<u64>,
pub stop_id: Option<&'a str>,
}
fn enc_varint(out: &mut Vec<u8>, mut v: u64) {
loop {
let mut byte = (v & 0x7f) as u8;
v >>= 7;
if v != 0 {
byte |= 0x80;
}
out.push(byte);
if v == 0 {
break;
}
}
}
fn enc_tag(out: &mut Vec<u8>, field: u32, wire: u8) {
enc_varint(out, ((field as u64) << 3) | wire as u64);
}
fn enc_string(out: &mut Vec<u8>, field: u32, s: &str) {
enc_tag(out, field, WIRE_LEN);
enc_varint(out, s.len() as u64);
out.extend_from_slice(s.as_bytes());
}
fn enc_len_field(out: &mut Vec<u8>, field: u32, bytes: &[u8]) {
enc_tag(out, field, WIRE_LEN);
enc_varint(out, bytes.len() as u64);
out.extend_from_slice(bytes);
}
fn enc_float(out: &mut Vec<u8>, field: u32, v: f32) {
enc_tag(out, field, WIRE_I32);
out.extend_from_slice(&v.to_bits().to_le_bytes());
}
fn enc_varint_field(out: &mut Vec<u8>, field: u32, v: u64) {
enc_tag(out, field, WIRE_VARINT);
enc_varint(out, v);
}
/// **Encode** a GTFS-RT `FeedMessage` (with a realistic `FeedHeader` so the parse
/// path exercises the header-skip) from vehicle fixtures — the tests + the offline
/// sample synthesise a feed with this, so no binary blob is committed. Round-trips
/// through [`parse_gtfs_rt`].
pub fn build_gtfs_rt(vehicles: &[VehicleFixture]) -> Vec<u8> {
let mut feed = Vec::new();
// FeedHeader (field 1): gtfs_realtime_version="2.0" (1), timestamp (3).
let mut header = Vec::new();
enc_string(&mut header, 1, "2.0");
enc_varint_field(&mut header, 3, 1_752_400_000);
enc_len_field(&mut feed, 1, &header);
for v in vehicles {
// TripDescriptor: trip_id=1, route_id=5.
let mut trip = Vec::new();
if let Some(t) = v.trip_id {
enc_string(&mut trip, 1, t);
}
if let Some(r) = v.route_id {
enc_string(&mut trip, 5, r);
}
// Position: latitude=1, longitude=2, bearing=3, speed=5.
let mut pos = Vec::new();
enc_float(&mut pos, 1, v.latitude);
enc_float(&mut pos, 2, v.longitude);
if let Some(b) = v.bearing {
enc_float(&mut pos, 3, b);
}
if let Some(s) = v.speed {
enc_float(&mut pos, 5, s);
}
// VehicleDescriptor: id=1.
let mut vdesc = Vec::new();
if let Some(id) = v.vehicle_id {
enc_string(&mut vdesc, 1, id);
}
// VehiclePosition: trip=1, position=2, timestamp=5, stop_id=7, vehicle=8.
let mut vp = Vec::new();
if !trip.is_empty() {
enc_len_field(&mut vp, 1, &trip);
}
enc_len_field(&mut vp, 2, &pos);
if let Some(ts) = v.timestamp {
enc_varint_field(&mut vp, 5, ts);
}
if let Some(stop) = v.stop_id {
enc_string(&mut vp, 7, stop);
}
if !vdesc.is_empty() {
enc_len_field(&mut vp, 8, &vdesc);
}
// FeedEntity: id=1, vehicle=4.
let mut entity = Vec::new();
enc_string(&mut entity, 1, v.entity_id);
enc_len_field(&mut entity, 4, &vp);
// FeedMessage.entity = 2.
enc_len_field(&mut feed, 2, &entity);
}
feed
}
// ── collector ───────────────────────────────────────────────────────────────────
/// The GTFS-Realtime vehicle-positions collector. `cfg.base_url` is the feed's
/// `.pb` endpoint (a great many agencies publish one; some behind a self-serve
/// key carried by [`AuthConfig`]).
#[derive(Default)]
pub struct GtfsRtCollector;
#[async_trait::async_trait]
impl Collector for GtfsRtCollector {
fn meta(&self) -> CollectorMeta {
CollectorMeta {
name: "gtfs-rt-vehicles".into(),
source_kind: SourceKind::Gtfs,
path: CollectorPath::RustNative,
direction: Direction::Source,
formats: vec!["gtfs-rt".into(), "application/x-protobuf".into()],
license: "per-source (typically CC-BY / open transit)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None,
join_keys: vec!["vehicle_id".into(), "route_id".into(), "trip_id".into()],
description: "GTFS-Realtime VehiclePositions (protobuf FeedMessage) → live \
transit vehicle position table (id/lat/lon/route/trip/timestamp) \
via a hand-rolled protobuf reader (no prost dep)."
.into(),
}
}
/// **There is no single authority here, and saying so IS the answer.** This is a
/// GENERIC reader: the authority is whichever agency runs the `.pb` endpoint an
/// operator configures, and korp cannot know it before the URL is set. What korp
/// CAN point at — and what actually helps — is the wire specification, so
/// `docs_url` is the GTFS-Realtime reference and `terms_url` is the
/// specification's own licence. The DATA licence is the feed operator's, and the
/// first quirk says so before anything else.
fn human_doc(&self) -> Option<crate::human_doc::HumanDoc> {
use crate::human_doc::{Contact, HumanDoc, RateLimit, Registration};
Some(
HumanDoc::new(
"whoever runs the GTFS-Realtime feed you configure — this is a generic \
reader, and the specification (gtfs.org, repository github.com/google/transit) \
governs the WIRE FORMAT only, never the vehicles",
"the transit agency whose vehicles the feed reports",
"https://github.com/google/transit/blob/master/LICENSE",
"https://gtfs.org/documentation/realtime/reference/",
)
.with_contact(Contact::Unknown)
.with_registration(Registration::NotNeeded)
.with_rate_limit(RateLimit::Documented(
"not set by the GTFS-Realtime specification — the agency serving the `.pb` \
sets it. `FeedHeader.timestamp` says when the snapshot was taken, so \
polling faster than the feed refreshes only re-reads the same positions."
.into(),
))
.with_quirk(
"`terms_url` is the SPECIFICATION's licence, not the feed's. The data \
licence and the contact for questions belong to whoever publishes the \
feed; the matching GTFS static feed states them in `feed_info.txt` \
(`feed_publisher_name`, `feed_publisher_url`, `feed_contact_email`, \
`feed_contact_url`) and `agency.txt` (`agency_name`, `agency_url`).",
)
.with_quirk(
"The declared source `https://gtfs-rt.example/vehiclepositions.pb` is an \
RFC 2606 documentation placeholder, not a feed. Until a real URL is \
configured the collector reports `Missing::PlaceholderSource` rather than \
running — see `Requirements::is_placeholder`.",
)
.with_quirk(
"Only an entity carrying a `vehicle` with a locatable `Position` (lat AND \
lon) becomes a row. A `trip_update` or `alert` entity, or a vehicle with \
no position, contributes NO row — an all-`trip_update` feed lands zero \
rows and comes back as a reason, not an error.",
)
.with_quirk(
"The hand-rolled protobuf reader walks `FeedMessage.entity` and SKIPS field \
1 (`FeedHeader`) along with every other unknown field. `incrementality` is \
therefore never read: a DIFFERENTIAL feed's partial update would land as \
if it were the whole world. Point this collector at FULL_DATASET feeds.",
)
.with_quirk(
"`AuthKind::None` is the DEFAULT, not a promise — some agencies put their \
realtime feed behind a self-service key. The collector resolves \
`cfg.auth` before the GET, so a keyed feed works; it is the card that \
cannot name the key, because only the configured endpoint knows it.",
),
)
}
fn source_target(&self) -> SourceTarget {
// Live vehicles are a moving-map overlay — same animated `osm3d-flight`
// scene role as AIS boats / OpenSky flights / Trafikverket trains.
SourceTarget::map_scene(
SourceEndpoint::GtfsRt,
"https://gtfs-rt.example/vehiclepositions.pb",
"gtfs_rt_vehicles",
"osm3d-flight",
)
}
/// **One vehicle position per vehicle per instant.**
///
/// The derived key was `(vehicle_id, route_id, trip_id)` — the vehicle's ASSIGNMENT,
/// which is constant for a whole journey. Every position report along that journey
/// keys identically, so an Upsert would have kept one point per trip and discarded
/// the movement, which is the only thing a realtime feed carries.
fn business_key(&self) -> crate::history::BusinessKey {
crate::history::BusinessKey::of(
&["vehicle_id", "timestamp"],
"one position per vehicle per instant — `(vehicle_id, route_id, trip_id)` is \
the ASSIGNMENT and is constant for a whole journey",
)
}
fn schema(&self, _cfg: &CollectorCfg) -> Result<Schema> {
Ok(vehicles_schema())
}
async fn collect(
&self,
cfg: &CollectorCfg,
_bounds: Bounds,
sink: &BatchSink,
) -> Result<CollectStats> {
let http = cfg.http(10);
let auth = cfg.auth.resolve_async().await?;
// The feed is binary protobuf — pull raw bytes, not text.
let bytes = http.get_bytes(&cfg.base_url, auth.as_ref()).await?;
let rows = parse_gtfs_rt(&bytes)?;
if rows.is_empty() {
return Ok(CollectStats {
reason: Some(
"GTFS-RT feed carried no locatable vehicle positions (feed gap / all \
trip_update/alert entities?)"
.into(),
),
..Default::default()
});
}
let n = rows.len() as u64;
sink.push(vehicles_batch(&rows)?)?;
Ok(CollectStats {
rows: n,
batches: 1,
table: cfg.table.clone(),
reason: None,
})
}
async fn checkpoint(&self) -> Result<Option<Checkpoint>> {
Ok(None)
}
}
#[cfg(test)]
mod tests {
use super::*;
use arrow_array::{Array, Float64Array, Int64Array, StringArray};
/// A two-vehicle GTFS-RT `FeedMessage` (with a `FeedHeader`) round-trips
/// through the hand-rolled protobuf reader → a vehicle-position table: a real
/// vehicle row decodes to the RIGHT id / lat / lon / route (RED-when-broken).
#[test]
fn gtfs_rt_vehicle_positions_decode_from_protobuf() {
let pb = build_gtfs_rt(&[
VehicleFixture {
entity_id: "veh-1",
vehicle_id: Some("9031001"),
trip_id: Some("trip-A"),
route_id: Some("17"),
latitude: 59.33107,
longitude: 18.05977,
bearing: Some(137.5),
speed: Some(11.4),
timestamp: Some(1_752_400_123),
stop_id: Some("9022001"),
},
VehicleFixture {
entity_id: "veh-2",
vehicle_id: Some("9031002"),
trip_id: Some("trip-B"),
route_id: Some("96"),
latitude: 57.70750,
longitude: 11.96750,
bearing: None,
speed: None,
timestamp: Some(1_752_400_130),
stop_id: None,
},
]);
let rows = parse_gtfs_rt(&pb).expect("parse gtfs-rt feed");
assert_eq!(rows.len(), 2, "two vehicle entities → two rows: {rows:#?}");
// RED-when-broken: the first vehicle decodes to the exact id/lat/lon/route.
let v0 = &rows[0];
assert_eq!(v0.entity_id, "veh-1");
assert_eq!(v0.vehicle_id.as_deref(), Some("9031001"));
assert_eq!(v0.route_id.as_deref(), Some("17"));
assert_eq!(v0.trip_id.as_deref(), Some("trip-A"));
assert!(
(v0.latitude - 59.33107).abs() < 1e-3,
"lat decoded from float: {}",
v0.latitude
);
assert!(
(v0.longitude - 18.05977).abs() < 1e-3,
"lon decoded from float: {}",
v0.longitude
);
assert!((v0.bearing.unwrap() - 137.5).abs() < 1e-3);
assert!((v0.speed.unwrap() - 11.4).abs() < 1e-3);
assert_eq!(v0.timestamp, Some(1_752_400_123));
assert_eq!(v0.stop_id.as_deref(), Some("9022001"));
// The second vehicle: bearing/speed/stop absent → None (not fabricated).
let v1 = &rows[1];
assert_eq!(v1.route_id.as_deref(), Some("96"));
assert!(v1.bearing.is_none() && v1.speed.is_none() && v1.stop_id.is_none());
// Arrow contract: types + a value probe.
let rb = vehicles_batch(&rows).expect("vehicles batch");
assert_eq!(rb.num_rows(), 2);
let schema = rb.schema();
let cols: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
for want in [
"entity_id",
"vehicle_id",
"route_id",
"latitude",
"longitude",
"timestamp",
] {
assert!(
cols.contains(&want),
"vehicle table carries `{want}`: {cols:?}"
);
}
assert_eq!(
schema.field_with_name("latitude").unwrap().data_type(),
&DataType::Float64
);
assert_eq!(
schema.field_with_name("timestamp").unwrap().data_type(),
&DataType::Int64
);
let lat = rb
.column_by_name("latitude")
.and_then(|a| a.as_any().downcast_ref::<Float64Array>())
.expect("latitude f64");
assert!((lat.value(0) - 59.33107).abs() < 1e-3);
let route = rb
.column_by_name("route_id")
.and_then(|a| a.as_any().downcast_ref::<StringArray>())
.expect("route_id utf8");
assert_eq!(route.value(1), "96");
let ts = rb
.column_by_name("timestamp")
.and_then(|a| a.as_any().downcast_ref::<Int64Array>())
.expect("timestamp i64");
assert_eq!(ts.value(0), 1_752_400_123);
crate::functional_status(
"korp-collectors/gtfs-rt-vehicles",
"protobuf_vehicle_positions",
true,
&format!(
"{} vehicle rows decoded from a GTFS-RT protobuf FeedMessage",
rb.num_rows()
),
);
}
/// The feed carries a `FeedHeader`, a vehicle entity, a NON-vehicle entity (a
/// `trip_update`, field 3 — no `vehicle`), and an entity whose vehicle has no
/// `Position`. Only the one locatable vehicle becomes a row — the header +
/// non-vehicle + position-less entities are all skipped (RED-when-broken).
#[test]
fn gtfs_rt_skips_header_non_vehicle_and_positionless_entities() {
// One real vehicle via the encoder.
let mut pb = build_gtfs_rt(&[VehicleFixture {
entity_id: "veh-1",
vehicle_id: Some("V1"),
route_id: Some("17"),
latitude: 59.0,
longitude: 18.0,
..Default::default()
}]);
// Append a trip_update entity (FeedEntity.id=1, trip_update=3 sub-message,
// NO vehicle field) — must contribute no row.
let mut trip_update = Vec::new();
enc_string(&mut trip_update, 1, "some-trip");
let mut tu_entity = Vec::new();
enc_string(&mut tu_entity, 1, "tu-1");
enc_len_field(&mut tu_entity, 3, &trip_update); // FeedEntity.trip_update = 3
enc_len_field(&mut pb, 2, &tu_entity); // FeedMessage.entity = 2
// Append a vehicle entity whose VehiclePosition has an EMPTY position
// (no lat/lon) — a vehicle with no locatable point → no row.
let empty_pos: Vec<u8> = Vec::new();
let mut vp = Vec::new();
enc_len_field(&mut vp, 2, &empty_pos); // VehiclePosition.position (empty)
let mut noloc_entity = Vec::new();
enc_string(&mut noloc_entity, 1, "noloc-1");
enc_len_field(&mut noloc_entity, 4, &vp); // FeedEntity.vehicle = 4
enc_len_field(&mut pb, 2, &noloc_entity);
let rows = parse_gtfs_rt(&pb).expect("parse mixed feed");
assert_eq!(
rows.len(),
1,
"only the locatable vehicle is a row (header + trip_update + positionless skipped): {rows:#?}"
);
assert_eq!(rows[0].vehicle_id.as_deref(), Some("V1"));
assert_eq!(rows[0].route_id.as_deref(), Some("17"));
crate::functional_status(
"korp-collectors/gtfs-rt-vehicles",
"skip_header_and_non_vehicle_entities",
true,
"1 row; FeedHeader, trip_update entity, and positionless vehicle all skipped",
);
}
/// An empty feed is an empty (not an error); a truncated varint / truncated
/// length-delimited field is a hard `Err` (RAGNARÖK), never a panic — the
/// malformed-input guards.
#[test]
fn gtfs_rt_empty_is_ok_and_truncated_is_error() {
assert!(parse_gtfs_rt(b"").expect("empty feed parses").is_empty());
// A tag that starts a length-delimited FeedMessage.entity (field 2, wire 2)
// but whose declared length runs past the buffer → truncated → Err.
// 0x12 = field 2, wire 2; 0x7f = length 127 with only a few bytes present.
assert!(
parse_gtfs_rt(&[0x12, 0x7f, 0x01, 0x02]).is_err(),
"over-long length → Err"
);
// A tag byte with the continuation bit set but no following byte → the
// varint is truncated → Err (not a panic).
assert!(parse_gtfs_rt(&[0x80]).is_err(), "truncated varint → Err");
}
}