qnetbench.characterize

Demand-signature extraction: the single-trace axes, the swept fidelity and staleness curves, the cross-application table, and run provenance.

Narrative guide: Characterization.

Signature and report

qnetbench.characterize.report

Per-application demand-signature report: the single-trace signature plus the summarised fidelity/staleness curves, and a cross-application table. This is the machine-readable form of the characterization (Deliverable 2).

AppSignature

Bases: BaseModel

The full demand signature for one application.

app instance-attribute

app: str

trace instance-attribute

trace: TraceSignature

fidelity_threshold class-attribute instance-attribute

fidelity_threshold: float | None = None

fidelity_range class-attribute instance-attribute

fidelity_range: float | None = None

staleness_halflife class-attribute instance-attribute

staleness_halflife: float | None = None

staleness_halflife_range class-attribute instance-attribute

staleness_halflife_range: float | None = None

fidelity_threshold_std class-attribute instance-attribute

fidelity_threshold_std: float | None = None

staleness_halflife_std class-attribute instance-attribute

staleness_halflife_std: float | None = None

fidelity_threshold_bracket class-attribute instance-attribute

fidelity_threshold_bracket: float | None = None

staleness_halflife_bracket class-attribute instance-attribute

staleness_halflife_bracket: float | None = None

fidelity_threshold_seed_median class-attribute instance-attribute

fidelity_threshold_seed_median: float | None = None

staleness_halflife_seed_median class-attribute instance-attribute

staleness_halflife_seed_median: float | None = None

fidelity_threshold_seed_count class-attribute instance-attribute

fidelity_threshold_seed_count: int = 0

staleness_halflife_seed_count class-attribute instance-attribute

staleness_halflife_seed_count: int = 0

n_seeds class-attribute instance-attribute

n_seeds: int = 0

characterize_app

characterize_app(
    app: str,
    *,
    coherence_time: float = 0.001,
    seeds: Sequence[int] = range(SEEDS),
) -> tuple[AppSignature, CharacterizationCurves]

Characterize one application: returns its signature and the raw curves.

Source code in qnetbench/characterize/report.py

def characterize_app(
    app: str,
    *,
    coherence_time: float = 1e-3,
    seeds: Sequence[int] = range(SEEDS),
) -> tuple[AppSignature, CharacterizationCurves]:
    """Characterize one application: returns its signature and the raw curves."""
    trace = characterize_trace(run_once(app, seed=0))
    curves = characterize_curves(app, coherence_time=coherence_time, seeds=seeds)
    signature = AppSignature(
        app=app,
        trace=trace,
        fidelity_threshold=curves.fidelity_threshold,
        fidelity_range=curves.fidelity_range,
        staleness_halflife=curves.staleness_halflife,
        staleness_halflife_range=curves.staleness_halflife_range,
        fidelity_threshold_std=curves.fidelity_threshold_std,
        staleness_halflife_std=curves.staleness_halflife_std,
        fidelity_threshold_bracket=curves.fidelity_threshold_bracket,
        staleness_halflife_bracket=curves.staleness_halflife_bracket,
        fidelity_threshold_seed_median=curves.fidelity_threshold_seed_median,
        staleness_halflife_seed_median=curves.staleness_halflife_seed_median,
        fidelity_threshold_seed_count=curves.fidelity_threshold_seed_count,
        staleness_halflife_seed_count=curves.staleness_halflife_seed_count,
        n_seeds=curves.n_seeds,
    )
    return signature, curves

render_table

render_table(signatures: list[AppSignature]) -> str

A compact cross-application demand-signature table.

Source code in qnetbench/characterize/report.py

def render_table(signatures: list[AppSignature]) -> str:
    """A compact cross-application demand-signature table."""
    header = (
        f"{'app':24} {'parties':>7} {'cv':>6} {'fano':>6} {'msg/pair':>9} "
        f"{'B/pair':>7} {'deadline':>8} {'F½util (±std)':>15} {'ΔF':>6} "
        f"{'stale½(ms, ±std)':>18} "
        f"{'stale½r(ms)':>11}"
    )
    lines = [header, "-" * len(header)]
    seeded = False
    for s in signatures:
        t = s.trace
        half_ms, half_ms_std, seed_note = _half_life_ms(s)
        half_ms_r = (
            s.staleness_halflife_range * 1e3 if s.staleness_halflife_range is not None else None
        )
        half_cell = _fmt_pm(half_ms, half_ms_std, ".3f")
        if seed_note:
            half_cell = "~" + half_cell + seed_note
            seeded = True
        lines.append(
            f"{s.app:24} {t.n_parties:>7} {t.request_cv:>6.2f} {t.fano_factor:>6.2f} "
            f"{t.msgs_per_pair:>9.2f} {t.bytes_per_pair:>7.2f} "
            f"{t.deadline_fraction:>8.2f} "
            f"{_fmt_pm(s.fidelity_threshold, s.fidelity_threshold_std, '.3f'):>15} "
            f"{_fmt(s.fidelity_range, '.2f'):>6} "
            f"{half_cell:>18} "
            f"{_fmt(half_ms_r, '.3f'):>11}"
        )
    if seeded:
        lines.append(
            "~ = the mean curve does not cross; value is the median over the "
            "(crossing/total) seeds that do."
        )
    return "\n".join(lines)

render_latex

render_latex(signatures: list[AppSignature]) -> str

The same table as a booktabs tabular, for \input{} into a paper.

Emits the tabular only — no table environment, caption or label — so the surrounding float stays in the manuscript and only the numbers are generated. Requires booktabs and a \code macro, both of which the manuscript defines.

Source code in qnetbench/characterize/report.py

def render_latex(signatures: list[AppSignature]) -> str:
    """The same table as a booktabs `tabular`, for \\input{} into a paper.

    Emits the tabular only — no table environment, caption or label — so the
    surrounding float stays in the manuscript and only the numbers are generated.
    Requires booktabs and a \\code macro, both of which the manuscript defines.
    """
    head = [
        "% Generated by `qnetbench characterize --out`. Do not edit by hand.",
        "% Requires: booktabs (\\toprule etc.) and a \\code{} macro.",
        r"\begin{tabular}{@{}lrrrrrrrrrr@{}}",
        r"\toprule",
        "Application & parties & $cv$ & Fano & msg/pair & B/pair & deadline & "
        r"$F_{1/2}$ & $\Delta_F$ & $t_{1/2}$ (ms) & $t_{1/2}^{r}$ \\",
        r"\midrule",
    ]
    rows = []
    seeded = False
    for s in signatures:
        t = s.trace
        half_ms, half_ms_std, seed_note = _half_life_ms(s)
        half_ms_r = (
            s.staleness_halflife_range * 1e3 if s.staleness_halflife_range is not None else None
        )
        if half_ms is None:
            half_cell = "---"
        elif seed_note:
            seeded = True
            half_cell = rf"$\sim${half_ms:.3f}{seed_note}"
        elif half_ms_std:
            half_cell = rf"{half_ms:.3f} $\pm$ {half_ms_std:.3f}"
        else:
            half_cell = f"{half_ms:.3f}"
        if s.fidelity_threshold is None:
            fid_cell = "---"
        elif s.fidelity_threshold_std:
            fid_cell = rf"{s.fidelity_threshold:.3f} $\pm$ {s.fidelity_threshold_std:.3f}"
        else:
            fid_cell = f"{s.fidelity_threshold:.3f}"
        rows.append(
            rf"\code{{{_tex_escape(s.app)}}} & {t.n_parties} & {t.request_cv:.2f} & "
            rf"{t.fano_factor:.2f} & {t.msgs_per_pair:.2f} & {t.bytes_per_pair:.2f} & "
            rf"{t.deadline_fraction:.2f} & "
            rf"{fid_cell} & {_fmt_tex(s.fidelity_range, '.2f')} & {half_cell} & "
            rf"{_fmt_tex(half_ms_r, '.3f')} \\"
        )
    tail = [r"\bottomrule", r"\end{tabular}"]
    if seeded:
        tail.append(
            "% $\\sim$ marks a half-life the mean curve does not resolve: the value is "
            "the median over the (crossing/total) seeds that do."
        )
    return "\n".join(head + rows + tail) + "\n"

Single-trace axes

qnetbench.characterize.signature

Single-trace demand-signature extraction.

The dimensions here are read directly from one run's trace (no parameter sweep): burstiness of entanglement requests, classical-communication coupling, deadline-criticality, staleness-intolerance, and multipartiteness. The fidelity/staleness curves (which need sweeps) live in curves.py.

TraceSignature

Bases: BaseModel

Demand-signature dimensions measurable from a single trace.

request_cv class-attribute instance-attribute

request_cv: float = 0.0

fano_factor class-attribute instance-attribute

fano_factor: float = 0.0

msgs_per_pair class-attribute instance-attribute

msgs_per_pair: float = 0.0

bytes_per_pair class-attribute instance-attribute

bytes_per_pair: float = 0.0

deadline_fraction class-attribute instance-attribute

deadline_fraction: float = 0.0

min_latency_budget class-attribute instance-attribute

min_latency_budget: float | None = None

min_staleness_tolerance class-attribute instance-attribute

min_staleness_tolerance: float | None = None

n_parties class-attribute instance-attribute

n_parties: int = 0

characterize_trace

characterize_trace(events: list[Event]) -> TraceSignature

Source code in qnetbench/characterize/signature.py

def characterize_trace(events: list[Event]) -> TraceSignature:
    requests = [e for e in events if isinstance(e, EntanglementRequested)]
    delivered = sum(1 for e in events if isinstance(e, EntanglementDelivered))
    sent = [e for e in events if isinstance(e, QubitSent)]  # single-qubit transmissions
    classical = [e for e in events if isinstance(e, ClassicalMessage)]

    sig = TraceSignature()

    # --- burstiness (over the demand events: pair requests and qubit sends) ---
    demand_times = sorted([e.t for e in requests] + [e.t for e in sent])
    gaps = [b - a for a, b in zip(demand_times, demand_times[1:], strict=False)]
    if gaps and statistics.mean(gaps) > 0:
        sig.request_cv = statistics.pstdev(gaps) / statistics.mean(gaps)
    sig.fano_factor = _fano_factor(demand_times)

    # --- classical coupling (per delivered pair or transmitted qubit) ---
    demand_units = delivered + len(sent)
    if demand_units:
        sig.msgs_per_pair = len(classical) / demand_units
        sig.bytes_per_pair = sum(e.n_bytes for e in classical) / demand_units

    # --- deadline-criticality & staleness ---
    if requests:
        with_deadline = sum(
            1
            for e in requests
            if e.demand.deadline is not None or e.demand.latency_budget is not None
        )
        sig.deadline_fraction = with_deadline / len(requests)
        budgets = [e.demand.latency_budget for e in requests if e.demand.latency_budget is not None]
        sig.min_latency_budget = min(budgets) if budgets else None
        tolerances = [
            e.demand.staleness_tolerance
            for e in requests
            if e.demand.staleness_tolerance is not None
        ]
        sig.min_staleness_tolerance = min(tolerances) if tolerances else None

    # --- multipartiteness ---
    nodes = {e.src for e in requests} | {e.dst for e in requests}
    nodes |= {e.src for e in sent} | {e.dst for e in sent}
    sig.n_parties = len(nodes)
    return sig

Swept curves

qnetbench.characterize.curves

Parametric demand-signature curves.

These need parameter sweeps rather than a single trace: the fidelity-sensitivity curve (utility vs delivered fidelity) and the staleness-tolerance curve (utility vs age of a pre-generated pair — directly feeding Issue #5). Both run on the reference backend, which is deterministic and fast, and both are regenerable from source so the characterization figures never drift from the code.

SEEDS module-attribute

SEEDS = 32

FIDELITY_TOL module-attribute

FIDELITY_TOL = 0.01

STALENESS_ABS_TOL module-attribute

STALENESS_ABS_TOL = 1e-05

STALENESS_REL_TOL module-attribute

STALENESS_REL_TOL = 0.05

Curve dataclass

Curve(
    x: list[float],
    y: list[float],
    xlabel: str,
    ylabel: str = "utility",
    y_std: list[float] = list(),
    seed_utils: list[list[float]] = list(),
)

x instance-attribute

x: list[float]

y instance-attribute

y: list[float]

xlabel instance-attribute

xlabel: str

ylabel class-attribute instance-attribute

ylabel: str = 'utility'

y_std class-attribute instance-attribute

y_std: list[float] = field(default_factory=list)

seed_utils class-attribute instance-attribute

seed_utils: list[list[float]] = field(default_factory=list)

insert

insert(xi: float, utils: list[float]) -> None

Insert one evaluated sweep point, keeping the curve sorted in x. Used by the bisection refinement, which adds points only where a crossing lies.

Source code in qnetbench/characterize/curves.py

def insert(self, xi: float, utils: list[float]) -> None:
    """Insert one evaluated sweep point, keeping the curve sorted in x. Used by
    the bisection refinement, which adds points only where a crossing lies."""
    i = bisect.bisect_left(self.x, xi)
    self.x.insert(i, xi)
    self.y.insert(i, statistics.mean(utils))
    self.y_std.insert(i, _stdev(utils))
    self.seed_utils.insert(i, utils)

as_rows

as_rows() -> list[dict[str, float]]

Source code in qnetbench/characterize/curves.py

def as_rows(self) -> list[dict[str, float]]:
    std = self.y_std or [0.0] * len(self.x)
    return [
        {self.xlabel: xi, self.ylabel: yi, f"{self.ylabel}_std": si}
        for xi, yi, si in zip(self.x, self.y, std, strict=True)
    ]

CharacterizationCurves dataclass

CharacterizationCurves(
    app: str,
    fidelity: Curve,
    staleness: Curve,
    fidelity_threshold: float | None = None,
    staleness_halflife: float | None = None,
    fidelity_threshold_std: float | None = None,
    staleness_halflife_std: float | None = None,
    fidelity_threshold_seed_median: float | None = None,
    staleness_halflife_seed_median: float | None = None,
    fidelity_threshold_seed_count: int = 0,
    staleness_halflife_seed_count: int = 0,
    n_seeds: int = 0,
    fidelity_range: float | None = None,
    staleness_halflife_range: float | None = None,
    fidelity_threshold_bracket: float | None = None,
    staleness_halflife_bracket: float | None = None,
)

app instance-attribute

app: str

fidelity instance-attribute

fidelity: Curve

staleness instance-attribute

staleness: Curve

fidelity_threshold class-attribute instance-attribute

fidelity_threshold: float | None = field(default=None)

staleness_halflife class-attribute instance-attribute

staleness_halflife: float | None = field(default=None)

fidelity_threshold_std class-attribute instance-attribute

fidelity_threshold_std: float | None = field(default=None)

staleness_halflife_std class-attribute instance-attribute

staleness_halflife_std: float | None = field(default=None)

fidelity_threshold_seed_median class-attribute instance-attribute

fidelity_threshold_seed_median: float | None = field(
    default=None
)

staleness_halflife_seed_median class-attribute instance-attribute

staleness_halflife_seed_median: float | None = field(
    default=None
)

fidelity_threshold_seed_count class-attribute instance-attribute

fidelity_threshold_seed_count: int = field(default=0)

staleness_halflife_seed_count class-attribute instance-attribute

staleness_halflife_seed_count: int = field(default=0)

n_seeds class-attribute instance-attribute

n_seeds: int = field(default=0)

fidelity_range class-attribute instance-attribute

fidelity_range: float | None = field(default=None)

staleness_halflife_range class-attribute instance-attribute

staleness_halflife_range: float | None = field(default=None)

fidelity_threshold_bracket class-attribute instance-attribute

fidelity_threshold_bracket: float | None = field(
    default=None
)

staleness_halflife_bracket class-attribute instance-attribute

staleness_halflife_bracket: float | None = field(
    default=None
)

CrossingStats dataclass

CrossingStats(
    median: float | None = None,
    std: float | None = None,
    n_crossing: int = 0,
    n_seeds: int = 0,
)

Where the individual seeds cross, as distinct from where the mean curve does.

The two can disagree, and the disagreement is informative. A curve that plateaus just above the level never crosses in the mean while a minority of seeds dip below it; reporting only the mean crossing then yields a spread with no location to be a spread of. median carries the location the crossing seeds agree on and n_crossing says how many of them resolved one at all, so a missing aggregate crossing can be reported as what it is — a curve that mostly plateaus — rather than as a bare dash.

median class-attribute instance-attribute

median: float | None = None

std class-attribute instance-attribute

std: float | None = None

n_crossing class-attribute instance-attribute

n_crossing: int = 0

n_seeds class-attribute instance-attribute

n_seeds: int = 0

fidelity_curve

fidelity_curve(
    app: str,
    fidelities: Sequence[float],
    seeds: Sequence[int] = range(SEEDS),
) -> Curve

Source code in qnetbench/characterize/curves.py

def fidelity_curve(
    app: str, fidelities: Sequence[float], seeds: Sequence[int] = range(SEEDS)
) -> Curve:
    evaluate = _fidelity_evaluator(app, seeds)
    x: list[float] = []
    y: list[float] = []
    y_std: list[float] = []
    seed_utils: list[list[float]] = []
    for f in fidelities:
        utils = evaluate(f)
        x.append(f)
        y.append(statistics.mean(utils))
        y_std.append(_stdev(utils))
        seed_utils.append(utils)
    return Curve(x=x, y=y, y_std=y_std, xlabel="delivered_fidelity", seed_utils=seed_utils)

staleness_curve

staleness_curve(
    app: str,
    ages: Sequence[float],
    coherence_time: float,
    base_fidelity: float = 1.0,
    seeds: Sequence[int] = range(SEEDS),
) -> Curve

Source code in qnetbench/characterize/curves.py

def staleness_curve(
    app: str,
    ages: Sequence[float],
    coherence_time: float,
    base_fidelity: float = 1.0,
    seeds: Sequence[int] = range(SEEDS),
) -> Curve:
    evaluate = _staleness_evaluator(app, coherence_time, base_fidelity, seeds)
    x: list[float] = []
    y: list[float] = []
    y_std: list[float] = []
    seed_utils: list[list[float]] = []
    for age in ages:
        utils = evaluate(age)
        x.append(age)
        y.append(statistics.mean(utils))
        y_std.append(_stdev(utils))
        seed_utils.append(utils)
    return Curve(x=x, y=y, y_std=y_std, xlabel="pair_age_s", seed_utils=seed_utils)

characterize_curves

characterize_curves(
    app: str,
    *,
    coherence_time: float = 0.001,
    seeds: Sequence[int] = range(SEEDS),
) -> CharacterizationCurves

Source code in qnetbench/characterize/curves.py

def characterize_curves(
    app: str,
    *,
    coherence_time: float = 1e-3,
    seeds: Sequence[int] = range(SEEDS),
) -> CharacterizationCurves:
    fid = fidelity_curve(app, [0.5, 0.6, 0.7, 0.8, 0.9, 0.95, 1.0], seeds)
    ages = [0.0, 2e-4, 5e-4, 1e-3, 2e-3, 5e-3, 1e-2]
    stale = staleness_curve(app, ages, coherence_time=coherence_time, seeds=seeds)

    # Thresholds are relative to each app's own utility *range*, so they compare
    # apps whose peak utility differs (e.g. QKD's utility is a key fraction < 0.5)
    # and, crucially, apps whose utility floor is nonzero. Crossing against fmax/2
    # would mark any application with u(F_min) > fmax/2 as having no threshold at
    # all, however strongly its utility depends on fidelity: entanglement swapping
    # runs 0.52 -> 1.00 across the swept range and would be reported as
    # fidelity-insensitive. Taking the midpoint of [fmin, fmax] removes the floor
    # and leaves apps with fmin = 0 (QKD and the other thresholded protocols)
    # exactly where they were.
    # Crossing levels are fixed from the coarse sweep and held there while the
    # curve is refined, so that adding points cannot move the target underneath us.
    fmax = max(fid.y) if fid.y else 0.0
    fmin = min(fid.y) if fid.y else 0.0
    frange = (fid.y[-1] - fid.y[0]) if fid.y else None
    level = (fmin + fmax) / 2
    resolved = bool(fid.y) and fmax > fmin
    threshold = None
    threshold_std = None
    threshold_bracket = None
    fid_stats = CrossingStats()
    if resolved:
        threshold_bracket = _refine_crossing(
            fid, _fidelity_evaluator(app, seeds), level, rising=True, abs_tol=FIDELITY_TOL
        )
        threshold = _first_crossing(fid.x, fid.y, level, rising=True)
        fid_stats = _crossing_stats(fid.x, fid.seed_utils, level, rising=True)
        threshold_std = fid_stats.std

    half = None
    half_std = None
    half_range = None
    half_bracket = None
    stale_stats = CrossingStats()
    if stale.y and stale.y[0] > 0:
        stale_eval = _staleness_evaluator(app, coherence_time, 1.0, seeds)
        smin = min(stale.y)
        half_bracket = _refine_crossing(
            stale, stale_eval, stale.y[0] / 2, rising=False,
            abs_tol=STALENESS_ABS_TOL, rel_tol=STALENESS_REL_TOL,
        )
        half = _first_crossing(stale.x, stale.y, stale.y[0] / 2, rising=False)
        stale_stats = _crossing_stats(stale.x, stale.seed_utils, stale.y[0] / 2, rising=False)
        half_std = stale_stats.std
        if stale.y[0] > smin:
            # Refined to the same tolerance, on the same (now denser) curve.
            _refine_crossing(
                stale, stale_eval, (stale.y[0] + smin) / 2, rising=False,
                abs_tol=STALENESS_ABS_TOL, rel_tol=STALENESS_REL_TOL,
            )
            half_range = _first_crossing(stale.x, stale.y, (stale.y[0] + smin) / 2, rising=False)
    return CharacterizationCurves(
        app=app,
        fidelity=fid,
        staleness=stale,
        fidelity_threshold=threshold,
        fidelity_range=frange,
        staleness_halflife=half,
        staleness_halflife_range=half_range,
        fidelity_threshold_std=threshold_std,
        staleness_halflife_std=half_std,
        fidelity_threshold_bracket=threshold_bracket,
        staleness_halflife_bracket=half_bracket,
        fidelity_threshold_seed_median=fid_stats.median,
        staleness_halflife_seed_median=stale_stats.median,
        fidelity_threshold_seed_count=fid_stats.n_crossing,
        staleness_halflife_seed_count=stale_stats.n_crossing,
        n_seeds=len(seeds),
    )

Provenance

qnetbench.characterize.provenance

Run provenance for generated characterization data.

A characterization run writes one JSON file per application, incrementally, over several minutes. If it dies half way the directory still looks finished: the files present are individually valid, they are simply from two different runs, and nothing in them says so. A later consumer — a plotting script, a paper table — then silently mixes them.

That is not hypothetical. It is how this project's own Table III came to hold numbers from two runs at once, and the only reason it was caught is that somebody re-derived the numbers by hand. So every per-app file records the id of the run that wrote it, the directory carries a manifest describing that run, and the consumers refuse to build from a directory where the two disagree.

The manifest is written twice: once at the start with complete=False, and once at the end with complete=True. An interrupted run therefore leaves a directory that positively declares itself unfinished, rather than one that is merely missing something nobody thought to check.

MANIFEST_NAME module-attribute

MANIFEST_NAME = 'manifest.json'

RunConsistencyError

Bases: RuntimeError

A data directory does not hold exactly one complete characterization run.

RunManifest

Bases: BaseModel

What produced the data in one output directory.

run_id instance-attribute

run_id: str

generated_at instance-attribute

generated_at: str

qnetbench_version instance-attribute

qnetbench_version: str

git_commit class-attribute instance-attribute

git_commit: str | None = None

git_dirty class-attribute instance-attribute

git_dirty: bool | None = None

seeds class-attribute instance-attribute

seeds: int = 0

apps class-attribute instance-attribute

apps: list[str] = []

complete class-attribute instance-attribute

complete: bool = False

new_run_id

new_run_id() -> str

Source code in qnetbench/characterize/provenance.py

def new_run_id() -> str:
    return uuid.uuid4().hex

git_state

git_state() -> tuple[str | None, bool | None]

(commit, dirty). Both None outside a git checkout — the data is still usable, it just cannot be tied back to a revision.

Source code in qnetbench/characterize/provenance.py

def git_state() -> tuple[str | None, bool | None]:
    """(commit, dirty). Both None outside a git checkout — the data is still usable,
    it just cannot be tied back to a revision."""
    commit = _git("rev-parse", "HEAD")
    if commit is None:
        return None, None
    status = _git("status", "--porcelain")
    return commit, bool(status) if status is not None else None

start_run

start_run(
    out_dir: Path, apps: list[str], seeds: int
) -> RunManifest

Stamp a directory as belonging to a new, not-yet-finished run.

Source code in qnetbench/characterize/provenance.py

def start_run(out_dir: Path, apps: list[str], seeds: int) -> RunManifest:
    """Stamp a directory as belonging to a new, not-yet-finished run."""
    commit, dirty = git_state()
    manifest = RunManifest(
        run_id=new_run_id(),
        generated_at=datetime.now(timezone.utc).isoformat(timespec="seconds"),
        qnetbench_version=qnetbench.__version__,
        git_commit=commit,
        git_dirty=dirty,
        seeds=seeds,
        apps=list(apps),
        complete=False,
    )
    write_manifest(out_dir, manifest)
    return manifest

finish_run

finish_run(out_dir: Path, manifest: RunManifest) -> None

Source code in qnetbench/characterize/provenance.py

def finish_run(out_dir: Path, manifest: RunManifest) -> None:
    write_manifest(out_dir, manifest.model_copy(update={"complete": True}))

write_manifest

write_manifest(
    out_dir: Path, manifest: RunManifest
) -> None

Source code in qnetbench/characterize/provenance.py

def write_manifest(out_dir: Path, manifest: RunManifest) -> None:
    write_atomic(out_dir / MANIFEST_NAME, manifest.model_dump_json(indent=2) + "\n")

write_atomic

write_atomic(path: Path, text: str) -> None

Write via a temporary file in the same directory, then rename.

A plain > redirect truncates its target the moment the shell opens it, so a run that dies before printing leaves an empty file where a reader expects a table. Renaming into place means the file is either the previous contents or the new ones, never a half-written state.

Source code in qnetbench/characterize/provenance.py

def write_atomic(path: Path, text: str) -> None:
    """Write via a temporary file in the same directory, then rename.

    A plain `>` redirect truncates its target the moment the shell opens it, so a
    run that dies before printing leaves an empty file where a reader expects a
    table. Renaming into place means the file is either the previous contents or
    the new ones, never a half-written state.
    """
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(text)
    os.replace(tmp, path)

load_manifest

load_manifest(out_dir: Path) -> RunManifest

Source code in qnetbench/characterize/provenance.py

def load_manifest(out_dir: Path) -> RunManifest:
    path = Path(out_dir) / MANIFEST_NAME
    if not path.exists():
        raise RunConsistencyError(
            f"{path} not found: this directory predates run manifests, or was not "
            f"written by `qnetbench characterize --out`. Regenerate it with "
            f"`qnetbench characterize --out {out_dir}`."
        )
    return RunManifest.model_validate_json(path.read_text())

app_files

app_files(out_dir: Path) -> list[Path]

The per-app signature files in a directory, manifest excluded.

Source code in qnetbench/characterize/provenance.py

def app_files(out_dir: Path) -> list[Path]:
    """The per-app signature files in a directory, manifest excluded."""
    return sorted(p for p in Path(out_dir).glob("*.json") if p.name != MANIFEST_NAME)

verify_run

verify_run(out_dir: Path) -> RunManifest

Check that a directory holds exactly one complete run, and return its manifest.

Raises RunConsistencyError describing the specific failure, because the useful thing to tell somebody whose figure is about to be wrong is which files came from where.

Source code in qnetbench/characterize/provenance.py

def verify_run(out_dir: Path) -> RunManifest:
    """Check that a directory holds exactly one complete run, and return its manifest.

    Raises RunConsistencyError describing the specific failure, because the useful
    thing to tell somebody whose figure is about to be wrong is which files came
    from where.
    """
    out_dir = Path(out_dir)
    manifest = load_manifest(out_dir)
    files = app_files(out_dir)

    foreign: dict[str, list[str]] = {}
    for path in files:
        try:
            payload = json.loads(path.read_text())
        except (OSError, json.JSONDecodeError) as exc:
            raise RunConsistencyError(f"{path} is not readable JSON: {exc}") from exc
        run_id = payload.get("run_id")
        if run_id != manifest.run_id:
            foreign.setdefault(str(run_id), []).append(path.stem)
    if foreign:
        detail = "; ".join(
            f"{len(names)} file(s) "
            + ("unstamped (written before run manifests)" if rid == "None" else f"from run {rid}")
            + f": {', '.join(sorted(names)[:6])}{'...' if len(names) > 6 else ''}"
            for rid, names in foreign.items()
        )
        raise RunConsistencyError(
            f"{out_dir} does not hold a single characterization run. Manifest says "
            f"{manifest.run_id}, but {detail}. Regenerate the whole directory in one "
            f"run: `qnetbench characterize --out {out_dir}`."
        )

    present = {p.stem for p in files}
    missing = [a for a in manifest.apps if a not in present]
    if not manifest.complete or missing:
        raise RunConsistencyError(
            f"{out_dir} holds an unfinished run ({len(present)}/{len(manifest.apps)} apps"
            + (f", missing {', '.join(missing[:6])}" if missing else "")
            + "). Re-run `qnetbench characterize --out "
            + f"{out_dir}` to completion before building anything from it."
        )
    return manifest