Battle

This module defines the Battle class and thus what each battle type can do to customize how it runs and scores the programs. If you are implementing your own custom battle types, make sure they adhere to the api specifications laid out here.

algobattle.battle.Battle

Bases: BaseModel

Base for classes that execute a specific kind of battle.

Each battle type determines what parameters each fight will be fought with, how many fights are fought, and how they will ultimately be scored.

Source code in algobattle/battle.py

class Battle(BaseModel):
    """Base for classes that execute a specific kind of battle.

    Each battle type determines what parameters each fight will be fought with, how many fights are fought, and how
    they will ultimately be scored.
    """

    fights: list[Fight] = Field(default_factory=list)
    """The list of fights that have been fought in this battle."""
    runtime_error: ExceptionInfo | None = None
    """The description of an otherwise unhandeled exception that occured during the execution of :meth:`Battle.run`."""

    _battle_types: ClassVar[dict[str, type[Self]]] = {}
    """Dictionary mapping the names of all registered battle types to their python classes."""

    class Config(BaseModel):
        """Config object for each specific battle type.

        A custom battle type can override this class to specify config options it uses. They will be parsed from a
        dictionary located at `battle` in the main config file. The created object will then be passed to the
        :meth:`Battle.run` method with its fields set accordingly.
        """

        type: Any
        """Type of battle that will be used."""

        @classmethod
        def __get_pydantic_core_schema__(cls, source: Type, handler: GetCoreSchemaHandler) -> CoreSchema:
            # there's two bugs we need to catch:
            # 1. this function is called during the pydantic BaseModel metaclass's __new__, so the BattleConfig class
            # won't be ready at that point and be missing in the namespace
            # 2. pydantic uses the core schema to build child classes core schema. for them we want to behave like a
            # normal model, only our own schema gets modified
            try:
                if cls != Battle.Config:
                    return handler(source)
            except NameError:
                return handler(source)

            match len(Battle._battle_types):
                case 0:
                    subclass_schema = handler(source)
                case 1:
                    subclass_schema = handler(next(iter(Battle._battle_types.values())))
                case _:
                    subclass_schema = tagged_union_schema(
                        choices={
                            battle.Config.model_fields["type"].default: battle.Config.__pydantic_core_schema__
                            for battle in Battle._battle_types.values()
                        },
                        discriminator="type",
                    )

            # we want to validate into the actual battle type's config, so we need to treat them as a tagged union
            # but if we're initializing a project the type might not be installed yet, so we want to also parse
            # into an unspecified dummy object. This wrap validator will efficiently and transparently act as a tagged
            # union when ignore_uninstalled is not set. If it is set it catches only the error of a missing tag, other
            # errors are passed through
            def check_installed(val: object, handler: ValidatorFunctionWrapHandler, info: ValidationInfo) -> object:
                try:
                    return handler(val)
                except ValidationError as e:
                    union_err = next(filter(lambda err: err["type"] == "union_tag_invalid", e.errors()), None)
                    if union_err is None:
                        raise
                    if info.context is not None and info.context.get("ignore_uninstalled", False):
                        if info.config is not None:
                            settings: dict[str, Any] = {
                                "strict": info.config.get("strict", None),
                                "from_attributes": info.config.get("from_attributes"),
                            }
                        else:
                            settings = {}
                        return Battle.FallbackConfig.model_validate(val, context=info.context, **settings)
                    else:
                        passed = union_err["input"]["type"]
                        installed = ", ".join(b.name() for b in Battle._battle_types.values())
                        raise ValueError(
                            f"The specified battle type '{passed}' is not installed. Installed types are: {installed}"
                        )

            return with_info_wrap_validator_function(check_installed, subclass_schema)

    class FallbackConfig(Config):
        """Fallback config object to parse into if the proper battle typ isn't installed and we're ignoring installs."""

        type: str

        model_config = ConfigDict(extra="allow")

        if TYPE_CHECKING:
            # to hint that we're gonna fill this with arbitrary data belonging to some supposed battle type
            def __getattr__(self, __attr: str) -> Any:
                ...

    class UiData(BaseModel):
        """Object containing custom diplay data.

        The display data object will be displayed as key-value pairs generated from the :meth:`.field` method.
        You can use the normally available pydantic config options to customize what these will look like.
        """

    @staticmethod
    def all() -> dict[str, type["Battle"]]:
        """Returns a dictionary mapping the names of all registered battle types to their python classes.

        It includes all subclasses of :class:`Battle` that have been initialized so far, including ones exposed to the
        algobattle module via the `algobattle.battle` entrypoint hook.
        """
        return Battle._battle_types

    @classmethod
    def load_entrypoints(cls) -> None:
        """Loads all battle types presented via entrypoints."""
        for entrypoint in entry_points(group="algobattle.battle"):
            battle = entrypoint.load()
            if not (isclass(battle) and issubclass(battle, Battle)):
                raise ValueError(f"Entrypoint {entrypoint.name} targets something other than a Battle type")

    @classmethod
    def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
        if cls.name() not in Battle._battle_types:
            Battle._battle_types[cls.name()] = cls
            Battle.Config.model_rebuild(force=True)
        return super().__pydantic_init_subclass__(**kwargs)

    @abstractmethod
    def score(self, config: _BattleConfig) -> float:
        """Calculates the score the solver has achieved during this battle.

        Should always be a nonnegative float, with higher values indicating a better performance of the solver.
        """
        raise NotImplementedError

    @staticmethod
    def format_score(score: float) -> str:
        """Formats a given score nicely.

        Purely auxialiary method that can be used to customize how a score will be rendered.
        """
        return f"{score:.2f}"

    @classmethod
    def name(cls) -> str:
        """Name of this battle type.

        Defaults to the battle class's name. Can be used to customize this behaviour if e.g. a battle type should have a
        name that is not a valid python identifier.
        """
        return cls.__name__

    @abstractmethod
    async def run_battle(self, fight: FightHandler, config: _BattleConfig, min_size: int, ui: BattleUi) -> None:
        """Executes one battle.

        Args:
            fight: The :class:`FightHandler` used to run each fight of this battle. It already contains information
                about the participating teams, default config settings, etc. Each fight can be executed using the
                :meth:`FightHandler.run` method.
            config: An instance of this battle type's :class:`BattleConfig` class, parsed from the corresponding section
                of the config file.
            min_size: The minimum size valid for this problem.
            ui: An interface to interact with the ui.
        """
        raise NotImplementedError

UiData

Bases: BaseModel

Object containing custom diplay data.

The display data object will be displayed as key-value pairs generated from the :meth:.field method. You can use the normally available pydantic config options to customize what these will look like.

Source code in algobattle/battle.py

class UiData(BaseModel):
    """Object containing custom diplay data.

    The display data object will be displayed as key-value pairs generated from the :meth:`.field` method.
    You can use the normally available pydantic config options to customize what these will look like.
    """

score(config) abstractmethod

Calculates the score the solver has achieved during this battle.

Should always be a nonnegative float, with higher values indicating a better performance of the solver.

Source code in algobattle/battle.py

@abstractmethod
def score(self, config: _BattleConfig) -> float:
    """Calculates the score the solver has achieved during this battle.

    Should always be a nonnegative float, with higher values indicating a better performance of the solver.
    """
    raise NotImplementedError

format_score(score) staticmethod

Formats a given score nicely.

Purely auxialiary method that can be used to customize how a score will be rendered.

Source code in algobattle/battle.py

@staticmethod
def format_score(score: float) -> str:
    """Formats a given score nicely.

    Purely auxialiary method that can be used to customize how a score will be rendered.
    """
    return f"{score:.2f}"

name() classmethod

Name of this battle type.

Defaults to the battle class's name. Can be used to customize this behaviour if e.g. a battle type should have a name that is not a valid python identifier.

Source code in algobattle/battle.py

@classmethod
def name(cls) -> str:
    """Name of this battle type.

    Defaults to the battle class's name. Can be used to customize this behaviour if e.g. a battle type should have a
    name that is not a valid python identifier.
    """
    return cls.__name__

run_battle(fight, config, min_size, ui) abstractmethod async

Executes one battle.

Parameters:

Name	Type	Description	Default
fight	FightHandler	The :class:FightHandler used to run each fight of this battle. It already contains information about the participating teams, default config settings, etc. Each fight can be executed using the :meth:FightHandler.run method.	required
config	_BattleConfig	An instance of this battle type's :class:BattleConfig class, parsed from the corresponding section of the config file.	required
min_size	int	The minimum size valid for this problem.	required
ui	BattleUi	An interface to interact with the ui.	required

Source code in algobattle/battle.py

@abstractmethod
async def run_battle(self, fight: FightHandler, config: _BattleConfig, min_size: int, ui: BattleUi) -> None:
    """Executes one battle.

    Args:
        fight: The :class:`FightHandler` used to run each fight of this battle. It already contains information
            about the participating teams, default config settings, etc. Each fight can be executed using the
            :meth:`FightHandler.run` method.
        config: An instance of this battle type's :class:`BattleConfig` class, parsed from the corresponding section
            of the config file.
        min_size: The minimum size valid for this problem.
        ui: An interface to interact with the ui.
    """
    raise NotImplementedError

algobattle.battle.FightHandler dataclass

Helper class to run fights of a given battle.

Source code in algobattle/battle.py

@dataclass
class FightHandler:
    """Helper class to run fights of a given battle."""

    problem: Problem
    generator: Generator
    solver: Solver
    battle: "Battle"
    ui: FightUi
    set_cpus: str | None
    log_config: ProgramLogConfigView

    @overload
    async def run(
        self,
        max_size: int,
        *,
        with_results: Literal[False] = False,
        **kwargs: Unpack[RunKwargs],
    ) -> Fight:
        ...

    @overload
    async def run(
        self,
        max_size: int,
        *,
        with_results: Literal[True],
        **kwargs: Unpack[RunKwargs],
    ) -> tuple[Fight, GeneratorResult, SolverResult | None]:
        ...

    async def run(
        self,
        max_size: int,
        *,
        with_results: bool = False,
        **kwargs: Unpack[RunKwargs],
    ) -> Fight | tuple[Fight, GeneratorResult, SolverResult | None]:
        """Execute a single fight of a battle.

        First the generator will be run and its output parsed. Then the solver will be given the created instance
        and run. Its output gets parsed into a solution, which will then be scored.
        The timeout, space, and cpu arguments each override the corresponding match config options if set. Leaving them
        unset results in the config options being used.

        Args:
            max_size: The maximum instance size the generator is allowed to create.
            timeout_generator: Timeout in seconds for the generator to finish running. `None` means it is given an
                unlimited amount of time.
            space_generator: Memory space in MB the generator has access to. `None` means it is given an unlimited
                amount of space.
            cpus_generator: Number of physical cpu cores the generator can use.
            timeout_solver: Timeout in seconds for the solver to finish running. `None` means it is given an unlimited
                amount of time.
            space_solver: Memory space in MB the solver has access to. `None` means it is given
                an unlimited amount of space.
            cpus_solver: Number of physical cpu cores the solver can use.
            generator_battle_input: Additional data the generator will be provided with.
            solver_battle_input: Additional data the solver will be provided with.
            generator_battle_output: Class used to parse additional data the generator outputs into a python object.
            solver_battle_output: Class used to parse additional data the solver outputs into a python object.
            with_results: Whether to return the raw result objects.

        Returns:
            The resulting info about the executed fight, and the results if the flag has been set.
        """
        gen_result, sol_result = await self.run_raw(max_size=max_size, **kwargs)
        if gen_result.instance is None or gen_result.solution is None:
            score = 1
        elif sol_result is None or sol_result.solution is None:
            score = 0
        else:
            score = self.calculate_score(gen_result, sol_result)
        fight = Fight.from_results(
            score=score,
            max_size=max_size,
            generator=gen_result,
            solver=sol_result,
            config=self.log_config,
        )
        self.battle.fights.append(fight)
        self.ui.end_fight()
        if with_results:
            return fight, gen_result, sol_result
        else:
            return fight

    async def run_raw(
        self,
        max_size: int,
        *,
        timeout_generator: float | None | EllipsisType = ...,
        space_generator: int | None | EllipsisType = ...,
        cpus_generator: int | EllipsisType = ...,
        timeout_solver: float | None | EllipsisType = ...,
        space_solver: int | None | EllipsisType = ...,
        cpus_solver: int | EllipsisType = ...,
        generator_battle_input: Encodable | None = None,
        solver_battle_input: Encodable | None = None,
        generator_battle_output: type[Encodable] | None = None,
        solver_battle_output: type[Encodable] | None = None,
    ) -> tuple[GeneratorResult, SolverResult | None]:
        """Runs a fight and returns the unprocessed results."""
        min_size = self.problem.min_size
        if max_size < min_size:
            raise ValueError(
                f"Cannot run battle at size {max_size} since it is smaller than the smallest "
                f"size the problem allows ({min_size})."
            )
        ui = self.ui
        ui.start_fight(max_size)
        gen_result = await self.generator.run(
            max_size=max_size,
            timeout=timeout_generator,
            space=space_generator,
            cpus=cpus_generator,
            battle_input=generator_battle_input,
            battle_output=generator_battle_output,
            set_cpus=self.set_cpus,
            ui=ui,
        )
        if gen_result.error is not None:
            return gen_result, None
        assert gen_result.instance is not None

        sol_result = await self.solver.run(
            gen_result.instance,
            max_size=max_size,
            timeout=timeout_solver,
            space=space_solver,
            cpus=cpus_solver,
            battle_input=solver_battle_input,
            battle_output=solver_battle_output,
            set_cpus=self.set_cpus,
            ui=ui,
        )
        return gen_result, sol_result

    def calculate_score(self, gen_result: GeneratorResult, sol_result: SolverResult) -> float:
        """Calculates the score achieved by the solver in this fight.

        Both results need to contain all instance and/or solution data required.

        Args:
            gen_result: The generator's result.
            sol_result: The solver's result

        Returns:
            A number in [0, 1] with higher numbers meaning the solver performed better.
        """
        assert gen_result.instance is not None
        assert sol_result.solution is not None
        if self.problem.with_solution:
            assert gen_result.solution is not None
            score = self.problem.score(
                gen_result.instance, solver_solution=sol_result.solution, generator_solution=gen_result.solution
            )
        else:
            score = self.problem.score(gen_result.instance, solution=sol_result.solution)
        return max(0, min(1, float(score)))

run(max_size, *, with_results=False, **kwargs) async

Execute a single fight of a battle.

First the generator will be run and its output parsed. Then the solver will be given the created instance and run. Its output gets parsed into a solution, which will then be scored. The timeout, space, and cpu arguments each override the corresponding match config options if set. Leaving them unset results in the config options being used.

Parameters:

Name	Type	Description	Default
max_size	int	The maximum instance size the generator is allowed to create.	required
timeout_generator		Timeout in seconds for the generator to finish running. None means it is given an unlimited amount of time.	required
space_generator		Memory space in MB the generator has access to. None means it is given an unlimited amount of space.	required
cpus_generator		Number of physical cpu cores the generator can use.	required
timeout_solver		Timeout in seconds for the solver to finish running. None means it is given an unlimited amount of time.	required
space_solver		Memory space in MB the solver has access to. None means it is given an unlimited amount of space.	required
cpus_solver		Number of physical cpu cores the solver can use.	required
generator_battle_input		Additional data the generator will be provided with.	required
solver_battle_input		Additional data the solver will be provided with.	required
generator_battle_output		Class used to parse additional data the generator outputs into a python object.	required
solver_battle_output		Class used to parse additional data the solver outputs into a python object.	required
with_results	bool	Whether to return the raw result objects.	False

Returns:

Type	Description
Fight | tuple[Fight, GeneratorResult, SolverResult | None]	The resulting info about the executed fight, and the results if the flag has been set.

Source code in algobattle/battle.py

async def run(
    self,
    max_size: int,
    *,
    with_results: bool = False,
    **kwargs: Unpack[RunKwargs],
) -> Fight | tuple[Fight, GeneratorResult, SolverResult | None]:
    """Execute a single fight of a battle.

    First the generator will be run and its output parsed. Then the solver will be given the created instance
    and run. Its output gets parsed into a solution, which will then be scored.
    The timeout, space, and cpu arguments each override the corresponding match config options if set. Leaving them
    unset results in the config options being used.

    Args:
        max_size: The maximum instance size the generator is allowed to create.
        timeout_generator: Timeout in seconds for the generator to finish running. `None` means it is given an
            unlimited amount of time.
        space_generator: Memory space in MB the generator has access to. `None` means it is given an unlimited
            amount of space.
        cpus_generator: Number of physical cpu cores the generator can use.
        timeout_solver: Timeout in seconds for the solver to finish running. `None` means it is given an unlimited
            amount of time.
        space_solver: Memory space in MB the solver has access to. `None` means it is given
            an unlimited amount of space.
        cpus_solver: Number of physical cpu cores the solver can use.
        generator_battle_input: Additional data the generator will be provided with.
        solver_battle_input: Additional data the solver will be provided with.
        generator_battle_output: Class used to parse additional data the generator outputs into a python object.
        solver_battle_output: Class used to parse additional data the solver outputs into a python object.
        with_results: Whether to return the raw result objects.

    Returns:
        The resulting info about the executed fight, and the results if the flag has been set.
    """
    gen_result, sol_result = await self.run_raw(max_size=max_size, **kwargs)
    if gen_result.instance is None or gen_result.solution is None:
        score = 1
    elif sol_result is None or sol_result.solution is None:
        score = 0
    else:
        score = self.calculate_score(gen_result, sol_result)
    fight = Fight.from_results(
        score=score,
        max_size=max_size,
        generator=gen_result,
        solver=sol_result,
        config=self.log_config,
    )
    self.battle.fights.append(fight)
    self.ui.end_fight()
    if with_results:
        return fight, gen_result, sol_result
    else:
        return fight

algobattle.battle.Fight

Bases: BaseModel

The result of one fight between the participating teams.

For a more detailed description of what each fight looks like, see :meth:FightHandler.run.

Source code in algobattle/battle.py

class Fight(BaseModel):
    """The result of one fight between the participating teams.

    For a more detailed description of what each fight looks like, see :meth:`FightHandler.run`.
    """

    score: float
    """The solving Team's score.

    Always a number in [0, 1]. 0 indicates a total failure of the solver, 1 that it succeeded perfectly.
    """
    max_size: int
    """The maximum size of an instance the generator was allowed to create."""
    generator: ProgramRunInfo
    """Data about the generator's execution."""
    solver: ProgramRunInfo | None
    """Data about the solver's execution."""

    @classmethod
    def from_results(
        cls,
        max_size: int,
        score: float,
        generator: GeneratorResult,
        solver: SolverResult | None,
        *,
        config: ProgramLogConfigView,
    ) -> Self:
        """Turns the involved result objects into a jsonable model."""
        inline_output = config.when == "always" or (
            config.when == "error"
            and (generator.error is not None or (solver is not None and solver.error is not None))
        )
        return cls(
            max_size=max_size,
            score=score,
            generator=ProgramRunInfo.from_result(generator, inline_output=inline_output),
            solver=ProgramRunInfo.from_result(solver, inline_output=inline_output) if solver is not None else None,
        )

score: float instance-attribute

The solving Team's score.

Always a number in [0, 1]. 0 indicates a total failure of the solver, 1 that it succeeded perfectly.

max_size: int instance-attribute

The maximum size of an instance the generator was allowed to create.

generator: ProgramRunInfo instance-attribute

Data about the generator's execution.

solver: ProgramRunInfo | None instance-attribute

Data about the solver's execution.

from_results(max_size, score, generator, solver, *, config) classmethod

Turns the involved result objects into a jsonable model.

Source code in algobattle/battle.py

@classmethod
def from_results(
    cls,
    max_size: int,
    score: float,
    generator: GeneratorResult,
    solver: SolverResult | None,
    *,
    config: ProgramLogConfigView,
) -> Self:
    """Turns the involved result objects into a jsonable model."""
    inline_output = config.when == "always" or (
        config.when == "error"
        and (generator.error is not None or (solver is not None and solver.error is not None))
    )
    return cls(
        max_size=max_size,
        score=score,
        generator=ProgramRunInfo.from_result(generator, inline_output=inline_output),
        solver=ProgramRunInfo.from_result(solver, inline_output=inline_output) if solver is not None else None,
    )