ModuleBuilt incomes with cpbenchy
cpbenchy.backend.run(submissions)- Version
- cpbenchy 0.1.0.dev0
- Last updated
- 6 Oct 2026 · 1 commit
- Authors
- ThomSerg
- Tags
- integrationapi
What it does
Lets another experiment tool use cpbenchy to measure runs. The tool decides which runs exist, which still need doing and where results are stored; cpbenchy measures the runs it is given and hands back the stats, tagged with the tool’s own ids.
Use it
from cpbenchy import Instance, Limits, RunSpec
from cpbenchy.backend import Submission, run
spec = RunSpec(Instance.from_path("instance.opb"), "ortools", Limits(time_s=60, mem_mib=4096), seed=1)
for item in run([Submission(spec, key="my-run-42")], jobs=4):
print(item.key, item.result.status, item.result.walltime_s)Using cpbenchy from your framework has a full example and the details.
Implementation
The run in src/cpbenchy/backend.py, lines 48–85 of 96, as of this version of the docs.
def run(
submissions: Iterable[Submission],
*,
jobs: int = 1,
executor: str = "auto",
plugins: Iterable[Any] = (),
args: Iterable[str] = (),
out: str | Path | None = None,
quiet: bool = False,
on_result: OnBackendResult | None = None,
) -> list[BackendResult]:
"""Measure `submissions` in one session, and return one `BackendResult` per submission.
`jobs`, `executor`, `plugins` and `args` are the session settings of `Experiment`. The runner
submits only what it wants done: runs already stored in `out` are measured again. `out` is a
fresh temporary directory when not given. `on_result` is called as each measurement finishes,
once per submission that shares it.
"""
pending = list(submissions)
if not pending:
return []
by_id: dict[str, list[Submission]] = {}
for submission in pending:
by_id.setdefault(submission.spec.run_id, []).append(submission)
experiment = Experiment(out, jobs=jobs, executor=executor, rerun=True, quiet=quiet, plugins=plugins, args=args)
experiment.add_runs(submission.spec for submission in _one_per_measurement(by_id))
tagged: list[BackendResult] = []
def report(result: RunResult) -> None:
log = _log_text(experiment.out, result.run_id)
for submission in by_id.get(result.run_id, []):
item = BackendResult(submission.key, dict(submission.metadata), result, log)
tagged.append(item)
if on_result is not None:
on_result(item)
experiment.run(on_result=report)
return tagged