ruff: re-enable basic checks

graph/common.py

@@ -3,6 +3,6 @@
 
 def fatal(reason):
     """Print the error and exit 1."""
-    sys.stderr.write("Fatal: {}\n".format(reason))
+    sys.stderr.write(f"Fatal: {reason}\n")
     sys.stderr.flush()
     sys.exit(1)

graph/graph.py

@@ -1,4 +1,4 @@
-from typing import Optional
+from __future__ import annotations
 
 import matplotlib
 import matplotlib.pyplot as plt
@@ -36,7 +36,7 @@ def __init__(
         square=False,
         show_source_file=None,
     ) -> None:
-        self.ax2: Optional[Axes] = None
+        self.ax2: Axes | None = None
         self.args = args
         self.fig, self.ax = plt.subplots()
         self.dpi = 100

graph/trace.py

@@ -132,7 +132,7 @@ def get_metric_unit(self, metric_type: Metrics):
     def get_all_metrics(self, metric_type: Metrics, filter=None) -> list[MonitorMetric]:
         """Return all metrics of a given type."""
         metrics = []
-        for _, metric in self.get_monitoring_metric(metric_type).items():
+        for metric in self.get_monitoring_metric(metric_type).values():
             for component_name, component in metric.items():
                 if not filter:
                     metrics.append(component)
@@ -349,7 +349,7 @@ def get_psu_power(self):
         power = 0
         if psus:
             power = [0] * len(psus[next(iter(psus))].get_samples())
-            for _, psu in psus.items():
+            for psu in psus.values():
                 count = 0
                 for value in psu.get_mean():
                     power[count] = power[count] + value

hwbench/bench/benchmark.py

@@ -112,28 +112,20 @@ def pre_run(self):
         p = self.parameters
         cpu_location = ""
         if p.get_pinned_cpu():
-            if isinstance(p.get_pinned_cpu(), (int, str)):
-                cpu_location = " on CPU {:3d}".format(p.get_pinned_cpu())
+            if isinstance(p.get_pinned_cpu(), int | str):
+                cpu_location = f" on CPU {p.get_pinned_cpu():3d}"
             elif isinstance(p.get_pinned_cpu(), list):
-                cpu_location = " on CPU [{}]".format(h.cpu_list_to_range(p.get_pinned_cpu()))
+                cpu_location = f" on CPU [{h.cpu_list_to_range(p.get_pinned_cpu())}]"
             else:
-                h.fatal("Unsupported get_pinned_cpu() format :{}".format(type(p.get_pinned_cpu())))
+                h.fatal(f"Unsupported get_pinned_cpu() format :{type(p.get_pinned_cpu())}")
 
         monitoring = ""
         if self.parameters.get_monitoring():
             monitoring = "(M)"
         print(
-            "[{}] {}/{}/{}{}: {:3d} stressor{} for {}s{}".format(
-                p.get_name(),
-                self.engine_module.get_engine().get_name(),
-                self.engine_module.get_name(),
-                p.get_engine_module_parameter(),
-                monitoring,
-                p.get_engine_instances_count(),
-                cpu_location,
-                p.get_runtime(),
-                status,
-            )
+            f"[{p.get_name()}] {self.engine_module.get_engine().get_name()}/"
+            f"{self.engine_module.get_name()}/{p.get_engine_module_parameter()}{monitoring}: "
+            f"{p.get_engine_instances_count():3d} stressor{cpu_location} for {p.get_runtime()}s{status}"
         )
 
     def post_run(self, run):

hwbench/bench/benchmarks.py

@@ -1,7 +1,8 @@
+from __future__ import annotations
+
 import datetime
 import time
 from datetime import timedelta
-from typing import Optional
 
 from ..environment.hardware import BaseHardware
 from ..utils import helpers as h
@@ -101,7 +102,7 @@ def parse_jobs_config(self, validate_parameters=True):
                     h.fatal("hosting_cpu_cores is not module hosting_cpu_cores_scaling !")
                 pinned_cpu = []
                 while len(hosting_cpu_cores):
-                    for step in range(steps):
+                    for _step in range(steps):
                         for cpu in hosting_cpu_cores.pop():
                             pinned_cpu.append(cpu)
                     self.__schedule_benchmarks(
@@ -111,7 +112,7 @@ def parse_jobs_config(self, validate_parameters=True):
                         validate_parameters,
                     )
             elif hosting_cpu_cores_scaling == "iterate":
-                for iteration in range(len(hosting_cpu_cores)):
+                for _iteration in range(len(hosting_cpu_cores)):
                     # Pick the last CPU of the list
                     pinned_cpu = hosting_cpu_cores.pop()
                     self.__schedule_benchmarks(job, stressor_range_scaling, pinned_cpu, validate_parameters)
@@ -242,7 +243,7 @@ def run(self):
                 print(f"hwbench: [{bench_name}]: started at {datetime.datetime.utcnow()}")
 
             # Save each benchmark result
-            results["{}_{}".format(benchmark.get_parameters().get_name(), benchmark.get_job_number())] = benchmark.run()
+            results[f"{benchmark.get_parameters().get_name()}_{benchmark.get_job_number()}"] = benchmark.run()
         return results
 
     def dump(self):
@@ -272,7 +273,7 @@ def dump(self):
                 print(f"cmdline={' '.join(em.run_cmd(param))}", file=f)
                 print("", file=f)
 
-    def get_monitoring(self) -> Optional[Monitoring]:
+    def get_monitoring(self) -> Monitoring | None:
         """Return the monitoring object"""
         return self.monitoring
 

hwbench/bench/engine.py

@@ -1,6 +1,7 @@
+from __future__ import annotations
+
 import abc
 import pathlib
-from typing import Optional
 
 from ..utils.external import External
 from ..utils.helpers import fatal
@@ -41,7 +42,9 @@ def run(self, params: BenchmarkParameters):
 
 
 class EngineBase(External):
-    def __init__(self, name: str, binary: str, modules: dict[str, EngineModuleBase] = {}):
+    def __init__(self, name: str, binary: str, modules: dict[str, EngineModuleBase] | None = None):
+        if modules is None:
+            modules = {}
         External.__init__(self, pathlib.Path(""))
         self.engine_name = name
         self.binary = binary
@@ -69,7 +72,7 @@ def add_module(self, engine_module: EngineModuleBase):
     def get_modules(self) -> dict[str, EngineModuleBase]:
         return self.modules
 
-    def get_module(self, module_name: str) -> Optional[EngineModuleBase]:
+    def get_module(self, module_name: str) -> EngineModuleBase | None:
         return self.modules.get(module_name)
 
     def module_exists(self, module_name) -> bool:

hwbench/bench/monitoring.py

@@ -123,7 +123,7 @@ def __compact(self):
             # Do not compact metadata
             if metric_name in MonitoringMetadata.list_str():
                 continue
-            for _, component in metric_type.items():
+            for component in metric_type.values():
                 for metric_name, metric in component.items():
                     metric.compact()
 

hwbench/bench/test_benchmarks_common.py

@@ -66,7 +66,7 @@ def parse_jobs_config(self, validate_parameters=True):
             with patch("hwbench.environment.turbostat.Turbostat.check_version") as cv:
                 cv.return_value = True
                 with patch("hwbench.environment.turbostat.Turbostat.run") as ts:
-                    with open("hwbench/tests/parsing/turbostat/run", "r") as f:
+                    with open("hwbench/tests/parsing/turbostat/run") as f:
                         ts.return_value = ast.literal_eval(f.read())
                         return self.benches.parse_jobs_config(validate_parameters)
 

hwbench/config/config.py

@@ -86,7 +86,7 @@ def get_engine(self, section_name) -> str:
 
     def load_engine(self, engine_name) -> EngineBase:
         """Return the engine from <engine_name> type."""
-        module = importlib.import_module("..engines.{}".format(engine_name), package="hwbench.engines")
+        module = importlib.import_module(f"..engines.{engine_name}", package="hwbench.engines")
         return module.Engine()
 
     def get_engine_module(self, section_name) -> str:
@@ -220,11 +220,11 @@ def validate_section(self, section_name):
         """Validate <section_name> section of a config file."""
         for directive in self.get_section(section_name):
             if not self.is_valid_keyword(directive):
-                h.fatal("job {}: invalid keyword {}".format(section_name, directive))
+                h.fatal(f"job {section_name}: invalid keyword {directive}")
             # Execute the validations_<function> from config_syntax file
             # It will validate the syntax of this particular function.
             # An invalid syntax is fatal and halts the program
-            validate_function = getattr(config_syntax, "validate_{}".format(directive))
+            validate_function = getattr(config_syntax, f"validate_{directive}")
             message = validate_function(self, section_name, self.get_section(section_name)[directive])
             if message:
                 h.fatal(f"Job {section_name}: keyword {directive} : {message}")

hwbench/config/test_parse.py

@@ -44,7 +44,7 @@ def test_keywords(self):
                     iba.return_value = True
                     self.get_jobs_config().validate_sections()
         except Exception as exc:
-            assert False, f"'validate_sections' detected a syntax error {exc}"
+            raise AssertionError(f"'validate_sections' detected a syntax error {exc}")
 
     def test_defaults(self):
         """Check if default values are properly set."""

hwbench/engines/spike.py

@@ -134,7 +134,9 @@ def get_fans_speed(self):
         )
         return sum([fan.get_values()[-1] for _, fan in raw_fans.items()])
 
-    def __spawn_stressor(self, additional_args=[], wait_stressor=False):
+    def __spawn_stressor(self, additional_args=None, wait_stressor=False):
+        if additional_args is None:
+            additional_args = []
         args = [
             self.engine_module.engine.get_binary(),
             "-c",

hwbench/engines/stressng.py

@@ -1,5 +1,6 @@
+from __future__ import annotations
+
 import re
-from typing import Optional
 
 from ..bench.benchmark import ExternalBench
 from ..bench.engine import EngineBase, EngineModuleBase
@@ -61,7 +62,7 @@ def version_minor(self) -> int:
             return int(self.version.split(b".")[2])
         return 0
 
-    def get_version(self) -> Optional[str]:
+    def get_version(self) -> str | None:
         if self.version:
             return self.version.decode("utf-8")
         return None

hwbench/engines/stressng_stream.py

@@ -1,5 +1,7 @@
+from __future__ import annotations
+
 import re
-from typing import Any, Optional
+from typing import Any
 
 from ..bench.parameters import BenchmarkParameters
 from .stressng import EngineBase, EngineModulePinnable, StressNG
@@ -23,7 +25,7 @@ def run_cmd(self) -> list[str]:
             str(self.parameters.get_engine_instances_count()),
         ]
 
-        self.stream_l3_size: Optional[int] = None
+        self.stream_l3_size: int | None = None
         if self.stream_l3_size is not None:
             ret.extend(["--stream-l3-size", str(self.stream_l3_size)])
         return ret

hwbench/engines/stressng_vnni.py

@@ -1,5 +1,5 @@
-from collections.abc import Iterable
-from typing import Callable, NamedTuple
+from collections.abc import Callable, Iterable
+from typing import NamedTuple
 
 from ..bench.parameters import BenchmarkParameters
 from ..environment.hardware import BaseHardware

hwbench/environment/base.py

@@ -2,7 +2,6 @@
 
 import pathlib
 from abc import ABC, abstractmethod
-from typing import Optional
 
 
 # This is the interface of Environment
@@ -13,5 +12,5 @@ def __init__(self, out_dir: pathlib.Path):
         pass
 
     @abstractmethod
-    def dump(self) -> dict[str, Optional[str | int] | dict]:
+    def dump(self) -> dict[str, str | int | None | dict]:
         return {}

hwbench/environment/cpu.py

@@ -1,4 +1,4 @@
-from typing import Optional
+from __future__ import annotations
 
 from .cpu_cores import CPU_CORES
 from .cpu_info import CPU_INFO
@@ -54,7 +54,7 @@ def get_peer_siblings(self, logical_cpu) -> list[int]:
         """Return the list of logical cores running on the same physical core."""
         return self.cpu_cores.get_peer_siblings(logical_cpu)
 
-    def get_peer_sibling(self, logical_cpu) -> Optional[int]:
+    def get_peer_sibling(self, logical_cpu) -> int | None:
         """Return sibling of a logical core."""
         return self.cpu_cores.get_peer_sibling(logical_cpu)
 

hwbench/environment/cpu_cores.py

@@ -1,5 +1,6 @@
+from __future__ import annotations
+
 import pathlib
-from typing import Optional
 
 from ..utils.external import External
 
@@ -81,7 +82,7 @@ def get_peer_siblings(self, logical_cpu) -> list[int]:
                     return self.get_cores(socket, core)
         return []
 
-    def get_peer_sibling(self, logical_cpu) -> Optional[int]:
+    def get_peer_sibling(self, logical_cpu) -> int | None:
         """Return sibling of a logical core."""
         # Let's find the associated core/ht of a given logical_cpu
         for core in self.get_peer_siblings(logical_cpu):

hwbench/environment/dmi.py

@@ -2,7 +2,6 @@
 
 import os
 import pathlib
-from typing import Optional
 
 from ..utils.archive import create_tar_from_directory, extract_file_from_tar
 from ..utils.external import External
@@ -18,19 +17,19 @@ def __init__(self, out_dir: pathlib.Path):
         create_tar_from_directory(self.SYS_DMI, pathlib.Path(self.tarfilename.as_posix()))
 
     @staticmethod
-    def bytes_to_dmi_info(payload: Optional[bytes]) -> Optional[str]:
+    def bytes_to_dmi_info(payload: bytes | None) -> str | None:
         if payload is None:
             return None
         return payload.decode("utf-8", "strict").replace("\n", "")
 
     @staticmethod
-    def extract_dmi_payload(tarfile: pathlib.Path, file: str, root_path=SYS_DMI) -> Optional[bytes]:
+    def extract_dmi_payload(tarfile: pathlib.Path, file: str, root_path=SYS_DMI) -> bytes | None:
         return extract_file_from_tar(tarfile.as_posix(), os.path.join(root_path, file))
 
-    def info(self, name: str) -> Optional[str]:
+    def info(self, name: str) -> str | None:
         return self.bytes_to_dmi_info(self.extract_dmi_payload(self.tarfilename, name))
 
-    def dump(self) -> dict[str, Optional[str | int] | dict]:
+    def dump(self) -> dict[str, str | int | None | dict]:
         return {
             "vendor": self.info("sys_vendor"),
             "product": self.info("product_name"),

hwbench/environment/hardware.py

@@ -2,7 +2,6 @@
 
 import pathlib
 from abc import abstractmethod
-from typing import Optional
 
 from ..utils.external import External_Simple
 from .base import BaseEnvironment
@@ -49,7 +48,7 @@ def __init__(self, out_dir: pathlib.Path, monitoring_config):
         DmidecodeRaw(out_dir).run()
         External_Simple(self.out_dir, ["ipmitool", "sdr"], "ipmitool-sdr")
 
-    def dump(self) -> dict[str, Optional[str | int] | dict]:
+    def dump(self) -> dict[str, str | int | None | dict]:
         dump = {
             "dmi": self.dmi.dump(),
             "cpu": self.cpu.dump(),

hwbench/environment/mock.py

@@ -1,9 +1,13 @@
+from __future__ import annotations
+
 from .hardware import Hardware
 from .vendors.mock import MockVendor
 
 
 class MockHardware(Hardware):
-    def __init__(self, flags: list[str] = [], cores: int = 0, cpu=None):
+    def __init__(self, flags: list[str] | None = None, cores: int = 0, cpu=None):
+        if flags is None:
+            flags = []
         self.cpu = cpu
         self.flags = flags
         self.cores = cores

hwbench/environment/test_parse.py

@@ -8,7 +8,7 @@
 path = pathlib.Path("")
 
 
-class TestParseCPU(object):
+class TestParseCPU:
     def test_ami_aptio(self):
         d = pathlib.Path("./hwbench/tests/parsing/ami_aptio/v5")
         print(f"parsing test {d.name}")