k4bench.regression.report_builder

k4bench.regression.report_builder

Assemble the nightly regression report from the EOS run history.

Walks every (detector, platform, sample) triple found under the WebEOS data URL (the same hierarchy the dashboard's sidebar cascades through — these triples have independent baselines and are never pooled), pulls a trailing window of runs into the local cache, rebuilds the trend frames with :mod:k4bench.analysis.trend, attaches per-run reliability verdicts with :mod:k4bench.results.reliability_evidence, and runs the step detector in :mod:k4bench.regression.engine over every metric series.

unjudged_value_verdicts

unjudged_value_verdicts(*, detector: str, platform: str, sample: str, results_df: DataFrame | None, event_df: DataFrame | None, tonight: str, already: set[tuple[str, str]]) -> list[MetricVerdict]

Raw metric values for tonight's run as unjudged UNKNOWN verdicts.

The engine skips unreliable runs (they must not pollute baselines or flags), so their metrics get no verdict and their values would never reach the report the dashboard's Overview tab reads — leaving that tab unable to plot them even with "Exclude unreliable runs" off. This records tonight's raw value for every (label, metric) not already judged, marked UNKNOWN (never a flag), so the value is preserved for display. A normally-judged (reliable) run already has a verdict per metric, so already covers it and this adds nothing.

Source code in k4bench/regression/report_builder.py

def unjudged_value_verdicts(
    *,
    detector: str,
    platform: str,
    sample: str,
    results_df: pd.DataFrame | None,
    event_df: pd.DataFrame | None,
    tonight: str,
    already: set[tuple[str, str]],
) -> list[MetricVerdict]:
    """Raw metric values for *tonight*'s run as unjudged ``UNKNOWN`` verdicts.

    The engine skips unreliable runs (they must not pollute baselines or flags),
    so their metrics get no verdict and their values would never reach the
    report the dashboard's Overview tab reads — leaving that tab unable to plot
    them even with "Exclude unreliable runs" off. This records tonight's raw
    value for every ``(label, metric)`` not *already* judged, marked ``UNKNOWN``
    (never a flag), so the value is preserved for display. A normally-judged
    (reliable) run already has a verdict per metric, so *already* covers it and
    this adds nothing.
    """
    out: list[MetricVerdict] = []

    def _emit(df: pd.DataFrame | None, metrics: dict[str, str]) -> None:
        if df is None or df.empty:
            return
        tonight_rows = df[df["run_id"] == tonight]
        for label in sorted(tonight_rows["label"].dropna().unique()):
            row = tonight_rows[tonight_rows["label"] == label]
            for metric, family in metrics.items():
                if metric not in row.columns or (str(label), metric) in already:
                    continue
                val = row[metric].iloc[0]
                if pd.isna(val) or not math.isfinite(float(val)):
                    continue
                out.append(MetricVerdict(
                    detector=detector, platform=platform, sample=sample,
                    label=str(label), metric_family=family, metric=metric,
                    sub_detector=None, run_id=tonight, run_date=tonight,
                    value=float(val), baseline_median=None, baseline_mad=None,
                    pct_change=None, z_score=None,
                    severity=Severity.UNKNOWN, direction=Direction.NONE,
                    reason="unreliable host — value recorded but not judged",
                ))

    results = _with_cpu_efficiency(results_df) if results_df is not None else None
    _emit(results, RUN_METRICS)
    _emit(event_df, EVENT_METRICS)
    return out

evaluate_group_series

evaluate_group_series(*, detector: str, platform: str, sample: str, results_df: DataFrame | None, event_df: DataFrame | None, reliability: dict[str, bool | None], hosts: dict[str, HostFact] | None = None) -> dict[SeriesId, list[MetricVerdict]]

Run the step detector over every run/event metric series of one run group. Region timings are not walked.

Returns the full verdict series per :class:SeriesId — the nightly report takes each series' verdict for the report night, while the dashboard drill-down and the retrospective threshold validation consume the whole walk.

hosts (from :func:~k4bench.regression.history.host_facts) names the machine behind each run, and only reaches the history tails attached to confirmed verdicts; it never enters the judgement itself. Omitted, the tails simply carry no host.

Source code in k4bench/regression/report_builder.py

def evaluate_group_series(
    *,
    detector: str,
    platform: str,
    sample: str,
    results_df: pd.DataFrame | None,
    event_df: pd.DataFrame | None,
    reliability: dict[str, bool | None],
    hosts: dict[str, HostFact] | None = None,
) -> dict[SeriesId, list[MetricVerdict]]:
    """Run the step detector over every run/event metric series of one run
    group. Region timings are not walked.

    Returns the **full verdict series** per :class:`SeriesId` — the nightly
    report takes each series' verdict for the report night, while the
    dashboard drill-down and the retrospective threshold validation consume
    the whole walk.

    *hosts* (from :func:`~k4bench.regression.history.host_facts`) names the
    machine behind each run, and only reaches the history tails attached to
    confirmed verdicts; it never enters the judgement itself. Omitted, the tails
    simply carry no host.
    """
    out: dict[SeriesId, list[MetricVerdict]] = {}

    def _run(df, mask, series):
        history = _series_history(df, mask, series.metric, reliability)
        verdicts = evaluate_series(history, series=series)
        if verdicts:
            out[series] = _with_history(history, verdicts, hosts or {})

    if results_df is not None and not results_df.empty:
        df = _with_cpu_efficiency(results_df)
        for label in sorted(df["label"].dropna().unique()):
            mask = df["label"] == label
            for metric, family in RUN_METRICS.items():
                if metric not in df.columns:
                    continue
                sid = SeriesId(detector, platform, sample, str(label), family, metric)
                _run(df, mask, sid)

    if event_df is not None and not event_df.empty:
        for label in sorted(event_df["label"].dropna().unique()):
            mask = event_df["label"] == label
            for metric, family in EVENT_METRICS.items():
                if metric not in event_df.columns:
                    continue
                sid = SeriesId(detector, platform, sample, str(label), family, metric)
                _run(event_df, mask, sid)

    return out

build_group_report

build_group_report(data_url: str, cache_dir: str | None, detector: str, platform: str, sample: str, *, fetch_window_runs: int = FETCH_WINDOW_RUNS, as_of: str | None = None) -> RunGroupReport | None

Build one triple's report from its trailing run window, or None when the triple has no fetchable runs at all.

as_of (a YYYY-MM-DD night) truncates the run history to runs on or before that night before the trailing window is taken, reproducing the report that night's runs would have produced — the seam the historical backfill drives. None judges the full history (the nightly CI case).

Source code in k4bench/regression/report_builder.py

def build_group_report(
    data_url: str,
    cache_dir: str | None,
    detector: str,
    platform: str,
    sample: str,
    *,
    fetch_window_runs: int = FETCH_WINDOW_RUNS,
    as_of: str | None = None,
) -> RunGroupReport | None:
    """Build one triple's report from its trailing run window, or ``None``
    when the triple has no fetchable runs at all.

    *as_of* (a ``YYYY-MM-DD`` night) truncates the run history to runs on or
    before that night before the trailing window is taken, reproducing the
    report that night's runs would have produced — the seam the historical
    backfill drives. ``None`` judges the full history (the nightly CI case).
    """
    stacks_dates = list_run_dates_all_stacks(data_url, detector, platform, sample)
    pairs = sorted(
        (date, stack) for stack, dates in stacks_dates.items() for date in dates
        if as_of is None or date <= as_of
    )[-fetch_window_runs:]
    if not pairs:
        return None
    window: dict[str, list[str]] = {}
    for date, stack in pairs:
        window.setdefault(stack, []).append(date)
    runs = fetch_runs_windowed(data_url, detector, platform, sample, window, cache_root=cache_dir)
    if not runs:
        return None
    run_dirs = tuple(r["run_dir"] for r in sorted(runs, key=lambda r: r["date"]))
    return group_report_from_run_dirs(detector, platform, sample, run_dirs)

group_report_from_run_dirs

group_report_from_run_dirs(detector: str, platform: str, sample: str, run_dirs: tuple[str, ...]) -> RunGroupReport | None

Build one triple's report from already-local run directories (ordered oldest → newest; each directory's name is its nightly date).

Source code in k4bench/regression/report_builder.py

def group_report_from_run_dirs(
    detector: str,
    platform: str,
    sample: str,
    run_dirs: tuple[str, ...],
) -> RunGroupReport | None:
    """Build one triple's report from already-local run directories (ordered
    oldest → newest; each directory's name is its nightly date)."""
    if not run_dirs:
        return None
    results_df = build_results_trend(run_dirs)
    event_df = build_event_timing_trend(run_dirs)
    machine_df = build_machine_info_trend(run_dirs)
    reliability = run_reliability_map(results_df, machine_df)
    tonight = max(Path(d).name for d in run_dirs)
    group = _group_report_from_frames(
        detector, platform, sample,
        results_df=results_df, event_df=event_df,
        reliability=reliability, tonight=tonight,
        hosts=host_facts(machine_df),
    )
    return None if group is None else _with_region_deltas(group, run_dirs)

build_nightly_report

build_nightly_report(data_url: str, cache_dir: str | None = None, *, fetch_window_runs: int = FETCH_WINDOW_RUNS, as_of: str | None = None) -> NightlyReport

Build the cross-detector report for the most recent nightly.

The report night is the newest run date seen across all triples. A triple dated earlier is still reported normally when its CI run says it came from the report night's own batch, whatever the gap between the two dates (see :func:_same_batch); for a night whose runs carry no CI run at all, a lag of up to :data:SAME_BATCH_LAG_DAYS stands in for that. Anything else gets a missing run job failure (a hard crash uploads nothing, so absence is itself the failure signal) — unless it is stale by more than :data:MISSING_RUN_GRACE_DAYS, in which case it is treated as retired and dropped.

as_of truncates every triple's history to runs on or before that night (see :func:build_group_report), making the report night the newest run ≤ as_of — the historical-backfill seam.

Source code in k4bench/regression/report_builder.py

def build_nightly_report(
    data_url: str,
    cache_dir: str | None = None,
    *,
    fetch_window_runs: int = FETCH_WINDOW_RUNS,
    as_of: str | None = None,
) -> NightlyReport:
    """Build the cross-detector report for the most recent nightly.

    The report night is the newest run date seen across all triples. A triple
    dated earlier is still reported normally when its CI run says it came from
    the report night's own batch, whatever the gap between the two dates (see
    :func:`_same_batch`); for a night whose runs carry no CI run at all, a lag
    of up to :data:`SAME_BATCH_LAG_DAYS` stands in for that. Anything else gets
    a *missing run* job failure (a hard crash uploads nothing, so absence is
    itself the failure signal) — unless it is stale by more than
    :data:`MISSING_RUN_GRACE_DAYS`, in which case it is treated as retired and
    dropped.

    *as_of* truncates every triple's history to runs on or before that night
    (see :func:`build_group_report`), making the report night the newest run
    ≤ *as_of* — the historical-backfill seam.
    """
    groups: list[RunGroupReport] = []
    for detector in list_detectors(data_url):
        for platform in list_platforms(data_url, detector):
            stack_samples = scan_stack_samples(data_url, detector, platform)
            samples = sorted({s for ss in stack_samples.values() for s in ss})
            for sample in samples:
                try:
                    group = build_group_report(
                        data_url, cache_dir, detector, platform, sample,
                        fetch_window_runs=fetch_window_runs, as_of=as_of,
                    )
                except Exception:
                    _log.exception(
                        "build_nightly_report: failed for %s/%s/%s",
                        detector, platform, sample,
                    )
                    continue
                if group is not None:
                    groups.append(group)

    return _finalize_report(groups)

build_nightly_report_local

build_nightly_report_local(data_dir: str, *, fetch_window_runs: int = FETCH_WINDOW_RUNS, as_of: str | None = None) -> NightlyReport

Like :func:build_nightly_report, but over a local directory tree with the same {detector}/{platform}/{stack}/{sample}/{date} layout as EOS (used by the integration test and for offline dry-runs; no network). as_of truncates each sample's runs the same way.

Source code in k4bench/regression/report_builder.py

def build_nightly_report_local(
    data_dir: str,
    *,
    fetch_window_runs: int = FETCH_WINDOW_RUNS,
    as_of: str | None = None,
) -> NightlyReport:
    """Like :func:`build_nightly_report`, but over a local directory tree with
    the same ``{detector}/{platform}/{stack}/{sample}/{date}`` layout as EOS
    (used by the integration test and for offline dry-runs; no network).
    *as_of* truncates each sample's runs the same way."""
    root = Path(data_dir)
    groups: list[RunGroupReport] = []
    for det_dir in sorted(p for p in root.iterdir() if p.is_dir()):
        if det_dir.name.startswith(("_", ".")):
            continue
        for plat_dir in sorted(p for p in det_dir.iterdir() if p.is_dir()):
            # Collect each sample's run dirs across all stacks.
            per_sample: dict[str, list[Path]] = {}
            for stack_dir in sorted(p for p in plat_dir.iterdir() if p.is_dir()):
                for sample_dir in sorted(p for p in stack_dir.iterdir() if p.is_dir()):
                    per_sample.setdefault(sample_dir.name, []).extend(
                        p for p in sample_dir.iterdir() if p.is_dir()
                    )
            for sample, run_paths in sorted(per_sample.items()):
                run_dirs = tuple(
                    str(p) for p in sorted(run_paths, key=lambda p: p.name)
                    if as_of is None or p.name <= as_of
                )[-fetch_window_runs:]
                group = group_report_from_run_dirs(
                    det_dir.name, plat_dir.name, sample, run_dirs
                )
                if group is not None:
                    groups.append(group)
    return _finalize_report(groups)