Skip to content

FetchResult

FetchResult dataclass 

FetchResult(bench_df: DataFrame, trace_df: DataFrame, opcounts: dict[str, dict[str, float]], meta: dict[str, Any], output_flags: dict[str, bool] = (lambda: {'estimator_inputs': True, 'merged_parquet': True, 'trace_parquet': True})())

In-memory result of a pipeline run.

Holds the parsed bench DataFrame, the derived trace DataFrame, and the opcount mapping needed to materialise the artifact bundle on disk (runtimes.csv, opcounts.json, bench_data.parquet, trace.parquet, meta.json).

The output_flags dict gates which artifacts write() produces; meta.json is always written.

default_out_dirname 

default_out_dirname() -> str

Return {earliest_run_ts}_{latest_run_ts} formatted YYYY-MM-DDTHH-MM-SSZ.

Source code in src/benchmarkoor_fetch/result.py 
36
37
38
39
40
41
def default_out_dirname(self) -> str:
    """Return `{earliest_run_ts}_{latest_run_ts}` formatted YYYY-MM-DDTHH-MM-SSZ."""
    window = self.meta.get("data_window") or {}
    start = _format_ts(window.get("start"))
    end = _format_ts(window.get("end"))
    return f"{start}_{end}"

write 

write(out_dir: Path) -> None

Write the artifact bundle to out_dir (created if missing).

Source code in src/benchmarkoor_fetch/result.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def write(self, out_dir: Path) -> None:
    """Write the artifact bundle to `out_dir` (created if missing)."""
    out_dir = Path(out_dir)
    out_dir.mkdir(parents=True, exist_ok=True)

    flags = self.output_flags
    row_counts: dict[str, int] = {"bench_data": int(len(self.bench_df))}

    if flags.get("estimator_inputs", True):
        runtimes = _runtimes_frame(self.bench_df)
        runtimes.to_csv(out_dir / "runtimes.csv", index=False)
        (out_dir / "opcounts.json").write_text(
            json.dumps(self.opcounts, indent=2, sort_keys=True)
        )
        row_counts["runtimes"] = int(len(runtimes))
        row_counts["opcounts"] = int(len(self.opcounts))

    if flags.get("merged_parquet", True):
        self.bench_df.to_parquet(out_dir / "bench_data.parquet", index=False)

    if flags.get("trace_parquet", True):
        self.trace_df.to_parquet(out_dir / "trace.parquet", index=False)
        row_counts["trace"] = int(len(self.trace_df))

    meta = dict(self.meta)
    meta["row_counts"] = row_counts
    (out_dir / "meta.json").write_text(
        json.dumps(meta, indent=2, sort_keys=True, default=str)
    )