feat(wire): S4/#109 — §5 GAUL sidecar builder + delivery/parity invariant

tests/test_parity.py

@@ -0,0 +1,24 @@
+"""Gid-set parity invariant (delivery/parity.py) — §5.2, primitives only."""
+
+import pytest
+
+from views_postprocessing.delivery.parity import GidParityError, assert_gid_set_parity
+
+
+def test_identical_sets_pass():
+    assert assert_gid_set_parity([1, 2, 3], {3, 2, 1}) is None
+
+
+def test_sidecar_missing_a_forecast_cell_raises_naming_it():
+    with pytest.raises(GidParityError, match=r"missing.*\[3\]"):
+        assert_gid_set_parity([1, 2, 3], [1, 2])
+
+
+def test_extra_sidecar_cell_raises_naming_it():
+    with pytest.raises(GidParityError, match=r"extra.*\[9\]"):
+        assert_gid_set_parity([1, 2], [1, 2, 9])
+
+
+def test_label_appears_in_message():
+    with pytest.raises(GidParityError, match="ocha-sidecar"):
+        assert_gid_set_parity([1], [2], label="ocha-sidecar")

tests/test_wire_sidecar.py

@@ -0,0 +1,71 @@
+"""§5 sidecar builder (wire/sidecar.py) — byte parity with the fixture sidecar via
+an injected synthetic lookup, plus pinned-schema property tests against the real
+ADR-011 lookup. Toolchain pin as in test_wire_shard (fail loud, never skip)."""
+
+import hashlib
+from pathlib import Path
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+import pytest
+
+from views_postprocessing.unfao.gaul_schema import CODE_COLS, METADATA_COLS
+from views_postprocessing.unfao.wire import sidecar
+
+_FIX = Path(__file__).resolve().parent / "fixtures" / "wire_contract"
+_GIDS = [100001, 100002, 100003, 100004, 100005, 100006]
+
+
+def _synthetic_lookup() -> pa.Table:
+    """The fixture sidecar's content, shaped like the REAL lookup (priogrid_gid key,
+    int64 codes with nulls, dictionary-encoded strings) — proving the builder's
+    casts, not echoing its output."""
+    return pa.table(
+        {
+            "priogrid_gid": pa.array(_GIDS, pa.int64()),
+            "pg_xcoord": pa.array([10.25, 10.75, 11.25, 11.75, 12.25, 12.75], pa.float64()),
+            "pg_ycoord": pa.array([5.25, 5.25, 5.25, 5.75, 5.75, 5.75], pa.float64()),
+            "country_iso_a3": pa.array(["AAA", "AAA", "AAA", "BBB", "BBB", None]).dictionary_encode(),
+            "admin1_gaul1_code": pa.array([11, 11, 12, 21, 21, None], pa.int64()),
+            "admin1_gaul1_name": pa.array(["A-one", "A-one", "A-two", "B-one", "B-one", None]).dictionary_encode(),
+            "admin1_gaul0_code": pa.array([1, 1, 1, 2, 2, None], pa.int64()),
+            "admin1_gaul0_name": pa.array(["Aland", "Aland", "Aland", "Bland", "Bland", None]).dictionary_encode(),
+            "admin2_gaul2_code": pa.array([111, 112, 121, 211, 212, None], pa.int64()),
+            "admin2_gaul2_name": pa.array(["A-1-1", "A-1-2", "A-2-1", "B-1-1", "B-1-2", None]).dictionary_encode(),
+        }
+    )
+
+
+def test_byte_parity_with_the_fixture_sidecar(tmp_path):
+    name, sha = sidecar.write_sidecar(
+        _synthetic_lookup(), _GIDS, run_id="fixture_run_0", directory=tmp_path
+    )
+    canonical = (_FIX / "fixture_run_0__sidecar.parquet").read_bytes()
+    assert name == "fixture_run_0__sidecar.parquet"
+    assert (tmp_path / name).read_bytes() == canonical
+    assert sha == hashlib.sha256(canonical).hexdigest()
+
+
+def test_missing_forecast_gid_fails_loud():
+    with pytest.raises(sidecar.SidecarError, match="absent from the GAUL lookup"):
+        sidecar.build_sidecar(_synthetic_lookup(), [*_GIDS, 999999])
+
+
+def test_rows_restricted_to_declared_gids_and_ascending():
+    table = sidecar.build_sidecar(_synthetic_lookup(), [100003, 100001])
+    got = table.column("priogrid_id").to_pylist()
+    assert got == [100001, 100003]  # subset only, ascending
+
+
+def test_real_lookup_produces_the_pinned_schema():
+    lookup = pq.read_table(
+        Path(__file__).resolve().parents[1] / "views_postprocessing" / "data" / "gaul_lookup.parquet"
+    )
+    some_gids = lookup.column("priogrid_gid").to_pylist()[:50]
+    table = sidecar.build_sidecar(lookup, some_gids)
+    assert table.column_names == ["priogrid_id"] + METADATA_COLS
+    for col in CODE_COLS:
+        assert table.schema.field(col).type == pa.float64()  # ALWAYS float64 (§5.1)
+    for col in ("country_iso_a3", "admin1_gaul1_name"):
+        assert table.schema.field(col).type == pa.string()  # plain, not dictionary
+    assert table.num_rows == 50

views_postprocessing/delivery/parity.py

@@ -0,0 +1,33 @@
+"""Gid-set parity invariant: the sidecar covers exactly the forecast's cells
+(ADR-013 §5.2).
+
+Representation-free — primitives only (two iterables of ints). A sidecar missing
+a forecast cell would silently strip that cell's geography downstream; a sidecar
+carrying extra cells claims geography for forecasts that don't exist. Both fail
+loud here, naming the offenders.
+
+Separate from ``coverage.py`` on purpose: coverage's reason to change is the
+region product definition (expected cell counts, exclusion pins); parity's is the
+wire contract. Different closures, different files.
+"""
+
+from __future__ import annotations
+
+
+class GidParityError(ValueError):
+    """Sidecar and forecast gid sets differ."""
+
+
+def assert_gid_set_parity(forecast_gids, sidecar_gids, *, label: str = "sidecar") -> None:
+    """Raise unless the two gid sets are identical (§5.2)."""
+    forecast = {int(g) for g in forecast_gids}
+    sidecar = {int(g) for g in sidecar_gids}
+    if forecast == sidecar:
+        return
+    missing = sorted(forecast - sidecar)
+    extra = sorted(sidecar - forecast)
+    raise GidParityError(
+        f"{label}: gid sets differ (§5.2) — {len(missing)} forecast cell(s) missing "
+        f"from the sidecar (first: {missing[:5]}), {len(extra)} extra sidecar cell(s) "
+        f"(first: {extra[:5]})."
+    )

views_postprocessing/unfao/wire/sidecar.py

@@ -0,0 +1,81 @@
+"""The §5 GAUL geography sidecar builder (ADR-013).
+
+One sidecar per run: a 10-column parquet — ``priogrid_id`` (int64) as the FIRST
+COLUMN, then the 9 contract columns in ``gaul_schema.METADATA_COLS`` order (the
+SSOT; never restated here). Pinned by §5.1: ``*_code`` columns **always float64**
+(missing codes become NaN values, matching the canonical fixture bytes), string
+columns plain (dictionary encodings are flattened; missing stays null), rows
+ascending by ``priogrid_id``.
+
+The lookup is a **parameter** (DIP): production passes the ADR-011
+``data/gaul_lookup.parquet`` table; tests inject a synthetic one. The builder is
+restricted to the declared forecast gid set, and a declared gid absent from the
+lookup FAILS LOUD — geography for a forecast cell must never silently vanish
+(§5.2 parity direction).
+
+pyarrow-only — the delivery's pandas-free path stays pandas-free.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from pathlib import Path
+
+import numpy as np
+import pyarrow as pa
+import pyarrow.compute as pc
+import pyarrow.parquet as pq
+
+from views_postprocessing.unfao.gaul_schema import CODE_COLS, COORD_COLS, METADATA_COLS
+from views_postprocessing.unfao.wire.naming import sidecar_name
+
+
+class SidecarError(ValueError):
+    """The sidecar cannot be built as declared."""
+
+
+def build_sidecar(lookup: pa.Table, gids) -> pa.Table:
+    """The §5.1 table for exactly the declared ``gids``, schema pinned.
+
+    ``lookup`` carries ``priogrid_gid`` plus the 9 contract columns (the ADR-011
+    lookup's shape). Output renames the key to the wire's ``priogrid_id``.
+    """
+    missing = [c for c in ("priogrid_gid", *METADATA_COLS) if c not in lookup.column_names]
+    if missing:
+        raise SidecarError(f"lookup lacks required columns: {missing}.")
+
+    wanted = np.asarray(sorted(int(g) for g in set(gids)), dtype=np.int64)
+    lookup_gids = lookup.column("priogrid_gid").to_numpy(zero_copy_only=False)
+    absent = np.setdiff1d(wanted, lookup_gids)
+    if absent.size:
+        raise SidecarError(
+            f"{absent.size} forecast cell(s) absent from the GAUL lookup — geography "
+            f"must never silently vanish (§5.2). First missing gids: {absent[:5].tolist()}."
+        )
+
+    mask = pa.array(np.isin(lookup_gids, wanted))
+    selected = lookup.filter(mask).sort_by("priogrid_gid")  # ascending rows (§5.1)
+
+    columns: dict[str, pa.Array] = {
+        "priogrid_id": selected.column("priogrid_gid").cast(pa.int64()).combine_chunks()
+    }
+    for col in METADATA_COLS:
+        array = selected.column(col)
+        if col in CODE_COLS:
+            # Always float64 (§5.1 dtype ruling); missing codes are NaN VALUES,
+            # matching the canonical fixture bytes (not arrow nulls).
+            array = pc.fill_null(array.cast(pa.float64()), float("nan"))
+        elif col in COORD_COLS:
+            array = array.cast(pa.float64())
+        else:
+            array = array.cast(pa.string())  # flatten dictionary encodings; nulls stay
+        columns[col] = array.combine_chunks()
+    return pa.table(columns)
+
+
+def write_sidecar(lookup: pa.Table, gids, *, run_id: str, directory: Path) -> tuple[str, str]:
+    """Write the run's sidecar; return ``(file_name, sha256_of_bytes)`` for §4.2."""
+    name = sidecar_name(run_id)
+    path = Path(directory) / name
+    pq.write_table(build_sidecar(lookup, gids), path)
+    return name, hashlib.sha256(path.read_bytes()).hexdigest()