src/wfs.rs
//! **OGC WFS** (Web Feature Service) — the standard geodata pull for Swedish +
//! EU agencies (Lantmäteriet, SGU, Naturvårdsverket, SMHI geodata). A
//! `GetFeature` request answers a **GML `FeatureCollection`**: a list of
//! `wfs:member` (WFS 2.0) / `gml:featureMember` (WFS 1.x) wrappers, each holding
//! ONE feature element whose children are the feature's properties, one of which
//! carries a `gml` geometry (`gml:Point` / `gml:LineString` / `gml:Polygon`).
//!
//! [`parse_wfs_gml`] streams the collection with `quick-xml` (same engine as the
//! ENTSO-E parser) and maps each member → one row: `feature_id` (`gml:id`),
//! `feature_type` (the feature element's local name), `srs_name`, a **WKT**
//! `geometry` string built from the gml coordinates (preserving the source
//! coordinate order + `srsName` so downstream owns axis interpretation), and the
//! remaining leaf properties serialised to a compact JSON `properties` string
//! (the same "everything else → JSON" shape the OSM Overpass collector uses for
//! `tags`). Pure-Rust, offline-testable — no geo engine, no network.
use anyhow::Result;
use arrow_array::RecordBatch;
use arrow_schema::{DataType, Schema};
use quick_xml::events::Event;
use super::arrow_util::{batch, schema, str_col};
use super::{
AuthKind, BatchSink, Bounds, CollectStats, Collector, CollectorCfg, CollectorMeta,
CollectorPath, Direction, GdprClass, SourceEndpoint, SourceKind, SourceTarget,
};
/// One parsed WFS feature.
#[derive(Debug, Clone, PartialEq)]
pub struct WfsFeature {
/// The `gml:id` of the feature element, if present.
pub feature_id: Option<String>,
/// The feature element's local name (the feature TYPE, e.g. `AddressPoint`).
pub feature_type: String,
/// The geometry's `srsName` (e.g. `urn:ogc:def:crs:EPSG::4326`), if declared.
pub srs_name: Option<String>,
/// The geometry as WKT (`POINT (x y)` / `LINESTRING (...)` / `POLYGON ((...))`),
/// coordinates in source order. `None` when the feature has no gml geometry.
pub geometry_wkt: Option<String>,
/// The non-geometry leaf properties as a compact JSON object string
/// (`{"name":"…","population":"1234"}`), keys sorted (BTreeMap-backed).
pub properties_json: String,
}
/// GML geometry container kinds this parser builds WKT for.
#[derive(Debug, Clone, Copy, PartialEq)]
enum GeomKind {
Point,
LineString,
Polygon,
}
/// The set of member-wrapper local names (WFS 2.0 `member`, WFS 1.x `featureMember`,
/// and the plural `featureMembers` some servers emit).
fn is_member(local: &str) -> bool {
matches!(local, "member" | "featureMember" | "featureMembers")
}
/// A gml geometry container → its kind (only the leading element matters; nested
/// rings/positions feed the same accumulator).
fn geom_kind(local: &str) -> Option<GeomKind> {
match local {
"Point" => Some(GeomKind::Point),
"LineString" | "Curve" => Some(GeomKind::LineString),
"Polygon" | "Surface" => Some(GeomKind::Polygon),
_ => None,
}
}
/// A gml coordinate-bearing leaf element.
fn is_coord_elem(local: &str) -> bool {
matches!(local, "pos" | "posList" | "coordinates")
}
fn local_name(name: quick_xml::name::QName) -> String {
String::from_utf8_lossy(name.local_name().as_ref()).into_owned()
}
/// Does this raw qname carry the `gml` prefix?
fn is_gml_prefixed(name: quick_xml::name::QName) -> bool {
name.as_ref().starts_with(b"gml:")
}
/// Read an attribute value by its local name (namespace-prefix-insensitive).
fn attr_local<'a>(e: &'a quick_xml::events::BytesStart, want: &str) -> Option<String> {
for a in e.attributes().flatten() {
let key = a.key;
let local = key.local_name();
if local.as_ref() == want.as_bytes() {
return Some(String::from_utf8_lossy(a.value.as_ref()).into_owned());
}
}
None
}
/// Chunk `pos`/`posList` numbers (whitespace-separated) into `"x y"` pairs (2-D).
fn pairs_from_poslist(text: &str) -> Vec<String> {
let toks: Vec<&str> = text.split_whitespace().collect();
toks.chunks(2)
.filter(|c| c.len() == 2)
.map(|c| format!("{} {}", c[0], c[1]))
.collect()
}
/// Chunk deprecated `gml:coordinates` (`"x,y x,y"`) into `"x y"` pairs.
fn pairs_from_coordinates(text: &str) -> Vec<String> {
text.split_whitespace()
.filter_map(|tuple| {
let mut it = tuple.split(',');
let x = it.next()?;
let y = it.next()?;
Some(format!("{x} {y}"))
})
.collect()
}
/// Build the WKT string for a geometry kind from accumulated `"x y"` pairs.
fn wkt(kind: GeomKind, pairs: &[String]) -> Option<String> {
if pairs.is_empty() {
return None;
}
Some(match kind {
GeomKind::Point => format!("POINT ({})", pairs[0]),
GeomKind::LineString => format!("LINESTRING ({})", pairs.join(", ")),
GeomKind::Polygon => format!("POLYGON (({}))", pairs.join(", ")),
})
}
/// A stack frame per open element — tracks whether it turned out to be a
/// non-geometry LEAF (text, no child element) so it can become a property.
struct Frame {
local: String,
is_gml: bool,
text: String,
had_child: bool,
}
/// Parse a GML `FeatureCollection` (WFS `GetFeature` response) → feature rows.
pub fn parse_wfs_gml(xml: &str) -> Result<Vec<WfsFeature>> {
let mut reader = quick_xml::Reader::from_str(xml);
let mut features = Vec::new();
let mut stack: Vec<Frame> = Vec::new();
let mut in_member = false;
// Depth (stack len) at which the feature element sits; None until we're in one.
let mut feature_depth: Option<usize> = None;
// Per-feature accumulators.
let mut feature_id: Option<String> = None;
let mut feature_type = String::new();
let mut srs_name: Option<String> = None;
let mut geom_kind_cur: Option<GeomKind> = None;
let mut geom_pairs: Vec<String> = Vec::new();
let mut props: std::collections::BTreeMap<String, String> = std::collections::BTreeMap::new();
let reset = |feature_id: &mut Option<String>,
feature_type: &mut String,
srs_name: &mut Option<String>,
geom_kind_cur: &mut Option<GeomKind>,
geom_pairs: &mut Vec<String>,
props: &mut std::collections::BTreeMap<String, String>| {
*feature_id = None;
feature_type.clear();
*srs_name = None;
*geom_kind_cur = None;
geom_pairs.clear();
props.clear();
};
loop {
match reader.read_event()? {
Event::Start(e) => {
let local = local_name(e.name());
let is_gml = is_gml_prefixed(e.name());
if let Some(parent) = stack.last_mut() {
parent.had_child = true;
}
if is_member(&local) {
in_member = true;
feature_depth = None;
reset(
&mut feature_id,
&mut feature_type,
&mut srs_name,
&mut geom_kind_cur,
&mut geom_pairs,
&mut props,
);
} else if in_member && feature_depth.is_none() && !is_gml {
// First non-gml element inside a member wrapper = the feature.
feature_type = local.clone();
feature_id = attr_local(&e, "id").or_else(|| attr_local(&e, "fid"));
feature_depth = Some(stack.len());
} else if let Some(k) = geom_kind(&local) {
// Entering a geometry container — remember its kind + srsName.
if geom_kind_cur.is_none() {
geom_kind_cur = Some(k);
}
if let Some(s) = attr_local(&e, "srsName") {
srs_name = Some(s);
}
}
stack.push(Frame {
local,
is_gml,
text: String::new(),
had_child: false,
});
}
Event::Text(t) => {
if let Some(f) = stack.last_mut() {
f.text.push_str(&t.unescape()?);
}
}
Event::End(_) => {
let Some(frame) = stack.pop() else { continue };
let txt = frame.text.trim();
if is_coord_elem(&frame.local) && geom_kind_cur.is_some() {
let pairs = match frame.local.as_str() {
"coordinates" => pairs_from_coordinates(txt),
_ => pairs_from_poslist(txt), // pos / posList
};
geom_pairs.extend(pairs);
} else if feature_depth.is_some()
&& !frame.is_gml
&& !frame.had_child
&& !txt.is_empty()
&& stack.len() > feature_depth.unwrap()
{
// A non-gml leaf strictly INSIDE the feature = a property.
props.insert(frame.local.clone(), txt.to_string());
}
// The feature element itself just closed → finalise a row.
if let Some(fd) = feature_depth {
if stack.len() == fd && frame.local == feature_type {
let geometry_wkt = geom_kind_cur.and_then(|k| wkt(k, &geom_pairs));
let properties_json =
serde_json::to_string(&props).unwrap_or_else(|_| "{}".to_string());
features.push(WfsFeature {
feature_id: feature_id.clone(),
feature_type: feature_type.clone(),
srs_name: srs_name.clone(),
geometry_wkt,
properties_json,
});
feature_depth = None;
}
}
if is_member(&frame.local) {
in_member = false;
}
}
Event::Eof => break,
_ => {}
}
}
Ok(features)
}
/// WFS features → one Arrow `RecordBatch`.
pub fn features_batch(features: &[WfsFeature]) -> Result<RecordBatch> {
let cols = vec![
str_col(
&features
.iter()
.map(|f| f.feature_id.clone())
.collect::<Vec<_>>(),
),
str_col(
&features
.iter()
.map(|f| Some(f.feature_type.clone()))
.collect::<Vec<_>>(),
),
str_col(
&features
.iter()
.map(|f| f.srs_name.clone())
.collect::<Vec<_>>(),
),
str_col(
&features
.iter()
.map(|f| f.geometry_wkt.clone())
.collect::<Vec<_>>(),
),
str_col(
&features
.iter()
.map(|f| Some(f.properties_json.clone()))
.collect::<Vec<_>>(),
),
];
batch(&wfs_schema(), cols)
}
pub fn wfs_schema() -> Schema {
schema(&[
("feature_id", DataType::Utf8, true),
("feature_type", DataType::Utf8, false),
("srs_name", DataType::Utf8, true),
("geometry", DataType::Utf8, true),
("properties", DataType::Utf8, false),
])
}
/// The OGC WFS collector. `cfg.base_url` is the full `GetFeature` request URL
/// (service/version/typeNames/outputFormat live in the query string); the GML
/// response is streamed → feature rows.
#[derive(Default)]
pub struct WfsCollector;
#[async_trait::async_trait]
impl Collector for WfsCollector {
fn meta(&self) -> CollectorMeta {
CollectorMeta {
name: "wfs".into(),
source_kind: SourceKind::Ogc,
path: CollectorPath::RustNative,
direction: Direction::Source,
formats: vec!["gml".into(), "xml".into()],
license: "per-source (typically CC-BY / open geodata)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None,
join_keys: vec!["feature_id".into(), "coordinate".into()],
description: "OGC WFS GetFeature GML FeatureCollection → feature table \
(WKT geometry + properties JSON), pure-Rust GML parse."
.into(),
}
}
/// **The authority is whoever runs the endpoint you configure — and WFS is the one
/// generic reader where that authority is machine-discoverable.**
///
/// A WFS server's `GetCapabilities` response carries an `ows:ServiceProvider` /
/// `ServiceContact` block: the operating body's name and contact details, published
/// by the standard itself. So the honest card here is not a shrug — it is a
/// pointer to the ONE request that answers "who runs this and what will it let me
/// have" for the endpoint actually configured. `docs_url` is the WFS 2.0.2
/// implementation standard because that is the document that tells an operator how
/// to build the `GetFeature` URL this collector wants in `base_url`.
fn human_doc(&self) -> Option<crate::human_doc::HumanDoc> {
use crate::human_doc::{Contact, HumanDoc, RateLimit, Registration};
Some(
HumanDoc::new(
"whoever operates the WFS endpoint you configure — this is a generic \
reader. OGC owns the STANDARD, never the features. The Swedish services \
this parser was written against are Lantmäteriet, SGU, Naturvårdsverket \
and SMHI geodata.",
"the agency whose features that endpoint serves",
"https://www.ogc.org/standard/wfs/",
"https://docs.ogc.org/is/09-025r2/09-025r2.html",
)
.with_contact(Contact::Unknown)
.with_registration(Registration::NotNeeded)
.with_rate_limit(RateLimit::Documented(
"set by the endpoint's operator, not by the standard. A WFS 2.0 server \
publishes its own page size as the `CountDefault` constraint in \
`GetCapabilities`, so ask the server before assuming one request returns \
the whole layer."
.into(),
))
.with_quirk(
"ASK THE ENDPOINT FIRST: a `GetCapabilities` request returns an \
`ows:ServiceProvider`/`ServiceContact` block naming the operating body \
and how to reach it, plus the `CountDefault` page size and the feature \
types available. That is the contact and the terms for the feed you \
configured — korp cannot know them, and `terms_url` here points at the \
STANDARD, not at anyone's data licence.",
)
.with_quirk(
"The declared source `https://example.wfs/geoserver/ows?…` is an RFC 2606 \
documentation placeholder, not a service: until a real `GetFeature` URL \
is configured the collector reports `Missing::PlaceholderSource` rather \
than running.",
)
.with_quirk(
"`base_url` must be the COMPLETE `GetFeature` request — \
`service`/`version`/`typeNames`/`outputFormat` (and any `bbox` or \
`count`) live in the query string. This collector adds no parameters of \
its own, so an endpoint URL with no `typeNames` returns an empty \
collection and lands as the reason `WFS GetFeature returned no members`.",
)
.with_quirk(
"Coordinates are passed through in SOURCE ORDER with the declared \
`srsName` kept in its own column — nothing is reprojected and no axis is \
swapped. `urn:ogc:def:crs:EPSG::4326` is lat/lon while \
`EPSG:4326` in the older form is commonly served lon/lat, so downstream \
owns the axis interpretation and must read `srs_name` to do it.",
)
.with_quirk(
"`AuthKind::None` is the default, not a promise: several national geodata \
WFS services require an account. The collector resolves `cfg.auth` before \
the GET so a keyed endpoint works — the card simply cannot name a key for \
an endpoint it has not seen.",
),
)
}
fn source_target(&self) -> SourceTarget {
// Feature geometry lands in a warehouse table a facett `osm3d` scene can
// render as a static feature layer (points/lines/polygons).
SourceTarget::map_scene(
SourceEndpoint::Wfs,
"https://example.wfs/geoserver/ows?service=WFS&version=2.0.0&request=GetFeature",
"wfs_features",
"osm3d",
)
}
/// **One row per WFS feature.** `feature_id` is the `gml:id` the service assigns;
/// the derived key also named `coordinate`, which the table does not carry (its
/// geometry is a `geometry` blob).
fn business_key(&self) -> crate::history::BusinessKey {
crate::history::BusinessKey::of(
&["feature_id"],
"one row per feature — `feature_id` is the service's own `gml:id`",
)
}
fn schema(&self, _cfg: &CollectorCfg) -> Result<Schema> {
Ok(wfs_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?;
let xml = http.get_text(&cfg.base_url, auth.as_ref()).await?;
let features = parse_wfs_gml(&xml)?;
if features.is_empty() {
return Ok(CollectStats {
reason: Some("WFS GetFeature returned no members (empty typeNames/bbox?)".into()),
..Default::default()
});
}
let n = features.len() as u64;
sink.push(features_batch(&features)?)?;
Ok(CollectStats {
rows: n,
batches: 1,
table: cfg.table.clone(),
reason: None,
})
}
}
#[cfg(test)]
mod tests {
use super::*;
/// A WFS 2.0 `wfs:member` collection with a Point feature (gml:pos) and a
/// LineString feature (gml:posList) maps to two rows: geometry → WKT, other
/// leaves → the properties JSON, srsName captured. Offline DATA contract.
#[test]
fn wfs_gml_members_map_to_feature_rows_with_wkt_and_properties() {
let xml = r#"<?xml version="1.0" encoding="UTF-8"?>
<wfs:FeatureCollection xmlns:wfs="http://www.opengis.net/wfs/2.0"
xmlns:gml="http://www.opengis.net/gml/3.2"
xmlns:ad="http://example.se/ad">
<wfs:member>
<ad:AddressPoint gml:id="AP.1">
<ad:name>Sergels Torg</ad:name>
<ad:population>1234</ad:population>
<ad:geometry>
<gml:Point srsName="urn:ogc:def:crs:EPSG::4326"><gml:pos>59.3326 18.0649</gml:pos></gml:Point>
</ad:geometry>
</ad:AddressPoint>
</wfs:member>
<wfs:member>
<ad:Road gml:id="RD.7">
<ad:name>E4</ad:name>
<ad:geometry>
<gml:LineString srsName="urn:ogc:def:crs:EPSG::4326">
<gml:posList>59.10 18.00 59.20 18.05 59.30 18.10</gml:posList>
</gml:LineString>
</ad:geometry>
</ad:Road>
</wfs:member>
</wfs:FeatureCollection>"#;
let feats = parse_wfs_gml(xml).expect("parse WFS GML");
assert_eq!(feats.len(), 2, "two members → two features");
// Feature 1 — Point.
let p = &feats[0];
assert_eq!(p.feature_id.as_deref(), Some("AP.1"));
assert_eq!(p.feature_type, "AddressPoint");
assert_eq!(p.srs_name.as_deref(), Some("urn:ogc:def:crs:EPSG::4326"));
assert_eq!(p.geometry_wkt.as_deref(), Some("POINT (59.3326 18.0649)"));
assert!(
p.properties_json.contains("\"name\":\"Sergels Torg\""),
"props: {}",
p.properties_json
);
assert!(
p.properties_json.contains("\"population\":\"1234\""),
"props: {}",
p.properties_json
);
// The geometry did NOT leak into the properties JSON.
assert!(
!p.properties_json.contains("geometry"),
"geometry excluded from props: {}",
p.properties_json
);
assert!(
!p.properties_json.contains("pos"),
"gml leaves excluded: {}",
p.properties_json
);
// Feature 2 — LineString.
let l = &feats[1];
assert_eq!(l.feature_id.as_deref(), Some("RD.7"));
assert_eq!(l.feature_type, "Road");
assert_eq!(
l.geometry_wkt.as_deref(),
Some("LINESTRING (59.10 18.00, 59.20 18.05, 59.30 18.10)")
);
assert!(l.properties_json.contains("\"name\":\"E4\""));
// Arrow round-trip.
let rb = features_batch(&feats).expect("features batch");
assert_eq!(rb.num_rows(), 2);
let rb_schema = rb.schema();
let cols: Vec<&str> = rb_schema
.fields()
.iter()
.map(|f| f.name().as_str())
.collect();
for want in [
"feature_id",
"feature_type",
"srs_name",
"geometry",
"properties",
] {
assert!(
cols.contains(&want),
"feature table carries `{want}`: {cols:?}"
);
}
let geom = rb
.column_by_name("geometry")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::StringArray>())
.expect("geometry is utf8");
assert!(geom.value(0).starts_with("POINT"));
assert!(geom.value(1).starts_with("LINESTRING"));
crate::functional_status(
"korp-collectors/wfs",
"gml_features_parse",
true,
&format!(
"{} WFS features → WKT geometry + properties JSON",
rb.num_rows()
),
);
}
/// The deprecated `gml:coordinates` form and a WFS 1.x `gml:featureMember`
/// wrapper are both handled; a feature with no geometry still yields a row
/// (geometry null, properties present).
#[test]
fn wfs_gml_coordinates_form_and_featuremember_and_no_geometry() {
let xml = r#"<gml:FeatureCollection xmlns:gml="http://www.opengis.net/gml"
xmlns:t="http://example.se/t">
<gml:featureMember>
<t:Site gml:id="S.1">
<t:label>Old form</t:label>
<t:geom><gml:Point srsName="EPSG:4326"><gml:coordinates>18.06,59.33</gml:coordinates></gml:Point></t:geom>
</t:Site>
</gml:featureMember>
<gml:featureMember>
<t:Note gml:id="N.1">
<t:text>no geometry here</t:text>
</t:Note>
</gml:featureMember>
</gml:FeatureCollection>"#;
let feats = parse_wfs_gml(xml).expect("parse WFS 1.x GML");
assert_eq!(feats.len(), 2, "two featureMembers → two features");
assert_eq!(
feats[0].geometry_wkt.as_deref(),
Some("POINT (18.06 59.33)"),
"coordinates x,y → x y"
);
assert_eq!(feats[0].srs_name.as_deref(), Some("EPSG:4326"));
assert!(feats[0].properties_json.contains("\"label\":\"Old form\""));
// The geometry-less feature: a row with null geometry, its text a property.
assert_eq!(feats[1].feature_type, "Note");
assert!(
feats[1].geometry_wkt.is_none(),
"no gml geometry → null WKT"
);
assert!(
feats[1]
.properties_json
.contains("\"text\":\"no geometry here\"")
);
crate::functional_status(
"korp-collectors/wfs",
"gml_coordinates_and_featuremember",
true,
"coordinates x,y form + WFS 1.x featureMember + geometry-less feature",
);
}
/// An empty FeatureCollection yields zero features (the collector turns that
/// into a RAGNARÖK reason) — the RED-when-broken empty guard.
#[test]
fn wfs_empty_collection_yields_no_features() {
let xml = r#"<wfs:FeatureCollection xmlns:wfs="http://www.opengis.net/wfs/2.0" numberReturned="0"/>"#;
assert!(parse_wfs_gml(xml).expect("parse").is_empty());
}
}