[DE-8690] Local EvaluationV2 over parquet: nucleus/evalv2 + nu evalv2 run (WIP)

cli/nu.py

@@ -37,6 +37,15 @@ def nu(ctx, web):
 nu.add_command(slices)  # type: ignore
 nu.add_command(tests)  # type: ignore
 
+# Local EvaluationV2 (`nu evalv2 run`). Guarded so `nu` still works if the optional evalv2-core
+# dependency isn't installed yet (ships from a private index; see pyproject wiring).
+try:
+    from nucleus.evalv2.cli import evalv2
+
+    nu.add_command(evalv2)  # type: ignore
+except ImportError:
+    pass
+
 if __name__ == "__main__":
     """To debug, run this script followed by request command tree e.g. `cli/nu.py datasets list`"""
     nu()

nucleus/evalv2/__init__.py

@@ -0,0 +1,6 @@
+"""Local EvaluationV2 harness: run the shared evalv2-core kernel over parquet on a laptop.
+
+This subpackage is intentionally import-light — it does NOT import the heavy top-level ``nucleus``
+SDK. The parquet adapter depends only on ``evalv2_core`` + ``pyarrow``, mapping parquet columns
+straight to the kernel's storage-agnostic types.
+"""

nucleus/evalv2/_tests/test_cli.py

@@ -0,0 +1,138 @@
+"""Step 10 — run_local_eval + `nu evalv2 run` CLI, isolated from the heavy nucleus SDK.
+
+The evalv2 modules use proper absolute imports (`from nucleus.evalv2.X import ...`). To load them
+without executing the heavy ``nucleus/__init__``, we register empty stub packages for ``nucleus`` and
+``nucleus.evalv2`` in ``sys.modules``, then load each submodule from its file in dependency order.
+"""
+
+import importlib.util
+import json
+import pathlib
+import sys
+import types
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+import pytest
+from click.testing import CliRunner
+
+_EVALV2 = pathlib.Path(__file__).resolve().parent.parent
+
+
+def _stub(name: str) -> None:
+    if name not in sys.modules:
+        m = types.ModuleType(name)
+        m.__path__ = []  # mark as package
+        sys.modules[name] = m
+
+
+def _load(mod_name: str, filename: str):
+    _stub("nucleus")
+    _stub("nucleus.evalv2")
+    spec = importlib.util.spec_from_file_location(mod_name, _EVALV2 / filename)
+    mod = importlib.util.module_from_spec(spec)
+    sys.modules[mod_name] = mod  # register before exec so downstream imports resolve
+    spec.loader.exec_module(mod)
+    return mod
+
+
+_load("nucleus.evalv2.parquet_io", "parquet_io.py")
+run_mod = _load("nucleus.evalv2.run", "run.py")
+cli_mod = _load("nucleus.evalv2.cli", "cli.py")
+
+
+def _write_predictions(path, rows):
+    cols = {k: [r[k] for r in rows] for k in ["prediction_id", "dataset_item_id", "label", "confidence", "x", "y", "width", "height"]}
+    cols["metadata"] = [None] * len(rows)
+    pq.write_table(pa.table(cols), path)
+
+
+def _write_gt(path, rows):
+    cols = {k: [r[k] for r in rows] for k in ["ground_truth_id", "dataset_item_id", "label", "x", "y", "width", "height"]}
+    cols["metadata"] = [None] * len(rows)
+    pq.write_table(pa.table(cols), path)
+
+
+def _fixture(tmp_path):
+    preds = tmp_path / "p.parquet"
+    gt = tmp_path / "g.parquet"
+    _write_predictions(
+        preds,
+        [
+            {"prediction_id": "p1", "dataset_item_id": "d1", "label": "car", "confidence": 0.9, "x": 0.0, "y": 0.0, "width": 20.0, "height": 20.0},
+            {"prediction_id": "p2", "dataset_item_id": "d2", "label": "truck", "confidence": 0.3, "x": 0.0, "y": 0.0, "width": 20.0, "height": 20.0},
+        ],
+    )
+    _write_gt(
+        gt,
+        [
+            {"ground_truth_id": "g1", "dataset_item_id": "d1", "label": "car", "x": 0.0, "y": 0.0, "width": 20.0, "height": 20.0},
+            {"ground_truth_id": "g2", "dataset_item_id": "d2", "label": "truck", "x": 0.0, "y": 0.0, "width": 20.0, "height": 20.0},
+        ],
+    )
+    return preds, gt
+
+
+# ---- config_from_dict -----------------------------------------------------------------------------
+
+
+def test_config_from_dict_parses_all() -> None:
+    cfg = run_mod.config_from_dict(
+        {
+            "rollup_groups": [{"class_name": "vehicle", "labels": ["car", "truck"]}],
+            "exclusion_rules": [{"type": "confidence", "min_confidence": 0.4}],
+            "class_agnostic": True,
+            "iou_type": "bbox",
+            "min_prediction_score": 0.2,
+        }
+    )
+    assert cfg.rollup_groups[0].class_name == "vehicle"
+    assert cfg.exclusion_rules and cfg.exclusion_rules[0].type == "confidence"
+    assert cfg.class_agnostic is True and cfg.min_prediction_score == 0.2
+
+
+# ---- run_local_eval -------------------------------------------------------------------------------
+
+
+def test_run_local_eval_writes_bundle(tmp_path) -> None:
+    preds, gt = _fixture(tmp_path)
+    out = tmp_path / "out"
+    summary = run_mod.run_local_eval(preds, gt, out)
+    assert summary.total_gt == 2
+    assert (out / "matches.parquet").exists() and (out / "summary.json").exists()
+
+
+# ---- CLI ------------------------------------------------------------------------------------------
+
+
+def test_cli_run_smoke(tmp_path) -> None:
+    preds, gt = _fixture(tmp_path)
+    out = tmp_path / "out"
+    res = CliRunner().invoke(cli_mod.evalv2, ["run", "--predictions", str(preds), "--ground-truth", str(gt), "--out", str(out)])
+    assert res.exit_code == 0, res.output
+    echoed = json.loads(res.output)
+    assert echoed["total_gt"] == 2 and echoed["out"] == str(out)
+    assert (out / "charts.json").exists()
+
+
+def test_cli_config_and_flags_honored(tmp_path) -> None:
+    preds, gt = _fixture(tmp_path)
+    out = tmp_path / "out"
+    cfg = tmp_path / "cfg.json"
+    # rollup car+truck -> vehicle; without it d1(car) and d2(truck) each match same-label GTs anyway,
+    # so exercise min_prediction_score: drop p2 (0.3) via the flag -> d2 GT becomes an FN.
+    cfg.write_text(json.dumps({"rollup_groups": [{"class_name": "vehicle", "labels": ["car", "truck"]}]}))
+    res = CliRunner().invoke(
+        cli_mod.evalv2,
+        ["run", "--predictions", str(preds), "--ground-truth", str(gt), "--out", str(out),
+         "--config", str(cfg), "--min-prediction-score", "0.5"],
+    )
+    assert res.exit_code == 0, res.output
+    echoed = json.loads(res.output)
+    # p1(0.9) kept -> TP; p2(0.3) dropped -> d2 GT is FN
+    assert echoed["total_tp"] == 1 and echoed["total_fn"] == 1
+
+
+def test_cli_missing_file_errors() -> None:
+    res = CliRunner().invoke(cli_mod.evalv2, ["run", "--predictions", "/nope.parquet", "--ground-truth", "/nope2.parquet", "--out", "/tmp/x"])
+    assert res.exit_code != 0

nucleus/evalv2/_tests/test_parquet_io.py

@@ -0,0 +1,228 @@
+"""Round-trip test for the parquet adapter.
+
+Isolated from the heavy top-level ``nucleus`` package: parquet_io is loaded directly from its file
+(so importing it never triggers ``nucleus/__init__``), and pytest is run with ``--confcutdir`` so the
+repo-root ``conftest.py`` (which imports nucleus + requires an API key) is never collected.
+"""
+
+import importlib.util
+import json
+import pathlib
+from collections.abc import Iterable
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+import pytest
+
+import evalv2_core
+from evalv2_core.types import (
+    BoxGeometry,
+    EvalConfig,
+    EvalSummary,
+    GroundTruth,
+    ItemBundle,
+    Prediction,
+)
+
+# --- load parquet_io.py directly, bypassing nucleus/__init__ --------------------------------------
+_PARQUET_IO = pathlib.Path(__file__).resolve().parent.parent / "parquet_io.py"
+_spec = importlib.util.spec_from_file_location("evalv2_parquet_io", _PARQUET_IO)
+_mod = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(_mod)
+ParquetSource = _mod.ParquetSource
+ParquetSink = _mod.ParquetSink
+
+
+# --- in-memory reference source/sink (for round-trip comparison) -----------------------------------
+class _MemSource:
+    def __init__(self, cfg, items):
+        self._cfg, self._items = cfg, items
+
+    def read_config(self):
+        return self._cfg
+
+    def read_items(self):
+        return iter(self._items)
+
+
+class _MemSink:
+    def __init__(self):
+        self.summary = None
+
+    def begin(self):
+        ...
+
+    def write_matches(self, m, p):
+        ...
+
+    def write_summary(self, summary, metrics, fields, charts):
+        self.summary = summary
+
+    def finalize(self):
+        ...
+
+
+def _write_predictions(path):
+    pq.write_table(
+        pa.table(
+            {
+                "prediction_id": ["p1", "p2"],
+                "dataset_item_id": ["d1", "d2"],
+                "label": ["car", "car"],
+                "confidence": [0.9, 0.8],
+                "x": [0.0, 0.0],
+                "y": [0.0, 0.0],
+                "width": [20.0, 20.0],
+                "height": [20.0, 20.0],
+                "metadata": [None, None],
+            }
+        ),
+        path,
+    )
+
+
+def _write_ground_truth(path):
+    pq.write_table(
+        pa.table(
+            {
+                "ground_truth_id": ["g1", "g2", "g3"],
+                "dataset_item_id": ["d1", "d2", "d3"],
+                "label": ["car", "person", "car"],
+                "x": [0.0, 0.0, 0.0],
+                "y": [0.0, 0.0, 0.0],
+                "width": [20.0, 20.0, 20.0],
+                "height": [20.0, 20.0, 20.0],
+                "metadata": [None, None, None],
+            }
+        ),
+        path,
+    )
+
+
+def _write_items(path):
+    pq.write_table(
+        pa.table(
+            {
+                "dataset_item_id": ["d1", "d2", "d3"],
+                "metadata": [json.dumps({"weather": "rain"}), json.dumps({"weather": "sun"}), json.dumps({"weather": "sun"})],
+            }
+        ),
+        path,
+    )
+
+
+def _reference_items():
+    def box(*b):
+        return BoxGeometry(*b)
+
+    return [
+        ItemBundle("d1", [Prediction("p1", "d1", "car", 0.9, box(0, 0, 20, 20))], [GroundTruth("g1", "d1", "car", box(0, 0, 20, 20))], {"weather": "rain"}),
+        ItemBundle("d2", [Prediction("p2", "d2", "car", 0.8, box(0, 0, 20, 20))], [GroundTruth("g2", "d2", "person", box(0, 0, 20, 20))], {"weather": "sun"}),
+        ItemBundle("d3", [], [GroundTruth("g3", "d3", "car", box(0, 0, 20, 20))], {"weather": "sun"}),
+    ]
+
+
+def test_parquet_round_trip_matches_in_memory(tmp_path) -> None:
+    pred_p = tmp_path / "predictions.parquet"
+    gt_p = tmp_path / "ground_truth.parquet"
+    items_p = tmp_path / "items.parquet"
+    out = tmp_path / "out"
+    _write_predictions(pred_p)
+    _write_ground_truth(gt_p)
+    _write_items(items_p)
+
+    # run through the parquet adapter
+    src = ParquetSource(pred_p, gt_p, items_p, config=EvalConfig())
+    sink = ParquetSink(out)
+    parquet_summary = evalv2_core.run(src, sink)
+
+    # run the same items purely in memory
+    mem_sink = _MemSink()
+    evalv2_core.run(_MemSource(EvalConfig(), _reference_items()), mem_sink)
+    mem_summary = mem_sink.summary
+
+    assert isinstance(parquet_summary, EvalSummary)
+    for key in ("map_50", "map_50_95", "total_tp", "total_fp", "total_fn", "total_gt"):
+        a, b = getattr(parquet_summary, key), getattr(mem_summary, key)
+        assert (a is None and b is None) or a == pytest.approx(b), key
+
+
+def test_source_yields_joined_bundles(tmp_path) -> None:
+    _write_predictions(tmp_path / "p.parquet")
+    _write_ground_truth(tmp_path / "g.parquet")
+    _write_items(tmp_path / "i.parquet")
+    src = ParquetSource(tmp_path / "p.parquet", tmp_path / "g.parquet", tmp_path / "i.parquet")
+    items = list(src.read_items())
+    assert [it.dataset_item_id for it in items] == ["d1", "d2", "d3"]
+    d1 = items[0]
+    assert d1.predictions[0].id == "p1" and d1.ground_truths[0].id == "g1"
+    assert d1.item_metadata == {"weather": "rain"}
+    assert isinstance(d1.predictions[0].geometry, BoxGeometry)
+
+
+def test_output_bundle_files_written_and_readable(tmp_path) -> None:
+    _write_predictions(tmp_path / "p.parquet")
+    _write_ground_truth(tmp_path / "g.parquet")
+    out = tmp_path / "out"
+    src = ParquetSource(tmp_path / "p.parquet", tmp_path / "g.parquet")
+    evalv2_core.run(src, ParquetSink(out))
+
+    assert (out / "matches.parquet").exists()
+    assert (out / "per_threshold.parquet").exists()
+    matches = pq.read_table(out / "matches.parquet").to_pylist()
+    assert matches and {m["match_type"] for m in matches} <= {"TP", "FP", "FN"}
+    per_thr = pq.read_table(out / "per_threshold.parquet")
+    assert "size_bucket" in per_thr.column_names
+
+    summary_doc = json.loads((out / "summary.json").read_text())
+    assert "summary" in summary_doc and len(summary_doc["metrics_at_confidence"]) == 17
+    charts = json.loads((out / "charts.json").read_text())
+    for key in ("mapSummary", "prCurve", "confusionMatrix", "tideAttribution", "f1Curve"):
+        assert key in charts
+
+
+def test_streaming_groups_across_batch_boundaries(tmp_path) -> None:
+    # force many tiny batches so an item's rows and item boundaries span batches
+    _write_predictions(tmp_path / "p.parquet")
+    _write_ground_truth(tmp_path / "g.parquet")
+    _write_items(tmp_path / "i.parquet")
+    src = ParquetSource(
+        tmp_path / "p.parquet", tmp_path / "g.parquet", tmp_path / "i.parquet", batch_size=1
+    )
+    items = list(src.read_items())
+    assert [it.dataset_item_id for it in items] == ["d1", "d2", "d3"]
+    assert items[0].predictions[0].id == "p1" and items[0].ground_truths[0].id == "g1"
+    assert items[2].predictions == [] and items[2].ground_truths[0].id == "g3"  # d3 gt-only
+
+
+def test_unsorted_input_raises(tmp_path) -> None:
+    # predictions out of dataset_item_id order -> loud error, not a silent mis-join
+    pq.write_table(
+        pa.table(
+            {
+                "prediction_id": ["p2", "p1"],
+                "dataset_item_id": ["d2", "d1"],  # descending -> unsorted
+                "label": ["car", "car"],
+                "confidence": [0.8, 0.9],
+                "x": [0.0, 0.0], "y": [0.0, 0.0], "width": [20.0, 20.0], "height": [20.0, 20.0],
+                "metadata": [None, None],
+            }
+        ),
+        tmp_path / "p.parquet",
+    )
+    _write_ground_truth(tmp_path / "g.parquet")
+    src = ParquetSource(tmp_path / "p.parquet", tmp_path / "g.parquet", batch_size=1)
+    with pytest.raises(ValueError, match="sorted by dataset_item_id"):
+        list(src.read_items())
+
+
+def test_item_metadata_round_trips_into_match_rows(tmp_path) -> None:
+    _write_predictions(tmp_path / "p.parquet")
+    _write_ground_truth(tmp_path / "g.parquet")
+    _write_items(tmp_path / "i.parquet")
+    out = tmp_path / "out"
+    src = ParquetSource(tmp_path / "p.parquet", tmp_path / "g.parquet", tmp_path / "i.parquet")
+    evalv2_core.run(src, ParquetSink(out))
+    matches = pq.read_table(out / "matches.parquet").to_pylist()
+    d1_rows = [m for m in matches if m["dataset_item_id"] == "d1"]
+    assert d1_rows and json.loads(d1_rows[0]["item_metadata"]) == {"weather": "rain"}