refactor: govdata

Author: TKaltofen

README.md

@@ -10,26 +10,50 @@ Connectors for German open government data, built on [mloda](https://github.com/
 
 Three example datasets cover the M1 themes: population (GovData CSV), elections (Bundeswahlleiterin `kerg.csv`), and environment (UBA Air Data JSON).
 
+## Status
+
+Young but working. All three example readers run end to end, with paginated dataset search, cached downloads with retries, and unit plus property-based tests behind them. Every reader is a thin subclass of `BaseGovDataReader` that overrides only the parse step; new datasets follow the same path (see [docs/adding-a-reader.md](docs/adding-a-reader.md)). Development happens in a 6-month Prototype Fund stage (June to November 2026), so the API may still shift between releases.
+
 ## Usage
 
 Read the Stuttgart population dataset (via GovData) as a typed PyArrow table:
 
 ```python
 from mloda.user import Feature, mloda
-from mloda_plugin_govdata.feature_groups.govdata import GovDataReader
+from mloda_plugin_govdata.feature_groups.govdata import StuttgartPopulationReader
 
 slug = "einwohner-nach-altersgruppen-und-stadtbezirken"
 result = mloda.run_all(
     [
-        Feature("Einwohner", options={GovDataReader.__name__: slug}),
-        Feature("Stadtbezirk", options={GovDataReader.__name__: slug}),
+        Feature("Einwohner", options={StuttgartPopulationReader.__name__: slug}),
+        Feature("Stadtbezirk", options={StuttgartPopulationReader.__name__: slug}),
     ],
     compute_frameworks=["PyArrowTable"],
 )
 table = result[0]  # pyarrow.Table with the requested columns
 ```
 
-The option value is a GovData dataset slug or a direct distribution URL. The license is read from the CKAN distribution metadata. Set `GovDataReader.cache_dir` to control where downloads are cached.
+The option value is a GovData dataset slug or a direct distribution URL. The license is read from the CKAN distribution metadata. Set `BaseGovDataReader.cache_dir` to control where downloads are cached. For any other GovData CSV dataset, `GovDataReader` works out of the box and reads every column as a string; subclass it and set `schema` for typed columns.
+
+Don't know the slug yet? Search GovData with the paginated CKAN `package_search` API:
+
+```python
+from mloda_plugin_govdata.feature_groups.govdata import build_client, search_datasets
+
+with build_client() as client:
+    for dataset in search_datasets(client, "einwohner stuttgart", max_results=10):
+        print(dataset.name, "|", dataset.title)
+```
+
+`search_datasets` walks the result pages lazily (`page_size` per request) and stops at `max_results` or the end of the result set.
+
+Got a slug but not the column names? `peek` lists what you can request as features:
+
+```python
+StuttgartPopulationReader.peek(slug)  # {"Stichtag": "date32[day]", "Stadtbezirk": "string", ...}
+```
+
+It works on every reader (`BundeswahlleiterinReader.peek(kerg)`, `UbaAirReader.peek(url)`) and downloads through the cache, so the actual feature request reuses the file. A typo in a feature name fails with the available columns and a close-match suggestion instead of a raw KeyError.
 
 The elections reader handles a direct CSV URL whose file has a multi-row merged header (Bundeswahlleiterin `kerg.csv`):
 
@@ -60,11 +84,24 @@ result = mloda.run_all(
 
 Columns are `station_id`, `date_start`, `component_id`, `scope_id`, `value`, `date_end`, and `index` (the air-quality index). Component and scope ids come from the UBA `components` and `scopes` endpoints.
 
+## Demo
+
+An interactive [marimo](https://marimo.io) notebook walks through dataset discovery and all three example datasets. The notebook lives in the repository (not in the published package), so run it from a source checkout:
+
+```bash
+git clone https://github.com/TomKaltofen/mloda-plugin-govdata.git
+cd mloda-plugin-govdata
+uv sync --all-extras
+uv run marimo edit demos/govdata_demo.py
+```
+
+The notebook hits the live GovData, Bundeswahlleiterin, and UBA endpoints; downloads are cached locally after the first run.
+
 ## Related Repositories
 
-- **[mloda](https://github.com/mloda-ai/mloda)**: The core library for open data access. Declaratively define what data you need, not how to get it. mloda handles feature resolution, dependency management, and compute framework abstraction automatically.
+- **[mloda](https://github.com/mloda-ai/mloda)**: the core library this plugin builds on. You declare which features you need; mloda resolves how to compute them.
 
-- **[mloda-registry](https://github.com/mloda-ai/mloda-registry)**: The central hub for discovering and sharing mloda plugins. Browse community-contributed FeatureGroups, find integration guides, and publish your own plugins for others to use.
+- **[mloda-registry](https://github.com/mloda-ai/mloda-registry)**: plugin registry and development guides for the mloda ecosystem.
 
 ## Funding
 

demos/govdata_demo.py

@@ -0,0 +1,149 @@
+"""Marimo demo: discover GovData datasets and read the three M1 example datasets.
+
+Run with: marimo edit demos/govdata_demo.py (needs network access; install the
+"demo" extra for marimo itself).
+"""
+
+import marimo
+
+__generated_with = "0.14.16"
+app = marimo.App(width="medium")
+
+
+@app.cell
+def _():
+    import marimo as mo
+
+    return (mo,)
+
+
+@app.cell
+def _(mo):
+    mo.md(
+        """
+        # mloda-plugin-govdata demo
+
+        German open government data as mloda features: search GovData via the
+        paginated CKAN API, then read the three M1 example datasets (population,
+        elections, environment) as typed Arrow tables. Every cell below talks to
+        the live endpoints; downloads are cached locally after the first run.
+
+        Part of the Prototype Fund project mloda-plugin-govdata (FKZ 16IS26S11).
+        """
+    )
+    return
+
+
+@app.cell
+def _():
+    import pandas as pd
+
+    from mloda.user import Feature, mloda
+    from mloda_plugin_govdata.feature_groups.govdata import (
+        BundeswahlleiterinReader,
+        StuttgartPopulationReader,
+        UbaAirReader,
+        build_client,
+        search_datasets,
+        uba_measures_url,
+    )
+
+    return (
+        BundeswahlleiterinReader,
+        Feature,
+        StuttgartPopulationReader,
+        UbaAirReader,
+        build_client,
+        mloda,
+        pd,
+        search_datasets,
+        uba_measures_url,
+    )
+
+
+@app.cell
+def _(mo):
+    mo.md("""## 1. Discover datasets (paginated CKAN `package_search`)""")
+    return
+
+
+@app.cell
+def _(mo):
+    query = mo.ui.text(value="einwohner stuttgart altersgruppen", label="GovData search", full_width=True)
+    query
+    return (query,)
+
+
+@app.cell
+def _(build_client, pd, query, search_datasets):
+    with build_client() as _client:
+        _hits = list(search_datasets(_client, query.value, max_results=10))
+    hits = pd.DataFrame({"name": [d.name for d in _hits], "title": [d.title for d in _hits]})
+    hits
+    return
+
+
+@app.cell
+def _(mo):
+    mo.md("""## 2. Population: Stuttgart residents by age group (GovData CSV)""")
+    return
+
+
+@app.cell
+def _(Feature, StuttgartPopulationReader, mloda):
+    _slug = "einwohner-nach-altersgruppen-und-stadtbezirken"
+    _result = mloda.run_all(
+        [
+            Feature("Stichtag", options={StuttgartPopulationReader.__name__: _slug}),
+            Feature("Stadtbezirk", options={StuttgartPopulationReader.__name__: _slug}),
+            Feature("Alter in 10 Gruppen", options={StuttgartPopulationReader.__name__: _slug}),
+            Feature("Einwohner", options={StuttgartPopulationReader.__name__: _slug}),
+        ],
+        compute_frameworks=["PyArrowTable"],
+    )
+    population = _result[0].to_pandas()
+    population
+    return
+
+
+@app.cell
+def _(mo):
+    mo.md("""## 3. Elections: Bundestagswahl 2025 results (`kerg.csv`, merged header)""")
+    return
+
+
+@app.cell
+def _(BundeswahlleiterinReader, Feature, mloda):
+    _kerg = "https://www.bundeswahlleiterin.de/bundestagswahlen/2025/ergebnisse/opendata/btw25/csv/kerg.csv"
+    _result = mloda.run_all(
+        [Feature("Gebiet", options={BundeswahlleiterinReader.__name__: _kerg})],
+        compute_frameworks=["PyArrowTable"],
+    )
+    elections = _result[0].to_pandas()
+    elections.head(20)
+    return
+
+
+@app.cell
+def _(mo):
+    mo.md("""## 4. Environment: hourly ozone at one station (UBA Air Data JSON)""")
+    return
+
+
+@app.cell
+def _(Feature, UbaAirReader, mloda, uba_measures_url):
+    _url = uba_measures_url(station=143, component=3, scope=2, date_from="2025-01-01", date_to="2025-01-01")
+    _result = mloda.run_all(
+        [
+            Feature("date_start", options={UbaAirReader.__name__: _url}),
+            Feature("value", options={UbaAirReader.__name__: _url}),
+        ],
+        compute_frameworks=["PyArrowTable"],
+    )
+    environment = _result[0].to_pandas()
+    environment
+    return
+
+
+if __name__ == "__main__":
+    app.run()

docs/adding-a-reader.md

@@ -0,0 +1,12 @@
+# Adding a Reader
+
+`BaseGovDataReader` (in `mloda_plugin_govdata/feature_groups/govdata/reader.py`) does the plumbing: locator coercion, CKAN discovery, cached download with retries, and column selection. Subclasses implement `_parse` (or, for plain CSVs, just set `schema` on a `GovDataReader` subclass), plus `suffix` when the payload is not CSV. Each data source lives in its own module (`population.py`, `bundeswahlleiterin.py`, `uba.py`); shared source-agnostic code (client, cache, discovery, locator, CSV parsing) lives in `core/`.
+
+To connect a new dataset:
+
+1. Check whether `GovDataReader` already handles it. A GovData slug or a direct CSV URL with a regular single-row header needs no code; every column is read as a string. For typed columns, subclass `GovDataReader` in a new module and set `schema` (see `population.py`).
+2. For a different payload shape, subclass `BaseGovDataReader` and implement `_parse` (and `suffix` for non-CSV payloads); `bundeswahlleiterin.py` and `uba.py` show the pattern.
+3. Keep source-specific parse logic in the source's module as a `parse_*_bytes` function plus a path wrapper, like `uba.py`. Generic parsing belongs in `core/parse.py`. That keeps it testable from fixture files without network access.
+4. Add tests under `mloda_plugin_govdata/feature_groups/govdata/tests/` with a small real sample in `tests/fixtures/`.
+5. Export the reader from `feature_groups/govdata/__init__.py` and add a usage snippet to the README.
+6. Run `tox` (pytest, ruff, mypy strict, bandit); it must pass before a PR.

mloda_plugin_govdata/compute_frameworks/__init__.py

@@ -1,13 +1,27 @@
-"""GovData connector: read German open-government CSV distributions into mloda."""
+"""GovData connector: read German open-government data into mloda."""
 
-from .discovery import Dataset, ResolvedDistribution, Resource, normalize_license, resolve_distribution
-from .locator import GovDataLocator
-from .parse import ColumnType, parse_german_csv, parse_german_csv_bytes, parse_multi_header_csv
-from .reader import BundeswahlleiterinReader, GovDataFeature, GovDataReader, UbaAirReader
-from .uba import UBA_AIR_BASE, parse_uba_measures, parse_uba_measures_bytes, uba_measures_url
+from .bundeswahlleiterin import BundeswahlleiterinReader
+from .core.client import build_client
+from .core.discovery import (
+    Dataset,
+    ResolvedDistribution,
+    Resource,
+    normalize_license,
+    resolve_distribution,
+    search_datasets,
+)
+from .core.locator import GovDataLocator
+from .core.parse import ColumnType, parse_german_csv, parse_german_csv_bytes, parse_multi_header_csv
+from .feature import GovDataFeature
+from .population import POPULATION_SCHEMA, POPULATION_SLUG, StuttgartPopulationReader
+from .reader import BaseGovDataReader, GovDataReader
+from .uba import UBA_AIR_BASE, UbaAirReader, parse_uba_measures, parse_uba_measures_bytes, uba_measures_url
 
 __all__ = [
+    "POPULATION_SCHEMA",
+    "POPULATION_SLUG",
     "UBA_AIR_BASE",
+    "BaseGovDataReader",
     "BundeswahlleiterinReader",
     "ColumnType",
     "Dataset",
@@ -16,13 +30,16 @@
     "GovDataReader",
     "Resource",
     "ResolvedDistribution",
+    "StuttgartPopulationReader",
     "UbaAirReader",
+    "build_client",
     "normalize_license",
     "parse_german_csv",
     "parse_german_csv_bytes",
     "parse_multi_header_csv",
     "parse_uba_measures",
     "parse_uba_measures_bytes",
     "resolve_distribution",
+    "search_datasets",
     "uba_measures_url",
 ]

mloda_plugin_govdata/extenders/__init__.py

@@ -0,0 +1,20 @@
+"""Bundeswahlleiterin federal election results (M1 elections theme, kerg.csv)."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pyarrow as pa
+
+from .core.discovery import ResolvedDistribution
+from .core.locator import GovDataLocator
+from .core.parse import ColumnType, parse_multi_header_csv
+from .reader import BaseGovDataReader
+
+
+class BundeswahlleiterinReader(BaseGovDataReader):
+    """Reads the Bundeswahlleiterin kerg.csv result file (5-line preamble, 3-row merged header)."""
+
+    @classmethod
+    def _parse(cls, path: Path, locator: GovDataLocator, distribution: ResolvedDistribution) -> pa.Table:
+        return parse_multi_header_csv(path, skiprows=5, header_rows=3, label_columns=4, value_type=ColumnType.INTEGER)

mloda_plugin_govdata/feature_groups/govdata/__init__.py

@@ -0,0 +1 @@
+"""Source-agnostic plumbing: HTTP client, download cache, CKAN discovery, locator, CSV parsing."""