ruff: Fix some extra warnings from pyupgrade plugin

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,5 +1,3 @@
-from typing import Optional
-
 import matplotlib
 import matplotlib.pyplot as plt
 import numpy as np
@@ -36,7 +34,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,6 @@
 import datetime
 import time
 from datetime import timedelta
-from typing import Optional
 
 from ..environment.hardware import BaseHardware
 from ..utils import helpers as h
@@ -242,7 +241,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 +271,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,5 @@
 import abc
 import pathlib
-from typing import Optional
 
 from ..utils.external import External
 from ..utils.helpers import fatal
@@ -71,7 +70,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/engines/stressng.py

@@ -1,5 +1,4 @@
 import re
-from typing import Optional
 
 from ..bench.benchmark import ExternalBench
 from ..bench.engine import EngineBase, EngineModuleBase
@@ -61,7 +60,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,5 @@
 import re
-from typing import Any, Optional
+from typing import Any
 
 from ..bench.parameters import BenchmarkParameters
 from .stressng import EngineBase, EngineModulePinnable, StressNG
@@ -23,7 +23,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,5 +1,3 @@
-from typing import Optional
-
 from .cpu_cores import CPU_CORES
 from .cpu_info import CPU_INFO
 from .numa import NUMA
@@ -54,7 +52,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,4 @@
 import pathlib
-from typing import Optional
 
 from ..utils.external import External
 
@@ -81,7 +80,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/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}")

hwbench/environment/test_vendors.py

@@ -83,7 +83,7 @@ def tearDown(self):
     def sample(self, name):
         """Return the samples for this test."""
         output = None
-        file = open(self.__get_samples_file_name(name), "r")
+        file = open(self.__get_samples_file_name(name))
         output = file.readlines()
         # If the file is empty but json output is requested, let's return an empty string
         if not len(output):

hwbench/environment/vendors/hpe/ilorest.py

@@ -87,18 +87,18 @@ def login(self, last_try=False):
         # We cannot login because of CreateLimitReachedForResource
         # Let's reset the bmc and retry
         if return_code == 32:
-            h.fatal("Cannot login to local ilo, return_code = {}".format(return_code))
+            h.fatal(f"Cannot login to local ilo, return_code = {return_code}")
         elif return_code == 64:
             h.fatal("Cannot login to local ilo", details="BMC is missing")
         elif return_code == 0:
             self.logged = True
             return True
         else:
-            h.fatal("Cannot login to local ilo, return_code = {}".format(return_code))
+            h.fatal(f"Cannot login to local ilo, return_code = {return_code}")
 
     def raw_get(self, endpoint, to_json=False):
         """Perform a raw get."""
-        command = "rawget /redfish/v1{}".format(endpoint)
+        command = f"rawget /redfish/v1{endpoint}"
         return_code, rawget = self.__ilorest(command)
         if return_code != 0:
             raise subprocess.CalledProcessError(returncode=return_code, cmd=command)
@@ -108,11 +108,11 @@ def raw_get(self, endpoint, to_json=False):
 
     def get(self, endpoint, select=None, filter=None, to_json=False):
         """Perform a get."""
-        command = "get {}".format(endpoint)
+        command = f"get {endpoint}"
         if select:
-            command += " --select {}".format(select)
+            command += f" --select {select}"
         if filter:
-            command += ' --filter "{}"'.format(filter)
+            command += f' --filter "{filter}"'
         command += " -j"
         return_code, get = self.__ilorest(command)
         if return_code != 0:
@@ -125,9 +125,9 @@ def list(self, select, filter=None, to_json=False):
         """Perform a get."""
         command = "list "
         if select:
-            command += " --select {}".format(select)
+            command += f" --select {select}"
         if filter:
-            command += ' --filter "{}"'.format(filter)
+            command += f' --filter "{filter}"'
         command += " -j"
         return_code, get = self.__ilorest(command)
         if return_code != 0:

hwbench/environment/vendors/monitoring_device.py

@@ -105,13 +105,13 @@ def connect_redfish(self, username: str, password: str, device_url: str):
             self.redfish_obj.login()
             self.logged = True
         except json.decoder.JSONDecodeError:
-            h.fatal("JSONDecodeError on {}".format(device_url))
+            h.fatal(f"JSONDecodeError on {device_url}")
         except redfish.rest.v1.RetriesExhaustedError:
-            h.fatal("RetriesExhaustedError on {}".format(device_url))
+            h.fatal(f"RetriesExhaustedError on {device_url}")
         except redfish.rest.v1.BadRequestError:
-            h.fatal("BadRequestError on {}".format(device_url))
+            h.fatal(f"BadRequestError on {device_url}")
         except redfish.rest.v1.InvalidCredentialsError:
-            h.fatal("Invalid credentials for {}".format(device_url))
+            h.fatal(f"Invalid credentials for {device_url}")
         except Exception as exception:
             h.fatal(type(exception))
 

hwbench/environment/vendors/vendor.py

@@ -41,7 +41,7 @@ def _load_vendor(self, directory: str, vendor: str):
 
         vendor_modulename = f"hwbench.environment.vendors.{directory}.{vendor}"
         if not find_spec(vendor_modulename):
-            h.fatal("cannot_find vendor module {}".format(vendor_modulename))
+            h.fatal(f"cannot_find vendor module {vendor_modulename}")
 
         return import_module(vendor_modulename)
 

hwbench/tests/test_env.py

@@ -1,3 +1,3 @@
-class TestEnv(object):
+class TestEnv:
     def test_dummy(self):
         print("dummy test")