Merge pull request #73 from ImogenBits/problem

rework problem and battle wrapper classes

Author: Benezivas

.github/workflows/python-app.yml

@@ -17,7 +17,7 @@ jobs:
     - name: Set up Python
       uses: actions/setup-python@v2
       with:
-        python-version: "3.10"
+        python-version: "3.11"
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
@@ -36,7 +36,7 @@ jobs:
     - name: Set up Python
       uses: actions/setup-python@v2
       with:
-        python-version: "3.10"
+        python-version: "3.11"
     - name: install Docker
       if: runner.os == 'macOS'
       uses: docker-practice/actions-setup-docker@master

algobattle/battle.py

@@ -15,11 +15,13 @@
 # for now we need to manually import the default wrappers to make sure they're initialized
 from algobattle.battle_wrappers.iterated import Iterated    # type: ignore # noqa: F401
 from algobattle.battle_wrappers.averaged import Averaged    # type: ignore # noqa: F401
+from algobattle.docker_util import DockerConfig, RunParameters
 
-from algobattle.match import MatchConfig, run_match
+from algobattle.match import MatchConfig, Match
+from algobattle.problem import Problem
 from algobattle.team import TeamHandler, TeamInfo
 from algobattle.ui import Ui
-from algobattle.util import check_path, getattr_set, import_problem_from_path
+from algobattle.util import check_path, getattr_set
 
 
 def setup_logging(logging_path: Path, verbose_logging: bool, silent: bool):
@@ -79,7 +81,7 @@ class ProgramConfig:
     logs: Path = Path.home() / ".algobattle_logs"
 
 
-def parse_cli_args(args: list[str]) -> tuple[ProgramConfig, MatchConfig, BattleWrapper.Config]:
+def parse_cli_args(args: list[str]) -> tuple[ProgramConfig, DockerConfig, MatchConfig, BattleWrapper.Config]:
     """Parse a given CLI arg list into config objects."""
     parser = ArgumentParser()
     parser.add_argument("problem", type=check_path, help="Path to a folder with the problem file.")
@@ -99,7 +101,8 @@ def parse_cli_args(args: list[str]) -> tuple[ProgramConfig, MatchConfig, BattleW
     parser.add_argument("--timeout_solver", type=float, help="Time limit for the solver execution.")
     parser.add_argument("--space_generator", type=int, help="Memory limit for the generator execution, in MB.")
     parser.add_argument("--space_solver", type=int, help="Memory limit the solver execution, in MB.")
-    parser.add_argument("--cpus", type=int, help="Number of cpu cores used for each docker container execution.")
+    parser.add_argument("--cpus_generator", type=int, help="Number of cpu cores used for generator container execution.")
+    parser.add_argument("--cpus_solver", type=int, help="Number of cpu cores used for solver container execution.")
 
     # battle wrappers have their configs automatically added to the CLI args
     for wrapper_name, wrapper in BattleWrapper.all().items():
@@ -110,7 +113,7 @@ def parse_cli_args(args: list[str]) -> tuple[ProgramConfig, MatchConfig, BattleW
     parsed = parser.parse_args(args)
 
     if parsed.battle_type is not None:
-        parsed.battle_type = BattleWrapper.get_wrapper(parsed.battle_type)
+        parsed.battle_type = BattleWrapper.all()[parsed.battle_type]
     cfg_path = parsed.config if parsed.config is not None else parsed.problem / "config.toml"
     if cfg_path.is_file():
         with open(cfg_path, "rb") as file:
@@ -125,7 +128,7 @@ def parse_cli_args(args: list[str]) -> tuple[ProgramConfig, MatchConfig, BattleW
         team_specs = config["teams"]
     else:
         team_specs = [{
-            "name": "team 0",
+            "name": "team_0",
             "generator": parsed.problem / "generator",
             "solver": parsed.problem / "solver",
         }]
@@ -141,42 +144,57 @@ def parse_cli_args(args: list[str]) -> tuple[ProgramConfig, MatchConfig, BattleW
 
     program_config = ProgramConfig(teams=teams, **getattr_set(parsed, "problem", "display", "logs"))
 
-    match_config = MatchConfig.from_dict(config.get("algobattle", {}))
+    match_config = MatchConfig.from_dict(config.get("match", {}))
     for name in vars(match_config):
         if getattr(parsed, name) is not None:
             setattr(match_config, name, getattr(parsed, name))
 
+    docker_params = config.get("docker", {})
+    docker_config = DockerConfig(
+        build_timeout=docker_params.get("build_timeout"),
+        generator=RunParameters(**docker_params.get("generator", {})),
+        solver=RunParameters(**docker_params.get("solver", {})),
+    )
+    if getattr(parsed, "timeout_build") is not None:
+        object.__setattr__(docker_config, "build_timeout", parsed.timeout_build)
+    for role in ("generator", "solver"):
+        role_config = getattr(docker_config, role)
+        for name in vars(role_config):
+            cli_name = f"{name}_{role}"
+            if getattr(parsed, cli_name) is not None:
+                object.__setattr__(role_config, name, getattr(parsed, cli_name))
+
     wrapper_config = match_config.battle_type.Config(**config.get(match_config.battle_type.name().lower(), {}))
     for name in vars(wrapper_config):
         cli_name = f"{match_config.battle_type.name().lower()}_{name}"
         if getattr(parsed, cli_name) is not None:
             setattr(wrapper_config, name, getattr(parsed, cli_name))
 
-    return program_config, match_config, wrapper_config
+    return program_config, docker_config, match_config, wrapper_config
 
 
 def main():
     """Entrypoint of `algobattle` CLI."""
     try:
-        program_config, match_config, wrapper_config = parse_cli_args(sys.argv[1:])
+        program_config, docker_config, match_config, wrapper_config = parse_cli_args(sys.argv[1:])
         logger = setup_logging(program_config.logs, match_config.verbose, program_config.display != "logs")
 
     except KeyboardInterrupt:
         raise SystemExit("Received keyboard interrupt, terminating execution.")
 
     try:
-        problem = import_problem_from_path(program_config.problem)
-        with TeamHandler.build(program_config.teams) as teams, ExitStack() as stack:
+        problem = Problem.import_from_path(program_config.problem)
+        with TeamHandler.build(program_config.teams, problem, docker_config) as teams, ExitStack() as stack:
             if program_config.display == "ui":
                 ui = Ui()
                 stack.enter_context(ui)
             else:
                 ui = None
 
-            result = run_match(match_config, wrapper_config, problem, teams, ui)
+            result = Match.run(match_config, wrapper_config, problem, teams, ui)
 
             logger.info('#' * 78)
-            logger.info(str(result))
+            logger.info(result.display())
             if match_config.points > 0:
                 points = result.calculate_points(match_config.points)
                 for team, pts in points.items():

algobattle/battle_wrapper.py

@@ -5,114 +5,172 @@
 characteristic that they are responsible for updating some match data during
 their run, such that it contains the current state of the match.
 """
-from __future__ import annotations
-from dataclasses import dataclass
+from dataclasses import dataclass, field as dataclass_field, fields
 from importlib.metadata import entry_points
 import logging
 from abc import abstractmethod, ABC
-from importlib import import_module
-from typing import Type
+from typing import (
+    Any,
+    Callable,
+    ClassVar,
+    Literal,
+    Mapping,
+    TypeAlias,
+    TypeVar,
+    dataclass_transform,
+    get_origin,
+    get_type_hints,
+)
+from algobattle.docker_util import DockerError, Generator, Solver, GeneratorResult, SolverResult
+from algobattle.observer import Subject
+from algobattle.util import Encodable, Role
 
-from algobattle.fight_handler import FightHandler
-from algobattle.team import Matchup
-from algobattle.observer import Observer, Subject
-from algobattle.util import CLIParsable
+logger = logging.getLogger("algobattle.battle_wrapper")
 
-logger = logging.getLogger('algobattle.battle_wrapper')
 
+_Config: TypeAlias = Any
+T = TypeVar("T")
 
-class BattleWrapper(ABC):
+
+def argspec(*, default: T, help: str = "", parser: Callable[[str], T] | None = None) -> T:
+    """Structure specifying the CLI arg."""
+    metadata = {"help": help, "parser": parser}
+    return dataclass_field(default=default, metadata={key: val for key, val in metadata.items() if val is not None})
+
+
+@dataclass
+class CombinedResults:
+    """The result of one execution of the generator and the solver with the generated instance."""
+
+    score: float
+    generator: GeneratorResult | DockerError
+    solver: SolverResult | DockerError | None
+
+
+class BattleWrapper(Subject, ABC):
     """Abstract Base class for wrappers that execute a specific kind of battle."""
 
-    @dataclass
-    class Config(CLIParsable):
-        """Object containing the config variables the wrapper will use."""
+    _wrappers: ClassVar[dict[str, type["BattleWrapper"]]] = {}
+
+    scoring_team: ClassVar[Role] = "solver"
 
-        pass
+    @dataclass_transform(field_specifiers=(argspec,))
+    class Config:
+        """Object containing the config variables the wrapper will use."""
 
-    _wrappers: dict[str, Type[BattleWrapper]] = {}
+        def __init_subclass__(cls) -> None:
+            dataclass(cls)
+            super().__init_subclass__()
+
+        # providing a dummy default impl that will be overriden, to get better static analysis
+        def __init__(self, **kwargs) -> None:
+            super().__init__()
+
+        @classmethod
+        def as_argparse_args(cls) -> list[tuple[str, dict[str, Any]]]:
+            """Constructs a list of argument names and `**kwargs` that can be passed to `ArgumentParser.add_argument()`."""
+            arguments: list[tuple[str, dict[str, Any]]] = []
+            resolved_annotations = get_type_hints(cls)
+            for field in fields(cls):
+                kwargs = {
+                    "type": field.metadata.get("parser", resolved_annotations[field.name]),
+                    "help": field.metadata.get("help", "") + f" Default: {field.default}",
+                }
+                if field.type == bool:
+                    kwargs["action"] = "store_const"
+                    kwargs["const"] = not field.default
+                elif get_origin(field.type) == Literal:
+                    kwargs["choices"] = field.type.__args__
+
+                arguments.append((field.name, kwargs))
+            return arguments
 
     @staticmethod
-    def all() -> dict[str, Type[BattleWrapper]]:
+    def all() -> dict[str, type["BattleWrapper"]]:
         """Returns a list of all registered wrappers."""
         for entrypoint in entry_points(group="algobattle.wrappers"):
             if entrypoint.name not in BattleWrapper._wrappers:
-                wrapper: Type[BattleWrapper] = entrypoint.load()
-                BattleWrapper._wrappers[wrapper.name()] = wrapper
+                wrapper: type[BattleWrapper] = entrypoint.load()
+                BattleWrapper._wrappers[wrapper.name().lower()] = wrapper
         return BattleWrapper._wrappers
 
-    def __init_subclass__(cls) -> None:
+    def __init_subclass__(cls, notify_var_changes: bool = False) -> None:
         if cls.name() not in BattleWrapper._wrappers:
-            BattleWrapper._wrappers[cls.name()] = cls
-        return super().__init_subclass__()
+            BattleWrapper._wrappers[cls.name().lower()] = cls
+        return super().__init_subclass__(notify_var_changes)
 
-    @staticmethod
-    def get_wrapper(wrapper_name: str) -> Type[BattleWrapper]:
-        """Try to import a Battle Wrapper from a given name.
-
-        For this to work, a BattleWrapper module with the same name as the argument
-        needs to be present in the algobattle/battle_wrappers folder.
-
-        Parameters
-        ----------
-        wrapper_name : str
-            Name of a battle wrapper module in algobattle/battle_wrappers.
-
-        Returns
-        -------
-        BattleWrapper
-            A BattleWrapper of the given wrapper_name.
-
-        Raises
-        ------
-        ValueError
-            If the wrapper does not exist in the battle_wrappers folder.
-        """
-        try:
-            wrapper_module = import_module("algobattle.battle_wrappers." + wrapper_name)
-            return getattr(wrapper_module, wrapper_name.capitalize())
-        except ImportError as e:
-            logger.critical(f"Importing a wrapper from the given path failed with the following exception: {e}")
-            raise ValueError from e
+    @abstractmethod
+    def score(self) -> float:
+        """The score achieved by the scored team during this battle."""
+        raise NotImplementedError
 
-    def __init__(self, fight_handler: FightHandler, config: BattleWrapper.Config) -> None:
-        super().__init__()
-        self.fight_handler = fight_handler
-        self.config = config
+    @staticmethod
+    def format_score(score: float) -> str:
+        """Formats a score nicely."""
+        return f"{score:.2f}"
 
     @abstractmethod
-    def run_round(self, matchup: Matchup, observer: Observer | None = None) -> BattleWrapper.Result:
-        """Execute a full round of fights between two teams configured in the fight_handler.
-
-        During execution, the concrete BattleWrapper should update the round_data dict
-        to which Observers can subscribe in order to react to new intermediate results.
-        """
+    def display(self) -> str:
+        """Nicely formats the object."""
         raise NotImplementedError
 
     @classmethod
     def name(cls) -> str:
         """Name of the type of this battle wrapper."""
         return cls.__name__
 
-    class Result(Subject):
-        """Result of a single battle."""
-
-        @property
-        @abstractmethod
-        def score(self) -> float:
-            """The score achieved by the solver of this battle."""
-            raise NotImplementedError
-
-        @staticmethod
-        @abstractmethod
-        def format_score(score: float) -> str:
-            """Formats a score nicely."""
-            raise NotImplementedError
+    @abstractmethod
+    def run_battle(self, generator: Generator, solver: Solver, config: _Config, min_size: int) -> None:
+        """Calculates the next instance size that should be fought over."""
+        raise NotImplementedError
 
-        def __str__(self) -> str:
-            return self.format_score(self.score)
+    def run_programs(
+        self,
+        generator: Generator,
+        solver: Solver,
+        size: int,
+        *,
+        timeout_generator: float | None = ...,
+        space_generator: int | None = ...,
+        cpus_generator: int = ...,
+        timeout_solver: float | None = ...,
+        space_solver: int | None = ...,
+        cpus_solver: int = ...,
+        generator_battle_input: Mapping[str, Encodable] = {},
+        solver_battle_input: Mapping[str, Encodable] = {},
+        generator_battle_output: Mapping[str, type[Encodable]] = {},
+        solver_battle_output: Mapping[str, type[Encodable]] = {},
+    ) -> CombinedResults:
+        """Execute a single fight of a battle, running the generator and solver and handling any errors gracefully."""
+        self.notify()
+        try:
+            gen_result = generator.run(
+                size=size,
+                timeout=timeout_generator,
+                space=space_generator,
+                cpus=cpus_generator,
+                battle_input=generator_battle_input,
+                battle_output=generator_battle_output,
+            )
+        except DockerError as e:
+            return CombinedResults(score=1, generator=e, solver=None)
 
-        @abstractmethod
-        def display(self) -> str:
-            """Nicely formats the object."""
-            raise NotImplementedError
+        try:
+            sol_result = solver.run(
+                gen_result.problem,
+                size=size,
+                timeout=timeout_solver,
+                space=space_solver,
+                cpus=cpus_solver,
+                battle_input=solver_battle_input,
+                battle_output=solver_battle_output,
+            )
+        except DockerError as e:
+            return CombinedResults(score=0, generator=gen_result, solver=e)
+
+        score = gen_result.problem.calculate_score(
+            solution=sol_result.solution, generator_solution=gen_result.solution, size=size
+        )
+        score = max(0, min(1, float(score)))
+        logger.info(f"The solver achieved a score of {score}.")
+        return CombinedResults(score, gen_result, sol_result)