CompetitionOutput

The starting point for the output of another competition: you describe the solution, the rest is done.

ObserverBuilt incomes with cpbenchy

from cpbenchy.observers import CompetitionOutput

Version: cpbenchy 0.1.0.dev0
Last updated: 6 Oct 2026 · 1 commit
Authors: ThomSerg
Source: src/cpbenchy/observers.py:36
Tags: competitionoutputbase class

What it does

The base of the competition output observers. It writes the o lines while solving and the s line at the end; a subclass only says which file it writes, which instance formats it applies to, and how the solution looks.

Use it

Subclass it, set name (the output file is logs/<run_id>.<name>.out) and formats, and return the solution’s lines (without v ) from solution:

from cpbenchy import formats
from cpbenchy.observers import CompetitionOutput

class SchedulingOutput(CompetitionOutput):
    name = "schedule"
    formats = ("jsplib", "rcpsp")

    def solution(self, ctx):  # v start[0] = 3, v start[1] = 0, ...
        solution = formats.solution_dict(formats.instance_variables(ctx.model))
        return [f"{name} = {value}" for name, value in solution.items()]

Use it like any observer: -p myoutput.py:SchedulingOutput, or in rules. Under cpbenchy solve, it writes to stdout instead, for submissions.

What it gives you


`solution(ctx)`	override: the solution, as the lines after `v`
`on_solution`	writes `o <objective>` for each solution found while solving
`on_finish`	writes the `s` line and the solution, or `s UNSATISFIABLE`, or a comment and `s UNKNOWN`
`write(ctx, *lines)`	appends lines to the output (or stdout)
`output(ctx_or_run)`	the output file, from the worker’s `ctx` or the parent’s `run`

cpbenchy.formats has the pieces for common formats.

Implementation

The CompetitionOutput in src/cpbenchy/observers.py, lines 36–71 of 222, as of this version of the docs.

src/cpbenchy/observers.pypython

class CompetitionOutput(Observer):
    """Writes `o` lines for each solution while solving, then an `s` line and the solution as `v` lines,
    to `<run_id>.<name>.out`. Override `solution` (and set `name` and `formats`) for a new format."""

    name = "competition"
    formats = ()  # observes nothing by itself

    def solution(self, ctx) -> list[str]:
        """The solution, as the lines after `v `."""
        raise NotImplementedError

    def on_solution(self, ctx, objective):
        if objective is not None:
            self.write(ctx, f"o {fmt.number(objective)}")

    def on_finish(self, ctx):
        if ctx.status in ("optimal", "feasible"):
            status = fmt.status_line(ctx.status, ctx.model.has_objective())
            self.write(ctx, status, *(f"v {line}" for line in self.solution(ctx)))
        elif ctx.status == "unsat":
            self.write(ctx, "s UNSATISFIABLE")
        else:
            self.write(ctx, f"c {ctx.error or 'no solution within the limits'}", "s UNKNOWN")

    def output(self, ctx_or_run) -> Path:
        """The output file, from the worker's `ctx` or the parent's `run`."""
        return ctx_or_run.artifact(f"{self.name}.out")

    def write(self, ctx, *lines: str) -> None:
        text = "".join(line + "\n" for line in lines)
        if ctx.options.get("stdout"):
            sys.stdout.write(text)
            sys.stdout.flush()
        else:
            with open(self.output(ctx), "a") as f:
                f.write(text)