diff --git a/graph/common.py b/graph/common.py index 7a9978d..e6b81ea 100644 --- a/graph/common.py +++ b/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) diff --git a/graph/graph.py b/graph/graph.py index b367752..927fe2d 100644 --- a/graph/graph.py +++ b/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 diff --git a/graph/trace.py b/graph/trace.py index b4bae4f..33cd509 100644 --- a/graph/trace.py +++ b/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 diff --git a/hwbench/bench/benchmark.py b/hwbench/bench/benchmark.py index f792b3f..cc8c56d 100644 --- a/hwbench/bench/benchmark.py +++ b/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): diff --git a/hwbench/bench/benchmarks.py b/hwbench/bench/benchmarks.py index 767f10a..47e57da 100644 --- a/hwbench/bench/benchmarks.py +++ b/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 diff --git a/hwbench/bench/engine.py b/hwbench/bench/engine.py index 261c4f3..04873bb 100644 --- a/hwbench/bench/engine.py +++ b/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: diff --git a/hwbench/bench/monitoring.py b/hwbench/bench/monitoring.py index c35f10c..9a8d798 100644 --- a/hwbench/bench/monitoring.py +++ b/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() diff --git a/hwbench/bench/test_benchmarks_common.py b/hwbench/bench/test_benchmarks_common.py index 98108e9..5c8e394 100644 --- a/hwbench/bench/test_benchmarks_common.py +++ b/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) diff --git a/hwbench/config/config.py b/hwbench/config/config.py index 7944302..925d1a1 100644 --- a/hwbench/config/config.py +++ b/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 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 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_ 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}") diff --git a/hwbench/engines/stressng.py b/hwbench/engines/stressng.py index a9eacef..dc2fcbb 100644 --- a/hwbench/engines/stressng.py +++ b/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 diff --git a/hwbench/engines/stressng_stream.py b/hwbench/engines/stressng_stream.py index cd63844..c813614 100644 --- a/hwbench/engines/stressng_stream.py +++ b/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 diff --git a/hwbench/engines/stressng_vnni.py b/hwbench/engines/stressng_vnni.py index 228caf3..ac8ef23 100644 --- a/hwbench/engines/stressng_vnni.py +++ b/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 diff --git a/hwbench/environment/base.py b/hwbench/environment/base.py index 738db49..46de2e4 100644 --- a/hwbench/environment/base.py +++ b/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 {} diff --git a/hwbench/environment/cpu.py b/hwbench/environment/cpu.py index 247858e..4ac50a6 100644 --- a/hwbench/environment/cpu.py +++ b/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) diff --git a/hwbench/environment/cpu_cores.py b/hwbench/environment/cpu_cores.py index 90c81bf..c0366a7 100644 --- a/hwbench/environment/cpu_cores.py +++ b/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): diff --git a/hwbench/environment/dmi.py b/hwbench/environment/dmi.py index 8ef3252..679c84b 100644 --- a/hwbench/environment/dmi.py +++ b/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"), diff --git a/hwbench/environment/hardware.py b/hwbench/environment/hardware.py index 72ee6b4..dda8a2e 100644 --- a/hwbench/environment/hardware.py +++ b/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(), diff --git a/hwbench/environment/test_parse.py b/hwbench/environment/test_parse.py index 9ad188e..383ff3b 100644 --- a/hwbench/environment/test_parse.py +++ b/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}") diff --git a/hwbench/environment/test_vendors.py b/hwbench/environment/test_vendors.py index d544f8c..76e3121 100644 --- a/hwbench/environment/test_vendors.py +++ b/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): diff --git a/hwbench/environment/vendors/hpe/ilorest.py b/hwbench/environment/vendors/hpe/ilorest.py index 0d20a03..d552e4c 100644 --- a/hwbench/environment/vendors/hpe/ilorest.py +++ b/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: diff --git a/hwbench/environment/vendors/monitoring_device.py b/hwbench/environment/vendors/monitoring_device.py index aac2fbe..e707d95 100644 --- a/hwbench/environment/vendors/monitoring_device.py +++ b/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)) diff --git a/hwbench/environment/vendors/vendor.py b/hwbench/environment/vendors/vendor.py index b222cfb..4dd9154 100644 --- a/hwbench/environment/vendors/vendor.py +++ b/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) diff --git a/hwbench/tests/test_env.py b/hwbench/tests/test_env.py index 5a69ef1..33edbfd 100644 --- a/hwbench/tests/test_env.py +++ b/hwbench/tests/test_env.py @@ -1,3 +1,3 @@ -class TestEnv(object): +class TestEnv: def test_dummy(self): print("dummy test") diff --git a/hwbench/utils/archive.py b/hwbench/utils/archive.py index 582c880..83ca186 100644 --- a/hwbench/utils/archive.py +++ b/hwbench/utils/archive.py @@ -3,7 +3,6 @@ import os import pathlib import tarfile -from typing import Optional def create_tar_from_directory(dir: str, tarfilename: pathlib.Path) -> None: @@ -27,7 +26,7 @@ def create_tar_from_directory(dir: str, tarfilename: pathlib.Path) -> None: return None -def extract_file_from_tar(tarfilename: str, filename: str) -> Optional[bytes]: +def extract_file_from_tar(tarfilename: str, filename: str) -> bytes | None: """return a specific file in a tar archive as bytes if the file exists.""" # may raise tarfile.ReadError if tarfilename is not a tar file diff --git a/hwbench/utils/external.py b/hwbench/utils/external.py index 63b7b08..b96735b 100644 --- a/hwbench/utils/external.py +++ b/hwbench/utils/external.py @@ -89,7 +89,7 @@ def run_cmd(self) -> list[str]: return self.cmd_list def parse_version(self, stdout: bytes, _stderr: bytes) -> bytes: - return bytes() + return b"" def parse_cmd(self, stdout: bytes, stderr: bytes): return {} diff --git a/hwbench/utils/helpers.py b/hwbench/utils/helpers.py index 651d347..2373e3b 100644 --- a/hwbench/utils/helpers.py +++ b/hwbench/utils/helpers.py @@ -3,7 +3,7 @@ import sys from datetime import timedelta from shutil import which -from typing import NoReturn, Optional +from typing import NoReturn def fatal(message) -> NoReturn: @@ -42,7 +42,7 @@ def cpu_list_to_range(cpu_list: list[int]) -> str: """ cpu_list.sort() output: list[str] = [] - previous_entry: Optional[int] = cpu_list[0] + previous_entry: int | None = cpu_list[0] for i in range(1, len(cpu_list)): current_entry = cpu_list[i]