Comparison & Batch Reports

Multi-run comparison (before/after deltas) and batch (cross-DUT yield) analysis and report building

Multi-run input model for comparison and batch reports.

A Run is a set of capture files plus per-run metadata; a RunSet is the ordered collection the ComparisonAnalyzer consumes. Comparison mode measures deltas against a baseline run; batch mode aggregates across runs (DUTs).

RunMetadata dataclass

RunMetadata(dut_id: Optional[str] = None, condition: Optional[str] = None, operator: Optional[str] = None, timestamp: Optional[datetime] = None, notes: Optional[str] = None)

Descriptive, optional context for one run.

Run dataclass

Run(label: str, files: List[Path], metadata: RunMetadata = RunMetadata(), waveforms: List[WaveformData] = list(), measurements: List[MeasurementResult] = list(), passed: Optional[bool] = None, incomplete: bool = False, load_errors: List[str] = list())

One test run: capture files in, analyzed waveforms/measurements out.

RunSet dataclass

RunSet(runs: List[Run], mode: str = MODE_COMPARISON, baseline_index: int = 0, criteria_set: Optional[CriteriaSet] = None)

The full input to a comparison or batch analysis.

validate

validate() -> None

Raise ValueError on structural problems (before any file I/O).

Source code in scpi_control/report_generator/models/comparison.py

def validate(self) -> None:
    """Raise ValueError on structural problems (before any file I/O)."""
    if len(self.runs) < 2:
        raise ValueError(f"A RunSet needs at least 2 runs, got {len(self.runs)}")
    labels = [run.label for run in self.runs]
    if len(set(labels)) != len(labels):
        raise ValueError(f"Run labels must be unique, got {labels}")
    if self.mode not in (MODE_COMPARISON, MODE_BATCH):
        raise ValueError(f"Unknown mode {self.mode!r}; use {MODE_COMPARISON!r} or {MODE_BATCH!r}")
    if not (0 <= self.baseline_index < len(self.runs)):
        raise ValueError(f"baseline_index {self.baseline_index} out of range for {len(self.runs)} runs")

DeltaEntry dataclass

DeltaEntry(run_label: str, value: Optional[float], delta: Optional[float], pct: Optional[float])

One non-baseline run's value for one statistic, relative to baseline.

AggregateStats dataclass

AggregateStats(mean: float, std: Optional[float], min: float, max: float, n: int)

Cross-run aggregate of one statistic (batch mode).

ComparisonResult dataclass

ComparisonResult(runset: RunSet, matched_channels: List[str], deltas: Dict[str, Dict[str, List[DeltaEntry]]] = dict(), aggregates: Dict[str, Dict[str, AggregateStats]] = dict(), yield_passed: Optional[int] = None, yield_total: Optional[int] = None, yield_incomplete: Optional[int] = None, warnings: List[str] = list())

Everything the report builder needs, fully computed.

Loads and analyzes a RunSet into a ComparisonResult.

Strict by default: a bad file fails the whole analysis with the run and file named. skip_bad_runs=True demotes that to a warning and drops the run, except when fewer than two runs survive or the comparison baseline itself is lost.

ComparisonAnalysisError

Bases: Exception

Analysis could not produce a usable result.

ComparisonAnalyzer

Turns a RunSet into a fully computed ComparisonResult.

evaluate_measurements

evaluate_measurements(waveforms: List[WaveformData], criteria_set: Optional[CriteriaSet], warnings: List[str], *, label: str) -> Tuple[List[MeasurementResult], Optional[bool], bool]

Turn waveforms' .statistics plus a CriteriaSet into MeasurementResult objects, and compute the resulting pass/fail verdict.

This is the shared core of ComparisonAnalyzer._apply_criteria, extracted so a single-capture (non-comparison) reporting path can apply the exact same criteria-evaluation semantics without a second, divergent implementation.

Only critical-severity criteria gate the verdict; warning/info criteria still produce MeasurementResults but never flip passed or incomplete. A criterion that cannot be evaluated (missing/non-numeric statistic, or not fully specified e.g. no min/max set) is surfaced as an entry appended to the caller-supplied, mutable warnings list -- never silently dropped. If such a criterion is critical, the verdict becomes incomplete rather than a false PASS or FAIL.

Parameters:

Name	Type	Description	Default
waveforms	List[WaveformData]	Waveforms to evaluate; each must already have .statistics populated (i.e. .analyze() already called).	required
criteria_set	Optional[CriteriaSet]	Criteria to apply, or None to skip evaluation entirely.	required
warnings	List[str]	Mutable list that un-evaluable-criterion messages are appended to, matching ComparisonAnalyzer.analyze's shared-warnings-list pattern -- not returned, since the caller already holds the list.	required
label	str	Identifies the source in warning text (e.g. "Run 'baseline'"), keyed-word argument so call sites read clearly at the call site.	required

Returns:

Type	Description
(measurements, passed, incomplete)
Optional[bool]	measurements: one MeasurementResult per criterion that could be resolved to a numeric statistic (regardless of severity).
bool	passed: False if any critical criterion failed; None if no critical criteria were evaluated (including when criteria_set is None); True if at least one critical criterion was evaluated and none failed.
Tuple[List[MeasurementResult], Optional[bool], bool]	incomplete: True only when a critical criterion could not be evaluated and no critical criterion definitively failed.

Source code in scpi_control/report_generator/analysis/comparison_analyzer.py

def evaluate_measurements(
    waveforms: List[WaveformData],
    criteria_set: Optional[CriteriaSet],
    warnings: List[str],
    *,
    label: str,
) -> Tuple[List[MeasurementResult], Optional[bool], bool]:
    """Turn waveforms' `.statistics` plus a `CriteriaSet` into `MeasurementResult`
    objects, and compute the resulting pass/fail verdict.

    This is the shared core of `ComparisonAnalyzer._apply_criteria`, extracted so
    a single-capture (non-comparison) reporting path can apply the exact same
    criteria-evaluation semantics without a second, divergent implementation.

    Only `critical`-severity criteria gate the verdict; `warning`/`info` criteria
    still produce `MeasurementResult`s but never flip `passed` or `incomplete`.
    A criterion that cannot be evaluated (missing/non-numeric statistic, or not
    fully specified e.g. no min/max set) is surfaced as an entry appended to the
    caller-supplied, mutable `warnings` list -- never silently dropped. If such a
    criterion is critical, the verdict becomes `incomplete` rather than a false
    PASS or FAIL.

    Args:
        waveforms: Waveforms to evaluate; each must already have `.statistics`
            populated (i.e. `.analyze()` already called).
        criteria_set: Criteria to apply, or None to skip evaluation entirely.
        warnings: Mutable list that un-evaluable-criterion messages are appended
            to, matching `ComparisonAnalyzer.analyze`'s shared-warnings-list
            pattern -- not returned, since the caller already holds the list.
        label: Identifies the source in warning text (e.g. "Run 'baseline'"),
            keyed-word argument so call sites read clearly at the call site.

    Returns:
        (measurements, passed, incomplete):
        - measurements: one `MeasurementResult` per criterion that could be
          resolved to a numeric statistic (regardless of severity).
        - passed: False if any critical criterion failed; None if no critical
          criteria were evaluated (including when criteria_set is None); True
          if at least one critical criterion was evaluated and none failed.
        - incomplete: True only when a critical criterion could not be
          evaluated and no critical criterion definitively failed.
    """
    measurements: List[MeasurementResult] = []
    if criteria_set is None:
        return measurements, None, False

    critical_pass: List[bool] = []
    critical_unevaluable = False
    for wf in waveforms:
        stats = wf.statistics or {}
        for criteria in criteria_set.criteria_list:
            if criteria.channel is not None and criteria.channel != wf.label:
                continue
            is_critical = (criteria.severity or "").lower() == "critical"
            key = ComparisonAnalyzer._resolve_stat_key(criteria.measurement_name, stats)
            value = stats.get(key) if key is not None else None
            if value is None or not isinstance(value, (int, float)) or isinstance(value, bool):
                warnings.append(f"{label}: criterion '{criteria.measurement_name}' on '{wf.label}' could not be evaluated (no numeric statistic)")
                if is_critical:
                    critical_unevaluable = True
                continue
            outcome = criteria.validate(float(value))
            measurements.append(
                MeasurementResult(
                    name=criteria.measurement_name,
                    value=float(value),
                    unit=STAT_UNITS.get(key, ""),
                    channel=wf.label,
                    passed=outcome.passed,
                    criteria_min=criteria.min_value,
                    criteria_max=criteria.max_value,
                )
            )
            if outcome.passed is None:
                warnings.append(f"{label}: criterion '{criteria.measurement_name}' on '{wf.label}' is not fully specified; not evaluated")
                if is_critical:
                    critical_unevaluable = True
            elif is_critical:
                critical_pass.append(outcome.passed)

    if any(p is False for p in critical_pass):
        return measurements, False, False
    if critical_unevaluable:
        return measurements, None, True
    if critical_pass:
        return measurements, True, False
    return measurements, None, False

Builds TestReports from analyzed RunSets, plus the manifest/sign-off helpers shared with the single-run report path.

build_manifest

build_manifest(runs: List[Run]) -> DataManifest

One entry per source file per run. Timestamp: provenance acquired_at, else waveform capture_timestamp, else None -- a file's mtime is when it was written or copied, not when the signal was acquired, so it is never substituted.

Source code in scpi_control/report_generator/comparison_report_builder.py

def build_manifest(runs: List[Run]) -> DataManifest:
    """One entry per source file per run. Timestamp: provenance acquired_at, else
    waveform capture_timestamp, else None -- a file's mtime is when it was written
    or copied, not when the signal was acquired, so it is never substituted."""
    entries: List[ManifestEntry] = []
    for run in runs:
        by_source: Dict[str, object] = {}
        for wf in run.waveforms:
            if wf.source_file is not None:
                by_source.setdefault(str(wf.source_file), wf)
        for filepath in run.files:
            path = Path(filepath)
            wf = by_source.get(str(path))
            provenance = getattr(wf, "provenance", None) if wf is not None else None
            timestamp = None
            if provenance is not None and provenance.acquired_at:
                timestamp = provenance.acquired_at
            elif wf is not None and getattr(wf, "capture_timestamp", None):
                timestamp = wf.capture_timestamp.isoformat()
            # No mtime fallback: a file's modification time is when it was written or
            # copied, not when the signal was acquired. Leaving this None renders as
            # "unknown", which is honest (audit M30).
            entries.append(
                ManifestEntry(
                    run_label=run.label,
                    file_path=str(path),
                    size_bytes=path.stat().st_size if path.exists() else 0,
                    sha256=_sha256(path) if path.exists() else "",
                    capture_timestamp=timestamp,
                    instrument=_instrument_string(provenance),
                )
            )
    return DataManifest(entries=entries)

build_comparison_report

build_comparison_report(result: ComparisonResult, metadata: ReportMetadata, template: Optional[ReportTemplate] = None, *, include_appendix: Optional[bool] = None, include_signoff: Optional[bool] = None) -> TestReport

Assemble a standard TestReport from an analyzed RunSet.

Explicit kwargs win; otherwise template settings; with neither, both the appendix and sign-off are included (comparison reports exist for traceability).

Source code in scpi_control/report_generator/comparison_report_builder.py

def build_comparison_report(
    result: ComparisonResult,
    metadata: ReportMetadata,
    template: Optional[ReportTemplate] = None,
    *,
    include_appendix: Optional[bool] = None,
    include_signoff: Optional[bool] = None,
) -> TestReport:
    """Assemble a standard TestReport from an analyzed RunSet.

    Explicit kwargs win; otherwise template settings; with neither, both the
    appendix and sign-off are included (comparison reports exist for
    traceability)."""
    runset = result.runset
    if include_appendix is None:
        include_appendix = template.include_raw_data_appendix if template is not None else True
    if include_signoff is None:
        include_signoff = template.include_signoff if template is not None else True

    report = TestReport(metadata=metadata)
    report.add_section(_overview_section(result))
    report.add_section(_overlay_section(result))
    if runset.mode == MODE_COMPARISON:
        report.add_section(_comparison_section(result))
    else:
        batch, aggregates = _batch_section(result)
        report.add_section(batch)
        report.add_section(aggregates)

    next_order = len(report.sections)
    if include_appendix:
        appendix = TestSection(title="Raw Data Appendix", order=next_order)
        appendix.manifest = build_manifest(runset.runs)
        appendix.comparison_table = _full_stats_table(result)
        for run in runset.runs:
            for m in run.measurements:
                channel = f"{run.label} · {m.channel}" if m.channel else run.label
                appendix.measurements.append(replace(m, channel=channel))
        report.add_section(appendix)
        next_order += 1
    if include_signoff:
        roles = template.signoff_roles if template is not None else list(DEFAULT_SIGNOFF_ROLES)
        names = template.signoff_names if template is not None else None
        signoff_section = TestSection(title="Sign-Off", order=next_order)
        signoff_section.signoff = build_signoff(roles, names)
        report.add_section(signoff_section)

    report.executive_summary = _executive_summary(result)
    report.summary_source = SUMMARY_SOURCE_COMPUTED
    evaluated = [run.passed for run in runset.runs if run.passed is not None]
    if False in evaluated:
        report.overall_result = "FAIL"
    elif any(run.incomplete for run in runset.runs):
        report.overall_result = "INCONCLUSIVE"
    elif evaluated:
        report.overall_result = "PASS"
    else:
        report.overall_result = "INCONCLUSIVE"
    return report

append_signoff_and_appendix

append_signoff_and_appendix(report: TestReport, template: ReportTemplate) -> None

Single-run path: honor the template's appendix/sign-off flags on an already-built report. Manifest is derived from section waveforms.

Source code in scpi_control/report_generator/comparison_report_builder.py

def append_signoff_and_appendix(report: TestReport, template: ReportTemplate) -> None:
    """Single-run path: honor the template's appendix/sign-off flags on an
    already-built report. Manifest is derived from section waveforms."""
    next_order = max((s.order for s in report.sections), default=-1) + 1
    if template.include_raw_data_appendix:
        pseudo = Run(label=report.metadata.title, files=[])
        seen_sources = set()
        for section in report.sections:
            for wf in section.waveforms:
                if wf.source_file is not None and str(wf.source_file) not in seen_sources:
                    seen_sources.add(str(wf.source_file))
                    pseudo.files.append(Path(wf.source_file))
                    pseudo.waveforms.append(wf)
        appendix = TestSection(title="Raw Data Appendix", order=next_order)
        appendix.manifest = build_manifest([pseudo])
        report.add_section(appendix)
        next_order += 1
    if template.include_signoff:
        signoff_section = TestSection(title="Sign-Off", order=next_order)
        signoff_section.signoff = build_signoff(template.signoff_roles, template.signoff_names)
        report.add_section(signoff_section)

See Also

• Waveform - Waveform acquisition and data handling
• Analysis - Signal analysis (FFT, THD, SNR)
• Signal Synthesis - Parameterized synthetic waveforms for hardware-free examples
• Report Generator: API Reference - Single-run report building blocks (ReportMetadata, TestReport, generators)