nordisk/korp-collectors

korp-collectors

The standalone LEAF of korp's open-data collector layer. One object-safe Collector trait + per-access-METHOD import collectors (Swedish + EU open data), parsing pure-Rust → Apache Arrow → (optionally) Apache Iceberg via skade. Plus a RegionSource seam that FETCHES OSM extracts on demand.

This crate depends on neither korp nor facett, so both can consume it — which breaks the facett ⇄ korp dependency cycle. (Mirrors how znippy-zoomies/osm-katana and Skidbladnir were extracted as leaves.) The ravens Hugin & Munin collect — hence "collectors" (née connectors).

Collectors (all Direction::Source)

Everything parses in pure Rust → Arrow RecordBatch. The memory BatchSink buffers batches for assertion; the skade-sink feature adds the Iceberg backend

• the path-agnostic run() runner.

The RegionSource seam (feature osm)

pub trait RegionSource: Send {
    /// Ensure the OSM source data for a named region is available locally,
    /// FETCHING it if missing. Returns the local .osm.pbf path.
    fn ensure(&self, region: &str) -> Result<std::path::PathBuf, String>;
    fn name(&self) -> &str;
}

• CachedLocalSource { dir } — resolves an already-present {region}-latest.osm.pbf (no fetch).
• HttpOsmSource { base_url, cache_dir } — the REAL connector: downloads {base_url}/{region}-latest.osm.pbf (Geofabrik-style) into cache_dir, skipping the download when already cached.

This is what lets a facett region picker be "fed by a connector fetching data" without facett depending on korp.

Features

• parsers (default) — trait/registry surface + shared HTTP core + all light pure-Rust parsers + the pure OSM admin tag-filter/trie. No heavy geo dep.
• osm — the OSM geometry collectors (via osm-katana) + the RegionSource seam.
• csv — the RFC-4180 (+ dialects) CSV collector (csv 1.4 + encoding_rs).
• mainframe — the EBCDIC + COBOL-copybook + COMP-3 fixed-width collector and its ASCII fixed-width cousin (fully hand-rolled, no new deps).
• gribnc — the NetCDF/GRIB weather-model collector (Copernicus CDS + DWD ICON): hand-rolled pure-Rust GRIB2 + NetCDF-classic decoders (no new deps) → a variable/time/bbox metadata catalog and the data-array unpack. GRIB2 Section 5/6/7 decodes simple (5.0), complex (5.2) and complex-with-spatial-differencing (5.3) packing with Section-6 bitmaps (a 0-bit is a missing point, never a fake 0.0); NetCDF-classic reads variable data arrays (fixed + record dims, numeric types). Only NetCDF-4/HDF5 stays deferred behind a would-be optional, non-default native netcdf (libnetcdf C) feature.
• skade-sink — BatchSink::iceberg + the run() Iceberg runner (skade 0.5).
• znippy — the collected-data converter (znippy_pack): pack parsed Arrow batches (or raw byte blobs) into a znippy archive — compact, blake3-checksummed, OpenZL-compressed storage/transport — and read them back byte-identical. Reuses the znippy-common pack format (no bespoke compression) + arrow-ipc. pack_tables/unpack_tables for named RecordBatch streams; pack_blobs/ unpack_blob for raw payloads.
• daemon — the collector daemon architecture (daemon.rs + iceberg_daemons.rs): the object-safe Config + State + Collector trio, the CollectorCtx seams (WarehouseSink, the GET/POST Fetcher, the SecretStore (API keys via an injectable seam — env default, vault later), the jera phase handle, the BIOS bus) and the Offline → Downloading → Indexing → Online state machine (empty-is-RED → Failed). Each registered daemon self-registers its live phase + facet into collector-register.redb (register.rs), so a RegisterReader in ANOTHER process — korp's Collectors panel, or korp-collectors status — sees it. daemon_registry() lists 21 warehouse-sink daemons under this feature (+ gtfs-stops under gtfs, + osm-cut under osm-daemon), among them the Swedish real-time trio polisen-events and smhi-weather (GET JSON) and trafikverket-trainpos (a POST XML <REQUEST> with its key in the body). Pure parsers — no korp, no facett; a daemon runs standalone (produces Arrow to the warehouse with no korp server) and, when korp is present, is visible through the shared register.
• service / systemd — run a daemon (or a whole nursery of them) as its own supervised process / systemd unit (collector-<name>.service), via Skidbladnir; the korp-collectors binary is the runner (serve / status / run / install).
• testmatrix — the tests' functional_status self-reports become real nornir test-matrix rows. Compiled OUT of the default build (no nornir dep otherwise).

Test

cargo test                                 # parsers (offline inject/assert)
cargo test --features osm                  # + OSM ways + RegionSource fetch/cache/parse
cargo test --features znippy               # + znippy converter byte-identical round-trip
cargo test --all-features                  # + skade Iceberg round-trip + testmatrix

The RegionSource test spins an in-process HTTP server serving a bundled 454-byte .osm.pbf fixture, proves fetch → cache-hit (a request counter) → osm-katana parses ≥1 real way. Deterministic, no external network.

License

MIT OR Apache-2.0.