feat(track-a-source): ADR-013 leg 2b — Hop-A source adapter (Track-A archive → PredictionFrame)

tests/test_track_a_source.py

@@ -0,0 +1,187 @@
+"""The Hop-A source adapter against the §10 golden fixture — happy path from the
+canonical bytes, tamper paths from surgically corrupted copies of them.
+
+Every tamper case starts from the REAL fixture and breaks exactly one declared fact
+(hash, member set, sample_count, dtype, major version, target, month set, cell count,
+identifier length) — proving the adapter rejects precisely what the contract says it
+must, not merely malformed junk.
+"""
+
+import io
+import json
+import zipfile
+from pathlib import Path
+
+import numpy as np
+import pytest
+
+from views_postprocessing.unfao import track_a_source as tas
+
+FIX = Path(__file__).resolve().parent / "fixtures" / "wire_contract"
+SHARD = (FIX / "fixture_run_0__lr_ged_sb__m000543.tap.zip").read_bytes()
+MANIFEST = json.loads((FIX / "fixture_run_0__lr_ged_sb__manifest.json").read_text())
+SHARD_NAME = MANIFEST["shards"][0]["name"]
+SHARD_SHA = MANIFEST["shards"][0]["sha256"]
+
+
+def _retouched_shard(**member_overrides: bytes) -> bytes:
+    """The fixture shard with named members replaced — everything else byte-identical."""
+    with zipfile.ZipFile(io.BytesIO(SHARD)) as zf:
+        members = {name: zf.read(name) for name in zf.namelist()}
+    members.update(member_overrides)
+    buf = io.BytesIO()
+    with zipfile.ZipFile(buf, "w", zipfile.ZIP_STORED) as zf:
+        for name, data in members.items():
+            zf.writestr(name, data)
+    return buf.getvalue()
+
+
+def _header(**overrides) -> bytes:
+    with zipfile.ZipFile(io.BytesIO(SHARD)) as zf:
+        header = json.loads(zf.read("metadata.json"))
+    header.update(overrides)
+    return json.dumps(header).encode()
+
+
+def _sha(data: bytes) -> str:
+    import hashlib
+
+    return hashlib.sha256(data).hexdigest()
+
+
+# --- happy path: the golden fixture is the oracle -----------------------------------
+
+
+def test_read_manifest_accepts_the_fixture_manifest():
+    manifest = tas.read_manifest((FIX / "fixture_run_0__lr_ged_sb__manifest.json").read_bytes())
+    assert manifest["run_id"] == "fixture_run_0"
+    assert manifest["sidecar_sha256"] is None  # Erratum E1
+
+
+def test_read_shard_round_trips_the_fixture():
+    frame, header = tas.read_shard(SHARD, expected_sha256=SHARD_SHA)
+    assert frame.n_rows == 6 and frame.sample_count == 4
+    assert frame.values.dtype == np.float32
+    np.testing.assert_array_equal(np.asarray(frame.index.unit), np.arange(100001, 100007))
+    np.testing.assert_array_equal(np.asarray(frame.index.time), np.full(6, 543))
+    assert header["contract_version"] == "1.5" and header["target"] == "lr_ged_sb"
+
+
+def test_frames_for_target_assembles_the_run():
+    frame = tas.frames_for_target(MANIFEST, {SHARD_NAME: SHARD})
+    assert frame.n_rows == 6 and frame.sample_count == 4
+
+
+# --- §3.2 hash verification -----------------------------------------------------------
+
+
+def test_tampered_bytes_rejected():
+    with pytest.raises(tas.TrackASourceError, match="sha256 mismatch"):
+        tas.read_shard(SHARD + b"\x00", expected_sha256=SHARD_SHA)
+
+
+# --- §3.1 member set -------------------------------------------------------------------
+
+
+def test_missing_member_rejected():
+    with zipfile.ZipFile(io.BytesIO(SHARD)) as zf:
+        members = {n: zf.read(n) for n in zf.namelist() if n != "identifiers.npz"}
+    buf = io.BytesIO()
+    with zipfile.ZipFile(buf, "w", zipfile.ZIP_STORED) as zf:
+        for name, data in members.items():
+            zf.writestr(name, data)
+    bad = buf.getvalue()
+    with pytest.raises(tas.TrackASourceError, match="archive members"):
+        tas.read_shard(bad, expected_sha256=_sha(bad))
+
+
+# --- header-vs-payload (per-hop ingest asserts) ----------------------------------------
+
+
+def test_header_lying_about_sample_count_rejected():
+    bad = _retouched_shard(**{"metadata.json": _header(sample_count=1024)})
+    with pytest.raises(tas.TrackASourceError, match="sample_count=1024"):
+        tas.read_shard(bad, expected_sha256=_sha(bad))
+
+
+def test_header_lying_about_dtype_rejected():
+    bad = _retouched_shard(**{"metadata.json": _header(dtype="float64")})
+    with pytest.raises(tas.TrackASourceError, match="dtype"):
+        tas.read_shard(bad, expected_sha256=_sha(bad))
+
+
+def test_identifier_length_mismatch_rejected():
+    with zipfile.ZipFile(io.BytesIO(SHARD)) as zf:
+        with zipfile.ZipFile(io.BytesIO(zf.read("identifiers.npz"))) as ids:
+            time = np.load(io.BytesIO(ids.read("time.npy")))
+            unit = np.load(io.BytesIO(ids.read("unit.npy")))
+    short_ids = io.BytesIO()
+    with zipfile.ZipFile(short_ids, "w", zipfile.ZIP_STORED) as zf:
+        for name, arr in (("time.npy", time[:-1]), ("unit.npy", unit)):
+            buf = io.BytesIO()
+            np.save(buf, arr)
+            zf.writestr(name, buf.getvalue())
+    bad = _retouched_shard(**{"identifiers.npz": short_ids.getvalue()})
+    with pytest.raises(tas.TrackASourceError, match="identifier lengths"):
+        tas.read_shard(bad, expected_sha256=_sha(bad))
+
+
+# --- §2.1 versioning -------------------------------------------------------------------
+
+
+def test_major_version_bump_rejected_in_shard_header():
+    bad = _retouched_shard(**{"metadata.json": _header(contract_version="2.0")})
+    with pytest.raises(tas.TrackASourceError, match="major"):
+        tas.read_shard(bad, expected_sha256=_sha(bad))
+
+
+def test_major_version_bump_rejected_in_manifest():
+    with pytest.raises(tas.TrackASourceError, match="major"):
+        tas.read_manifest(json.dumps({**MANIFEST, "contract_version": "2.0"}).encode())
+
+
+def test_minor_version_drift_accepted():
+    bumped = _retouched_shard(**{"metadata.json": _header(contract_version="1.9")})
+    frame, header = tas.read_shard(bumped, expected_sha256=_sha(bumped))
+    assert header["contract_version"] == "1.9" and frame.n_rows == 6
+
+
+# --- run assembly (§3.2 completeness; §3.3 manifest-is-identity) ----------------------
+
+
+def test_missing_shard_bytes_rejected():
+    with pytest.raises(tas.TrackASourceError, match="not provided"):
+        tas.frames_for_target(MANIFEST, {})
+
+
+def test_shard_target_disagreeing_with_manifest_rejected():
+    bad = _retouched_shard(**{"metadata.json": _header(target="lr_ged_ns")})
+    manifest = {**MANIFEST, "shards": [{"name": SHARD_NAME, "sha256": _sha(bad)}]}
+    with pytest.raises(tas.TrackASourceError, match="target"):
+        tas.frames_for_target(manifest, {SHARD_NAME: bad})
+
+
+def test_wrong_month_coverage_rejected():
+    manifest = {**MANIFEST, "expected_months": [543, 544]}
+    with pytest.raises(tas.TrackASourceError, match="not provided"):
+        # month 544's shard is absent entirely — caught at the bytes gate
+        tas.frames_for_target(
+            {**manifest, "shards": MANIFEST["shards"] + [{"name": "m544", "sha256": "0" * 64}]},
+            {SHARD_NAME: SHARD},
+        )
+    bad = _retouched_shard(**{"metadata.json": _header(time_id=999)})
+    manifest = {**MANIFEST, "shards": [{"name": SHARD_NAME, "sha256": _sha(bad)}]}
+    with pytest.raises(tas.TrackASourceError, match="months covered"):
+        tas.frames_for_target(manifest, {SHARD_NAME: bad})
+
+
+def test_wrong_cell_count_rejected():
+    manifest = {**MANIFEST, "expected_cell_count": 7}
+    with pytest.raises(tas.TrackASourceError, match="cells"):
+        tas.frames_for_target(manifest, {SHARD_NAME: SHARD})
+
+
+def test_manifest_missing_required_field_rejected():
+    truncated = {k: v for k, v in MANIFEST.items() if k != "expected_months"}
+    with pytest.raises(tas.TrackASourceError, match="expected_months"):
+        tas.read_manifest(json.dumps(truncated).encode())

views_postprocessing/unfao/track_a_source.py

@@ -0,0 +1,146 @@
+"""Hop-A source adapter: verified Track-A archives → interior ``PredictionFrame``s
+(ADR-013 §3 consumer side; the delivery *source* adapter required by the #88 review
+amendment — the manager calls this, representation logic never lives inline in it).
+
+Pandas-free by construction: a Track-A archive is zip + npy + json; this module turns it
+into declared numpy primitives and hands them to ``frames.build_prediction_frame``. The
+new wire path never touches pandas.
+
+Declare-don't-guess, per hop: every check here is a **mechanism** guard — the artifact
+must match what its own manifest and header *declare* (hashes, counts, shapes, versions).
+The §6 *policy* gate (``delivery.draws``, S_min) is separate and runs at the FAO-facing
+upload, not here.
+
+What this module verifies (fail loud, naming expected vs found):
+- manifest: contract major-version compatible (§2.1 — MINOR additive, MAJOR breaking);
+- shard bytes match the manifest's declared sha256 (§3.2: views-postprocessing verifies
+  Hop-A hashes on read);
+- archive members are exactly the §3.1 triple; header major-version compatible;
+- payload matches the header's own declaration: ``(N, sample_count)`` float32, identifier
+  arrays of length N (the per-hop ingest assert);
+- a run's shards cover exactly the manifest's expected months and cell count (§4.2a
+  feeds on this when the manager awaits all targets).
+"""
+
+from __future__ import annotations
+
+import hashlib
+import io
+import json
+import zipfile
+
+import numpy as np
+from views_frames import PredictionFrame
+
+from views_postprocessing.unfao.frames import build_prediction_frame
+
+_ARCHIVE_MEMBERS = {"y_pred.npy", "identifiers.npz", "metadata.json"}
+_SUPPORTED_CONTRACT_MAJOR = "1"
+
+
+class TrackASourceError(ValueError):
+    """A Track-A artifact does not match what its manifest or header declares."""
+
+
+def _require_contract_major(version: str, *, where: str) -> None:
+    major = str(version).split(".", 1)[0]
+    if major != _SUPPORTED_CONTRACT_MAJOR:
+        raise TrackASourceError(
+            f"{where}: contract_version {version!r} has major {major!r}; this consumer "
+            f"speaks major {_SUPPORTED_CONTRACT_MAJOR!r} (§2.1 — MAJOR bumps are breaking)."
+        )
+
+
+def read_manifest(manifest_bytes: bytes) -> dict:
+    """Parse and validate a Hop-A (run, target) manifest (§3.2)."""
+    manifest = json.loads(manifest_bytes)
+    _require_contract_major(manifest.get("contract_version", "?"), where="manifest")
+    for key in ("run_id", "target", "shards", "expected_months", "expected_cell_count"):
+        if key not in manifest:
+            raise TrackASourceError(f"manifest: required field {key!r} is missing.")
+    return manifest
+
+
+def read_shard(shard_bytes: bytes, *, expected_sha256: str) -> tuple[PredictionFrame, dict]:
+    """One verified Track-A shard → ``(PredictionFrame, header)``.
+
+    ``expected_sha256`` is the manifest's declaration for these bytes — the §3.2
+    hash verification this repo owns on read.
+    """
+    actual = hashlib.sha256(shard_bytes).hexdigest()
+    if actual != expected_sha256:
+        raise TrackASourceError(
+            f"shard: sha256 mismatch — manifest declares {expected_sha256}, "
+            f"bytes hash to {actual}. Refusing a tampered or torn artifact."
+        )
+    with zipfile.ZipFile(io.BytesIO(shard_bytes)) as zf:
+        members = set(zf.namelist())
+        if members != _ARCHIVE_MEMBERS:
+            raise TrackASourceError(
+                f"shard: archive members {sorted(members)} != required {sorted(_ARCHIVE_MEMBERS)} (§3.1)."
+            )
+        header = json.loads(zf.read("metadata.json"))
+        values = np.load(io.BytesIO(zf.read("y_pred.npy")))
+        with zipfile.ZipFile(io.BytesIO(zf.read("identifiers.npz"))) as ids:
+            time = np.load(io.BytesIO(ids.read("time.npy")))
+            unit = np.load(io.BytesIO(ids.read("unit.npy")))
+
+    _require_contract_major(header.get("contract_version", "?"), where="shard header")
+    declared_s = header.get("sample_count")
+    if values.ndim != 2 or values.shape[1] != declared_s:
+        found = values.shape if values.ndim == 2 else f"{values.ndim}-D"
+        raise TrackASourceError(
+            f"shard: header declares sample_count={declared_s} but y_pred is {found} — "
+            f"the artifact lies about its own draws (per-hop ingest assert)."
+        )
+    if str(values.dtype) != header.get("dtype"):
+        raise TrackASourceError(
+            f"shard: header declares dtype={header.get('dtype')!r}, y_pred is {values.dtype}."
+        )
+    n = values.shape[0]
+    if time.shape != (n,) or unit.shape != (n,):
+        raise TrackASourceError(
+            f"shard: identifier lengths (time={time.shape}, unit={unit.shape}) != N={n}."
+        )
+    return build_prediction_frame(values, time, unit), header
+
+
+def frames_for_target(manifest: dict, shard_bytes_by_name: dict) -> PredictionFrame:
+    """A (run, target)'s verified shards → one interior ``PredictionFrame``.
+
+    ``shard_bytes_by_name`` maps shard ``name`` → downloaded bytes; the manifest is the
+    only source of which shards exist (§3.3: names are locators, manifest content is
+    identity). Verifies run completeness against the manifest's own declarations —
+    months covered exactly, cell count per month — then stacks months into one frame.
+    """
+    frames, months_seen = [], []
+    for entry in manifest["shards"]:
+        name = entry["name"]
+        if name not in shard_bytes_by_name:
+            raise TrackASourceError(
+                f"run: manifest lists shard {name!r} but its bytes were not provided — "
+                f"an unmanifested or missing shard must not be silently skipped."
+            )
+        frame, header = read_shard(shard_bytes_by_name[name], expected_sha256=entry["sha256"])
+        if header.get("target") != manifest["target"]:
+            raise TrackASourceError(
+                f"run: shard header target {header.get('target')!r} != manifest target "
+                f"{manifest['target']!r}."
+            )
+        if frame.n_rows != manifest["expected_cell_count"]:
+            raise TrackASourceError(
+                f"run: shard {name!r} carries {frame.n_rows} cells, manifest expects "
+                f"{manifest['expected_cell_count']}."
+            )
+        frames.append(frame)
+        months_seen.append(int(header.get("time_id")))
+
+    if sorted(months_seen) != sorted(int(m) for m in manifest["expected_months"]):
+        raise TrackASourceError(
+            f"run: months covered {sorted(months_seen)} != manifest expected "
+            f"{sorted(manifest['expected_months'])} — a torn run must not be assembled."
+        )
+    values = np.concatenate([f.values for f in frames], axis=0)
+    time = np.concatenate([np.asarray(f.index.time) for f in frames])
+    unit = np.concatenate([np.asarray(f.index.unit) for f in frames])
+    return build_prediction_frame(values, time, unit)