REFACTOR: Load config with dataclasses, earlier validation

Author: mdboom

bench_runner/bases.py

@@ -12,7 +12,7 @@
 
 
 def get_bases() -> list[str]:
-    return config.get_bench_runner_config().get("bases", {}).get("versions", [])
+    return config.get_config().bases.versions
 
 
 @functools.cache

bench_runner/config.py

@@ -2,31 +2,130 @@
 Handles the loading of the bench_runner.toml configuration file.
 """
 
+import dataclasses
 import functools
 from pathlib import Path
 import tomllib
-from typing import Any
 
 
-from . import runners
+from . import flags as mflags
+from . import plot as mplot
+from . import runners as mrunners
 from .util import PathLike
 
 
+@dataclasses.dataclass
+class Bases:
+    # The base versions to compare every benchmark run to.
+    # Should be a full-specified version, e.g. "3.13.0".
+    versions: list[str]
+    # List of configuration flags that are compared against the default build of
+    # its commit merge base.
+    compare_to_default: list[str] = dataclasses.field(default_factory=list)
+
+    def __post_init__(self):
+        if len(self.versions) == 0:
+            raise RuntimeError(
+                "No `bases.versions` are defined in `bench_runner.toml`. "
+            )
+        mflags.normalize_flags(self.compare_to_default)
+
+
+@dataclasses.dataclass
+class Notify:
+    # The Github issue to use to send notification emails
+    notification_issue: int = 0
+
+
+@dataclasses.dataclass
+class PublishMirror:
+    # Whether to skip publishing to the mirror
+    skip: bool = False
+
+
+@dataclasses.dataclass
+class Benchmarks:
+    # Benchmarks to exclude from plots.
+    excluded_benchmarks: list[str] = dataclasses.field(default_factory=list)
+
+
+@dataclasses.dataclass
+class Weekly:
+    flags: list[str] = dataclasses.field(default_factory=list)
+    runners: list[str] = dataclasses.field(default_factory=list)
+
+    def __post_init__(self):
+        self.flags = mflags.normalize_flags(self.flags)
+
+
+@dataclasses.dataclass
+class Config:
+    bases: Bases
+    runners: dict[str, mrunners.Runner]
+    publish_mirror: PublishMirror = dataclasses.field(default_factory=PublishMirror)
+    benchmarks: Benchmarks = dataclasses.field(default_factory=Benchmarks)
+    notify: Notify = dataclasses.field(default_factory=Notify)
+    longitudinal_plot: mplot.LongitudinalPlotConfig | None = None
+    flag_effect_plot: mplot.FlagEffectPlotConfig | None = None
+    benchmark_longitudinal_plot: mplot.BenchmarkLongitudinalPlotConfig | None = None
+    weekly: dict[str, Weekly] = dataclasses.field(default_factory=dict)
+
+    def __post_init__(self):
+        self.bases = Bases(**self.bases)  # pyright: ignore[reportCallIssue]
+        if len(self.runners) == 0:
+            raise RuntimeError(
+                "No runners are defined in `bench_runner.toml`. "
+                "Please set up some runners first."
+            )
+        self.runners = {
+            name: mrunners.Runner(
+                nickname=name, **runner  # pyright: ignore[reportCallIssue]
+            )
+            for name, runner in self.runners.items()
+        }
+        if isinstance(self.publish_mirror, dict):
+            self.publish_mirror = PublishMirror(**self.publish_mirror)
+        if isinstance(self.benchmarks, dict):
+            self.benchmarks = Benchmarks(**self.benchmarks)
+        if isinstance(self.notify, dict):
+            self.notify = Notify(**self.notify)
+        self.longitudinal_plot = (
+            mplot.LongitudinalPlotConfig(
+                **self.longitudinal_plot  # pyright: ignore[reportCallIssue]
+            )
+            if self.longitudinal_plot
+            else None
+        )
+        self.flag_effect_plot = (
+            mplot.FlagEffectPlotConfig(
+                **self.flag_effect_plot  # pyright: ignore[reportCallIssue]
+            )
+            if self.flag_effect_plot
+            else None
+        )
+        self.benchmark_longitudinal_plot = (
+            mplot.BenchmarkLongitudinalPlotConfig(
+                **self.benchmark_longitudinal_plot  # pyright: ignore[reportCallIssue]
+            )
+            if self.benchmark_longitudinal_plot
+            else None
+        )
+        self.weekly = {
+            name: Weekly(**weekly)  # pyright: ignore[reportCallIssue]
+            for name, weekly in self.weekly.items()
+        }
+        if len(self.weekly) == 0:
+            self.weekly = {"default": Weekly(runners=list(self.runners.keys()))}
+
+
 @functools.cache
-def get_bench_runner_config(filepath: PathLike | None = None):
+def get_config(filepath: PathLike | None = None) -> Config:
     if filepath is None:
         filepath = Path("bench_runner.toml")
     else:
         filepath = Path(filepath)
 
     with filepath.open("rb") as fd:
-        return tomllib.load(fd)
-
+        content = tomllib.load(fd)
 
-def get_config_for_current_runner(filepath: PathLike | None = None) -> dict[str, Any]:
-    config = get_bench_runner_config(filepath)
-    runner = runners.get_runner_for_hostname(cfgpath=filepath)
-    all_runners = config.get("runners", {})
-    if len(all_runners) >= 1:
-        return all_runners.get(runner.nickname, {})
-    return {}
+    return Config(**content)

bench_runner/flags.py

@@ -56,3 +56,11 @@ def flags_to_human(flags: list[str]) -> Iterable[str]:
             if flag_descr.name == flag:
                 yield flag_descr.short_name
                 break
+
+
+def normalize_flags(flags: list[str]) -> list[str]:
+    result = sorted(set(flags))
+    for flag in result:
+        if flag not in FLAG_MAPPING.values():
+            raise ValueError(f"Invalid flag {flag}")
+    return result

bench_runner/gh.py

@@ -11,11 +11,11 @@
 
 from . import config
 from . import flags as mflags
-from . import runners
 
 
 def get_machines():
-    return [x.name for x in runners.get_runners() if x.available] + ["all"]
+    cfg = config.get_config()
+    return [x.name for x in cfg.runners.values() if x.available] + ["all"]
 
 
 def _get_flags(d: Mapping[str, Any]) -> list[str]:
@@ -71,8 +71,8 @@ def benchmark(
 
 
 def send_notification(body):
-    conf = config.get_bench_runner_config()
-    notification_issue = conf.get("notify", {}).get("notification_issue", 0)
+    cfg = config.get_config()
+    notification_issue = cfg.notify.notification_issue
 
     if notification_issue == 0:
         print("Not sending Github notification.")