diff --git a/src/rtichoke/_viz_spec_v2.py b/src/rtichoke/_viz_spec_v2.py new file mode 100644 index 00000000..1ec30b70 --- /dev/null +++ b/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"}], + } diff --git a/tests/test_viz_spec_v2.py b/tests/test_viz_spec_v2.py new file mode 100644 index 00000000..7ec46dea --- /dev/null +++ b/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"] + ]