bloc.testing.non_regression#

Non-regression snapshot utilities for Bloc model simulations.

This module provides tools to write and compare deterministic CSV snapshots derived from simulation scenario_results dicts, enabling model-level non-regression testing across code changes.

The snapshot schema mirrors the oo-refactor Calculation Note data model so that, once that branch merges, only the extraction adapter changes — not the snapshot format, the comparison rules, or the committed baselines.

Public API#

write_scenario_snapshots() – write five CSV files per scenario.
compare_scenario_snapshots() – compare produced vs. committed CSV files and return a ComparisonReport.
ComparisonReport – holds per-file mismatch details and exit status.
SnapshotConfig – controls tolerances, floor, and schema version.

CSV dialect (all files)#

UTF-8, no BOM, LF line endings, trailing newline.
csv.QUOTE_MINIMAL, comma delimiter, decimal point, no thousands separator.
Float format: repr(float(v)) for exact round-trip.
nan, inf, -inf serialized as lowercase literals.
All rows sorted by the set of key columns before writing.

Schema version#

snapshot_schema_version is recorded in metadata.csv. The comparator refuses to compare files with different schema versions and raises a clear error. Bump SNAPSHOT_SCHEMA_VERSION whenever a non-trivial schema change is made.

Attributes#

`SNAPSHOT_SCHEMA_VERSION`
`OPTIONAL_BASELINE_MODEL_IDS`
`DEFAULT_COMPOSITION_FLOOR`
`TIMING_METRIC_KEYS`

Classes#

`ToleranceRule`	Per-key numeric tolerance override loaded from `tolerances.yaml`.
`NodeRule`	Per-model node-name mapping loaded from `tolerances.yaml`.
`SnapshotConfig`	Configuration for snapshot writing and comparison.
`FileMismatch`	Mismatch details for a single CSV file.
`ComparisonReport`	Report produced by `compare_scenario_snapshots()`.
`ScenarioTiming`	Wall-clock timing for one non-regression scenario.
`ScenarioMemory`	Peak memory for one non-regression scenario.

Functions#

`format_timing_summary`(timings)	Return a compact timing digest for all compared scenarios.
`format_memory_summary`(memories)	Return a compact peak-memory digest for all compared scenarios.
`write_scenario_snapshots`(scenario_id, scenario_data, ...)	Write the five canonical CSV snapshot files for one scenario.
`read_metadata_float`(metadata_path, key)	Return the float value of key in metadata_path, or None when absent.
`read_metadata_elapsed_s`(metadata_path)	Return `elapsed_s` (solve wall-clock) or None when absent.
`read_metadata_collect_s`(metadata_path)	Return `collect_s` (results-generation wall-clock) or None when absent.
`read_metadata_peak_rss_mb`(metadata_path)	Return `peak_rss_mb` from metadata_path, or None when absent.
`read_metadata_yaml_path`(metadata_path)	Return the `yaml_path` provenance value from metadata_path, or None.
`conservative_elapsed_s`(*values)	Return the longest non-null wall-clock time from values.
`conservative_peak_rss_mb`(*values)	Return the largest non-null peak memory from values.
`write_metadata_float`(metadata_path, scenario_id, key, ...)	Set or replace key in an existing `metadata.csv` file.
`write_metadata_elapsed_s`(metadata_path, scenario_id, ...)	Set or replace `elapsed_s` in an existing `metadata.csv` file.
`write_metadata_collect_s`(metadata_path, scenario_id, ...)	Set or replace `collect_s` in an existing `metadata.csv` file.
`write_metadata_peak_rss_mb`(metadata_path, scenario_id, ...)	Set or replace `peak_rss_mb` in an existing `metadata.csv` file.
`check_scenario_memory`(*, model_id, scenario_id, ...[, cfg])	Compare scenario peak memory against the committed baseline.
`check_scenario_timing`(*, model_id, scenario_id, ...[, ...])	Compare one scenario wall-clock phase against the committed baseline.
`compare_scenario_snapshots`(scenario_id, baseline_dir, ...)	Compare produced snapshot CSV files against committed baselines.
`load_tolerance_config`(path)	Load a `SnapshotConfig` from a YAML tolerances file.
`non_regression_out_dir`(script_dir, scenario_id)	Return the canonical non-regression output path for one scenario.
`find_produced_snapshots`(script_dir)	Return `[(scenario_id, path), ...]` for all produced snapshot dirs.
`baseline_dir_for`(repo_root, model_id, scenario_id)	Return the committed baseline directory for one scenario.
`snapshot_from_ctsim`(ctsim, script_path[, scenario_id, cfg])	Write non-regression snapshots directly from a `ctwrap.Simulation` object.
`model_id_from_script`(script_path, repo_root)	Derive the `model_id` from a model script path.

Module Contents#

bloc.testing.non_regression.SNAPSHOT_SCHEMA_VERSION = 1#

bloc.testing.non_regression.OPTIONAL_BASELINE_MODEL_IDS: frozenset[str]#

bloc.testing.non_regression.DEFAULT_COMPOSITION_FLOOR = 1e-05#

class bloc.testing.non_regression.ToleranceRule#

Per-key numeric tolerance override loaded from tolerances.yaml.

model: str = '*'#

scenario: str = '*'#

key: str = '*'#

rtol: float = 0.0001#

atol: float = 1e-10#

class bloc.testing.non_regression.NodeRule#

Per-model node-name mapping loaded from tolerances.yaml.

Lets a renamed reactor network compare against an unchanged baseline without rewriting the committed CSV files.

model, scenario: fnmatch patterns selecting which scenarios the rule applies to.

aliases#: Map of baseline_node -> canonical_node. Node names are matched after the ``[N] `` display-number prefix is stripped. The same map is applied to both baseline and actual rows, so any entry whose key is the current node name is a harmless identity. Several baseline names may map to one canonical node (e.g. a node split that was later merged).

ignored#: Node names (after prefix stripping) present in the baseline but not yet implemented in the current model. Rows for these nodes are skipped on both sides instead of being reported as missing.

model: str = '*'#

scenario: str = '*'#

aliases: dict[str, str]#

ignored: frozenset[str]#

class bloc.testing.non_regression.SnapshotConfig#

Configuration for snapshot writing and comparison.

Parameters:

composition_floor – Drop species with |value| < composition_floor from compositions.csv.
default_rtol – Default relative tolerance for numeric comparisons.
default_atol – Default absolute tolerance for numeric comparisons.
overrides – List of ToleranceRule objects; first match wins.
max_diff_rows – Maximum number of differing rows shown per file in the report.
timing_max_ratio – Maximum allowed ratio actual_elapsed_s / baseline_elapsed_s before a timing non-regression failure. Only enforced when both sides record elapsed_s in metadata.csv.
memory_max_ratio – Maximum allowed ratio actual_peak_rss_mb / baseline_peak_rss_mb before a memory non-regression failure. Only enforced when both sides record peak_rss_mb in metadata.csv.
timing_floor_s – Baselines below this are reported but never fail. A fraction of a second of wall clock on a shared CI runner is scheduling noise, and timing_max_ratio applied to a 0.2 s baseline fails on a 0.05 s hiccup. Every committed solve baseline is far above this floor, so it only affects the cheap phases (a small model’s results generation).

composition_floor: float = 1e-05#

default_rtol: float = 0.0001#

default_atol: float = 1e-10#

overrides: list[ToleranceRule] = []#

node_rules: list[NodeRule] = []#

max_diff_rows: int = 50#

timing_max_ratio: float = 1.2#

timing_floor_s: float = 2.0#

memory_max_ratio: float = 1.2#

class bloc.testing.non_regression.FileMismatch#

Mismatch details for a single CSV file.

filename: str#

changed: list[dict[str, Any]] = []#

missing: list[dict[str, Any]] = []#

extra: list[dict[str, Any]] = []#

property ok: bool#: Return True when the file has no mismatches.

format_report(max_rows=50)#: Return a human-readable diff-style report for this file.

class bloc.testing.non_regression.ComparisonReport#

Report produced by compare_scenario_snapshots().

scenario_id: str#

model_id: str#

baseline_dir: pathlib.Path#

actual_dir: pathlib.Path#

file_mismatches: list[FileMismatch] = []#

schema_error: str = ''#

actual_yaml_path: str | None = None#

baseline_yaml_path: str | None = None#

property ok: bool#: Return True if the scenario matches the baseline fully.

format_report(max_rows=50)#: Return the full human-readable comparison report.

format_summary(top_n=5)#

Return a compact digest with the top-N mismatches by relative error.

Surfaces the rows with the largest numeric relative error (from all changed lists across all files) and up to top_n missing-in-actual entries. Type-change rows (no rel_err field) appear after numeric rows. extra rows are omitted from the summary as they are less immediately actionable than missing or changed ones.

class bloc.testing.non_regression.ScenarioTiming#

Wall-clock timing for one non-regression scenario.

model_id: str#

scenario_id: str#

actual_s: float | None#

baseline_s: float | None#

max_ratio: float = 1.2#

floor_s: float = 2.0#

metric: str = 'solve'#

property ratio: float | None#: Return actual/baseline when both timings are positive.

property below_floor: bool#: Return True when the baseline is too small to judge a ratio on.

property ok: bool#: Return True when timing is within tolerance or cannot be compared.

format_line()#: Return one summary line for this scenario.

format_failure()#: Return a verbose failure message when timing regresses.

class bloc.testing.non_regression.ScenarioMemory#

Peak memory for one non-regression scenario.

Mirrors ScenarioTiming. actual_mb and baseline_mb are process high-water marks recorded when the scenario finished, so in a multi-scenario run they include memory retained by earlier scenarios – see bloc.memory.peak_rss_mb().

model_id: str#

scenario_id: str#

actual_mb: float | None#

baseline_mb: float | None#

max_ratio: float = 1.2#

property ratio: float | None#: Return actual/baseline when both measurements are positive.

property ok: bool#: Return True when memory is within tolerance or cannot be compared.

format_line()#: Return one summary line for this scenario.

format_failure()#: Return a verbose failure message when memory regresses.

bloc.testing.non_regression.format_timing_summary(timings)#: Return a compact timing digest for all compared scenarios.

bloc.testing.non_regression.format_memory_summary(memories)#

Return a compact peak-memory digest for all compared scenarios.

The trailing worst-case line is the number that matters for CI capacity: a hosted runner has 16 GB, and the job is killed when the process approaches it (issue #331).

bloc.testing.non_regression.write_scenario_snapshots(scenario_id, scenario_data, out_dir, model_id, cfg=None)#

Write the five canonical CSV snapshot files for one scenario.

Parameters:

scenario_id – Scenario identifier used as the first column in every CSV file.
scenario_data – Dict with keys outputs, node_data, physical_node_names, energy_flows, metadata. Missing keys default to empty.
out_dir – Directory where the five CSV files are written.
model_id – Relative model identifier (e.g. "SPRING_A3"); stored in metadata.
cfg – Optional SnapshotConfig; defaults are used when None.

bloc.testing.non_regression.read_metadata_float(metadata_path, key)#: Return the float value of key in metadata_path, or None when absent.

bloc.testing.non_regression.read_metadata_elapsed_s(metadata_path)#: Return elapsed_s (solve wall-clock) or None when absent.

bloc.testing.non_regression.read_metadata_collect_s(metadata_path)#: Return collect_s (results-generation wall-clock) or None when absent.

bloc.testing.non_regression.read_metadata_peak_rss_mb(metadata_path)#: Return peak_rss_mb from metadata_path, or None when absent.

bloc.testing.non_regression.read_metadata_yaml_path(metadata_path)#: Return the yaml_path provenance value from metadata_path, or None.

bloc.testing.non_regression.conservative_elapsed_s(*values)#: Return the longest non-null wall-clock time from values.

bloc.testing.non_regression.conservative_peak_rss_mb(*values)#

Return the largest non-null peak memory from values.

Baselines take the maximum for the same reason timings do: a baseline that captured an unusually cheap run would make every later run look like a regression.

bloc.testing.non_regression.write_metadata_float(metadata_path, scenario_id, key, value_f)#: Set or replace key in an existing metadata.csv file.

bloc.testing.non_regression.write_metadata_elapsed_s(metadata_path, scenario_id, elapsed_s)#: Set or replace elapsed_s in an existing metadata.csv file.

bloc.testing.non_regression.write_metadata_collect_s(metadata_path, scenario_id, collect_s)#: Set or replace collect_s in an existing metadata.csv file.

bloc.testing.non_regression.write_metadata_peak_rss_mb(metadata_path, scenario_id, peak_rss_mb)#: Set or replace peak_rss_mb in an existing metadata.csv file.

bloc.testing.non_regression.check_scenario_memory(*, model_id, scenario_id, baseline_dir, actual_dir, cfg=None)#: Compare scenario peak memory against the committed baseline.

bloc.testing.non_regression.TIMING_METRIC_KEYS#

bloc.testing.non_regression.check_scenario_timing(*, model_id, scenario_id, baseline_dir, actual_dir, cfg=None, metric='solve')#

Compare one scenario wall-clock phase against the committed baseline.

Parameters:: metric – Which phase to compare – "solve" or "collect", see TIMING_METRIC_KEYS and ScenarioTiming.metric. The two regress independently: results generation can double while the solve is untouched, and only a separate measurement catches that.

bloc.testing.non_regression.compare_scenario_snapshots(scenario_id, baseline_dir, actual_dir, model_id, cfg=None)#

Compare produced snapshot CSV files against committed baselines.

Parameters:

scenario_id – Scenario identifier (used for tolerance resolution).
baseline_dir – Directory containing committed baseline CSV files.
actual_dir – Directory containing freshly produced CSV files.
model_id – Relative model identifier (e.g. "SPRING_A3").
cfg – Optional SnapshotConfig.

Returns:

Contains per-file mismatch details.

Return type:

ComparisonReport

bloc.testing.non_regression.load_tolerance_config(path)#

Load a SnapshotConfig from a YAML tolerances file.

The YAML file has the following structure:

defaults:
  rtol: 1.0e-4
  atol: 1.0e-10

overrides:
  - model: "SPRING_A3"
    scenario: "*"
    key: "X_*"
    rtol: 1.0e-4

Parameters:: path – Path to tolerances.yaml.
Returns:: Populated from the file; defaults applied when keys are absent.
Return type:: SnapshotConfig

bloc.testing.non_regression.non_regression_out_dir(script_dir, scenario_id)#

Return the canonical non-regression output path for one scenario.

Written to:

<script_dir>/Results/_non_regression/<scenario_id>/

This location is git-ignored; the parent runner discovers it after each model subprocess finishes.

Parameters:

script_dir – Directory of the model run script.
scenario_id – Scenario identifier used as the leaf directory name.

bloc.testing.non_regression.find_produced_snapshots(script_dir)#

Return [(scenario_id, path), ...] for all produced snapshot dirs.

Parameters:: script_dir – Directory of the model run script.

bloc.testing.non_regression.baseline_dir_for(repo_root, model_id, scenario_id)#

Return the committed baseline directory for one scenario.

Parameters:

repo_root – Root of the repository.
model_id – Relative model identifier (e.g. "SPRING_A3").
scenario_id – Scenario identifier.

bloc.testing.non_regression.snapshot_from_ctsim(ctsim, script_path, scenario_id='BASE', cfg=None)#

Write non-regression snapshots directly from a ctwrap.Simulation object.

This adapter is intended for model scripts that run a single scenario without going through bloc.calc_note.generate_calculation_note(). It extracts outputs from ctsim.data (res_dic if available, plus reactor state if sim is available) and writes the canonical CSV files.

Parameters:

ctsim – A ctwrap.Simulation object whose .run(...) has completed.
script_path – Path to the calling model script (used to derive model_id and the output directory).
scenario_id – Identifier for this scenario; defaults to "BASE".
cfg – Optional SnapshotConfig; defaults are used when None.

Notes

The function is best-effort: fields that cannot be extracted are silently omitted. Prefer using bloc.calc_note.generate_calculation_note() which hooks snapshots automatically and provides richer node_data.

bloc.testing.non_regression.model_id_from_script(script_path, repo_root)#

Derive the model_id from a model script path.

Returns the relative directory under models/, e.g. "SPRING_A3" for models/SPRING_A3/run_concept.py.

Parameters:

script_path – Absolute path to the model script.
repo_root – Root of the repository.