Add internal canonical ROC v2 spec builder

src/rtichoke/_viz_spec_v2.py

@@ -0,0 +1,117 @@
+"""Internal builders for canonical ``rtichoke_viz`` v2 specifications.
+
+These helpers are deliberately not wired into production rendering yet. They
+translate already-computed performance data plus semantic evaluation metadata
+into the canonical visualization contract.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+
+import polars as pl
+
+from rtichoke.processing.evaluation_semantics import _EvaluationMetadata
+
+_REQUIRED_ROC_COLUMNS = {
+    "reference_group",
+    "chosen_cutoff",
+    "sensitivity",
+    "specificity",
+}
+
+
+def _roc_v2_spec_from_performance_data(
+    performance_data: pl.DataFrame,
+    evaluation_metadata: Mapping[str, _EvaluationMetadata],
+) -> dict[str, object]:
+    """Build a canonical ROC-v2 spec without recalculating statistics.
+
+    ``reference_group`` is used only to join existing performance rows to the
+    semantic metadata established by the high-level production inputs. Stable
+    evaluation/series IDs are ordinal over that semantic metadata, so they do
+    not encode labels, compatibility grouping names, prevalence, coordinates,
+    colors, or other presentation/statistical values.
+    """
+    missing = _REQUIRED_ROC_COLUMNS.difference(performance_data.columns)
+    if missing:
+        missing_columns = ", ".join(sorted(missing))
+        raise ValueError(f"ROC performance data is missing columns: {missing_columns}")
+
+    rows = performance_data.select(
+        "reference_group",
+        "chosen_cutoff",
+        "sensitivity",
+        "specificity",
+    ).to_dicts()
+    row_groups = {str(row["reference_group"]) for row in rows}
+    metadata_groups = set(evaluation_metadata)
+
+    missing_metadata = row_groups.difference(metadata_groups)
+    if missing_metadata:
+        groups = ", ".join(sorted(missing_metadata))
+        raise ValueError(
+            f"ROC performance rows are missing evaluation metadata: {groups}"
+        )
+
+    ordered_groups = [group for group in evaluation_metadata if group in row_groups]
+    evaluation_ids = {
+        group: f"evaluation-{index}"
+        for index, group in enumerate(ordered_groups, start=1)
+    }
+    series_ids = {
+        group: f"series-{index}" for index, group in enumerate(ordered_groups, start=1)
+    }
+
+    evaluations: list[dict[str, object]] = []
+    series: list[dict[str, object]] = []
+
+    for group in ordered_groups:
+        metadata = evaluation_metadata[group]
+        evaluation: dict[str, object] = {
+            "id": evaluation_ids[group],
+            "population": metadata.population,
+        }
+        if metadata.model is not None:
+            evaluation["model"] = metadata.model
+        evaluations.append(evaluation)
+
+        if metadata.model is not None:
+            display_value = metadata.model
+            display_role = "model"
+        else:
+            display_value = metadata.population
+            display_role = "population"
+
+        series.append(
+            {
+                "id": series_ids[group],
+                "evaluationId": evaluation_ids[group],
+                "display": {
+                    "label": display_value,
+                    "group": display_value,
+                    "role": display_role,
+                },
+            }
+        )
+
+    return {
+        "schemaVersion": "2.0",
+        "type": "roc",
+        "evaluations": evaluations,
+        "series": series,
+        "data": [
+            {
+                "seriesId": series_ids[str(row["reference_group"])],
+                "cutoff": row["chosen_cutoff"],
+                "sensitivity": row["sensitivity"],
+                "specificity": row["specificity"],
+            }
+            for row in rows
+        ],
+        "x": "false_positive_rate",
+        "y": "sensitivity",
+        "xAxis": {"label": "1 - Specificity", "domain": [0, 1]},
+        "yAxis": {"label": "Sensitivity", "domain": [0, 1]},
+        "references": [{"type": "identity", "scope": "global"}],
+    }

tests/test_viz_spec_v2.py

@@ -0,0 +1,131 @@
+import numpy as np
+
+from rtichoke._viz_spec_v2 import _roc_v2_spec_from_performance_data
+from rtichoke.performance_data.performance_data import prepare_performance_data
+from rtichoke.processing.evaluation_semantics import (
+    _SHARED_POPULATION,
+    _build_evaluation_metadata,
+)
+
+
+def _static_metadata(probs, reals):
+    return _build_evaluation_metadata(probs, reals, np.array([]))
+
+
+def test_roc_v2_one_model_one_population():
+    probs = {"Model A": np.array([0.05, 0.2, 0.7, 0.95])}
+    reals = np.array([0, 0, 1, 1])
+    performance_data = prepare_performance_data(probs, reals, by=0.25)
+
+    spec = _roc_v2_spec_from_performance_data(
+        performance_data,
+        _static_metadata(probs, reals),
+    )
+
+    assert spec["schemaVersion"] == "2.0"
+    assert spec["type"] == "roc"
+    assert spec["evaluations"] == [
+        {
+            "id": "evaluation-1",
+            "model": "Model A",
+            "population": _SHARED_POPULATION,
+        }
+    ]
+    assert spec["series"] == [
+        {
+            "id": "series-1",
+            "evaluationId": "evaluation-1",
+            "display": {"label": "Model A", "group": "Model A", "role": "model"},
+        }
+    ]
+    assert {row["seriesId"] for row in spec["data"]} == {"series-1"}
+    assert spec["references"] == [{"type": "identity", "scope": "global"}]
+
+
+def test_roc_v2_multiple_models_share_one_population():
+    probs = {
+        "Model A": np.array([0.05, 0.2, 0.7, 0.95]),
+        "Model B": np.array([0.1, 0.4, 0.6, 0.9]),
+    }
+    reals = np.array([0, 0, 1, 1])
+    performance_data = prepare_performance_data(probs, reals, by=0.25)
+
+    spec = _roc_v2_spec_from_performance_data(
+        performance_data,
+        _static_metadata(probs, reals),
+    )
+
+    evaluations = spec["evaluations"]
+    assert [evaluation["id"] for evaluation in evaluations] == [
+        "evaluation-1",
+        "evaluation-2",
+    ]
+    assert {evaluation["population"] for evaluation in evaluations} == {
+        _SHARED_POPULATION
+    }
+    assert [evaluation["model"] for evaluation in evaluations] == ["Model A", "Model B"]
+    assert [series["id"] for series in spec["series"]] == ["series-1", "series-2"]
+    assert {series["display"]["role"] for series in spec["series"]} == {"model"}
+    assert {row["seriesId"] for row in spec["data"]} == {"series-1", "series-2"}
+
+
+def test_roc_v2_keyed_populations_keep_model_identity_unknown():
+    probs = {
+        "Population A": np.array([0.05, 0.2, 0.7, 0.95]),
+        "Population B": np.array([0.1, 0.4, 0.6, 0.9]),
+    }
+    reals = {
+        "Population A": np.array([0, 0, 1, 1]),
+        "Population B": np.array([0, 1, 0, 1]),
+    }
+    performance_data = prepare_performance_data(probs, reals, by=0.25)
+
+    spec = _roc_v2_spec_from_performance_data(
+        performance_data,
+        _static_metadata(probs, reals),
+    )
+
+    assert spec["evaluations"] == [
+        {"id": "evaluation-1", "population": "Population A"},
+        {"id": "evaluation-2", "population": "Population B"},
+    ]
+    assert [series["display"] for series in spec["series"]] == [
+        {"label": "Population A", "group": "Population A", "role": "population"},
+        {"label": "Population B", "group": "Population B", "role": "population"},
+    ]
+    assert {row["seriesId"] for row in spec["data"]} == {"series-1", "series-2"}
+
+
+def test_roc_v2_ids_do_not_encode_compatibility_group_labels():
+    reals = {
+        "Population A": np.array([0, 0, 1, 1]),
+        "Population B": np.array([0, 1, 0, 1]),
+    }
+    probs = {
+        "Population A": np.array([0.05, 0.2, 0.7, 0.95]),
+        "Population B": np.array([0.1, 0.4, 0.6, 0.9]),
+    }
+    spec = _roc_v2_spec_from_performance_data(
+        prepare_performance_data(probs, reals, by=0.25),
+        _static_metadata(probs, reals),
+    )
+
+    renamed_reals = {
+        "Cohort X": reals["Population A"],
+        "Cohort Y": reals["Population B"],
+    }
+    renamed_probs = {
+        "Cohort X": probs["Population A"],
+        "Cohort Y": probs["Population B"],
+    }
+    renamed_spec = _roc_v2_spec_from_performance_data(
+        prepare_performance_data(renamed_probs, renamed_reals, by=0.25),
+        _static_metadata(renamed_probs, renamed_reals),
+    )
+
+    assert [evaluation["id"] for evaluation in spec["evaluations"]] == [
+        evaluation["id"] for evaluation in renamed_spec["evaluations"]
+    ]
+    assert [series["id"] for series in spec["series"]] == [
+        series["id"] for series in renamed_spec["series"]
+    ]