feat(historical): B2/#126 — frame-native historical (pandas out of the construction)

tests/test_historical_builder.py

@@ -0,0 +1,123 @@
+"""The frame-native historical artifact builder (unfao/historical.py, #126) —
+reader-level parity with the legacy characterization golden, plus its own
+fail-loud properties."""
+
+from pathlib import Path
+
+import numpy as np
+import pyarrow as pa
+import pyarrow.parquet as pq
+import pytest
+from views_frames import FeatureFrame, SpatialLevel, SpatioTemporalIndex
+
+from views_postprocessing.unfao import historical
+from views_postprocessing.unfao.frame_extraction import drop_months_above
+from views_postprocessing.unfao.gaul_schema import METADATA_COLS
+
+import sys
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+from test_historical_parity import GOLDEN, REAL_TARGETS, reader_view  # noqa: E402
+
+_FIXDIR = Path(__file__).resolve().parent / "fixtures" / "historical_golden"
+_LOOKUP = Path(__file__).resolve().parents[1] / "views_postprocessing" / "data" / "gaul_lookup.parquet"
+
+
+def _slice_frame() -> FeatureFrame:
+    """The committed raw input slice as the FeatureFrame the loader would return."""
+    table = pq.read_table(_FIXDIR / "raw_slice_input.parquet")
+    time = np.asarray(table.column("month_id"), dtype=np.int64)
+    unit = np.asarray(table.column("priogrid_id"), dtype=np.int64)
+    values_2d = np.column_stack(
+        [np.asarray(table.column(t), dtype=np.float32) for t in REAL_TARGETS]
+    )
+    index = SpatioTemporalIndex(time=time, unit=unit, level=SpatialLevel.PGM)
+    return FeatureFrame.from_2d(values_2d, index, feature_names=list(REAL_TARGETS))
+
+
+def test_reader_level_parity_with_the_legacy_golden(tmp_path):
+    # The #126 acceptance oracle: same slice, frame-native chain, and faoapi's
+    # reader must see the same frame the legacy chain produced (minus junk).
+    table = historical.build_historical_table(_slice_frame(), pq.read_table(_LOOKUP))
+    historical.assert_metadata_complete(table)
+    name, _ = historical.write_historical_artifact(table, tmp_path / "new.parquet")
+    new = reader_view(tmp_path / "new.parquet")
+    old = reader_view(GOLDEN)
+    assert set(new["targets"]) == set(REAL_TARGETS)  # junk row/col GONE by design
+    assert new["rows"].keys() == old["rows"].keys()  # identical (month, cell) set
+    def norm(v):
+        # faoapi's reader semantics: pandas collapses parquet null/NaN to NaN,
+        # and _validate_feature_samples (handlers.py:374-376) normalizes scalar
+        # and single-element-list values to the same 1-array — the legacy chain
+        # shipped [v] (the C-40 list-in-cell disease), the frame builder ships v;
+        # both reach faoapi identical. The oracle compares post-normalization.
+        if isinstance(v, list) and len(v) == 1:
+            v = v[0]
+        return float("nan") if v is None else v
+
+    for key, new_cols in new["rows"].items():
+        old_cols = old["rows"][key]
+        for col in (*REAL_TARGETS, *METADATA_COLS):
+            a, b = norm(new_cols[col]), norm(old_cols[col])
+            if isinstance(a, float) and isinstance(b, float):
+                if a != a and b != b:  # both NaN — equal through the reader
+                    continue
+                assert a == pytest.approx(b, rel=1e-6), (key, col)
+            else:
+                assert a == b, (key, col)
+
+
+def test_junk_columns_are_dropped():
+    table = historical.build_historical_table(_slice_frame(), pq.read_table(_LOOKUP))
+    assert "row" not in table.column_names and "col" not in table.column_names
+    assert table.column_names == ["month_id", "priogrid_id", *REAL_TARGETS, *METADATA_COLS]
+
+
+def test_strings_are_plain_and_codes_float64():
+    table = historical.build_historical_table(_slice_frame(), pq.read_table(_LOOKUP))
+    assert table.schema.field("country_iso_a3").type == pa.string()
+    assert table.schema.field("admin1_gaul0_code").type == pa.float64()
+
+
+def test_cell_absent_from_lookup_fails_loud():
+    frame = _slice_frame()
+    bad_index = SpatioTemporalIndex(
+        time=np.asarray(frame.index.time),
+        unit=np.where(np.arange(frame.n_rows) == 0, 999_999, np.asarray(frame.index.unit)),
+        level=SpatialLevel.PGM,
+    )
+    bad = FeatureFrame(frame.values, bad_index, feature_names=list(frame.feature_names))
+    with pytest.raises(historical.HistoricalArtifactError, match="999999"):
+        historical.build_historical_table(bad, pq.read_table(_LOOKUP))
+
+
+def test_multi_sample_historical_refused():
+    frame = _slice_frame()
+    fat = FeatureFrame(
+        np.repeat(frame.values, 2, axis=2), frame.index, feature_names=list(frame.feature_names)
+    )
+    with pytest.raises(historical.HistoricalArtifactError, match="single-valued"):
+        historical.build_historical_table(fat, pq.read_table(_LOOKUP))
+
+
+def test_unmapped_cell_count_counts_distinct_gids():
+    table = historical.build_historical_table(_slice_frame(), pq.read_table(_LOOKUP))
+    assert historical.unmapped_cell_count(table) == 0
+    # poke one null into a metadata column across both months of one cell
+    poked = table.set_column(
+        table.column_names.index("country_iso_a3"),
+        "country_iso_a3",
+        pa.array(
+            [None if i < 2 else v for i, v in enumerate(table.column("country_iso_a3").to_pylist())],
+            pa.string(),
+        ),
+    )
+    assert historical.unmapped_cell_count(poked) >= 1
+
+
+def test_frame_native_month_clip():
+    frame = _slice_frame()  # months 121, 122
+    clipped = drop_months_above(frame, 121)
+    months = sorted(set(np.asarray(clipped.index.time).tolist()))
+    assert months == [121]
+    assert drop_months_above(frame, 122) is frame  # nothing to clip → identity

views_postprocessing/unfao/frame_extraction.py

@@ -39,6 +39,21 @@ def months_of(frame: PredictionFrame) -> NDArray[np.int64]:
     return np.unique(np.asarray(frame.index.time, dtype=np.int64))
 
 
+def drop_months_above(frame, last_valid_month_id: int):
+    """The frame clipped to observed months (``time <= last_valid_month_id``).
+
+    Frame-native sibling of ``extraction.drop_months_above`` (S7/#92): works on
+    any frame sharing the ``SpatioTemporalIndex`` surface (PredictionFrame,
+    FeatureFrame) via ``.select`` on a boolean row mask. The boundary value is
+    declared by the caller (producer-sourced, C-26) — never inferred here.
+    """
+    time = np.asarray(frame.index.time, dtype=np.int64)
+    mask = time <= int(last_valid_month_id)
+    if mask.all():
+        return frame
+    return frame.select(mask)
+
+
 def drop_units(frame: PredictionFrame, excluded: frozenset) -> PredictionFrame:
     """The frame without the DECLARED excluded cells (row filter on ``unit``).
 

views_postprocessing/unfao/historical.py

@@ -0,0 +1,115 @@
+"""The FAO historical (actuals) artifact, built pandas-free (#126).
+
+One concept: turn the historical ``views_frames.FeatureFrame`` plus the injected
+ADR-011 GAUL lookup into the SAME artifact faoapi's deployed reader already
+ingests — plain-column parquet with ``month_id, priogrid_id, <targets...>,
+<the 9 gaul_schema.METADATA_COLS>``. The wire artifact does not change shape
+(ADR-013 §4.1/§8); only its construction leaves pandas. Reader-level parity with
+the legacy chain is pinned by ``tests/test_historical_parity.py``'s golden.
+
+Rules carried over from the sidecar builder (§5.1 discipline):
+- lookup injected (DIP) — production passes ``data/gaul_lookup.parquet``;
+- a cell absent from the lookup FAILS LOUD (geography never silently vanishes);
+- string columns plain (never dictionary/categorical), code columns float64;
+- deliberately DROPPED: any non-feature payload columns (the legacy chain would
+  have shipped datafactory's ``row``/``col`` grid coordinates as bogus
+  auto-detected "targets" — pinned in the characterization golden).
+
+The null-gate (every metadata value present) is this module's fail-loud
+equivalent of the legacy ``_validate`` historical leg; ``unmapped_cell_count``
+feeds the same provenance field it always did (C-15).
+"""
+
+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
+
+
+class HistoricalArtifactError(ValueError):
+    """The historical artifact cannot be built as declared."""
+
+
+def build_historical_table(frame, lookup: pa.Table) -> pa.Table:
+    """FeatureFrame + GAUL lookup → the reader-compatible historical table.
+
+    ``frame.values`` is ``(N, F, S)``; actuals are single-valued (S=1) — a
+    multi-sample historical is a declaration error, refused loud.
+    """
+    if frame.sample_count != 1:
+        raise HistoricalArtifactError(
+            f"historical actuals must be single-valued; got sample_count={frame.sample_count}."
+        )
+    missing_cols = [c for c in ("priogrid_gid", *METADATA_COLS) if c not in lookup.column_names]
+    if missing_cols:
+        raise HistoricalArtifactError(f"lookup lacks required columns: {missing_cols}.")
+
+    unit = np.asarray(frame.index.unit, dtype=np.int64)
+    time = np.asarray(frame.index.time, dtype=np.int64)
+
+    lookup_gids = lookup.column("priogrid_gid").to_numpy(zero_copy_only=False)
+    order = np.argsort(lookup_gids)
+    sorted_gids = lookup_gids[order]
+    pos = np.searchsorted(sorted_gids, unit)
+    pos_valid = (pos < len(sorted_gids))
+    if not pos_valid.all() or not (sorted_gids[np.clip(pos, 0, len(sorted_gids) - 1)] == unit).all():
+        absent = np.unique(unit[~pos_valid | (sorted_gids[np.clip(pos, 0, len(sorted_gids) - 1)] != unit)])
+        raise HistoricalArtifactError(
+            f"{absent.size} historical cell(s) absent from the GAUL lookup — geography "
+            f"must never silently vanish. First missing gids: {absent[:5].tolist()}."
+        )
+    row_indices = pa.array(order[pos])
+
+    columns: dict = {
+        "month_id": pa.array(time, pa.int64()),
+        "priogrid_id": pa.array(unit, pa.int64()),
+    }
+    for i, name in enumerate(frame.feature_names):
+        columns[name] = pa.array(frame.values[:, i, 0])
+    for col in METADATA_COLS:
+        array = lookup.column(col).combine_chunks().take(row_indices)
+        if col in CODE_COLS:
+            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())  # plain, never dictionary (§5.1 discipline)
+        columns[col] = array
+    return pa.table(columns)
+
+
+def assert_metadata_complete(table: pa.Table) -> None:
+    """Fail loud on any missing metadata value (the legacy null-gate, kept)."""
+    for col in METADATA_COLS:
+        nulls = table.column(col).null_count
+        if nulls:
+            raise HistoricalArtifactError(
+                f"historical artifact has {nulls} null value(s) in required metadata "
+                f"column {col!r} ({nulls}/{table.num_rows} rows)."
+            )
+
+
+def unmapped_cell_count(table: pa.Table) -> int:
+    """Distinct cells with a null in ANY metadata column (the C-15 provenance count)."""
+    mask = None
+    for col in METADATA_COLS:
+        col_null = pc.is_null(table.column(col))
+        mask = col_null if mask is None else pc.or_(mask, col_null)
+    if not pc.any(mask).as_py():
+        return 0
+    gids = pc.filter(table.column("priogrid_id"), mask)
+    return len(pc.unique(gids))
+
+
+def write_historical_artifact(table: pa.Table, path: Path) -> tuple[str, str]:
+    """Serialize; return ``(file_name, sha256_of_bytes)``."""
+    path = Path(path)
+    pq.write_table(table, path)
+    return path.name, hashlib.sha256(path.read_bytes()).hexdigest()