Skip to content

pipekit-evaluate

Multidimensional evaluation framework. The benchmark subpackage ships the orchestration layer of the benchmark ladder — the level x rung x complexity cube, the carrier-agnostic protocols, and the "simpler thing is ground truth" reference rule. The taxonomy module carries the Unit x Lens x Stage tags, MetricScorer turns any metric operator into a tagged scorer, and evaluate collects a run's tagged scores into a content-hashed EvaluationReport. Design doc: packages/pipekit-evaluate/docs/design/benchmark_ladder.md.

Taxonomy

pipekit_evaluate.taxonomy

The Unit x Lens x Stage taxonomy, carried by scorers as data.

An evaluation picks one value on each of three axes:

  • Unit — what is scored: a field, a statistic, a trajectory, an event or a budget.
  • Lens — what kind of failure the score can see: point-wise, probabilistic, spectral, structural, detection or physical-constraint.
  • Stage — where in a model's life the score may run (training, validation, final evaluation, monitoring), which bounds how cheap and differentiable it must be.

The tags live on the scorer, not on the metric. The metrics stay in the carrier libraries (xrtoolz.metrics), which never import pipekit-evaluate; a scorer declares its tags as optional attributes (unit, lens, stages, differentiable) and :func:tags_of reads them, filling in the defaults. A scorer with only a name is therefore a point-wise field score for validation and final evaluation.

Lens

Bases: StrEnum

What kind of failure a score can see.

Source code in packages/pipekit-evaluate/src/pipekit_evaluate/taxonomy.py 
39
40
41
42
43
44
45
46
47
class Lens(StrEnum):
    """What kind of failure a score can see."""

    POINTWISE = "pointwise"
    PROBABILISTIC = "probabilistic"
    SPECTRAL = "spectral"
    STRUCTURAL = "structural"
    DETECTION = "detection"
    PHYSICAL = "physical"

ScorerTags dataclass

A scorer's position in the taxonomy.

Strings are accepted wherever an enum member is, and one stage may be given on its own, so ScorerTags(lens="structural", stages="final") is valid; the fields always hold enum members after construction.

Attributes:

Name Type Description
unit Unit

What the scorer scores.
lens Lens

What kind of failure it can see.
stages frozenset[Stage]

The stages it may run at; never empty.
differentiable bool

Whether its value can be differentiated with respect to the prediction (a requirement for use as a loss).
Source code in packages/pipekit-evaluate/src/pipekit_evaluate/taxonomy.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
@dataclass(frozen=True, init=False)
class ScorerTags:
    """A scorer's position in the taxonomy.

    Strings are accepted wherever an enum member is, and one stage may be
    given on its own, so ``ScorerTags(lens="structural", stages="final")``
    is valid; the fields always hold enum members after construction.

    Attributes:
        unit: What the scorer scores.
        lens: What kind of failure it can see.
        stages: The stages it may run at; never empty.
        differentiable: Whether its value can be differentiated with
            respect to the prediction (a requirement for use as a loss).
    """

    unit: Unit
    lens: Lens
    stages: frozenset[Stage]
    differentiable: bool

    # A hand-written __init__ so the accepted inputs (strings, one stage)
    # are in the signature, while the fields keep their normalised types.
    def __init__(
        self,
        unit: Unit | str = Unit.FIELD,
        lens: Lens | str = Lens.POINTWISE,
        stages: Stage | str | Iterable[Stage | str] = DEFAULT_STAGES,
        differentiable: bool = False,
    ) -> None:
        normalised = _as_stages(stages)
        if not normalised:
            raise ValueError("a scorer must run at one stage or more")
        object.__setattr__(self, "unit", Unit(unit))
        object.__setattr__(self, "lens", Lens(lens))
        object.__setattr__(self, "stages", normalised)
        object.__setattr__(self, "differentiable", bool(differentiable))

    def runs_at(self, stage: Stage | str) -> bool:
        """Whether the scorer may run at ``stage``."""
        return Stage(stage) in self.stages

runs_at(stage)

Whether the scorer may run at stage.

Source code in packages/pipekit-evaluate/src/pipekit_evaluate/taxonomy.py 
109
110
111
def runs_at(self, stage: Stage | str) -> bool:
    """Whether the scorer may run at ``stage``."""
    return Stage(stage) in self.stages

Stage

Bases: StrEnum

Where in a model's life a score may run.

Source code in packages/pipekit-evaluate/src/pipekit_evaluate/taxonomy.py 
50
51
52
53
54
55
56
class Stage(StrEnum):
    """Where in a model's life a score may run."""

    TRAINING = "training"
    VALIDATION = "validation"
    FINAL = "final"
    MONITORING = "monitoring"

Unit

Bases: StrEnum

What a score is computed on.

Source code in packages/pipekit-evaluate/src/pipekit_evaluate/taxonomy.py 
29
30
31
32
33
34
35
36
class Unit(StrEnum):
    """What a score is computed on."""

    FIELD = "field"
    STATISTIC = "statistic"
    TRAJECTORY = "trajectory"
    EVENT = "event"
    BUDGET = "budget"

tags_of(scorer)

Read a scorer's taxonomy tags, defaulting the ones it does not set.

Parameters:

Name Type Description Default
scorer Any

Any object; the optional attributes unit, lens, stages and differentiable are read when present.

 required

Returns:

Type Description
ScorerTags

The scorer's :class:ScorerTags. A scorer with none of the
ScorerTags

attributes is a point-wise field score for validation and final
ScorerTags

evaluation, and not differentiable.

Raises:

Type Description
ValueError

If an attribute holds a value outside its axis.
Source code in packages/pipekit-evaluate/src/pipekit_evaluate/taxonomy.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def tags_of(scorer: Any) -> ScorerTags:
    """Read a scorer's taxonomy tags, defaulting the ones it does not set.

    Args:
        scorer: Any object; the optional attributes ``unit``, ``lens``,
            ``stages`` and ``differentiable`` are read when present.

    Returns:
        The scorer's :class:`ScorerTags`. A scorer with none of the
        attributes is a point-wise field score for validation and final
        evaluation, and not differentiable.

    Raises:
        ValueError: If an attribute holds a value outside its axis.
    """
    defaults = ScorerTags()
    return ScorerTags(
        unit=getattr(scorer, "unit", defaults.unit),
        lens=getattr(scorer, "lens", defaults.lens),
        stages=getattr(scorer, "stages", defaults.stages),
        differentiable=bool(getattr(scorer, "differentiable", defaults.differentiable)),
    )

Scorers

pipekit_evaluate.scorers

Turn any metric operator into a tagged :class:Scorer.

The metrics live with their carriers (xrtoolz.metrics for xarray, geotoolz for GeoTensors); :class:MetricScorer wraps one, records its place in the Unit x Lens x Stage taxonomy, and returns the named scalar values every scorer returns. Nothing here imports a carrier library: the metric's output is read structurally.

MetricScorer dataclass

A metric operator as a tagged :class:Scorer.

The metric is any callable (prediction, reference) -> result, such as an xrtoolz.metrics operator. An :class:Estimate passed as the prediction or the reference is handed to the metric as from_estimate(estimate), by default its point estimate, so the same scorer serves :func:evaluate-style calls on raw carriers and the benchmark ladder's estimates. A probabilistic scorer has no default: a point would silently drop the uncertainty it is meant to score, so it must say what to take from an estimate (lambda e: e.sample(100)).

The result becomes {name: float}:

  • a Dataset-like result (one with data_vars) gives one entry per data variable;
  • a mapping gives one entry per key;
  • anything else is one value, keyed by its name when it has a string one (a named DataArray) and by "value" otherwise.

Every value must be a scalar with no dimensions, even singleton ones; a metric that keeps a dimension (a score per lead time or per region) must be reduced first. Because the values become Python floats, a :class:MetricScorer is never differentiable; a scorer meant for use as a loss implements :class:Scorer itself.

Attributes:

Name Type Description
metric Callable[[Any, Any], Any]

The metric operator.
name str

The scorer's name; defaults to the metric's __name__ or, failing that, its class name.
unit Unit | str

What the metric scores.
lens Lens | str

What kind of failure it can see.
stages Stage | str | Iterable[Stage | str]

The stages it may run at.
from_estimate Callable[[Any], Any] | None

What to hand the metric for an :class:Estimate; None means its point estimate, which a probabilistic scorer refuses.
references ReferenceSource | str | Iterable[ReferenceSource | str] | None

The benchmark reference sources the scorer is run against, as members or their string values; None means the cell's primary reference only.
Source code in packages/pipekit-evaluate/src/pipekit_evaluate/scorers.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
@dataclass(frozen=True)
class MetricScorer:
    """A metric operator as a tagged :class:`Scorer`.

    The metric is any callable ``(prediction, reference) -> result``, such
    as an ``xrtoolz.metrics`` operator. An :class:`Estimate` passed as the
    prediction or the reference is handed to the metric as
    ``from_estimate(estimate)``, by default its point estimate, so the same
    scorer serves :func:`evaluate`-style calls on raw carriers and the
    benchmark ladder's estimates. A probabilistic scorer has no default: a
    point would silently drop the uncertainty it is meant to score, so it
    must say what to take from an estimate (``lambda e: e.sample(100)``).

    The result becomes ``{name: float}``:

    * a Dataset-like result (one with ``data_vars``) gives one entry per
      data variable;
    * a mapping gives one entry per key;
    * anything else is one value, keyed by its ``name`` when it has a
      string one (a named ``DataArray``) and by ``"value"`` otherwise.

    Every value must be a scalar with no dimensions, even singleton ones; a
    metric that keeps a dimension (a score per lead time or per region)
    must be reduced first. Because the values become Python floats, a
    :class:`MetricScorer` is never differentiable; a scorer meant for use
    as a loss implements :class:`Scorer` itself.

    Attributes:
        metric: The metric operator.
        name: The scorer's name; defaults to the metric's ``__name__`` or,
            failing that, its class name.
        unit: What the metric scores.
        lens: What kind of failure it can see.
        stages: The stages it may run at.
        from_estimate: What to hand the metric for an :class:`Estimate`;
            ``None`` means its point estimate, which a probabilistic scorer
            refuses.
        references: The benchmark reference sources the scorer is run
            against, as members or their string values; ``None`` means the
            cell's primary reference only.
    """

    metric: Callable[[Any, Any], Any]
    name: str = ""
    unit: Unit | str = Unit.FIELD
    lens: Lens | str = Lens.POINTWISE
    stages: Stage | str | Iterable[Stage | str] = DEFAULT_STAGES
    from_estimate: Callable[[Any], Any] | None = None
    references: ReferenceSource | str | Iterable[ReferenceSource | str] | None = None

    def __post_init__(self) -> None:
        if not self.name:
            name = getattr(self.metric, "__name__", type(self.metric).__name__)
            object.__setattr__(self, "name", name)
        # Validate and normalise the tags through the one constructor that
        # owns the rules, so strings become enum members here too.
        tags = ScorerTags(self.unit, self.lens, self.stages)
        object.__setattr__(self, "unit", tags.unit)
        object.__setattr__(self, "lens", tags.lens)
        object.__setattr__(self, "stages", tags.stages)
        if self.references is not None:
            object.__setattr__(self, "references", as_sources(self.references))

    def __call__(self, prediction: Any, reference: Any) -> dict[str, float]:
        """Score ``prediction`` against ``reference``.

        Args:
            prediction: A carrier, or an :class:`Estimate`.
            reference: A carrier, or an :class:`Estimate` (an oracle's
                posterior).

        Returns:
            The metric's values by name.

        Raises:
            ValueError: If an estimate cannot be handed to the metric (a
                probabilistic scorer without ``from_estimate``, or an
                estimate with no point), or a metric value is not a scalar.
        """
        result = self.metric(self._carrier(prediction), self._carrier(reference))
        return _as_scores(result, scorer=self.name)

    def _carrier(self, value: Any) -> Any:
        if not isinstance(value, Estimate):
            return value
        if self.from_estimate is not None:
            return self.from_estimate(value)
        if self.lens is Lens.PROBABILISTIC:
            raise ValueError(
                f"scorer {self.name!r} is probabilistic, so its point estimate "
                "would drop the uncertainty it scores; pass from_estimate= to "
                "say what to take from an Estimate (e.g. lambda e: e.sample(100))"
            )
        point = value.point()
        if point is None:
            raise ValueError(
                f"MetricScorer needs a point estimate, but "
                f"{type(value).__name__}.point() returned None"
            )
        return point

__call__(prediction, reference)

Score prediction against reference.

Parameters:

Name Type Description Default
prediction Any

A carrier, or an :class:Estimate.

 required
reference Any

A carrier, or an :class:Estimate (an oracle's posterior).

 required

Returns:

Type Description
dict[str, float]

The metric's values by name.

Raises:

Type Description
ValueError

If an estimate cannot be handed to the metric (a probabilistic scorer without from_estimate, or an estimate with no point), or a metric value is not a scalar.
Source code in packages/pipekit-evaluate/src/pipekit_evaluate/scorers.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def __call__(self, prediction: Any, reference: Any) -> dict[str, float]:
    """Score ``prediction`` against ``reference``.

    Args:
        prediction: A carrier, or an :class:`Estimate`.
        reference: A carrier, or an :class:`Estimate` (an oracle's
            posterior).

    Returns:
        The metric's values by name.

    Raises:
        ValueError: If an estimate cannot be handed to the metric (a
            probabilistic scorer without ``from_estimate``, or an
            estimate with no point), or a metric value is not a scalar.
    """
    result = self.metric(self._carrier(prediction), self._carrier(reference))
    return _as_scores(result, scorer=self.name)

Evaluation report

pipekit_evaluate.report

The evaluation report: every score of a run, tagged and content-hashed.

:func:evaluate runs a list of scorers on one prediction and one reference and collects their values into an :class:EvaluationReport. Each :class:ReportEntry carries its scorer's place in the Unit x Lens x Stage taxonomy, so a report can be filtered by axis (report.filter(lens="structural")), compared with an earlier one (report.diff(baseline)) and identified by a content hash that ignores when it was made. A benchmark-ladder run becomes the same report through :meth:~pipekit_evaluate.benchmark.BenchmarkReport.to_report, with each entry naming its cell and reference.

EvaluationReport dataclass

The tagged scores of one evaluation run.

Attributes:

Name Type Description
entries tuple[ReportEntry, ...]

One entry per scorer value; keys are unique.
model_ref str | None

What was evaluated (a registry hash, a URI), if known.
dataset_ref str | None

What it was evaluated on, if known.
config Mapping[str, Any]

Free-form settings of the run, kept as a read-only JSON snapshot taken at construction: mappings at every depth are read-only, lists become tuples, and a value that is not JSON (a Path, an object) is stored as its repr. Changing the caller's config afterwards changes neither the report nor its hash.
timestamp datetime

When the report was made (UTC). Not part of :attr:content_hash.
Source code in packages/pipekit-evaluate/src/pipekit_evaluate/report.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
@dataclass(frozen=True)
class EvaluationReport:
    """The tagged scores of one evaluation run.

    Attributes:
        entries: One entry per scorer value; keys are unique.
        model_ref: What was evaluated (a registry hash, a URI), if known.
        dataset_ref: What it was evaluated on, if known.
        config: Free-form settings of the run, kept as a read-only JSON
            snapshot taken at construction: mappings at every depth are
            read-only, lists become tuples, and a value that is not JSON (a
            ``Path``, an object) is stored as its ``repr``. Changing the
            caller's config afterwards changes neither the report nor its
            hash.
        timestamp: When the report was made (UTC). Not part of
            :attr:`content_hash`.
    """

    entries: tuple[ReportEntry, ...]
    model_ref: str | None = None
    dataset_ref: str | None = None
    config: Mapping[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
    _config_json: str = field(init=False, repr=False, compare=False)

    def __post_init__(self) -> None:
        entries = tuple(self.entries)
        seen: set[str] = set()
        dups: set[str] = set()
        for entry in entries:
            if entry.key in seen:
                dups.add(entry.key)
            seen.add(entry.key)
        if dups:
            raise ValueError(
                f"duplicate report entries {sorted(dups)}; scorer names must "
                "be unique within a report"
            )
        object.__setattr__(self, "entries", entries)
        # The canonical JSON is the snapshot: the hash and to_dict read it,
        # so mutating a nested value the caller still holds cannot change a
        # report that has already been hashed.
        # Thawing first lets a frozen config (from filter or replace) hash
        # exactly as the plain config it was made from.
        config_json = stable_json(_thaw(self.config))
        object.__setattr__(self, "_config_json", config_json)
        object.__setattr__(self, "config", _freeze(json.loads(config_json)))

    @property
    def values(self) -> dict[str, float]:
        """Every value by entry key."""
        return {e.key: e.value for e in self.entries}

    @property
    def content_hash(self) -> str:
        """A sha256 over the entries, the references and the config.

        Entry order and the timestamp do not change it, so two runs that
        score the same values on the same references hash equal. An entry
        without a cell or reference hashes as it did before those fields
        existed, so an :func:`evaluate` report keeps its hash.
        """
        payload = {
            "entries": [
                _hash_record(e) for e in sorted(self.entries, key=lambda e: e.key)
            ],
            "model_ref": self.model_ref,
            "dataset_ref": self.dataset_ref,
            "config": json.loads(self._config_json),
        }
        return sha256_hex(stable_json(payload))

    def filter(
        self,
        *,
        unit: Unit | str | None = None,
        lens: Lens | str | None = None,
        stage: Stage | str | None = None,
        scorer: str | None = None,
        cell: str | None = None,
        reference: ReferenceSource | str | None = None,
    ) -> EvaluationReport:
        """The entries that match every given axis value.

        Args:
            unit: Keep entries of this unit.
            lens: Keep entries of this lens.
            stage: Keep entries whose scorer runs at this stage.
            scorer: Keep entries of the scorer with this name.
            cell: Keep entries of this benchmark cell label.
            reference: Keep entries scored against this reference source.

        Returns:
            A report with the matching entries and this report's
            references, config and timestamp.
        """
        unit = None if unit is None else Unit(unit)
        lens = None if lens is None else Lens(lens)
        stage = None if stage is None else Stage(stage)
        reference = None if reference is None else ReferenceSource(reference).value
        kept = tuple(
            e
            for e in self.entries
            if (unit is None or e.tags.unit is unit)
            and (lens is None or e.tags.lens is lens)
            and (stage is None or e.tags.runs_at(stage))
            and (scorer is None or e.scorer == scorer)
            and (cell is None or e.cell == cell)
            and (reference is None or e.reference == reference)
        )
        return replace(self, entries=kept)

    def diff(
        self, baseline: EvaluationReport
    ) -> dict[str, tuple[float | None, float | None]]:
        """The entries whose value differs from ``baseline``'s.

        Args:
            baseline: The report to compare against, such as last week's.

        Returns:
            ``{key: (baseline value, this value)}`` for every key whose
            value changed, with ``None`` on the side that lacks the key.
            Two NaNs count as equal, so a report's diff against itself is
            empty.
        """
        before, after = baseline.values, self.values
        changes: dict[str, tuple[float | None, float | None]] = {}
        for key in sorted(before.keys() | after.keys()):
            old, new = before.get(key), after.get(key)
            if not _same(old, new):
                changes[key] = (old, new)
        return changes

    def to_records(self) -> list[dict[str, Any]]:
        """One flat record per entry, in entry order."""
        return [e.to_record() for e in self.entries]

    def to_dict(self) -> dict[str, Any]:
        """A strict-JSON form, with the content hash.

        A non-finite float (in a value or in the config) is written as the
        string ``"NaN"``, ``"Infinity"`` or ``"-Infinity"``, so the result
        passes ``json.dumps(..., allow_nan=False)``; :meth:`to_records` and
        :meth:`to_pandas` keep the floats.
        """
        return _strict_json(
            {
                "entries": self.to_records(),
                "model_ref": self.model_ref,
                "dataset_ref": self.dataset_ref,
                "config": json.loads(self._config_json),
                "timestamp": self.timestamp.isoformat(),
                "content_hash": self.content_hash,
            }
        )

    def to_pandas(self) -> Any:
        """One row per entry, as a ``pandas.DataFrame``.

        Raises:
            ImportError: If pandas is not installed.
        """
        try:
            pd = importlib.import_module("pandas")
        except ImportError as err:
            raise ImportError(
                "EvaluationReport.to_pandas needs pandas; install it with "
                "`pip install pandas`"
            ) from err
        return pd.DataFrame.from_records(self.to_records())

content_hash property

A sha256 over the entries, the references and the config.

Entry order and the timestamp do not change it, so two runs that score the same values on the same references hash equal. An entry without a cell or reference hashes as it did before those fields existed, so an :func:evaluate report keeps its hash.

values property

Every value by entry key.

diff(baseline)

The entries whose value differs from baseline's.

Parameters:

Name Type Description Default
baseline EvaluationReport

The report to compare against, such as last week's.

 required

Returns:

Type Description
dict[str, tuple[float | None, float | None]]

{key: (baseline value, this value)} for every key whose
dict[str, tuple[float | None, float | None]]

value changed, with None on the side that lacks the key.
dict[str, tuple[float | None, float | None]]

Two NaNs count as equal, so a report's diff against itself is
dict[str, tuple[float | None, float | None]]

empty.
Source code in packages/pipekit-evaluate/src/pipekit_evaluate/report.py 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
def diff(
    self, baseline: EvaluationReport
) -> dict[str, tuple[float | None, float | None]]:
    """The entries whose value differs from ``baseline``'s.

    Args:
        baseline: The report to compare against, such as last week's.

    Returns:
        ``{key: (baseline value, this value)}`` for every key whose
        value changed, with ``None`` on the side that lacks the key.
        Two NaNs count as equal, so a report's diff against itself is
        empty.
    """
    before, after = baseline.values, self.values
    changes: dict[str, tuple[float | None, float | None]] = {}
    for key in sorted(before.keys() | after.keys()):
        old, new = before.get(key), after.get(key)
        if not _same(old, new):
            changes[key] = (old, new)
    return changes

filter(*, unit=None, lens=None, stage=None, scorer=None, cell=None, reference=None)

The entries that match every given axis value.

Parameters:

Name Type Description Default
unit Unit | str | None

Keep entries of this unit.

 None
lens Lens | str | None

Keep entries of this lens.

 None
stage Stage | str | None

Keep entries whose scorer runs at this stage.

 None
scorer str | None

Keep entries of the scorer with this name.

 None
cell str | None

Keep entries of this benchmark cell label.

 None
reference ReferenceSource | str | None

Keep entries scored against this reference source.

 None

Returns:

Type Description
EvaluationReport

A report with the matching entries and this report's
EvaluationReport

references, config and timestamp.
Source code in packages/pipekit-evaluate/src/pipekit_evaluate/report.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def filter(
    self,
    *,
    unit: Unit | str | None = None,
    lens: Lens | str | None = None,
    stage: Stage | str | None = None,
    scorer: str | None = None,
    cell: str | None = None,
    reference: ReferenceSource | str | None = None,
) -> EvaluationReport:
    """The entries that match every given axis value.

    Args:
        unit: Keep entries of this unit.
        lens: Keep entries of this lens.
        stage: Keep entries whose scorer runs at this stage.
        scorer: Keep entries of the scorer with this name.
        cell: Keep entries of this benchmark cell label.
        reference: Keep entries scored against this reference source.

    Returns:
        A report with the matching entries and this report's
        references, config and timestamp.
    """
    unit = None if unit is None else Unit(unit)
    lens = None if lens is None else Lens(lens)
    stage = None if stage is None else Stage(stage)
    reference = None if reference is None else ReferenceSource(reference).value
    kept = tuple(
        e
        for e in self.entries
        if (unit is None or e.tags.unit is unit)
        and (lens is None or e.tags.lens is lens)
        and (stage is None or e.tags.runs_at(stage))
        and (scorer is None or e.scorer == scorer)
        and (cell is None or e.cell == cell)
        and (reference is None or e.reference == reference)
    )
    return replace(self, entries=kept)

to_dict()

A strict-JSON form, with the content hash.

A non-finite float (in a value or in the config) is written as the string "NaN", "Infinity" or "-Infinity", so the result passes json.dumps(..., allow_nan=False); :meth:to_records and :meth:to_pandas keep the floats.

Source code in packages/pipekit-evaluate/src/pipekit_evaluate/report.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
def to_dict(self) -> dict[str, Any]:
    """A strict-JSON form, with the content hash.

    A non-finite float (in a value or in the config) is written as the
    string ``"NaN"``, ``"Infinity"`` or ``"-Infinity"``, so the result
    passes ``json.dumps(..., allow_nan=False)``; :meth:`to_records` and
    :meth:`to_pandas` keep the floats.
    """
    return _strict_json(
        {
            "entries": self.to_records(),
            "model_ref": self.model_ref,
            "dataset_ref": self.dataset_ref,
            "config": json.loads(self._config_json),
            "timestamp": self.timestamp.isoformat(),
            "content_hash": self.content_hash,
        }
    )

to_pandas()

One row per entry, as a pandas.DataFrame.

Raises:

Type Description
ImportError

If pandas is not installed.
Source code in packages/pipekit-evaluate/src/pipekit_evaluate/report.py 
246
247
248
249
250
251
252
253
254
255
256
257
258
259
def to_pandas(self) -> Any:
    """One row per entry, as a ``pandas.DataFrame``.

    Raises:
        ImportError: If pandas is not installed.
    """
    try:
        pd = importlib.import_module("pandas")
    except ImportError as err:
        raise ImportError(
            "EvaluationReport.to_pandas needs pandas; install it with "
            "`pip install pandas`"
        ) from err
    return pd.DataFrame.from_records(self.to_records())

to_records()

One flat record per entry, in entry order.

Source code in packages/pipekit-evaluate/src/pipekit_evaluate/report.py 
223
224
225
def to_records(self) -> list[dict[str, Any]]:
    """One flat record per entry, in entry order."""
    return [e.to_record() for e in self.entries]

ReportEntry dataclass

One named value from one scorer.

Attributes:

Name Type Description
scorer str

The scorer's name; it may not contain . or :, which separate the parts of :attr:key.
metric str

The value's name within the scorer's output.
value float

The value.
tags ScorerTags

The scorer's taxonomy tags.
cell str | None

The benchmark cell label ("L4/r5/c0") for a ladder run.
reference str | None

The reference source ("known_theta") for a ladder run.
Source code in packages/pipekit-evaluate/src/pipekit_evaluate/report.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@dataclass(frozen=True)
class ReportEntry:
    """One named value from one scorer.

    Attributes:
        scorer: The scorer's name; it may not contain ``.`` or ``:``, which
            separate the parts of :attr:`key`.
        metric: The value's name within the scorer's output.
        value: The value.
        tags: The scorer's taxonomy tags.
        cell: The benchmark cell label (``"L4/r5/c0"``) for a ladder run.
        reference: The reference source (``"known_theta"``) for a ladder
            run.
    """

    scorer: str
    metric: str
    value: float
    tags: ScorerTags = field(default_factory=ScorerTags)
    cell: str | None = None
    reference: str | None = None

    def __post_init__(self) -> None:
        _check_name(self.scorer)
        value = float(self.value)
        # -0.0 == 0.0, so diff() sees no change between them; storing +0.0
        # keeps the content hash in agreement with diff().
        object.__setattr__(self, "value", 0.0 if value == 0.0 else value)

    @property
    def key(self) -> str:
        """The entry's identifier in its report.

        ``"<scorer>.<metric>"``, prefixed by ``"<cell>:<reference>:"`` for a
        ladder run, as in ``"L4/r5/c0:known_theta:rmse.value"``. Unambiguous
        because neither a cell label, a reference source nor a scorer name
        has a ``:``, and a scorer name has no ``.``: the scorer ends at the
        first dot after the prefix, and the metric name may contain more.
        """
        parts = [p for p in (self.cell, self.reference) if p is not None]
        return ":".join([*parts, f"{self.scorer}.{self.metric}"])

    def to_record(self) -> dict[str, Any]:
        """A flat, JSON-serialisable record of the entry."""
        return {
            "scorer": self.scorer,
            "metric": self.metric,
            "value": self.value,
            "unit": self.tags.unit.value,
            "lens": self.tags.lens.value,
            "stages": sorted(s.value for s in self.tags.stages),
            "differentiable": self.tags.differentiable,
            "cell": self.cell,
            "reference": self.reference,
        }

key property

The entry's identifier in its report.

"<scorer>.<metric>", prefixed by "<cell>:<reference>:" for a ladder run, as in "L4/r5/c0:known_theta:rmse.value". Unambiguous because neither a cell label, a reference source nor a scorer name has a :, and a scorer name has no .: the scorer ends at the first dot after the prefix, and the metric name may contain more.

to_record()

A flat, JSON-serialisable record of the entry.

Source code in packages/pipekit-evaluate/src/pipekit_evaluate/report.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
def to_record(self) -> dict[str, Any]:
    """A flat, JSON-serialisable record of the entry."""
    return {
        "scorer": self.scorer,
        "metric": self.metric,
        "value": self.value,
        "unit": self.tags.unit.value,
        "lens": self.tags.lens.value,
        "stages": sorted(s.value for s in self.tags.stages),
        "differentiable": self.tags.differentiable,
        "cell": self.cell,
        "reference": self.reference,
    }

evaluate(prediction, reference, scorers, *, stage=None, model_ref=None, dataset_ref=None, config=None)

Score one prediction against one reference with every scorer.

Parameters:

Name Type Description Default
prediction Any

What is evaluated, in whatever carrier the scorers take.

 required
reference Any

What it is compared with.

 required
scorers Iterable[Scorer]

The scorers; their names must be unique.

 required
stage Stage | str | None

Run only the scorers tagged for this stage; all of them when None.

 None
model_ref str | None

What was evaluated, recorded on the report.

 None
dataset_ref str | None

What it was evaluated on, recorded on the report.

 None
config Mapping[str, Any] | None

Settings recorded on the report and in its hash.

 None

Returns:

Type Description
EvaluationReport

The report, one entry per value each scorer returned.

Raises:

Type Description
ValueError

If two scorers share a name, or a name contains . or :; checked before any scorer runs, because either would let two scorers' values share a key.
Source code in packages/pipekit-evaluate/src/pipekit_evaluate/report.py 
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
def evaluate(
    prediction: Any,
    reference: Any,
    scorers: Iterable[Scorer],
    *,
    stage: Stage | str | None = None,
    model_ref: str | None = None,
    dataset_ref: str | None = None,
    config: Mapping[str, Any] | None = None,
) -> EvaluationReport:
    """Score one prediction against one reference with every scorer.

    Args:
        prediction: What is evaluated, in whatever carrier the scorers take.
        reference: What it is compared with.
        scorers: The scorers; their names must be unique.
        stage: Run only the scorers tagged for this stage; all of them
            when ``None``.
        model_ref: What was evaluated, recorded on the report.
        dataset_ref: What it was evaluated on, recorded on the report.
        config: Settings recorded on the report and in its hash.

    Returns:
        The report, one entry per value each scorer returned.

    Raises:
        ValueError: If two scorers share a name, or a name contains ``.`` or
            ``:``; checked before any scorer runs, because either would let
            two scorers' values share a key.
    """
    scorers = list(scorers)
    _check_scorer_names(scorers)
    stage = None if stage is None else Stage(stage)
    entries: list[ReportEntry] = []
    for scorer in scorers:
        tags = tags_of(scorer)
        if stage is not None and not tags.runs_at(stage):
            continue
        for metric, value in scorer(prediction, reference).items():
            entries.append(ReportEntry(scorer.name, metric, value, tags))
    return EvaluationReport(
        tuple(entries),
        model_ref=model_ref,
        dataset_ref=dataset_ref,
        config=config or {},
    )

Protocols

pipekit_evaluate.benchmark.protocols

Carrier-agnostic protocols for the benchmark ladder.

These mirror the style of :mod:pipekit_cycle.protocols: small, runtime-checkable structural interfaces that domain packages (plumax, somax, xtremax, …) satisfy without importing pipekit-evaluate. They are the orchestration layer described in packages/pipekit-evaluate/docs/design/benchmark_ladder.md §4 — the "who plays which rung" contracts — and sit on top of (not inside) the Unit x Lens x Stage scorer taxonomy (:mod:pipekit_evaluate.taxonomy).

A benchmark Task reuses pipekit_cycle's ForwardModel / ObservationOperator as its simulator and observation operator, so a benchmark is assembled from the same parts a DACycle is. Those types are referenced under :data:typing.TYPE_CHECKING only, so importing this module never pulls in pipekit-cycle.

CovarianceEstimate

Bases: Estimate, Protocol

An :class:Estimate that also reports its posterior covariance.

A separate protocol rather than a member of :class:Estimate, so every existing estimate keeps satisfying :class:Estimate. A scorer that needs second moments (Gaussian coverage, a Mahalanobis distance, a spread-skill ratio) checks isinstance(estimate, CovarianceEstimate) instead of probing attributes or drawing samples.

Members

covariance(): The posterior covariance of :meth:point's quantity, as a (d, d) matrix for a (d,) point.

Source code in packages/pipekit-evaluate/src/pipekit_evaluate/benchmark/protocols.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@runtime_checkable
class CovarianceEstimate(Estimate, Protocol):
    """An :class:`Estimate` that also reports its posterior covariance.

    A separate protocol rather than a member of :class:`Estimate`, so every
    existing estimate keeps satisfying :class:`Estimate`. A scorer that needs
    second moments (Gaussian coverage, a Mahalanobis distance, a spread-skill
    ratio) checks ``isinstance(estimate, CovarianceEstimate)`` instead of
    probing attributes or drawing samples.

    Members:
        covariance(): The posterior covariance of :meth:`point`'s quantity,
            as a ``(d, d)`` matrix for a ``(d,)`` point.
    """

    def covariance(self) -> Any: ...

Estimate

Bases: Protocol

A point and/or a posterior produced by an estimator.

Scorers branch on what is present: a point-only estimate returns None from :meth:sample. Per benchmark_ladder.md §8 decision 1, :meth:sample is the required posterior surface and :meth:log_prob is optional.

Members

point(): Best point estimate (mean / MAP / MLE), or None. sample(n, key): n posterior draws, or None if point-only. log_prob(theta): Optional posterior log-density at theta.

Source code in packages/pipekit-evaluate/src/pipekit_evaluate/benchmark/protocols.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
@runtime_checkable
class Estimate(Protocol):
    """A point and/or a posterior produced by an estimator.

    Scorers branch on what is present: a point-only estimate returns
    ``None`` from :meth:`sample`. Per ``benchmark_ladder.md`` §8 decision
    1, :meth:`sample` is the required posterior surface and
    :meth:`log_prob` is optional.

    Members:
        point(): Best point estimate (mean / MAP / MLE), or ``None``.
        sample(n, key): ``n`` posterior draws, or ``None`` if point-only.
        log_prob(theta): Optional posterior log-density at ``theta``.
    """

    def point(self) -> Any: ...

    def sample(self, n: int, *, key: Any = None) -> Any: ...

    def log_prob(self, theta: Any) -> Any: ...

Estimator

Bases: Protocol

Any rung that maps observations → an :class:Estimate.

The same protocol covers the model-based oracle (rung 2), emulator-based inference (rung 4) and the amortized predictor (rung 5). rung locates the estimator on the staged axis (§2.2) and complexity locates it on the per-rung complexity axis (§2.3).

Members

rung: Position on the staged ladder (1-6). complexity: Position on the per-rung complexity axis (0 = simplest). Two estimators sharing a rung but differing in complexity are the cheap/rich pair scored against each other. fit(task): Return a fitted estimator (a no-op return self for an already-trained or closed-form rung-2 oracle). call(obs): Produce an :class:Estimate from observations.

Source code in packages/pipekit-evaluate/src/pipekit_evaluate/benchmark/protocols.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
@runtime_checkable
class Estimator(Protocol):
    """Any rung that maps observations → an :class:`Estimate`.

    The same protocol covers the model-based oracle (rung 2), emulator-based
    inference (rung 4) and the amortized predictor (rung 5). ``rung``
    locates the estimator on the staged axis (§2.2) and ``complexity``
    locates it on the per-rung complexity axis (§2.3).

    Members:
        rung: Position on the staged ladder (1-6).
        complexity: Position on the per-rung complexity axis (``0`` =
            simplest). Two estimators sharing a ``rung`` but differing in
            ``complexity`` are the cheap/rich pair scored against each other.
        fit(task): Return a fitted estimator (a no-op ``return self`` for an
            already-trained or closed-form rung-2 oracle).
        __call__(obs): Produce an :class:`Estimate` from observations.
    """

    rung: int
    complexity: int

    def fit(self, task: Task) -> Estimator: ...

    def __call__(self, obs: Any) -> Estimate: ...

Oracle

Bases: Protocol

The designated reference :class:Estimator for a task (usually rung 2).

Not yet consumed by Benchmark — reference selection currently lives inline in Benchmark._oracle_prediction; this seam exists for the planned scorer taxonomy, where reference policy becomes pluggable.

Members

reference_for(task): Return the estimator whose output is the reference posterior for the fast rungs of task.

Source code in packages/pipekit-evaluate/src/pipekit_evaluate/benchmark/protocols.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
@runtime_checkable
class Oracle(Protocol):
    """The designated reference :class:`Estimator` for a task (usually rung 2).

    Not yet consumed by `Benchmark` — reference selection currently
    lives inline in ``Benchmark._oracle_prediction``; this seam exists
    for the planned scorer taxonomy, where reference policy becomes
    pluggable.

    Members:
        reference_for(task): Return the estimator whose output is the
            reference posterior for the fast rungs of ``task``.
    """

    def reference_for(self, task: Task) -> Estimator: ...

Prior

Bases: Protocol

A minimal prior over task parameters θ.

Deliberately thin (see benchmark_ladder.md §8, decision 2): a sample surface with an optional log_prob, so neither NumPyro nor distrax is a hard dependency of the framework core.

Members

sample(n, key): Draw n samples of θ. log_prob(theta): Optional log-density at theta; implementations without a tractable density may raise NotImplementedError.

Source code in packages/pipekit-evaluate/src/pipekit_evaluate/benchmark/protocols.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
@runtime_checkable
class Prior(Protocol):
    """A minimal prior over task parameters ``θ``.

    Deliberately thin (see ``benchmark_ladder.md`` §8, decision 2): a
    ``sample`` surface with an optional ``log_prob``, so neither NumPyro
    nor ``distrax`` is a hard dependency of the framework core.

    Members:
        sample(n, key): Draw ``n`` samples of ``θ``.
        log_prob(theta): Optional log-density at ``theta``; implementations
            without a tractable density may raise ``NotImplementedError``.
    """

    def sample(self, n: int, *, key: Any = None) -> Any: ...

    def log_prob(self, theta: Any) -> Any: ...

Scorer

Bases: Protocol

Score one prediction against one reference.

This is the seam onto the Unit x Lens x Stage taxonomy: a concrete scorer is a (Unit, Lens) pair. The benchmark layer only requires a name and a callable that returns a mapping of named metric values. A scorer places itself in the taxonomy through the optional attributes unit, lens, stages and differentiable, which are not protocol members; :func:pipekit_evaluate.taxonomy.tags_of reads them with their defaults, and :class:pipekit_evaluate.scorers.MetricScorer sets them on any metric operator. A further optional attribute, references, names the reference sources a scorer can be run against in the benchmark ladder; without it a scorer is run against each cell's primary reference only (:func:pipekit_evaluate.benchmark.run.scores_against).

Members

name: Short scorer identifier, used to namespace its metrics. call(prediction, reference): Named metric values.

Source code in packages/pipekit-evaluate/src/pipekit_evaluate/benchmark/protocols.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
@runtime_checkable
class Scorer(Protocol):
    """Score one prediction against one reference.

    This is the seam onto the ``Unit x Lens x Stage`` taxonomy: a concrete
    scorer *is* a ``(Unit, Lens)`` pair. The benchmark layer only requires
    a name and a callable that returns a mapping of named metric values.
    A scorer places itself in the taxonomy through the optional attributes
    ``unit``, ``lens``, ``stages`` and ``differentiable``, which are not
    protocol members; :func:`pipekit_evaluate.taxonomy.tags_of` reads them
    with their defaults, and :class:`pipekit_evaluate.scorers.MetricScorer`
    sets them on any metric operator. A further optional attribute,
    ``references``, names the reference sources a scorer can be run
    against in the benchmark ladder; without it a scorer is run against
    each cell's primary reference only
    (:func:`pipekit_evaluate.benchmark.run.scores_against`).

    Members:
        name: Short scorer identifier, used to namespace its metrics.
        __call__(prediction, reference): Named metric values.
    """

    name: str

    def __call__(self, prediction: Estimate, reference: Any) -> Mapping[str, float]: ...

Task

Bases: Protocol

One benchmark scenario: prior + simulator + obs operator + data.

The simulator and observation operator are exactly the pipekit_cycle protocols, so no parallel hierarchy is introduced.

Members

name: Human-readable task identifier. domain: Domain tag ("plume" | "ocean" | "land" | …). prior(): The prior over task parameters θ (rung 1 input). simulator(): The rung-1 generative ForwardModel (θ → state). observation_operator(): The state → observations map. datasets(): The L0..L4 products by key, plus any known truth (e.g. "theta", "state", "obs") used as references.

Source code in packages/pipekit-evaluate/src/pipekit_evaluate/benchmark/protocols.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@runtime_checkable
class Task(Protocol):
    """One benchmark scenario: prior + simulator + obs operator + data.

    The simulator and observation operator are *exactly* the
    ``pipekit_cycle`` protocols, so no parallel hierarchy is introduced.

    Members:
        name: Human-readable task identifier.
        domain: Domain tag (``"plume"`` | ``"ocean"`` | ``"land"`` | …).
        prior(): The prior over task parameters ``θ`` (rung 1 input).
        simulator(): The rung-1 generative ``ForwardModel`` (``θ → state``).
        observation_operator(): The ``state → observations`` map.
        datasets(): The ``L0..L4`` products by key, plus any known truth
            (e.g. ``"theta"``, ``"state"``, ``"obs"``) used as references.
    """

    name: str
    domain: str

    def prior(self) -> Prior: ...

    def simulator(self) -> ForwardModel: ...

    def observation_operator(self) -> ObservationOperator: ...

    def datasets(self) -> Mapping[str, Any]: ...

The cube

pipekit_evaluate.benchmark.cube

The level x rung x complexity cube.

A benchmark cell is a (data level, inference rung, complexity step) triple — the single mental model of benchmark_ladder.md §2.4. The earlier level x rung matrix is one face of the cube, read at a fixed complexity step.

Cell dataclass

One cell of the benchmark cube.

Attributes:

Name Type Description
level str

The data-hierarchy level ("L0" … "L4").
rung int

The staged-ladder rung (1-6).
complexity int

The per-rung complexity step (0 = simplest).
Source code in packages/pipekit-evaluate/src/pipekit_evaluate/benchmark/cube.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
@dataclass(frozen=True)
class Cell:
    """One cell of the benchmark cube.

    Attributes:
        level: The data-hierarchy level (``"L0"`` … ``"L4"``).
        rung: The staged-ladder rung (1-6).
        complexity: The per-rung complexity step (``0`` = simplest).
    """

    level: str
    rung: int
    complexity: int = 0

    def __post_init__(self) -> None:
        if self.level not in LEVEL_NAMES:
            raise ValueError(
                f"unknown level {self.level!r}; expected one of {sorted(LEVEL_NAMES)}"
            )
        if self.rung not in RUNG_NAMES:
            raise ValueError(
                f"unknown rung {self.rung!r}; expected one of {sorted(RUNG_NAMES)}"
            )
        if self.complexity < 0:
            raise ValueError(f"complexity must be >= 0, got {self.complexity}")

    @property
    def rung_name(self) -> str:
        """The human-readable name of :attr:`rung`."""
        return RUNG_NAMES[self.rung]

    @property
    def label(self) -> str:
        """A compact ``level/rung/complexity`` string key for reports."""
        return f"{self.level}/r{self.rung}/c{self.complexity}"

label property

A compact level/rung/complexity string key for reports.

rung_name property

The human-readable name of :attr:rung.

Reference rule

pipekit_evaluate.benchmark.reference

The "simpler thing is ground truth" rule, encoded as data.

The single invariant that makes the ladder a benchmark and not five scripts (benchmark_ladder.md §4.3) runs along both axes of §2:

  • the staged axis — the previous rung is the reference;
  • the complexity axis — the adjacent (simpler) step is the reference.

Both are data, not code branches: the default tables below live on the :class:ReferenceRule, which a domain may override (e.g. an oracle-less domain falls back to known-θ coverage only). Beside each rung's primary reference, a rung may name further references it is also scored against (:data:DEFAULT_ALSO): the amortized rung is judged against the oracle's posterior and, in the same run, against the known truth.

Axis

Bases: Enum

Which benchmark axis a reference is resolved along.

Source code in packages/pipekit-evaluate/src/pipekit_evaluate/benchmark/reference.py 
24
25
26
27
28
class Axis(Enum):
    """Which benchmark axis a reference is resolved along."""

    STAGED = "staged"  # §2.2 inference ladder
    COMPLEXITY = "complexity"  # §2.3 per-rung depth

ReferenceRule dataclass

Resolve the reference source for a rung along a given axis.

Attributes:

Name Type Description
staged Mapping[int, ReferenceSource]

Per-rung primary reference sources on the staged axis.
complexity Mapping[int, ReferenceSource]

Per-rung primary reference sources on the complexity axis.
also Mapping[int, tuple[ReferenceSource, ...]]

Per-rung further reference sources, scored beside the primary one on either axis whenever the primary is not :attr:ReferenceSource.NONE.

A complexity step of 0 is the simplest step on its rung and has no simpler neighbour, so it is resolved as :attr:ReferenceSource.NONE on the complexity axis regardless of the table.

Source code in packages/pipekit-evaluate/src/pipekit_evaluate/benchmark/reference.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
@dataclass(frozen=True)
class ReferenceRule:
    """Resolve the reference source for a rung along a given axis.

    Attributes:
        staged: Per-rung primary reference sources on the staged axis.
        complexity: Per-rung primary reference sources on the complexity
            axis.
        also: Per-rung further reference sources, scored beside the
            primary one on either axis whenever the primary is not
            :attr:`ReferenceSource.NONE`.

    A complexity step of ``0`` is the simplest step on its rung and has no
    simpler neighbour, so it is resolved as :attr:`ReferenceSource.NONE`
    on the complexity axis regardless of the table.
    """

    staged: Mapping[int, ReferenceSource] = field(
        default_factory=lambda: dict(DEFAULT_STAGED)
    )
    complexity: Mapping[int, ReferenceSource] = field(
        default_factory=lambda: dict(DEFAULT_COMPLEXITY)
    )
    also: Mapping[int, tuple[ReferenceSource, ...]] = field(
        default_factory=lambda: dict(DEFAULT_ALSO)
    )

    def source(
        self, rung: int, *, axis: Axis = Axis.STAGED, complexity: int = 0
    ) -> ReferenceSource:
        """Return the reference source for ``rung`` along ``axis``.

        Args:
            rung: The staged-ladder rung (1-6).
            axis: Which axis to resolve the reference along.
            complexity: The complexity step; only consulted for
                :attr:`Axis.COMPLEXITY`, where step ``0`` has no simpler
                neighbour.

        Returns:
            The :class:`ReferenceSource` for that rung/step.
        """
        if axis is Axis.COMPLEXITY:
            if complexity <= 0:
                return ReferenceSource.NONE
            return self.complexity.get(rung, ReferenceSource.SIMPLER_STEP)
        return self.staged.get(rung, ReferenceSource.NONE)

    def sources(
        self, rung: int, *, axis: Axis = Axis.STAGED, complexity: int = 0
    ) -> tuple[ReferenceSource, ...]:
        """Every reference source for ``rung`` along ``axis``, primary first.

        Args:
            rung: The staged-ladder rung (1-6).
            axis: Which axis to resolve the primary reference along.
            complexity: The complexity step, as for :meth:`source`.

        Returns:
            The primary source followed by the rung's :attr:`also` sources,
            without repeats; empty when the primary source is
            :attr:`ReferenceSource.NONE`, because an unscored cell has no
            further references either.
        """
        primary = self.source(rung, axis=axis, complexity=complexity)
        if primary is ReferenceSource.NONE:
            return ()
        extra = [s for s in self.also.get(rung, ()) if s is not primary]
        return (primary, *dict.fromkeys(extra))

source(rung, *, axis=Axis.STAGED, complexity=0)

Return the reference source for rung along axis.

Parameters:

Name Type Description Default
rung int

The staged-ladder rung (1-6).

 required
axis Axis

Which axis to resolve the reference along.

 STAGED
complexity int

The complexity step; only consulted for :attr:Axis.COMPLEXITY, where step 0 has no simpler neighbour.

 0

Returns:

Name Type Description
The ReferenceSource

class:ReferenceSource for that rung/step.
Source code in packages/pipekit-evaluate/src/pipekit_evaluate/benchmark/reference.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def source(
    self, rung: int, *, axis: Axis = Axis.STAGED, complexity: int = 0
) -> ReferenceSource:
    """Return the reference source for ``rung`` along ``axis``.

    Args:
        rung: The staged-ladder rung (1-6).
        axis: Which axis to resolve the reference along.
        complexity: The complexity step; only consulted for
            :attr:`Axis.COMPLEXITY`, where step ``0`` has no simpler
            neighbour.

    Returns:
        The :class:`ReferenceSource` for that rung/step.
    """
    if axis is Axis.COMPLEXITY:
        if complexity <= 0:
            return ReferenceSource.NONE
        return self.complexity.get(rung, ReferenceSource.SIMPLER_STEP)
    return self.staged.get(rung, ReferenceSource.NONE)

sources(rung, *, axis=Axis.STAGED, complexity=0)

Every reference source for rung along axis, primary first.

Parameters:

Name Type Description Default
rung int

The staged-ladder rung (1-6).

 required
axis Axis

Which axis to resolve the primary reference along.

 STAGED
complexity int

The complexity step, as for :meth:source.

 0

Returns:

Type Description
ReferenceSource

The primary source followed by the rung's :attr:also sources,
...

without repeats; empty when the primary source is
tuple[ReferenceSource, ...]

attr:ReferenceSource.NONE, because an unscored cell has no
tuple[ReferenceSource, ...]

further references either.
Source code in packages/pipekit-evaluate/src/pipekit_evaluate/benchmark/reference.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def sources(
    self, rung: int, *, axis: Axis = Axis.STAGED, complexity: int = 0
) -> tuple[ReferenceSource, ...]:
    """Every reference source for ``rung`` along ``axis``, primary first.

    Args:
        rung: The staged-ladder rung (1-6).
        axis: Which axis to resolve the primary reference along.
        complexity: The complexity step, as for :meth:`source`.

    Returns:
        The primary source followed by the rung's :attr:`also` sources,
        without repeats; empty when the primary source is
        :attr:`ReferenceSource.NONE`, because an unscored cell has no
        further references either.
    """
    primary = self.source(rung, axis=axis, complexity=complexity)
    if primary is ReferenceSource.NONE:
        return ()
    extra = [s for s in self.also.get(rung, ()) if s is not primary]
    return (primary, *dict.fromkeys(extra))

ReferenceSource

Bases: Enum

Where a rung/step's ground truth comes from.

Source code in packages/pipekit-evaluate/src/pipekit_evaluate/benchmark/reference.py 
31
32
33
34
35
36
37
38
39
40
class ReferenceSource(Enum):
    """Where a rung/step's ground truth comes from."""

    NONE = "none"  # generative rung; not scored against a reference
    KNOWN_THETA = "known_theta"  # synthetic truth from the simulator
    SIMULATOR_OUTPUT = "simulator_output"  # rung-1 forward outputs
    ORACLE_POSTERIOR = "oracle_posterior"  # rung-2 oracle posterior
    PREVIOUS_RUNG = "previous_rung"  # the staged step just below
    SIMPLER_STEP = "simpler_step"  # the complexity step just below
    BEST_VALIDATED_BELOW = "best_validated_below"  # for rung 6 "improve"

as_sources(sources)

Normalise one reference source or several to a set of members.

Parameters:

Name Type Description Default
sources ReferenceSource | str | Iterable[ReferenceSource | str]

A :class:ReferenceSource, its string value, or an iterable of either.

 required

Returns:

Type Description
frozenset[ReferenceSource]

The sources as a frozenset of members.

Raises:

Type Description
ValueError

If a string is not a :class:ReferenceSource value.
Source code in packages/pipekit-evaluate/src/pipekit_evaluate/benchmark/reference.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def as_sources(
    sources: ReferenceSource | str | Iterable[ReferenceSource | str],
) -> frozenset[ReferenceSource]:
    """Normalise one reference source or several to a set of members.

    Args:
        sources: A :class:`ReferenceSource`, its string value, or an
            iterable of either.

    Returns:
        The sources as a frozenset of members.

    Raises:
        ValueError: If a string is not a :class:`ReferenceSource` value.
    """
    # A bare string is one source, not an iterable of characters.
    if isinstance(sources, ReferenceSource | str):
        sources = (sources,)
    return frozenset(ReferenceSource(s) for s in sources)

Benchmark runner

pipekit_evaluate.benchmark.run

The benchmark runner: fill cube cells, resolve references, score.

:class:Benchmark realises benchmark_ladder.md §4.2 (the Benchmark dataclass) and §4.3 (the reference rule). It is deliberately carrier- and scorer-agnostic: it fits each estimator, resolves the reference for that cell via :class:~pipekit_evaluate.benchmark.reference.ReferenceRule, and delegates the actual critique to the supplied :class:Scorers. Its :class:BenchmarkReport converts into the tagged, content-hashed :class:~pipekit_evaluate.report.EvaluationReport that a plain :func:~pipekit_evaluate.report.evaluate call returns. Writing a pipekit_experiment.Run per cell is left to a later slice.

Benchmark dataclass

Run the level x rung x complexity cube and score each cell.

Attributes:

Name Type Description
task Task

The scenario under test.
estimators Sequence[Estimator]

The rungs/steps to fill in (each carries rung and complexity).
scorers Sequence[Scorer]

The Unit x Lens critiques applied to every scored cell.
level str

The data-hierarchy level the run sits at (default "L4").
axis Axis

Which axis to resolve references along (default staged).
reference_rule ReferenceRule

The "simpler thing is ground truth" tables.
reference Estimator | None

An explicit oracle estimator; when None the rung-2 estimator among estimators with the highest complexity (the richest kernel, such as NUTS rather than OI) is used.
obs_key, (truth_key, state_key)

Keys into task.datasets() for the observations fed to estimators and the known-θ / simulator-state references.
Source code in packages/pipekit-evaluate/src/pipekit_evaluate/benchmark/run.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
@dataclass
class Benchmark:
    """Run the ``level x rung x complexity`` cube and score each cell.

    Attributes:
        task: The scenario under test.
        estimators: The rungs/steps to fill in (each carries ``rung`` and
            ``complexity``).
        scorers: The ``Unit x Lens`` critiques applied to every scored cell.
        level: The data-hierarchy level the run sits at (default ``"L4"``).
        axis: Which axis to resolve references along (default staged).
        reference_rule: The "simpler thing is ground truth" tables.
        reference: An explicit oracle estimator; when ``None`` the rung-2
            estimator among ``estimators`` with the highest complexity (the
            richest kernel, such as NUTS rather than OI) is used.
        obs_key, truth_key, state_key: Keys into ``task.datasets()`` for the
            observations fed to estimators and the known-θ / simulator-state
            references.
    """

    task: Task
    estimators: Sequence[Estimator]
    scorers: Sequence[Scorer]
    level: str = "L4"
    axis: Axis = Axis.STAGED
    reference_rule: ReferenceRule = field(default_factory=ReferenceRule)
    reference: Estimator | None = None
    obs_key: str = "obs"
    truth_key: str = "theta"
    state_key: str = "state"

    def run(self) -> BenchmarkReport:
        """Fit every estimator, resolve each cell's reference, and score it."""
        self._check_unique_cells()
        # Scores are namespaced by scorer name, so two scorers sharing one, or
        # a name holding a key separator, would overwrite values in a cell.
        _check_scorer_names(self.scorers)
        tags = {scorer.name: tags_of(scorer) for scorer in self.scorers}
        data = dict(self.task.datasets())
        if self.obs_key not in data:
            raise KeyError(
                f"Benchmark: obs_key {self.obs_key!r} not in task.datasets() "
                f"(available keys: {sorted(data)})."
            )
        obs = data[self.obs_key]

        # Pass 1 — fit and predict every estimator, keyed by (rung, complexity).
        predictions: dict[tuple[int, int], Estimate] = {}
        for est in self.estimators:
            predictions[(est.rung, est.complexity)] = est.fit(self.task)(obs)

        oracle = self._oracle_prediction(obs, predictions)

        # Pass 2 — resolve each cell's references and score against each. Process
        # in ascending cell order so a cell is only considered "validated" (and
        # thus usable as a BEST_VALIDATED_BELOW reference) once it has itself
        # resolved its primary reference. ``validated`` maps a cell to its
        # prediction.
        validated: dict[tuple[int, int], Estimate] = {}
        by_cell: dict[tuple[int, int], CellResult] = {}
        entries: list[ReportEntry] = []
        for est in sorted(self.estimators, key=lambda e: (e.rung, e.complexity)):
            key = (est.rung, est.complexity)
            cell = Cell(self.level, est.rung, est.complexity)
            primary = self.reference_rule.source(
                est.rung, axis=self.axis, complexity=est.complexity
            )
            # An unscored cell has no sources; resolving its NONE primary
            # still gives the note that says why.
            sources = self.reference_rule.sources(
                est.rung, axis=self.axis, complexity=est.complexity
            ) or (primary,)
            by_reference: dict[ReferenceSource, dict[str, float]] = {}
            notes: list[str] = []
            for source in sources:
                scorers = [
                    scorer
                    for scorer in self.scorers
                    if scores_against(scorer, source, primary=primary)
                ]
                # A further reference no scorer asks for is not resolved, so
                # it cannot add a note either.
                if source is not primary and not scorers:
                    continue
                reference, note = self._resolve(
                    source, est, data, predictions, oracle, validated
                )
                if reference is None:
                    # The primary's note reads as before; a further
                    # reference's note says which reference it is about.
                    notes.append(
                        note if source is primary else f"{source.value}: {note}"
                    )
                    continue
                by_reference[source] = self._score(
                    predictions[key], reference, scorers, tags, cell, source, entries
                )
            by_cell[key] = CellResult(
                cell,
                primary,
                scores=by_reference.get(primary, {}),
                note="; ".join(notes),
                by_reference=by_reference,
            )
            if primary in by_reference:
                validated[key] = predictions[key]

        # Emit one result per estimator, preserving the caller's order.
        results = [by_cell[(est.rung, est.complexity)] for est in self.estimators]
        return BenchmarkReport(
            task=self.task.name,
            domain=self.task.domain,
            axis=self.axis,
            results=results,
            entries=tuple(entries),
        )

    @staticmethod
    def _score(
        prediction: Estimate,
        reference: object,
        scorers: Sequence[Scorer],
        tags: Mapping[str, ScorerTags],
        cell: Cell,
        source: ReferenceSource,
        entries: list[ReportEntry],
    ) -> dict[str, float]:
        """The scorers' values against one reference, namespaced by scorer.

        Each value is also appended to ``entries`` as a tagged
        :class:`ReportEntry` for :meth:`BenchmarkReport.to_report`.
        """
        scores: dict[str, float] = {}
        for scorer in scorers:
            for name, value in scorer(prediction, reference).items():
                scores[f"{scorer.name}.{name}"] = value
                entries.append(
                    ReportEntry(
                        scorer.name,
                        name,
                        value,
                        tags[scorer.name],
                        cell=cell.label,
                        reference=source.value,
                    )
                )
        return scores

    def _check_unique_cells(self) -> None:
        """Reject two estimators occupying the same cube cell.

        Each ``(rung, complexity)`` cell must be filled by exactly one
        estimator; otherwise predictions would silently overwrite and
        results would collide (including in :meth:`BenchmarkReport.to_dict`).
        """
        seen: set[tuple[int, int]] = set()
        dups: set[tuple[int, int]] = set()
        for est in self.estimators:
            key = (est.rung, est.complexity)
            if key in seen:
                dups.add(key)
            seen.add(key)
        if dups:
            raise ValueError(
                "duplicate benchmark cells (rung, complexity): "
                f"{sorted(dups)}; each cell must be filled by exactly one estimator"
            )

    def _oracle_prediction(
        self, obs: object, predictions: Mapping[tuple[int, int], Estimate]
    ) -> Estimate | None:
        """The rung-2 reference posterior, explicit or inferred."""
        if self.reference is not None:
            # If the explicit oracle is itself one of the estimators, reuse its
            # already-fitted prediction rather than running it a second time —
            # a slow/stochastic oracle (MCMC, 4D-Var) would otherwise yield a
            # different posterior than the one reported for its own cell.
            for est in self.estimators:
                if est is self.reference:
                    return predictions[(est.rung, est.complexity)]
            return self.reference.fit(self.task)(obs)
        # No explicit oracle: the inferred oracle is the *most* sophisticated
        # rung-2 kernel (highest complexity) — e.g. NUTS / Strong-4D-Var, not
        # the cheap OI/MAP baseline at the bottom of the rung-2 ladder.
        rung2 = [e for e in self.estimators if e.rung == 2]
        if rung2:
            best = max(rung2, key=lambda e: e.complexity)
            return predictions[(best.rung, best.complexity)]
        return None

    def _resolve(
        self,
        source: ReferenceSource,
        est: Estimator,
        data: Mapping[str, object],
        predictions: Mapping[tuple[int, int], Estimate],
        oracle: Estimate | None,
        validated: Mapping[tuple[int, int], Estimate],
    ) -> tuple[object | None, str]:
        """Map a :class:`ReferenceSource` to a concrete reference object.

        Returns ``(reference, note)``; a ``None`` reference means the cell
        is not scored, with ``note`` explaining why. ``validated`` holds the
        cells resolved so far, used by ``BEST_VALIDATED_BELOW``.
        """
        if source is ReferenceSource.NONE:
            if self.axis is Axis.COMPLEXITY and est.complexity <= 0:
                return (
                    None,
                    "simplest step on its rung: no simpler step to compare against",
                )
            return None, "generative rung: not scored against a reference"
        if source is ReferenceSource.KNOWN_THETA:
            return self._require(data, self.truth_key)
        if source is ReferenceSource.SIMULATOR_OUTPUT:
            return self._require(data, self.state_key)
        if source is ReferenceSource.ORACLE_POSTERIOR:
            if oracle is None:
                return None, "no oracle (rung-2) estimator available"
            return oracle, ""
        if source is ReferenceSource.PREVIOUS_RUNG:
            return self._lookup(predictions, est.rung - 1, est.complexity)
        if source is ReferenceSource.SIMPLER_STEP:
            return self._lookup(predictions, est.rung, est.complexity - 1)
        if source is ReferenceSource.BEST_VALIDATED_BELOW:
            # Stay on the estimator's own complexity slice: comparing rung 6 to
            # a lower rung at a *different* complexity would co-vary both axes
            # at once (the framework walks one axis at a time).
            below = [
                rc for rc in validated if rc[0] < est.rung and rc[1] == est.complexity
            ]
            if not below:
                return None, "no validated rung below this one on the same complexity"
            best = max(below)  # highest validated rung on this complexity slice
            return validated[best], ""
        return None, f"unhandled reference source {source!r}"

    @staticmethod
    def _require(data: Mapping[str, object], key: str) -> tuple[object | None, str]:
        if key not in data:
            return None, f"reference key {key!r} absent from task.datasets()"
        return data[key], ""

    @staticmethod
    def _lookup(
        predictions: Mapping[tuple[int, int], Estimate], rung: int, complexity: int
    ) -> tuple[object | None, str]:
        key = (rung, complexity)
        if key not in predictions:
            return None, f"no estimator at rung {rung}, complexity {complexity}"
        return predictions[key], ""

run()

Fit every estimator, resolve each cell's reference, and score it.

Source code in packages/pipekit-evaluate/src/pipekit_evaluate/benchmark/run.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
def run(self) -> BenchmarkReport:
    """Fit every estimator, resolve each cell's reference, and score it."""
    self._check_unique_cells()
    # Scores are namespaced by scorer name, so two scorers sharing one, or
    # a name holding a key separator, would overwrite values in a cell.
    _check_scorer_names(self.scorers)
    tags = {scorer.name: tags_of(scorer) for scorer in self.scorers}
    data = dict(self.task.datasets())
    if self.obs_key not in data:
        raise KeyError(
            f"Benchmark: obs_key {self.obs_key!r} not in task.datasets() "
            f"(available keys: {sorted(data)})."
        )
    obs = data[self.obs_key]

    # Pass 1 — fit and predict every estimator, keyed by (rung, complexity).
    predictions: dict[tuple[int, int], Estimate] = {}
    for est in self.estimators:
        predictions[(est.rung, est.complexity)] = est.fit(self.task)(obs)

    oracle = self._oracle_prediction(obs, predictions)

    # Pass 2 — resolve each cell's references and score against each. Process
    # in ascending cell order so a cell is only considered "validated" (and
    # thus usable as a BEST_VALIDATED_BELOW reference) once it has itself
    # resolved its primary reference. ``validated`` maps a cell to its
    # prediction.
    validated: dict[tuple[int, int], Estimate] = {}
    by_cell: dict[tuple[int, int], CellResult] = {}
    entries: list[ReportEntry] = []
    for est in sorted(self.estimators, key=lambda e: (e.rung, e.complexity)):
        key = (est.rung, est.complexity)
        cell = Cell(self.level, est.rung, est.complexity)
        primary = self.reference_rule.source(
            est.rung, axis=self.axis, complexity=est.complexity
        )
        # An unscored cell has no sources; resolving its NONE primary
        # still gives the note that says why.
        sources = self.reference_rule.sources(
            est.rung, axis=self.axis, complexity=est.complexity
        ) or (primary,)
        by_reference: dict[ReferenceSource, dict[str, float]] = {}
        notes: list[str] = []
        for source in sources:
            scorers = [
                scorer
                for scorer in self.scorers
                if scores_against(scorer, source, primary=primary)
            ]
            # A further reference no scorer asks for is not resolved, so
            # it cannot add a note either.
            if source is not primary and not scorers:
                continue
            reference, note = self._resolve(
                source, est, data, predictions, oracle, validated
            )
            if reference is None:
                # The primary's note reads as before; a further
                # reference's note says which reference it is about.
                notes.append(
                    note if source is primary else f"{source.value}: {note}"
                )
                continue
            by_reference[source] = self._score(
                predictions[key], reference, scorers, tags, cell, source, entries
            )
        by_cell[key] = CellResult(
            cell,
            primary,
            scores=by_reference.get(primary, {}),
            note="; ".join(notes),
            by_reference=by_reference,
        )
        if primary in by_reference:
            validated[key] = predictions[key]

    # Emit one result per estimator, preserving the caller's order.
    results = [by_cell[(est.rung, est.complexity)] for est in self.estimators]
    return BenchmarkReport(
        task=self.task.name,
        domain=self.task.domain,
        axis=self.axis,
        results=results,
        entries=tuple(entries),
    )

BenchmarkReport dataclass

The collected results of a benchmark run.

Attributes:

Name Type Description
task str

The task name.
domain str

The task domain.
axis Axis

The axis the references were resolved along.
results Sequence[CellResult]

One :class:CellResult per estimator.
entries tuple[ReportEntry, ...]

Every scored value as a tagged :class:ReportEntry naming its cell and reference, in scoring order; filled by :meth:Benchmark.run.
Source code in packages/pipekit-evaluate/src/pipekit_evaluate/benchmark/run.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
@dataclass(frozen=True)
class BenchmarkReport:
    """The collected results of a benchmark run.

    Attributes:
        task: The task name.
        domain: The task domain.
        axis: The axis the references were resolved along.
        results: One :class:`CellResult` per estimator.
        entries: Every scored value as a tagged :class:`ReportEntry` naming
            its cell and reference, in scoring order; filled by
            :meth:`Benchmark.run`.
    """

    task: str
    domain: str
    axis: Axis
    results: Sequence[CellResult]
    entries: tuple[ReportEntry, ...] = ()

    def to_dict(self) -> dict[str, Any]:
        """A JSON-serialisable form that keeps every reference and note.

        Returns:
            ``{"task", "domain", "axis", "cells"}``, where ``cells`` maps
            each cell label to its ``rung_name``, primary ``reference``,
            ``scores`` by reference source (``{source: {metric: value}}``)
            and ``note``.
        """
        return {
            "task": self.task,
            "domain": self.domain,
            "axis": self.axis.value,
            "cells": {
                r.cell.label: {
                    "rung_name": r.cell.rung_name,
                    "reference": r.reference.value,
                    "scores": {
                        source.value: dict(scores)
                        for source, scores in r.by_reference.items()
                    },
                    "note": r.note,
                }
                for r in self.results
            },
        }

    def to_report(
        self,
        *,
        model_ref: str | None = None,
        dataset_ref: str | None = None,
        config: Mapping[str, Any] | None = None,
    ) -> EvaluationReport:
        """The run as an :class:`EvaluationReport`, one entry per scored value.

        Each entry carries its cell label and reference source, so
        ``report.filter(reference="known_theta")`` or
        ``report.filter(lens="structural")`` cuts across the cube.

        Args:
            model_ref: What was evaluated, recorded on the report.
            dataset_ref: What it was evaluated on, recorded on the report.
            config: Settings recorded on the report and in its hash, beside
                the run's ``task``, ``domain`` and ``axis``.

        Returns:
            The report; its config holds ``task``, ``domain`` and ``axis``
            unless ``config`` sets them.
        """
        run = {"task": self.task, "domain": self.domain, "axis": self.axis.value}
        return EvaluationReport(
            self.entries,
            model_ref=model_ref,
            dataset_ref=dataset_ref,
            config={**run, **(config or {})},
        )

to_dict()

A JSON-serialisable form that keeps every reference and note.

Returns:

Type Description
dict[str, Any]

{"task", "domain", "axis", "cells"}, where cells maps
dict[str, Any]

each cell label to its rung_name, primary reference,
dict[str, Any]

scores by reference source ({source: {metric: value}})
dict[str, Any]

and note.
Source code in packages/pipekit-evaluate/src/pipekit_evaluate/benchmark/run.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def to_dict(self) -> dict[str, Any]:
    """A JSON-serialisable form that keeps every reference and note.

    Returns:
        ``{"task", "domain", "axis", "cells"}``, where ``cells`` maps
        each cell label to its ``rung_name``, primary ``reference``,
        ``scores`` by reference source (``{source: {metric: value}}``)
        and ``note``.
    """
    return {
        "task": self.task,
        "domain": self.domain,
        "axis": self.axis.value,
        "cells": {
            r.cell.label: {
                "rung_name": r.cell.rung_name,
                "reference": r.reference.value,
                "scores": {
                    source.value: dict(scores)
                    for source, scores in r.by_reference.items()
                },
                "note": r.note,
            }
            for r in self.results
        },
    }

to_report(*, model_ref=None, dataset_ref=None, config=None)

The run as an :class:EvaluationReport, one entry per scored value.

Each entry carries its cell label and reference source, so report.filter(reference="known_theta") or report.filter(lens="structural") cuts across the cube.

Parameters:

Name Type Description Default
model_ref str | None

What was evaluated, recorded on the report.

 None
dataset_ref str | None

What it was evaluated on, recorded on the report.

 None
config Mapping[str, Any] | None

Settings recorded on the report and in its hash, beside the run's task, domain and axis.

 None

Returns:

Type Description
EvaluationReport

The report; its config holds task, domain and axis
EvaluationReport

unless config sets them.
Source code in packages/pipekit-evaluate/src/pipekit_evaluate/benchmark/run.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def to_report(
    self,
    *,
    model_ref: str | None = None,
    dataset_ref: str | None = None,
    config: Mapping[str, Any] | None = None,
) -> EvaluationReport:
    """The run as an :class:`EvaluationReport`, one entry per scored value.

    Each entry carries its cell label and reference source, so
    ``report.filter(reference="known_theta")`` or
    ``report.filter(lens="structural")`` cuts across the cube.

    Args:
        model_ref: What was evaluated, recorded on the report.
        dataset_ref: What it was evaluated on, recorded on the report.
        config: Settings recorded on the report and in its hash, beside
            the run's ``task``, ``domain`` and ``axis``.

    Returns:
        The report; its config holds ``task``, ``domain`` and ``axis``
        unless ``config`` sets them.
    """
    run = {"task": self.task, "domain": self.domain, "axis": self.axis.value}
    return EvaluationReport(
        self.entries,
        model_ref=model_ref,
        dataset_ref=dataset_ref,
        config={**run, **(config or {})},
    )

CellResult dataclass

The scored outcome for one cube cell.

Attributes:

Name Type Description
cell Cell

The (level, rung, complexity) cell.
reference ReferenceSource

The cell's primary reference source.
scores Mapping[str, float]

Named metric values (namespaced by scorer) against the primary reference, empty when the cell is not scored or its primary reference was unavailable.
note str

Human-readable explanation of every reference that could not be resolved; empty when all of them were.
by_reference Mapping[ReferenceSource, Mapping[str, float]]

The scores against every reference that resolved, primary included, keyed by source. When only scores is given, it is filled in as {reference: scores}.
Source code in packages/pipekit-evaluate/src/pipekit_evaluate/benchmark/run.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@dataclass(frozen=True)
class CellResult:
    """The scored outcome for one cube cell.

    Attributes:
        cell: The ``(level, rung, complexity)`` cell.
        reference: The cell's primary reference source.
        scores: Named metric values (namespaced by scorer) against the
            primary reference, empty when the cell is not scored or its
            primary reference was unavailable.
        note: Human-readable explanation of every reference that could not
            be resolved; empty when all of them were.
        by_reference: The scores against every reference that resolved,
            primary included, keyed by source. When only ``scores`` is
            given, it is filled in as ``{reference: scores}``.
    """

    cell: Cell
    reference: ReferenceSource
    scores: Mapping[str, float] = field(default_factory=dict)
    note: str = ""
    by_reference: Mapping[ReferenceSource, Mapping[str, float]] = field(
        default_factory=dict
    )

    def __post_init__(self) -> None:
        if not self.by_reference and self.scores:
            object.__setattr__(self, "by_reference", {self.reference: self.scores})

scores_against(scorer, source, *, primary)

Whether scorer is run against the reference source.

A scorer names the reference kinds it can score against in an optional references attribute (one source or several, as members or their string values). Without one it is scored against the cell's primary reference only, so a scorer written for one kind of reference (an oracle's posterior Estimate) is never handed another (the raw known truth) because a rung gained a further reference.

Parameters:

Name Type Description Default
scorer Scorer

Any scorer.

 required
source ReferenceSource

The reference source about to be scored against.

 required
primary ReferenceSource

The cell's primary reference source.

 required

Returns:

Type Description
bool

True when the scorer should be run against source.
Source code in packages/pipekit-evaluate/src/pipekit_evaluate/benchmark/run.py 
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
def scores_against(
    scorer: Scorer, source: ReferenceSource, *, primary: ReferenceSource
) -> bool:
    """Whether ``scorer`` is run against the reference ``source``.

    A scorer names the reference kinds it can score against in an optional
    ``references`` attribute (one source or several, as members or their
    string values). Without one it is scored against the cell's primary
    reference only, so a scorer written for one kind of reference (an
    oracle's posterior ``Estimate``) is never handed another (the raw
    known truth) because a rung gained a further reference.

    Args:
        scorer: Any scorer.
        source: The reference source about to be scored against.
        primary: The cell's primary reference source.

    Returns:
        ``True`` when the scorer should be run against ``source``.
    """
    declared = getattr(scorer, "references", None)
    if declared is None:
        return source is primary
    return source in as_sources(declared)