Something went wrong. Try again.
Live STIB-MIVB departures on the Pebble Time 2. It covers the bus, tram, and metro network of Brussels
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565use crate::{ provider::{FeedFailure, FeedStage, RawFeed}, quota::Quota, storage::{VilloDiscoveryBatch, VilloStatusBatch},};use anyhow::{Context, Result};use chrono::Utc;use serde::Deserialize;use std::collections::HashMap;
const DISCOVERY_URL: &str = "https://api.cyclocity.fr/contracts/bruxelles/gbfs/v3/gbfs.json";
pub type VilloInfo = HashMap<Box<str>, VilloStationMeta>;
#[derive(Clone, Debug, PartialEq)]pub struct VilloStationMeta { pub name: String, pub lat: f64, pub lon: f64,}
pub fn parse_discovery(body: &str) -> Result<(String, String)> { let doc: serde_json::Value = serde_json::from_str(body).context("discovery envelope")?; let data = doc.get("data").context("missing data")?; let feeds = data .get("feeds") .and_then(|f| f.as_array()) .cloned() .or_else(|| { let obj = data.as_object()?; let lang = obj.get("fr").or_else(|| obj.values().next())?; lang.get("feeds")?.as_array().cloned() }) .context("no feed set found")?;
let mut info_url = None; let mut status_url = None; for feed in &feeds { let name = feed.get("name").and_then(|n| n.as_str()).unwrap_or_default(); let url = feed.get("url").and_then(|u| u.as_str()); match (name, url) { ("station_information", Some(u)) => info_url = Some(u.to_string()), ("station_status", Some(u)) => status_url = Some(u.to_string()), _ => {} } } Ok(( info_url.context("station_information feed missing")?, status_url.context("station_status feed missing")?, ))}
#[derive(Deserialize)]struct InformationDoc { data: InformationData,}
#[derive(Deserialize)]struct InformationData { stations: Vec<RawStationInfo>,}
#[derive(Deserialize)]struct RawStationInfo { station_id: Option<String>, name: Option<NameField>, lat: Option<f64>, lon: Option<f64>,}
#[derive(Deserialize)]#[serde(untagged)]enum NameField { Plain(String), Localized(Vec<LocalizedName>),}
#[derive(Deserialize)]struct LocalizedName { text: String, language: String,}
fn resolve_name(field: NameField) -> String { match field { NameField::Plain(s) => s, NameField::Localized(names) => names .iter() .find(|n| n.language == "fr") .or_else(|| names.iter().find(|n| !n.text.trim().is_empty())) .map(|n| n.text.clone()) .unwrap_or_default(), }}
pub fn parse_information(body: &str) -> Result<VilloInfo> { let doc: InformationDoc = serde_json::from_str(body).context("station_information envelope")?; Ok(doc .data .stations .into_iter() .filter_map(|s| { let id = s.station_id.filter(|id| !id.trim().is_empty())?; let lat = s.lat?; let lon = s.lon?; let name = s.name.map(resolve_name).unwrap_or_default(); Some((id.into_boxed_str(), VilloStationMeta { name, lat, lon })) }) .collect())}
#[derive(Deserialize)]struct StatusDoc { data: StatusData,}
#[derive(Deserialize)]struct StatusData { stations: Vec<RawStationStatus>,}
#[derive(Deserialize)]struct RawStationStatus { station_id: Option<String>, num_vehicles_available: Option<i64>, num_bikes_available: Option<i64>, num_docks_available: Option<i64>, vehicle_types_available: Option<Vec<RawVehicleTypeCount>>,}
#[derive(Deserialize)]struct RawVehicleTypeCount { vehicle_type_id: Option<String>, count: Option<i64>,}
fn as_count(n: i64) -> Option<u16> { u16::try_from(n).ok()}
fn kind_count(types: &Option<Vec<RawVehicleTypeCount>>, kind: &str) -> Option<u16> { types .as_ref()? .iter() .find(|t| t.vehicle_type_id.as_deref() == Some(kind)) .and_then(|t| t.count) .and_then(as_count)}
pub type VilloStatus = HashMap<Box<str>, VilloStationCounts>;
#[derive(Clone, Debug, PartialEq)]pub struct VilloStationCounts { pub bikes: u16, pub docks: u16, pub bikes_mechanical: Option<u16>, pub bikes_electrical: Option<u16>,}
pub fn parse_status(body: &str) -> Result<VilloStatus> { let doc: StatusDoc = serde_json::from_str(body).context("station_status envelope")?; Ok(doc .data .stations .into_iter() .filter_map(|s| { let mechanical = kind_count(&s.vehicle_types_available, "mechanical"); let electrical = kind_count(&s.vehicle_types_available, "electrical"); let id = s.station_id.filter(|id| !id.trim().is_empty())?; let bikes = as_count(s.num_vehicles_available.or(s.num_bikes_available)?)?; let docks = as_count(s.num_docks_available?)?; Some((id.into_boxed_str(), VilloStationCounts { bikes, docks, bikes_mechanical: mechanical, bikes_electrical: electrical, })) }) .collect())}
async fn fetch_raw( http: &reqwest::Client, quota: &Quota, url: &str, feed: &'static str, scope: &str,) -> std::result::Result<RawFeed, FeedFailure> { if !quota.try_take(Utc::now()) { return Err(FeedFailure { stage: FeedStage::Provider, source: anyhow::anyhow!("villo quota exhausted"), }); } let response = http .get(url) .send() .await .with_context(|| format!("failed to fetch {url}")) .and_then(|response| { response .error_for_status() .with_context(|| format!("failed to fetch {url}")) }) .map_err(|source| FeedFailure { stage: FeedStage::Provider, source, })?; let content_type = response .headers() .get(reqwest::header::CONTENT_TYPE) .with_context(|| format!("missing content type for {url}")) .and_then(|value| { value .to_str() .with_context(|| format!("invalid content type for {url}")) }) .map_err(|source| FeedFailure { stage: FeedStage::Provider, source, })? .to_owned() .into_boxed_str(); let body = response .bytes() .await .with_context(|| format!("failed to read {url}")) .map_err(|source| FeedFailure { stage: FeedStage::Provider, source, })? .to_vec(); Ok(RawFeed { provider: "villo", feed, scope: scope.into(), captured_at: Utc::now(), source_at: None, content_type, body, })}
async fn fetch_discovery_at( http: &reqwest::Client, quota: &Quota, discovery_url: &str,) -> std::result::Result<VilloDiscoveryBatch, FeedFailure> { let discovery = fetch_raw(http, quota, discovery_url, "discovery", "gbfs").await?; let (information_url, status_url) = std::str::from_utf8(&discovery.body) .context("villo discovery encoding") .and_then(parse_discovery) .map_err(|source| FeedFailure { stage: FeedStage::Parse, source, })?; let information = fetch_raw( http, quota, &information_url, "discovery", "station_information", ) .await?; let stations = std::str::from_utf8(&information.body) .context("villo station information encoding") .and_then(parse_information) .and_then(|stations| { anyhow::ensure!(!stations.is_empty(), "villo stations empty"); Ok(stations) }) .map_err(|source| FeedFailure { stage: FeedStage::Parse, source, })?; Ok(VilloDiscoveryBatch { raw: vec![discovery, information], status_url, stations, })}
pub async fn fetch_discovery( http: &reqwest::Client, quota: &Quota,) -> std::result::Result<VilloDiscoveryBatch, FeedFailure> { fetch_discovery_at(http, quota, DISCOVERY_URL).await}
pub async fn fetch_status( http: &reqwest::Client, quota: &Quota, status_url: &str,) -> std::result::Result<VilloStatusBatch, FeedFailure> { let raw = fetch_raw(http, quota, status_url, "status", "").await?; let status = std::str::from_utf8(&raw.body) .context("villo station status encoding") .and_then(parse_status) .map_err(|source| FeedFailure { stage: FeedStage::Parse, source, })?; Ok(VilloStatusBatch { raw, status })}
#[cfg(test)]mod tests { use super::*;
const DISCOVERY_FLAT: &str = r#"{"last_updated":"2026-08-20T22:31:37Z","ttl":3600,"version":"3.0","data":{"feeds":[ {"name":"gbfs_versions","url":"https://api.cyclocity.fr/contracts/bruxelles/gbfs/v3/gbfs_versions.json"}, {"name":"station_information","url":"https://api.cyclocity.fr/contracts/bruxelles/gbfs/v3/station_information.json"}, {"name":"station_status","url":"https://api.cyclocity.fr/contracts/bruxelles/gbfs/v3/station_status.json"} ]}}"#;
const DISCOVERY_BY_LANGUAGE: &str = r#"{"data":{ "en":{"feeds":[{"name":"station_information","url":"https://example/en/info.json"},{"name":"station_status","url":"https://example/en/status.json"}]}, "fr":{"feeds":[{"name":"station_information","url":"https://example/fr/info.json"},{"name":"station_status","url":"https://example/fr/status.json"}]} }}"#;
#[test] fn parses_flat_v3_discovery() { let (info, status) = parse_discovery(DISCOVERY_FLAT).unwrap(); assert_eq!(info, "https://api.cyclocity.fr/contracts/bruxelles/gbfs/v3/station_information.json"); assert_eq!(status, "https://api.cyclocity.fr/contracts/bruxelles/gbfs/v3/station_status.json"); }
#[test] fn prefers_fr_feed_set_when_language_keyed() { let (info, status) = parse_discovery(DISCOVERY_BY_LANGUAGE).unwrap(); assert_eq!(info, "https://example/fr/info.json"); assert_eq!(status, "https://example/fr/status.json"); }
const INFORMATION: &str = r#"{"data":{"stations":[ {"station_id":"1","name":[{"text":"LEOPOLD II","language":"en"},{"text":"LEOPOLD II FR","language":"fr"}],"lat":50.836304,"lon":4.41505,"capacity":20}, {"station_id":"2","name":"PLAIN NAME","lat":50.836151,"lon":4.326019,"capacity":25}, {"station_id":"3","lat":50.85,"lon":4.35,"capacity":10}, {"station_id":"","name":"NO ID","lat":50.85,"lon":4.35}, {"station_id":"5","name":"NO COORDS","capacity":10} ]}}"#;
#[test] fn resolves_v3_localized_name_preferring_fr() { let stations = parse_information(INFORMATION).unwrap(); assert_eq!(stations["1"].name, "LEOPOLD II FR"); assert!((stations["1"].lat - 50.836304).abs() < 1e-9); assert!((stations["1"].lon - 4.41505).abs() < 1e-9); }
#[test] fn falls_back_to_plain_string_name_for_v2() { let stations = parse_information(INFORMATION).unwrap(); assert_eq!(stations["2"].name, "PLAIN NAME"); }
#[test] fn defaults_missing_name_to_empty() { let stations = parse_information(INFORMATION).unwrap(); assert_eq!(stations["3"].name, ""); }
#[test] fn skips_malformed_information_entries() { let stations = parse_information(INFORMATION).unwrap(); assert_eq!(stations.len(), 3); assert!(!stations.contains_key("")); assert!(!stations.contains_key("5")); }
const STATUS: &str = r#"{"data":{"stations":[ {"station_id":"1","num_vehicles_available":6,"num_docks_available":14,"is_installed":true,"vehicle_types_available":[{"vehicle_type_id":"mechanical","count":4},{"vehicle_type_id":"electrical","count":2}]}, {"station_id":"2","num_bikes_available":12,"num_docks_available":13,"is_installed":true}, {"station_id":"3","num_docks_available":0}, {"station_id":"","num_vehicles_available":1,"num_docks_available":1} ]}}"#;
#[test] fn reads_v3_num_vehicles_available() { let status = parse_status(STATUS).unwrap(); assert_eq!(status["1"], VilloStationCounts { bikes: 6, docks: 14, bikes_mechanical: Some(4), bikes_electrical: Some(2), }); }
#[test] fn falls_back_to_v2_num_bikes_available() { let status = parse_status(STATUS).unwrap(); assert_eq!(status["2"], VilloStationCounts { bikes: 12, docks: 13, bikes_mechanical: None, bikes_electrical: None, }); }
#[test] fn vehicle_type_breakdown_tolerates_unknown_ids_and_bad_counts() { let body = r#"{"data":{"stations":[ {"station_id":"1","num_vehicles_available":6,"num_docks_available":14,"vehicle_types_available":[{"vehicle_type_id":"cargo","count":1},{"vehicle_type_id":"electrical","count":-2},{"vehicle_type_id":"mechanical"}]} ]}}"#; let status = parse_status(body).unwrap(); assert_eq!(status["1"], VilloStationCounts { bikes: 6, docks: 14, bikes_mechanical: None, bikes_electrical: None, }); }
#[test] fn skips_malformed_status_entries() { let status = parse_status(STATUS).unwrap(); assert_eq!(status.len(), 2); assert!(!status.contains_key("3")); assert!(!status.contains_key("")); }
#[test] fn skips_a_negative_count_without_failing_the_batch() { let body = r#"{"data":{"stations":[ {"station_id":"1","num_vehicles_available":-3,"num_docks_available":14}, {"station_id":"2","num_vehicles_available":6,"num_docks_available":13} ]}}"#; let status = parse_status(body).unwrap(); assert_eq!(status.len(), 1); assert_eq!(status["2"], VilloStationCounts { bikes: 6, docks: 13, bikes_mechanical: None, bikes_electrical: None, }); }
async fn villo_server() -> (String, tokio::task::JoinHandle<()>) { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); let base = format!("http://{address}"); let discovery = format!( r#"{{"data":{{"feeds":[ {{"name":"station_information","url":"{base}/information"}}, {{"name":"station_status","url":"{base}/status"}} ]}}}}"#, ); let app = axum::Router::new() .route( "/gbfs", axum::routing::get(move || { let discovery = discovery.clone(); async move { ( [(axum::http::header::CONTENT_TYPE, "application/json")], discovery, ) } }), ) .route( "/information", axum::routing::get(|| async { ( [(axum::http::header::CONTENT_TYPE, "application/json")], INFORMATION, ) }), ) .route( "/status", axum::routing::get(|| async { ( [(axum::http::header::CONTENT_TYPE, "application/json")], STATUS, ) }), ); let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap(); }); (base, server) }
#[tokio::test] async fn fetches_discovery_and_status_as_raw_parsed_batches() { let (base, server) = villo_server().await; let http = reqwest::Client::new(); let quota = Quota::new(10, 5_000); let discovery = fetch_discovery_at(&http, "a, &format!("{base}/gbfs")) .await .unwrap(); assert_eq!(discovery.raw.len(), 2); assert_eq!(discovery.raw[0].provider, "villo"); assert_eq!(discovery.raw[0].feed, "discovery"); assert_eq!(discovery.raw[0].scope.as_ref(), "gbfs"); assert!(discovery.stations.contains_key("1")); assert_eq!(discovery.status_url, format!("{base}/status"));
let status = fetch_status(&http, "a, &discovery.status_url) .await .unwrap(); server.abort(); assert_eq!(status.raw.provider, "villo"); assert_eq!(status.raw.feed, "status"); assert!(status.raw.scope.is_empty()); assert_eq!(status.raw.body, STATUS.as_bytes()); assert_eq!(status.status["1"].bikes, 6); }
#[tokio::test] async fn fetch_discovery_maps_http_failures_to_provider_stage() { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); let app = axum::Router::new().fallback(|| async { axum::http::StatusCode::BAD_GATEWAY }); let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap(); }); let failure = match fetch_discovery_at( &reqwest::Client::new(), &Quota::new(10, 5_000), &format!("http://{address}/gbfs"), ) .await { Ok(_) => panic!("expected provider failure"), Err(failure) => failure, }; server.abort(); assert_eq!(failure.stage, crate::provider::FeedStage::Provider); }
#[tokio::test] async fn fetch_discovery_maps_invalid_bodies_to_parse_stage() { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); let app = axum::Router::new().route( "/gbfs", axum::routing::get(|| async { ( [(axum::http::header::CONTENT_TYPE, "application/json")], "invalid", ) }), ); let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap(); }); let failure = match fetch_discovery_at( &reqwest::Client::new(), &Quota::new(10, 5_000), &format!("http://{address}/gbfs"), ) .await { Ok(_) => panic!("expected parse failure"), Err(failure) => failure, }; server.abort(); assert_eq!(failure.stage, crate::provider::FeedStage::Parse); }
}