# 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`)

| name | kind | parses | feature |
|------|------|--------|---------|
| `dcat-ap-se` | DCAT-AP-SE crawl | RDF/XML + SPARQL-JSON → dataset/distribution registry | `parsers` |
| `pxweb-jsonstat2` | PxWeb / JSON-stat2 | statistical cube → code-col-per-dim + `value` | `parsers` |
| `eurostat-sdmx`, `oecd-sdmx`, `ecb-sdmx`, `insee-sdmx`, `ons-sdmx`, `istat-sdmx`, `worldbank-sdmx` | SDMX-JSON 2.1 / 1.0 | **generalised** series/observation dim split → per-`(dims × time × measure)` value rows — one collector, any SDMX provider (Eurostat/OECD/ECB/IMF/**INSEE** France/**ONS** UK/**ISTAT** Italy/**World Bank**/EEA), zero new parsing — presets differ only in branding + endpoint | `parsers` |
| `entsoe`, `entsoe-generation`, `entsoe_day_ahead`, `nordpool`, `energy-charts-power` | ENERGY market | ENTSO-E market-document XML → time-series points (SE1–SE4 ⇄ EIC): `entsoe` load/generic `quantity`, `entsoe-generation` per-`psrType` volume (A75), **`entsoe_day_ahead`** day-ahead SPOT PRICES (A44, `price.amount` per zone × hour → `entsoe_day_ahead_prices`; the open free-token price route, token via `ENTSOE_TOKEN`); Nord Pool day-ahead zones (licensed twin), Fraunhofer generation-mix | `parsers` |
| `opensky-flights`, `opensky-europe-bbox`, `opensky-tracks` | OPEN FLIGHT recorder (ADS-B) | OpenSky `states/all` positional-array → flight position table → `osm3d-flight` scene (**live auth: OAuth2 client-credentials**, 2025 Keycloak, via `OPENSKY_CLIENT_ID`/`OPENSKY_CLIENT_SECRET`). **`opensky-tracks`** = the recorded flight PATH: `/tracks` `path[]` → ordered `(time, lat, lon, baro_altitude, heading, on_ground)` waypoint trajectory → `opensky_tracks` table → `osm3d-flight` replay (basic auth via `OPENSKY_USER`/`OPENSKY_PASS`) | `parsers` |
| `smhi-hydroobs` | WATER | SMHI hydrological observations (river discharge / water level) — station `value` series → Arrow | `parsers` |
| `metno-forecast` | WEATHER | Met.no Locationforecast point forecast → per-time forecast table (temp, wind, humidity, pressure, cloud) → `osm3d` scene | `parsers` |
| `gbif-occurrences` | biodiversity | GBIF occurrence search (keyless, CC0/CC-BY) → occurrence point table (taxon + WGS84 fix + country) → `osm3d-poi` scene | `parsers` |
| `worldbank-indicators`, `restcountries` | reference / stats | World Bank Open Data (`[meta,data]` envelope) → per-`(country, year, indicator)`; RESTCountries → ISO-3166 country join spine | `parsers` |
| `mobility-database`, `mobility-database-rt` | transit META-catalogue | Mobility Database (CC0, the **TransitFeeds successor**) → typed feed-descriptor rows that **auto-drive** the GTFS collectors; bearer-token via `MOBILITY_DB_TOKEN`. `mobility-database` enumerates STATIC GTFS feeds (`/v1/gtfs_feeds`); `mobility-database-rt` enumerates **GTFS-Realtime** feeds (`/v1/gtfs_rt_feeds`, id, `entity_type` vp/tu/sa, provider, country, `.pb` URL) that drive the `gtfs-rt-vehicles` protobuf reader for live vehicle positions | `parsers` |
| `erddap-emodnet-physics`, `erddap-noaa-coastwatch` | ERDDAP scientific data | ERDDAP `.json` `table` (columnNames + columnTypes + rows matrix) → typed Arrow, uniform over any dataset id on any ERDDAP server (EMODnet ocean + global ocean/atmos); `.csv` variant via the `csv` collector | `parsers` |
| `eea-wise-water`, `destatis-genesis`, `cbs-odata` | EU open data | EEA WISE water quality (Waterbase via the Discodata SQL API, `/results`) → per-`(site, determinand, year)` rows; Destatis GENESIS-Online table catalogue (`/List`, token via `DESTATIS_GENESIS_TOKEN`); CBS Open Data (Statistics Netherlands) OData TypedDataSet (`/value`) → typed observations | `parsers` |
| `polisen-events`, `kolada-kpi`, `smhi-metobs`, `smhi-weather`, `ais-positions`, `marinetraffic-ais`, `gbfs-citybikes`, `osm-overpass`, `bolagsverket-organisationer` | generic REST/JSON | config-driven JSON field-map → Arrow | `parsers` |
| `usgs-earthquakes` | USGS GeoJSON feed | `FeatureCollection` Point features → quake rows (id, place, magnitude, depth, lon/lat, epoch-ms time) → `osm3d` scene | `parsers` |
| `rss-atom` | RSS/Atom feed | RSS 2.0 / Atom 1.0 syndication (dialect-agnostic `quick-xml`) → per-item rows (feed_title, id, title, link, summary, author, published, categories JSON) | `parsers` |
| `openaq-air-quality` | OpenAQ REST | air-quality `results[]` JSON → measurement rows (station, parameter, value, unit, lon/lat, country, city, UTC time; nested `coordinates`+`date.utc` flatten) → `osm3d` scene | `parsers` |
| `energy-charts-power` | Energy-Charts REST | generation-mix JSON — parallel `unix_seconds[]` axis ZIPPED against each `production_types[].data[]` by index → per-`(time, type)` power rows (Europe-wide via `?country=`; CC-BY) | `parsers` |
| `wfs` | OGC WFS | `GetFeature` GML `FeatureCollection` → feature rows (WKT geometry + properties JSON) → `osm3d` scene | `parsers` |
| `gtfs-rt-vehicles` | GTFS-Realtime feed | `VehiclePositions` **protobuf** `FeedMessage` → live transit vehicle position table (entity/vehicle/route/trip id, lat/lon, bearing, speed, timestamp, stop) via a hand-rolled protobuf reader (no `prost` dep) → `osm3d-flight` scene | `parsers` |
| `trafikverket-trainpos` | Trafikverket | POST XML `<REQUEST>` → JSON position table | `parsers` |
| `osm-geoparquet`, `osm-admin-boundaries` | OSM bulk file | PBF → WKB ways / admin MultiPolygons (via `osm-katana`) | `osm` |
| `csv` | flat file | RFC 4180 (+ `;`/TAB, BOM, Latin-1, decimal-comma) → typed Arrow via a two-pass schema sniff | `csv` |
| `mainframe-copybook` | flat file | IBM EBCDIC fixed-width decoded against a COBOL copybook (PIC X/9/S9V, COMP-3 packed, zoned, COMP binary) → Arrow (COMP-3 → `Decimal128`) | `mainframe` |
| `fixed-width` | flat file | ASCII/Latin-1 fixed-width decoded against a COBOL copybook layout (DISPLAY usage) | `mainframe` |
| `copernicus-cds`, `dwd-icon` | NetCDF/GRIB weather model | hand-rolled pure-Rust GRIB2 + NetCDF-classic decode → a variable/time/bbox **metadata catalog** (discipline/param, reference time, Ni/Nj, lat/lon bbox, num_points) **AND the data-ARRAY unpack**: GRIB2 Section 5/6/7 simple (5.0) + complex (5.2) + complex-with-spatial-differencing (5.3) packing with Section-6 bitmaps, NetCDF-classic variable arrays (fixed + record dims). `params.mode="values"` emits per-point grid values. Copernicus CDS (`retrieve/v1`, key from `CDSAPI_KEY`) + DWD ICON. Only NetCDF-4/HDF5 stays deferred | `gribnc` |
| `copernicus-era5`, `copernicus-cams` | Copernicus reanalysis (value-first) | ERA5 climate reanalysis (CDS `retrieve/v1`, key from `CDSAPI_KEY`) + CAMS atmosphere/air-quality (ADS `retrieve/v1`, key from `ADSAPI_KEY`) → **per-point grid VALUE rows** by default (reuses the `gribnc` GRIB2 Section 5/6/7 + NetCDF value unpack, `mode="values"`) | `gribnc` |

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`)

```rust
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.