src/trafikverket.rs
//! **Trafikverket** (Transport Administration) — real-time road + rail. The API
//! takes an **XML `<REQUEST>` body** POSTed to
//! `https://api.trafikinfo.trafikverket.se/v2/data.json` and answers JSON. The
//! authentication key lives **inside** the request (`<LOGIN authenticationkey=…>`)
//! — pluggable via env (`AuthConfig` is not used for the URL here; the key is
//! injected into the body), never hardcoded.
//!
//! This is a **moving-map** connector: `TrainPosition` lands a time-stamped
//! position table (`train_id`, WGS84 point, timestamp, bearing, speed) so
//! facett-map3d can overlay + animate trains and time-travel via skade AS-OF.
//! `RoadNetwork` / weather stations reuse the same client with a different
//! `objecttype` + field map.
//!
//! Offline contract: [`super::rest::parse_records`] over a canned `data.json`
//! response at pointer `/RESPONSE/RESULT/0/{objecttype}`.
use anyhow::Result;
use arrow_schema::Schema;
use super::rest::{FieldMap, FieldType, parse_records};
use super::{
AuthKind, BatchSink, Bounds, CollectStats, Collector, CollectorCfg, CollectorMeta,
CollectorPath, Direction, GdprClass, SourceEndpoint, SourceKind, SourceTarget,
};
/// Build the Trafikverket XML request body: a `<LOGIN>` carrying the key + the
/// configured `<QUERY>` fragment. The key is read from env (`key_env`); the
/// query XML comes from config (`params.query` — the `objecttype`, filters, …).
pub fn build_request(key: &str, query_xml: &str) -> String {
format!("<REQUEST><LOGIN authenticationkey=\"{key}\"/>{query_xml}</REQUEST>")
}
/// The Trafikverket collector. `params.objecttype` selects the result array
/// (e.g. `"TrainPosition"`); `params.query` is the `<QUERY …>…</QUERY>` XML; the
/// key env var defaults to `TRAFIKVERKET_API_KEY` (override via `params.key_env`).
pub struct TrafikverketCollector {
name: String,
objecttype: String,
fields: Vec<FieldMap>,
/// The facett map scene this arm feeds (`osm3d-flight` for the animated
/// train overlay, `osm3d` for the static NVDB ways layer).
scene: &'static str,
/// The representative warehouse/Iceberg table this arm lands.
table: String,
}
impl Default for TrafikverketCollector {
fn default() -> Self {
Self::train_positions()
}
}
impl TrafikverketCollector {
/// Real-time **train positions** → time-stamped position table.
pub fn train_positions() -> Self {
Self {
name: "trafikverket-trainpos".into(),
objecttype: "TrainPosition".into(),
scene: "osm3d-flight",
table: "trafikverket_trainpos".into(),
fields: vec![
FieldMap::new("train_id", "/Train/OperationalTrainNumber", FieldType::Str),
FieldMap::new("wgs84", "/Position/WGS84", FieldType::Str),
FieldMap::new("timestamp", "/TimeStamp", FieldType::Str),
FieldMap::new("bearing", "/Bearing", FieldType::F64),
FieldMap::new("speed", "/Speed", FieldType::F64),
FieldMap::new("status_active", "/Status/Active", FieldType::Str),
],
}
}
/// Real-time **train positions**, richly mapped (objecttype `TrainPosition`)
/// → a moving-map position table facett-map3d overlays + ANIMATES on the
/// `osm3d-flight` scene (the same role as AIS boats / OpenSky flights).
/// Distinct, more richly mapped sibling of [`Self::train_positions`]: POSTs
/// the same XML query and parses the same `/RESPONSE/RESULT/0/TrainPosition`
/// array, adding advertised number + modified time.
pub fn trains() -> Self {
Self {
name: "trafikverket_trains".into(),
objecttype: "TrainPosition".into(),
scene: "osm3d-flight",
table: "trafikverket_trains".into(),
fields: vec![
FieldMap::new(
"train_number",
"/Train/OperationalTrainNumber",
FieldType::Str,
),
FieldMap::new(
"advertised_number",
"/Train/AdvertisedTrainNumber",
FieldType::Str,
),
FieldMap::new("wgs84", "/Position/WGS84", FieldType::Str),
FieldMap::new("bearing", "/Bearing", FieldType::F64),
FieldMap::new("speed", "/Speed", FieldType::F64),
FieldMap::new("timestamp", "/TimeStamp", FieldType::Str),
FieldMap::new("modified_time", "/ModifiedTime", FieldType::Str),
FieldMap::new("status_active", "/Status/Active", FieldType::Str),
],
}
}
/// **NVDB road network** (`RoadGeometry`) → a static WaysLayer table. Each
/// record is a road segment whose line geometry arrives as a **WKT
/// LINESTRING** in `/Geometry/WGS84` (WGS84 lon/lat). Unlike osm.rs (which
/// WKB-encodes ways), Trafikverket hands back WKT text, so `FieldType::Str`
/// maps it 1:1 into a `geometry_wgs84` Utf8 column — no new codec. Lands in
/// the `osm3d` ways scene the facett map renders UNDER the moving overlays.
pub fn road_network() -> Self {
Self {
name: "trafikverket-roads".into(),
objecttype: "RoadGeometry".into(),
scene: "osm3d",
table: "trafikverket_roads".into(),
fields: vec![
FieldMap::new("element_id", "/Element_Id", FieldType::I64),
FieldMap::new("road_main_number", "/RoadMainNumber", FieldType::I64),
FieldMap::new("road_sub_number", "/RoadSubNumber", FieldType::I64),
FieldMap::new("direction", "/Direction", FieldType::I64),
FieldMap::new("length_m", "/Length", FieldType::F64),
FieldMap::new("county", "/County", FieldType::I64),
// The line geometry — a WKT LINESTRING in WGS84 lon/lat.
FieldMap::new("geometry_wgs84", "/Geometry/WGS84", FieldType::Str),
FieldMap::new(
"geometry_sweref99tm",
"/Geometry/SWEREF99TM",
FieldType::Str,
),
FieldMap::new("modified_time", "/ModifiedTime", FieldType::Str),
FieldMap::new("timestamp", "/TimeStamp", FieldType::Str),
],
}
}
pub fn fields(&self) -> &[FieldMap] {
&self.fields
}
pub fn objecttype(&self) -> &str {
&self.objecttype
}
/// RFC-6901 pointer to the result array for this object type.
pub fn records_pointer(&self) -> String {
format!("/RESPONSE/RESULT/0/{}", self.objecttype)
}
}
#[async_trait::async_trait]
impl Collector for TrafikverketCollector {
/// **Trafikverket's open API needs an agreement key**, and every arm of this
/// collector (train positions · announcements · road network) posts the same
/// `<LOGIN authenticationkey=…>` envelope. `TRAFIKVERKET_API_KEY` is the name the
/// host already uses (`korp::collect::populate::known_sources`), so the
/// declaration and the run path name ONE variable, not two.
fn requirements(&self) -> crate::requires::Requirements {
crate::requires::Requirements::open(self.source_target().source_endpoint).with_secrets(
vec![crate::requires::SecretReq::new(
"TRAFIKVERKET_API_KEY",
"a free Trafikverket API key (api.trafikinfo.trafikverket.se ▸ Skapa konto)",
)],
)
}
fn meta(&self) -> CollectorMeta {
CollectorMeta {
name: self.name.clone(),
source_kind: SourceKind::Rest,
path: CollectorPath::RustNative,
direction: Direction::Source,
formats: vec!["json".into(), "xml-query".into()],
license: "free but key + agreement".into(),
gdpr: GdprClass::Open,
auth: AuthKind::Agreement,
join_keys: vec!["train_id".into(), "coordinate".into(), "timestamp".into()],
description: "Trafikverket real-time road/rail (POST XML query → JSON); \
TrainPosition = moving-map position table."
.into(),
}
}
/// **The one collector in this file's neighbourhood with a real, single
/// authority** — and the card says the same thing [`Self::requirements`] already
/// says, in the same words, because two different phrasings of "you need a key"
/// are how a card starts lying.
///
/// Every fact here is published on Trafikverket's own open-API page
/// (`trafikverket.se/e-tjanster/trafikverkets-oppna-api-for-trafikinformation/`,
/// read 2026-09-04): the CC0 licence link, the registration steps (register →
/// accept the licence attached to the API → verify your e-mail), and the traffic
/// statement — Trafikverket monitors how much traffic each service generates and
/// CONTACTS you if you exceed the limits described in its documentation. That last
/// one is the honest rate limit: there is no number, there is a phone call.
///
/// The `road_network` arm differs in one field only — it serves NVDB, the national
/// road database, rather than live train positions — so it gets its own `owner`
/// line and its own geometry quirk, and shares everything else.
fn human_doc(&self) -> Option<crate::human_doc::HumanDoc> {
use crate::human_doc::{Contact, HumanDoc, RateLimit, Registration};
let roads = self.objecttype == "RoadGeometry";
let doc = HumanDoc::new(
"Trafikverket (the Swedish Transport Administration)",
if roads {
"Trafikverket — the road geometry comes from NVDB, its national road \
database"
} else {
"Trafikverket"
},
"https://www.trafikverket.se/e-tjanster/trafikverkets-oppna-api-for-trafikinformation/",
"https://data.trafikverket.se/documentation/datacache/intro",
)
.with_contact(Contact::Form(
"https://www.trafikverket.se/om-oss/kontakta-oss/".into(),
))
.with_registration(Registration::Agreement {
how: "a free Trafikverket API key (api.trafikinfo.trafikverket.se ▸ Skapa \
konto) — register at https://data.trafikverket.se/get-started, ACCEPT \
the licence attached to the API, then verify your e-mail address. \
Registering makes you a customer of Trafikverket and it stores your \
e-mail. Put the key in `TRAFIKVERKET_API_KEY`."
.into(),
})
.with_rate_limit(RateLimit::Documented(
"no published number. Trafikverket states that it monitors how much traffic \
each service generates and will CONTACT you if you exceed the limits \
described in its documentation."
.into(),
))
.with_quirk(
"The key goes in the REQUEST BODY, not a header or the URL: every call POSTs \
`<REQUEST><LOGIN authenticationkey=…/><QUERY …/></REQUEST>` as `text/xml` to \
a single endpoint (`api.trafikinfo.trafikverket.se/v2/data.json`) and gets \
JSON back. Nothing in `AuthConfig` reaches the wire here — the collector \
reads `TRAFIKVERKET_API_KEY` (override with `params.key_env`) itself.",
)
.with_quirk(
"The data is CC0 1.0 (Trafikverket links \
creativecommons.org/publicdomain/zero/1.0/ from its own API page), but the \
ACCESS still needs an accepted licence and a verified account — which is why \
this declares `AuthKind::Agreement` and not `AuthKind::None`. Free to use, \
not free to reach.",
)
.with_quirk(
"One endpoint, many object types: `params.objecttype` selects the arm and \
`params.query` carries the `<QUERY …>` XML. With no `params.query` the \
collector sends a bare `<QUERY objecttype=\"…\" schemaversion=\"1.0\"/>` — \
unfiltered, which on a national object type is a great deal of data. Write \
the filter.",
);
Some(if roads {
doc.with_quirk(
"NVDB hands back geometry as WKT TEXT, not WKB: `/Geometry/WGS84` is a \
`LINESTRING (lon lat, …)` string mapped 1:1 into the `geometry_wgs84` \
Utf8 column, with the SWEREF99TM variant kept beside it. That is the \
opposite of `crate::osm`, which WKB-encodes — do not reach for a WKB \
decoder on this table.",
)
} else {
doc.with_quirk(
"Positions arrive as WKT `POINT (lon lat)` strings in `wgs84` and \
timestamps as the source's own offset strings \
(`2026-07-13T10:15:00.000+02:00` — local Swedish offset, not UTC). \
Nothing is parsed or normalised here so downstream owns both; `speed` \
and `bearing` are the only numeric columns.",
)
})
}
fn source_target(&self) -> SourceTarget {
// Each arm declares its own scene/table: TrainPosition → the animated
// `osm3d-flight` overlay; RoadGeometry → the static `osm3d` ways layer.
SourceTarget::map_scene(
SourceEndpoint::Trafikverket,
"https://api.trafikinfo.trafikverket.se/v2/data.json",
self.table.clone(),
self.scene,
)
}
/// **The identity differs per Trafikinfo object type**, and the derived key did
/// not: all three inherited `["train_id", "coordinate", "timestamp"]`, so
/// `trafikverket-roads` — a table of ROAD ELEMENTS — was keyed on a train.
///
/// * a train POSITION is one fix per train per instant,
/// * `TrainAnnouncement` rows carry `train_number` rather than `train_id`,
/// * a road element is a permanent object with its own `element_id`, and keying it
/// on `timestamp` (the one column that survived the derivation) would have
/// collapsed every road in Sweden read in the same second into one row.
fn business_key(&self) -> crate::history::BusinessKey {
match self.name.as_str() {
"trafikverket-roads" => crate::history::BusinessKey::of(
&["element_id"],
"a road element is a permanent object — `element_id` is NVDB's own id",
),
"trafikverket_trains" => crate::history::BusinessKey::of(
&["train_number", "timestamp"],
"one announcement per train per instant — this feed spells the train \
`train_number`, not `train_id`",
),
_ => crate::history::BusinessKey::of(
&["train_id", "timestamp"],
"one position fix per train per instant — `wgs84` is WHERE it is, which \
is exactly the thing that changes",
),
}
}
/// **Train positions as a graph** (lane EC O7) — only the position feed has one.
///
/// A row is one FIX of one train and the table carries no key for it, so the fix's
/// identity is derived: `train_id|timestamp`, which is also its business key.
fn graph_mapping(&self) -> Option<crate::graph_mapping::GraphMapping> {
use crate::graph_mapping::{DerivedId, GraphMapping};
if self.name != "trafikverket-trainpos" {
return None;
}
Some(
GraphMapping::new(
"trafikverket_trains",
"Trafikverket — tågpositioner",
"trafikverket_trainpos",
"never drilled live — oden holds no TRAFIKVERKET_API_KEY, so this spec \
is proven on the collector's own fixture only. The feed carries a WGS84 \
point, not a line or a station, so there is no train→line edge.",
)
.derive_id(DerivedId::from_columns("position_id", &["train_id", "timestamp"]))
.vertex(
"TrainPosition",
"position_id",
&["wgs84", "timestamp", "bearing", "speed"],
)
.vertex("Train", "train_id", &["status_active"])
.edge(
"OF_TRAIN",
("TrainPosition", "position_id", "position_id"),
("Train", "train_id", "train_id"),
),
)
}
fn schema(&self, _cfg: &CollectorCfg) -> Result<Schema> {
let fdefs: Vec<(&str, arrow_schema::DataType, bool)> = self
.fields
.iter()
.map(|f| (f.column.as_str(), f.ty.data_type(), true))
.collect();
Ok(super::arrow_util::schema(&fdefs))
}
async fn collect(
&self,
cfg: &CollectorCfg,
_bounds: Bounds,
sink: &BatchSink,
) -> Result<CollectStats> {
let http = cfg.http(10);
// The key is in the BODY, not the URL — read it from env (pluggable).
let key_env = cfg
.params
.get("key_env")
.and_then(|v| v.as_str())
.unwrap_or("TRAFIKVERKET_API_KEY");
let key = std::env::var(key_env)
.map_err(|_| anyhow::anyhow!("missing Trafikverket key env var `{key_env}`"))?;
let query_xml = cfg
.params
.get("query")
.and_then(|v| v.as_str())
.unwrap_or_else(|| {
// a minimal default query for this object type
""
});
let query_xml = if query_xml.is_empty() {
format!(
"<QUERY objecttype=\"{}\" schemaversion=\"1.0\"/>",
self.objecttype
)
} else {
query_xml.to_string()
};
let body = build_request(&key, &query_xml);
let resp = http
.post_text(&cfg.base_url, body, "text/xml", &super::http::NoAuth)
.await?;
let json: serde_json::Value = serde_json::from_str(&resp)?;
let rb = parse_records(&json, &self.records_pointer(), &self.fields)?;
let n = rb.num_rows() as u64;
if n == 0 {
return Ok(CollectStats {
reason: Some(format!("Trafikverket {} returned no rows", self.objecttype)),
..Default::default()
});
}
sink.push(rb)?;
Ok(CollectStats {
rows: n,
batches: 1,
table: cfg.table.clone(),
reason: None,
})
}
}
#[cfg(test)]
mod tests {
use super::*;
/// `TrafikverketCollector::trains()` maps a canned `TrainPosition`
/// `data.json` response → the moving-map position table (offline DATA
/// contract, no network).
#[test]
fn trafikverket_trainposition_parses_into_a_moving_map_position_table() {
let json: serde_json::Value = serde_json::json!({
"RESPONSE": { "RESULT": [ { "TrainPosition": [
{
"Train": { "OperationalTrainNumber": "12345", "AdvertisedTrainNumber": "525" },
"Position": { "WGS84": "POINT (14.12345 57.98765)" },
"Bearing": 137.5,
"Speed": 118.0,
"TimeStamp": "2026-07-13T10:15:00.000+02:00",
"ModifiedTime": "2026-07-13T10:15:02.000Z",
"Status": { "Active": true }
},
{
"Train": { "OperationalTrainNumber": "22200", "AdvertisedTrainNumber": "1043" },
"Position": { "WGS84": "POINT (18.05000 59.33000)" },
"Bearing": 12.0,
"Speed": 0.0,
"TimeStamp": "2026-07-13T10:15:01.000+02:00",
"ModifiedTime": "2026-07-13T10:15:03.000Z",
"Status": { "Active": true }
}
] } ] }
});
let c = TrafikverketCollector::trains();
assert_eq!(c.objecttype(), "TrainPosition");
assert_eq!(c.records_pointer(), "/RESPONSE/RESULT/0/TrainPosition");
let rb =
parse_records(&json, &c.records_pointer(), c.fields()).expect("parse TrainPosition");
assert_eq!(rb.num_rows(), 2, "two train positions parsed");
let schema = rb.schema();
let cols: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
for want in [
"train_number",
"wgs84",
"bearing",
"speed",
"timestamp",
"status_active",
] {
assert!(
cols.contains(&want),
"position table carries `{want}`: {cols:?}"
);
}
let wgs = rb
.column_by_name("wgs84")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::StringArray>())
.expect("wgs84 is utf8");
assert_eq!(wgs.value(0), "POINT (14.12345 57.98765)");
let speed = rb
.column_by_name("speed")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::Float64Array>())
.expect("speed is f64");
assert!((speed.value(0) - 118.0).abs() < 1e-9, "first train speed");
let active = rb
.column_by_name("status_active")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::StringArray>())
.expect("status_active is utf8");
assert_eq!(active.value(0), "true");
crate::functional_status(
"korp-collectors/trafikverket_trains",
"trainposition_parse",
true,
&format!(
"{} train positions → moving-map position table",
rb.num_rows()
),
);
}
/// `TrafikverketCollector::road_network()` maps a canned `RoadGeometry`
/// `data.json` response → a WaysLayer table with a WKT LINESTRING geometry.
#[test]
fn nvdb_road_geometry_parses_into_a_ways_layer() {
let json: serde_json::Value = serde_json::json!({
"RESPONSE": { "RESULT": [ { "RoadGeometry": [
{
"Element_Id": 123456,
"RoadMainNumber": 40,
"RoadSubNumber": 0,
"Direction": 2,
"Length": 812.5,
"County": 14,
"Geometry": {
"WGS84": "LINESTRING (12.001 57.700, 12.003 57.702, 12.006 57.705)",
"SWEREF99TM": "LINESTRING (319000 6396000, 319120 6396210)"
},
"ModifiedTime": "2026-07-10T08:00:00.000+02:00",
"TimeStamp": "2026-07-13T06:00:00.000+02:00"
},
{
"Element_Id": 123457,
"RoadMainNumber": 27,
"RoadSubNumber": 0,
"Direction": 1,
"Length": 1503.0,
"County": 14,
"Geometry": {
"WGS84": "LINESTRING (12.100 57.800, 12.104 57.803)",
"SWEREF99TM": "LINESTRING (325000 6407000, 325260 6407330)"
},
"ModifiedTime": "2026-07-11T09:30:00.000+02:00",
"TimeStamp": "2026-07-13T06:00:00.000+02:00"
}
] } ] }
});
let c = TrafikverketCollector::road_network();
let rb =
parse_records(&json, &c.records_pointer(), c.fields()).expect("parse RoadGeometry");
assert_eq!(rb.num_rows(), 2, "two road segments parsed");
let schema = rb.schema();
let cols: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
for want in [
"element_id",
"road_main_number",
"geometry_wgs84",
"length_m",
] {
assert!(
cols.contains(&want),
"roads table carries `{want}`: {cols:?}"
);
}
let eid = rb
.column_by_name("element_id")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::Int64Array>())
.expect("element_id is i64");
assert_eq!(eid.value(0), 123456);
let geom = rb
.column_by_name("geometry_wgs84")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::StringArray>())
.expect("geometry_wgs84 is utf8");
assert!(
geom.value(0).starts_with("LINESTRING"),
"WKT line geometry: {}",
geom.value(0)
);
crate::functional_status(
"korp-collectors/trafikverket-roads",
"roadgeometry_parse",
true,
&format!(
"{} NVDB road segments → WaysLayer (WKT LINESTRING)",
rb.num_rows()
),
);
}
}