Merge branch 'draft' into absolute2

Author: mzuenni

.pre-commit-config.yaml

@@ -7,9 +7,9 @@ repos:
 -   repo: https://github.com/astral-sh/ruff-pre-commit
     rev: v0.9.9
     hooks:
-    -   id: ruff-format
     -   id: ruff
         args: [ --fix ]
+    -   id: ruff-format
 -   repo: https://github.com/pre-commit/mirrors-mypy
     rev: v1.15.0
     hooks:

bin/config.py

@@ -5,7 +5,7 @@
 import re
 from pathlib import Path
 from collections.abc import Mapping, Sequence
-from typing import Final, Literal, Optional
+from typing import Any, Final, Literal, Optional
 
 SPEC_VERSION: Final[str] = "2023-07-draft"
 
@@ -86,7 +86,6 @@
     "invalid_input",
     "invalid_answer",
     "invalid_output",
-    "bad",
 ]
 
 
@@ -105,7 +104,7 @@
 
 args = argparse.Namespace()
 
-DEFAULT_ARGS: Final[Mapping] = {
+DEFAULT_ARGS: Final[Mapping[str, Any]] = {
     "jobs": (os.cpu_count() or 1) // 2,
     "time": 600,  # Used for `bt fuzz`
     "verbose": 0,

bin/constraints.py

@@ -1,9 +1,11 @@
 import re
 from collections import defaultdict
+from typing import Optional
 
 import latex
 import validate
 from colorama import Fore, Style
+from problem import Problem
 
 # Local imports
 from util import *
@@ -16,7 +18,9 @@
 """
 
 
-def check_validators(problem):
+def check_validators(
+    problem: Problem,
+) -> tuple[set[int | float], list[str | tuple[int | float, str, int | float]]]:
     in_constraints: validate.ConstraintsDict = {}
     ans_constraints: validate.ConstraintsDict = {}
     problem.validate_data(validate.Mode.INPUT, constraints=in_constraints)
@@ -27,10 +31,10 @@ def check_validators(problem):
         log("No constraint validation of answer values found in answer or output validators.")
     print()
 
-    validator_values = set()
+    validator_values: set[int | float] = set()
     validator_defs: list[str | tuple[int | float, str, int | float]] = []
 
-    def f(cs):
+    def f(cs: validate.ConstraintsDict) -> None:
         for loc, value in sorted(cs.items()):
             name, has_low, has_high, vmin, vmax, low, high = value
             validator_defs.append((low, name, high))
@@ -45,12 +49,12 @@ def f(cs):
     return validator_values, validator_defs
 
 
-def check_statement(problem, language):
+def check_statement(problem: Problem, language: str) -> tuple[set[int | float], list[str]]:
     statement_file = problem.path / latex.PdfType.PROBLEM.path(language)
     statement = statement_file.read_text()
 
-    statement_values = set()
-    statement_defs = []
+    statement_values: set[int | float] = set()
+    statement_defs: list[str] = []
 
     defines = ["\\def", "\\newcommand"]
     sections = ["Input", "Output", "Interaction"]
@@ -67,15 +71,16 @@ def check_statement(problem, language):
     }
     relations = re.compile(r"(<=|!=|>=|<|=|>)")
 
-    def math_eval(text):
+    def math_eval(text: str) -> Optional[int | float]:
         try:
             # eval is dangerous, but on the other hand we run submission code so this is fine
             text = text.replace("^", "**")
-            return eval(text, {"__builtin__": None})
+            value = eval(text, {"__builtin__": None})
+            return value if isinstance(value, (int, float)) else None
         except (SyntaxError, NameError, TypeError, ZeroDivisionError):
             return None
 
-    def constraint(text):
+    def constraint(text: str) -> None:
         # handles $$math$$
         if len(text) == 0:
             return
@@ -132,13 +137,13 @@ def constraint(text):
     in_io = False
     end = None
 
-    def matches(text):
+    def matches(text: str) -> bool:
         nonlocal pos
         if pos + len(text) > len(statement):
             return False
         return statement[pos : pos + len(text)] == text
 
-    def parse_group():
+    def parse_group() -> str:
         nonlocal pos
         assert statement[pos] == "{"
         next = pos + 1
@@ -155,7 +160,7 @@ def parse_group():
         pos = next
         return name
 
-    def parse_command():
+    def parse_command() -> str:
         nonlocal pos
         assert statement[pos] == "\\"
         next = pos + 1
@@ -251,16 +256,16 @@ def parse_command():
     return statement_values, statement_defs
 
 
-def check_constraints(problem):
+def check_constraints(problem: Problem) -> bool:
     validator_values, validator_defs = check_validators(problem)
-    statement_values = defaultdict(set)
-    statement_defs = defaultdict(set)
+    statement_values: dict[int | float, set[str]] = defaultdict(set)
+    statement_defs: dict[str, set[str]] = defaultdict(set)
     for lang in problem.statement_languages:
         values, defs = check_statement(problem, lang)
-        for entry in values:
-            statement_values[entry].add(lang)
-        for entry in defs:
-            statement_defs[entry].add(lang)
+        for value_entry in values:
+            statement_values[value_entry].add(lang)
+        for def_entry in defs:
+            statement_defs[def_entry].add(lang)
 
     # print all the definitions.
     value_len = 12

bin/contest.py

@@ -1,14 +1,15 @@
 import config
 
 from pathlib import Path
+from typing import cast, Any, Optional
 
 from util import *
 
 # Read the contest.yaml, if available
-_contest_yaml = None
+_contest_yaml: Optional[dict[str, Any]] = None
 
 
-def contest_yaml():
+def contest_yaml() -> dict[str, Any]:
     global _contest_yaml
     if _contest_yaml is not None:
         return _contest_yaml
@@ -25,22 +26,22 @@ def contest_yaml():
 _problems_yaml = None
 
 
-def problems_yaml():
+def problems_yaml() -> Optional[list[dict[str, Any]]]:
     global _problems_yaml
-    if _problems_yaml:
-        return _problems_yaml
     if _problems_yaml is False:
         return None
+    if _problems_yaml:
+        return _problems_yaml
 
     problemsyaml_path = Path("problems.yaml")
     if not problemsyaml_path.is_file():
         _problems_yaml = False
         return None
     _problems_yaml = read_yaml(problemsyaml_path)
-    return _problems_yaml
+    return cast(list[dict[str, Any]], _problems_yaml)
 
 
-def get_api():
+def get_api() -> str:
     api = config.args.api or contest_yaml().get("api")
     if not api:
         fatal(
@@ -105,7 +106,7 @@ def call_api(method, endpoint, **kwargs):
     return r
 
 
-def call_api_get_json(url):
+def call_api_get_json(url: str):
     r = call_api("GET", url)
     r.raise_for_status()
     try:

bin/export.py

@@ -1,12 +1,13 @@
+import config
 import datetime
+import re
+import shutil
 import sys
+import util
 import yaml
-import re
 import zipfile
-import config
-import util
 from pathlib import Path
-from typing import Optional
+from typing import Any, Optional
 
 from contest import *
 from latex import PdfType

bin/fuzz.py

@@ -3,8 +3,11 @@
 import run
 import random
 import generate
+import signal
 import time
 import threading
+from pathlib import Path
+from typing import Any, Optional
 
 import parallel
 from util import *
@@ -22,9 +25,11 @@
 
 
 class GeneratorTask:
-    def __init__(self, fuzz: "Fuzz", t, i, tmp_id):
+    def __init__(self, fuzz: "Fuzz", t: generate.TestcaseRule, i: int, tmp_id: int):
         self.fuzz = fuzz
-        self.generator = t.generator
+        generator = t.generator
+        assert generator is not None
+        self.generator = generator
         self.solution = t.config.solution
         self.i = i
         self.tmp_id = tmp_id
@@ -37,13 +42,13 @@ def __init__(self, fuzz: "Fuzz", t, i, tmp_id):
         self.save_mutex = threading.Lock()
         self.saved = False
 
-    def run(self, bar):
+    def run(self, bar: ProgressBar) -> None:
         if self._run(bar):
             self.fuzz.finish_task(self.tmp_id)
         else:
             self.fuzz.finish_task(self.tmp_id, 1 + len(self.fuzz.submissions))
 
-    def _run(self, bar):
+    def _run(self, bar: ProgressBar) -> bool:
         # GENERATE THE TEST DATA
         dir = Path("fuzz") / f"tmp_id_{str(self.tmp_id)}"
         cwd = self.fuzz.problem.tmpdir / "tool_runs" / dir
@@ -102,7 +107,7 @@ def _run(self, bar):
                 self.fuzz.queue.put(SubmissionTask(self, submission, testcase, self.tmp_id))
         return True
 
-    def save_test(self, bar):
+    def save_test(self, bar: ProgressBar) -> None:
         if self.saved:
             return
         save = False
@@ -120,17 +125,23 @@ def save_test(self, bar):
 
 
 class SubmissionTask:
-    def __init__(self, generator_task, submission, testcase, tmp_id):
+    def __init__(
+        self,
+        generator_task: GeneratorTask,
+        submission: run.Submission,
+        testcase: Testcase,
+        tmp_id: int,
+    ):
         self.generator_task = generator_task
         self.submission = submission
         self.testcase = testcase
         self.tmp_id = tmp_id
 
-    def run(self, bar):
+    def run(self, bar: ProgressBar) -> None:
         self._run(bar)
         self.generator_task.fuzz.finish_task(self.tmp_id)
 
-    def _run(self, bar):
+    def _run(self, bar: ProgressBar) -> None:
         r = run.Run(self.generator_task.fuzz.problem, self.submission, self.testcase)
         localbar = bar.start(f"{self.generator_task.i}: {self.submission.name}")
         result = r.run(localbar)
@@ -153,10 +164,11 @@ def __init__(self, problem: problem.Problem):
         # Filter to only keep valid rules depending on seed without duplicates from count
         added_testcase_rules = set()
 
-        def add_testcase(t):
+        def add_testcase(t: generate.TestcaseRule) -> None:
             if (
                 t.in_is_generated
                 and t.parse_error is None
+                and t.generator is not None
                 and t.generator.uses_seed
                 and t.generator.command_string.strip() not in added_testcase_rules
             ):
@@ -175,7 +187,7 @@ def add_testcase(t):
         # SUBMISSIONS
         self.submissions = self.problem.selected_or_accepted_submissions()
 
-    def run(self):
+    def run(self) -> bool:
         if not has_ryaml:
             error("Fuzzing needs the ruamel.yaml python3 library. Install python[3]-ruamel.yaml.")
             return False
@@ -190,7 +202,7 @@ def run(self):
 
         message("Press CTRL+C to stop\n", "Fuzz", color_type=MessageType.LOG)
 
-        def runner(task: GeneratorTask):
+        def runner(task: GeneratorTask | SubmissionTask) -> None:
             task.run(bar)
 
         # config.args.no_bar = True
@@ -201,7 +213,7 @@ def runner(task: GeneratorTask):
         self.tasks = 0
         self.queue = parallel.new_queue(runner, pin=True)
 
-        def soft_exit(sig, frame):
+        def soft_exit(sig: Any, frame: Any) -> None:
             if self.queue.aborted:
                 fatal("Running interrupted", force=True)
             else:
@@ -238,7 +250,7 @@ def soft_exit(sig, frame):
 
     # finish task from generator with tmp_id
     # also add new tasks if queue becomes too empty
-    def finish_task(self, tmp_id=None, count=1):
+    def finish_task(self, tmp_id: Optional[int] = None, count: int = 1) -> None:
         with self.queue:
             # return tmp_id (and reuse it if all submissions are finished)
             if tmp_id is not None:
@@ -257,18 +269,18 @@ def finish_task(self, tmp_id=None, count=1):
                 self.iteration += 1
                 # 1 new generator tasks which will also create one task per submission
                 new_tasks = 1 + len(self.submissions)
-                tmp_id = min(self.free_tmp_id)
-                self.free_tmp_id.remove(tmp_id)
-                self.tmp_id_count[tmp_id] = new_tasks
+                new_tmp_id = min(self.free_tmp_id)
+                self.free_tmp_id.remove(new_tmp_id)
+                self.tmp_id_count[new_tmp_id] = new_tasks
                 self.tasks += new_tasks
                 self.queue.put(
-                    GeneratorTask(self, testcase_rule, self.iteration, tmp_id),
+                    GeneratorTask(self, testcase_rule, self.iteration, new_tmp_id),
                     priority=1,
                 )
 
     # Write new rule to yaml
     # lock between read and write to ensure that no rule gets lost
-    def save_test(self, command):
+    def save_test(self, command: str) -> None:
         with self.generators_yaml_mutex:
             generators_yaml = self.problem.path / "generators/generators.yaml"
             data = None