Reporting¶

Run archiving¶

Run archiving: auto-numbered run directories with full reproducibility snapshots.

A run is one estimation (or one batch study). Each run gets its own directory that is never overwritten — repeated runs with the same name get an auto-incremented ~NN suffix (biogeme convention: model, model~00, model~01, …). The directory contains everything needed to reproduce the run:

config.json — every estimator/solver config, seeds, spec description, package versions, and the git commit of the code that produced the run.
data.npz / data_ref.json — the estimation inputs, either snapshotted in full (compressed) or, above a size threshold, referenced by content hash and source path so staleness is detectable.
environment.json — machine info: hostname, platform, CPU count, native thread count, total RAM, BLAS-relevant env vars.
records/*.npz — per-task records for batch studies (one file per replication).
result.npz + compute.json — final estimates and the compute profile (wall time, peak RSS, CPU time).
report.html — human-readable report (see micropurc.reporting.html).

Example

>>> writer = RunWriter("runs", "sim_grid", config={
...     "estimation": est_cfg, "projection": proj_cfg, "seed": 0})
>>> writer.snapshot_data(y=y_csr, origins=origins, dests=dests)
>>> result = est.fit(...)
>>> writer.finalize(result={"beta_hat": result["beta_hat"]})

micropurc.reporting.runlog.DEFAULT_MAX_FULL_BYTES = 209715200¶: Size ceiling (bytes, summed over one snapshot) for copying inputs in full. A snapshot above it is recorded as a content-hash reference. Override per call via max_full_bytes.

micropurc.reporting.runlog.next_run_dir(base, name)[source]¶

Returns the first non-existing run directory for name under base.

Follows the biogeme naming convention: the first run is base/name; subsequent runs are base/name~00, base/name~01, … so no run is ever overwritten. The next index is one past the highest suffix present, so deleting an intermediate run leaves its number retired.

Parameters:

base (str | Path) – Parent directory that holds all runs (created if missing).
name (str) – Logical run name, e.g. "ema_ab".

Returns:

Path of a fresh, not-yet-created run directory.

Return type:

Path

micropurc.reporting.runlog.peak_rss_mb()[source]¶

Returns this process’s peak resident set size in MiB.

Uses getrusage on macOS/Linux and the Win32 API on Windows, so no external dependency is needed. The value is a high-water mark over the whole process lifetime, so it covers memory already released by the time it is read.

Returns:: Peak RSS in MiB (0.0 if it cannot be determined).
Return type:: float

micropurc.reporting.runlog.environment_info()[source]¶

Collects machine and software environment info for a run archive.

Every probe falls back to a placeholder on failure, so archiving a run survives an unavailable extension or missing package metadata.

Returns:: JSON-serializable dict with hostname, platform, Python/NumPy/SciPy versions, micropurc version + git commit, CPU count, native thread count, total RAM, and thread-related environment variables.
Return type:: dict[str, Any]

class micropurc.reporting.runlog.RunWriter(base, name, *, config=None, resume=None)[source]¶

Bases: object

Archives one estimation run into a fresh auto-numbered directory.

Variables:

run_dir – The directory created for this run.

Parameters:

base (str | Path)
name (str)
config (Mapping[str, Any] | None)
resume (str | Path | None)

Example

>>> writer = RunWriter("runs", "ema_ab", config={"seed": 0})
>>> writer.log("starting probe")
>>> writer.add_record("ab_N500_rep0", beta_hat=beta_hat)
>>> writer.finalize(result={"beta_hat": beta_hat})

__init__(base, name, *, config=None, resume=None)[source]¶

Creates or resumes a run directory and snapshots config and environment.

The wall-clock and CPU-time origins are taken here, so a resumed run reports the time spent since the resume.

Parameters:

base (str | Path) – Parent directory holding all runs.
name (str) – Logical run name; the directory auto-increments to avoid overwriting earlier runs with the same name.
config (Mapping[str, Any] | None) – Everything needed to reproduce the run: estimator/solver config dataclasses, seeds, protocol constants, data descriptions. Serialized to config.json.
resume (str | Path | None) – Existing run directory to continue (skips creating a new one and does not rewrite config.json).

Raises:

FileNotFoundError – If resume points to a missing directory.

Return type:

None

log(message)[source]¶

Appends a timestamped line to run.log.

Parameters:: message (str) – Text to log.
Return type:: None

snapshot_data(*, max_full_bytes=209715200, source_paths=None, name='data', **arrays)[source]¶

Snapshots estimation inputs for exact future replication.

Inputs totalling at most max_full_bytes are copied in full into {name}.npz (compressed). Above that, {name}_ref.json records SHA-256 content hashes and optional source paths, so a future load can verify it is reading identical data. Sparse inputs are flattened to their CSR components first, and each component is hashed on its own.

Parameters:

max_full_bytes (int) – Full-copy size threshold in bytes.
source_paths (Mapping[str, str | Path] | None) – Optional map from array name to the file it came from (stored alongside hashes for reference snapshots).
name (str) – Base filename for the snapshot.
**arrays (Any) – Arrays or scipy sparse matrices to snapshot.

Returns:

Path of the file written (.npz or _ref.json).

Return type:

Path

add_record(record_name, **arrays)[source]¶

Writes one per-task record under records/.

Each replication of a batch study gets its own file, so a crashed study resumes from the records already on disk (see record_exists()).

Parameters:

record_name (str) – Filename stem, e.g. "ab_N500_rep0".
**arrays (Any) – Scalars/arrays to store (strings allowed).

Returns:

Path of the record written.

Return type:

Path

record_exists(record_name)[source]¶

Reports whether a per-task record already exists, for resume.

Parameters:: record_name (str) – Filename stem used in add_record().
Returns:: True when the record file is present.
Return type:: bool

finalize(result=None, status='completed')[source]¶

Writes the final result, the compute profile, and the run status.

compute.json is the marker of a finished run, so a directory without it was interrupted.

Parameters:

result (Mapping[str, Any] | None) – Final estimates and diagnostics (arrays/scalars) stored in result.npz.
status (str) – Run status recorded in compute.json ("completed", "failed", …).

Returns:

The run directory.

Return type:

Path

Loading and analysis¶

Load archived runs and analyze them: predicted flows, active subnetworks.

The loader is the read side of RunWriter. Given a run directory it reconstructs the configuration, the dataset, and — when the network was snapshotted — a fully functional forward model, so that estimates can be interrogated without re-running the estimation:

Example

>>> from micropurc.reporting import load_run
>>> run = load_run("runs/ema_ab~03")
>>> run.beta_hat
array([0.501, 0.497, 0.503])
>>> flows = run.predicted_flow(od_pairs=[(3, 17)])       # forward QP at beta_hat
>>> sub = run.active_subnetwork((3, 17))                  # links carrying flow
>>> cmp = run.compare_routes(trips=range(10))             # predicted vs observed

The network snapshot convention (see snapshot_data): arrays named edges, Z, A (sparse), scale_m, and optionally attribute_names reconstruct the Network; y (sparse), origins, dests reconstruct the observations.

class micropurc.reporting.loader.RunArchive(run_dir, config=<factory>, environment=<factory>, compute=<factory>, result=<factory>, data=<factory>, data_ref=<factory>)[source]¶

Bases: object

An archived run loaded back into memory.

Variables:

run_dir (pathlib.Path) – Directory the run was loaded from.
config (dict[str, Any]) – Parsed config.json.
environment (dict[str, Any]) – Parsed environment.json (machine/software info).
compute (dict[str, Any]) – Parsed compute.json (wall time, peak RSS, …), if the run was finalized.
result (dict[str, Any]) – Arrays from result.npz (beta_hat, se_hat, …).
data (dict[str, Any]) – Snapshotted inputs from data.npz (empty when the run used a hash reference — see data_ref).
data_ref (dict[str, Any]) – Parsed data_ref.json when the inputs were too large to copy; contains SHA-256 hashes and source paths for verification.

Parameters:

run_dir (Path)
config (dict[str, Any])
environment (dict[str, Any])
compute (dict[str, Any])
result (dict[str, Any])
data (dict[str, Any])
data_ref (dict[str, Any])

run_dir: Path¶

config: dict[str, Any]¶

environment: dict[str, Any]¶

compute: dict[str, Any]¶

result: dict[str, Any]¶

data: dict[str, Any]¶

data_ref: dict[str, Any]¶

property beta_hat: ndarray¶

Point estimates from result.npz.

Returns:: The beta_hat array.
Raises:: KeyError – If the run has no stored beta_hat.

property se_hat: ndarray | None¶

Standard errors from result.npz (None when absent).

Returns:: The se_hat array or None.

records()[source]¶

Loads every per-task record of a batch study.

Returns:: Mapping from record name to its arrays.
Return type:: dict[str, dict[str, Any]]

network()[source]¶

Rebuilds the Network from the data snapshot.

Returns:: The snapshotted network.
Raises:: KeyError – If the snapshot lacks the network arrays (edges, Z, A).
Return type:: Network

solver()[source]¶

Rebuilds the forward solver with the snapshotted perturbation scale.

Returns:: A PIQP forward solver matching the archived run.
Raises:: KeyError – If the snapshot lacks scale_m.
Return type:: PIQPFlowSolver

predicted_flow(od_pairs=None, *, trips=None, beta=None)[source]¶

Solves the forward QP at beta for selected OD pairs.

Parameters:

od_pairs (Iterable[tuple[int, int]] | None) – (origin, dest) node pairs to solve.
trips (Iterable[int] | None) – Alternatively, trip indices whose OD pairs are used.
beta (ndarray | None) – Parameters; defaults to the run’s beta_hat.

Returns:

Mapping (origin, dest) -> x with the optimal link-flow vector (shape (L,)) of each pair.

Return type:

dict[tuple[int, int], ndarray]

Example

>>> flows = run.predicted_flow([(3, 17)])
>>> flows[(3, 17)].sum()

active_subnetwork(od, *, beta=None, slack_tol=0.0)[source]¶

Extracts the active subnetwork of one OD pair at beta.

Active links are identified with the dual-slack KKT test the estimator uses: link \(j\) is active when \(s_j = c_j + (A^{\top}\lambda)_j\) falls below the cutoff, which is zero and tightens to -slack_tol when a positive tolerance is passed.

Parameters:

od (tuple[int, int]) – (origin, dest) node pair.
beta (ndarray | None) – Parameters; defaults to the run’s beta_hat.
slack_tol (float) – Dual-slack dead-band passed to the active-set test.

Returns:

Dict with links (active link ids), tail/head (their endpoints), flow (their optimal flows), slack (per-link raw dual slack), margins (per-link active-set margin \(|s_j| / \lVert Z_j \rVert\) in beta-units), margin (the per-trip margin, the minimum over links with \(\lVert Z_j \rVert > 0\), which with the node potentials held fixed is the length of the smallest parameter perturbation that flips a link), and active_set (the full ActiveSet).

Return type:

dict[str, Any]

Example

>>> sub = run.active_subnetwork((3, 17))
>>> list(zip(sub["tail"], sub["head"], sub["flow"]))

compare_routes(trips, *, beta=None, flow_tol=1e-08)[source]¶

Compares predicted flows against the observed routes of trips.

For each trip the forward QP is solved at beta for the trip’s OD pair, and the predicted flow pattern is compared with the observed route (the nonzero links of the snapshotted y row).

Parameters:

trips (Sequence[int]) – Trip indices into the snapshotted dataset.
beta (ndarray | None) – Parameters; defaults to the run’s beta_hat.
flow_tol (float) – Flow threshold above which a link counts as used.

Returns:

trip, od, n_obs_links, n_pred_links, obs_links_carrying_flow (count of observed links with predicted flow > tol), flow_on_route (total predicted flow mass on observed links), flow_share_on_route (that mass over the total predicted flow mass — 1.0 means the prediction concentrates entirely on the observed route), and min_flow_on_route.

Return type:

One dict per trip with

Raises:

ValueError – If the run has no snapshotted y (observed routes).

Example

>>> rows = run.compare_routes(trips=[0, 1, 2])
>>> rows[0]["flow_share_on_route"]

micropurc.reporting.loader.load_run(run_dir)[source]¶

Loads an archived run directory written by RunWriter.

Parameters:: run_dir (str | Path) – Path of the run directory (e.g. runs/ema_ab~03).
Returns:: RunArchive with config, environment, compute profile, result arrays, and the data snapshot (when copied in full).
Raises:: FileNotFoundError – If run_dir does not exist.
Return type:: RunArchive

HTML reports¶

Self-contained HTML reports for archived runs, in the biogeme tradition.

Two renderers:

render_fit_report() — one estimation: parameter table with SE/t/p, convergence summary with a Q-history sparkline, dataset provenance, full config (collapsible), and the compute/environment profile.
render_batch_report() — a batch study (e.g. a Monte Carlo sweep): arbitrary summary tables, gate results, and the compute profile.

Reports are single files with inline CSS — no external assets — so they can be archived inside the run directory and opened anywhere.

Example

>>> from micropurc.reporting import load_run, render_fit_report
>>> run = load_run("runs/sim_grid")
>>> render_fit_report(run, param_names=["b_length", "b_time"])  # -> report.html

micropurc.reporting.html.render_fit_report(run, *, param_names=None, out_name='report.html')[source]¶

Writes the single-estimation HTML report into the run directory.

The reported statistic is \(t_k = \hat\beta_k / \mathrm{se}_k\) and its p-value is the two-sided standard-normal tail \(2\big(1 - \Phi(|t_k|)\big)\). Parameters whose standard error is missing, non-finite, or non-positive show a dash in the three test columns.

Parameters:

run (RunArchive) – Loaded run archive (must contain beta_hat in its result).
param_names (Sequence[str] | None) – Parameter names for the estimates table; defaults to beta_0, beta_1, ....
out_name (str) – Output filename inside the run directory.

Returns:

Path of the written report.

Return type:

Path

Example

>>> render_fit_report(load_run("runs/sim_grid"),
...                   param_names=["b_length", "b_time", "b_toll"])

micropurc.reporting.html.render_batch_report(run, *, title, tables, gates=(), out_name='report.html')[source]¶

Writes a batch-study HTML report (Monte Carlo sweeps, ablations, …).

Parameters:

run (RunArchive) – Loaded run archive of the batch run.
title (str) – Report title.
tables (Sequence[tuple[str, Sequence[str], Iterable[Sequence[Any]]]]) – Sequence of (section_title, headers, rows) summary tables.
gates (Sequence[tuple[str, bool]]) – (gate_name, passed) pairs rendered as a pass/fail list.
out_name (str) – Output filename inside the run directory.

Returns:

Path of the written report.

Return type:

Path

Example

>>> render_batch_report(run, title="EMA AB study",
...     tables=[("Goal A", ["N", "bias"], [[500, 0.01]])],
...     gates=[("RMSE decreasing", True)])