diff --git a/s_tui/sources/rapl_power_source.py b/s_tui/sources/rapl_power_source.py index 0cdcc9d..5ad180a 100644 --- a/s_tui/sources/rapl_power_source.py +++ b/s_tui/sources/rapl_power_source.py @@ -24,7 +24,7 @@ import logging import time -from s_tui.sources.rapl_read import get_power_reader +from s_tui.sources.rapl_read import get_power_reader, ZenpowerReader from s_tui.sources.source import Source @@ -53,6 +53,7 @@ def __init__(self): self.last_probe = self.reader.read_power() self.max_power = 1 self.last_measurement = [0.0] * len(self.last_probe) + self._direct_power = isinstance(self.reader, ZenpowerReader) multi_sensors = [] for item in self.last_probe: @@ -90,25 +91,31 @@ def update(self) -> None: for m_idx, _ in enumerate(self.last_probe): try: - joule_used = ( - current_measurement_value[m_idx].current - - self.last_probe[m_idx].current - ) / float(self.MICRO_JOULE_IN_JOULE) - - logging.debug("seconds passed %s", seconds_passed) - watts_used = float(joule_used) / float(seconds_passed) - logging.debug("watts used %s", watts_used) - logging.info( - "Joule_Used %f, seconds passed, %f", joule_used, seconds_passed - ) - - if watts_used > 0: - # The information on joules used elapses every once in a - # while, this might lead to negative readings. - # To prevent this, we keep the last value until the next - # update - self.last_measurement[m_idx] = watts_used - logging.info("Power reading elapsed") + if self._direct_power: + # ZenpowerReader returns direct power reading in microwatts + watts_used = current_measurement_value[m_idx].current / 1000000.0 + if watts_used >= 0: + self.last_measurement[m_idx] = watts_used + else: + joule_used = ( + current_measurement_value[m_idx].current + - self.last_probe[m_idx].current + ) / float(self.MICRO_JOULE_IN_JOULE) + + logging.debug("seconds passed %s", seconds_passed) + watts_used = float(joule_used) / float(seconds_passed) + logging.debug("watts used %s", watts_used) + logging.info( + "Joule_Used %f, seconds passed, %f", joule_used, seconds_passed + ) + + if watts_used > 0: + # The information on joules used elapses every once in a + # while, this might lead to negative readings. + # To prevent this, we keep the last value until the next + # update + self.last_measurement[m_idx] = watts_used + logging.info("Power reading elapsed") except (IndexError, AttributeError) as e: logging.warning("Error reading RAPL sensor %d: %s", m_idx, e) diff --git a/s_tui/sources/rapl_read.py b/s_tui/sources/rapl_read.py index b7992e3..e5aac54 100644 --- a/s_tui/sources/rapl_read.py +++ b/s_tui/sources/rapl_read.py @@ -76,6 +76,78 @@ def available() -> bool: return os.path.exists("/sys/class/powercap/intel-rapl") +class ZenpowerReader: + """Reader for ZenPower5 kernel driver (AMD Zen 1-5). + + ZenPower5 exposes power sensors via hwmon sysfs interface: + - power*_input: Power in microwatts (Zen 5 RAPL, or calculated for earlier gens) + - in*_input: Voltage in microvolts + - curr*_input: Current in microamps + + On Zen 5, RAPL provides direct power readings. + On earlier generations, power may be calculated from SVI2 voltage/current. + """ + + ZENPOWER_HWMON_DIR = "/sys/class/hwmon/" + + def __init__(self) -> None: + self.power_inputs: list[tuple[str, str]] = [] + self.basenames = glob.glob(f"{self.ZENPOWER_HWMON_DIR}hwmon*/") + + for path in self.basenames: + name_file = os.path.join(path, "name") + name = cat(name_file, fallback="", binary=False) + if name and "zenpower" in name.lower(): + power_files = sorted(glob.glob(os.path.join(path, "power*_input"))) + for idx, pf in enumerate(power_files): + label_file = pf.replace("_input", "_label") + label = cat(label_file, fallback="", binary=False) + + # Use label from file, or generate fallback + if label: + # Map zenpower labels to user-friendly names + if "SVI2_P_Core" in label: + display_label = "Core" + elif "SVI2_P_SoC" in label: + display_label = "SoC" + elif "RAPL_P_Package" in label: + display_label = "Package" + elif "RAPL_P_Core" in label: + display_label = "Core" + else: + display_label = label.strip() + else: + # No label file - map by position (power1=Core, power2=SoC for zenpower) + if idx == 0: + display_label = "Core" + elif idx == 1: + display_label = "SoC" + else: + display_label = f"power{idx + 1}" + + self.power_inputs.append((display_label, pf)) + + def read_power(self) -> list[RaplStats]: + ret = [] + for label, power_file in self.power_inputs: + try: + value = cat(power_file) + value_float = float(value) + ret.append(RaplStats(label, value_float, 0.0)) + except (OSError, ValueError) as err: + logging.warning("ignoring %r for file %r", (err, power_file), RuntimeWarning) + return ret + + @staticmethod + def available() -> bool: + basenames = glob.glob(f"{ZenpowerReader.ZENPOWER_HWMON_DIR}hwmon*/name") + for path in basenames: + name = cat(path, fallback="", binary=False) + if name and "zenpower" in name.lower(): + return True + return False + + class AMDEnergyReader: def __init__(self) -> None: self.inputs = list( @@ -210,8 +282,8 @@ def available() -> bool: return msr_available() -def get_power_reader() -> RaplReader | AMDEnergyReader | AMDRaplMsrReader | None: - for ReaderType in (RaplReader, AMDEnergyReader, AMDRaplMsrReader): +def get_power_reader() -> RaplReader | ZenpowerReader | AMDEnergyReader | AMDRaplMsrReader | None: + for ReaderType in (ZenpowerReader, RaplReader, AMDEnergyReader, AMDRaplMsrReader): if ReaderType.available(): return ReaderType() return None diff --git a/tests/test_rapl_read.py b/tests/test_rapl_read.py index 3390625..c0529fe 100644 --- a/tests/test_rapl_read.py +++ b/tests/test_rapl_read.py @@ -7,6 +7,7 @@ AMDRaplMsrReader, RaplReader, RaplStats, + ZenpowerReader, get_power_reader, ) @@ -103,6 +104,163 @@ def test_read_power_skips_none_name(self, mocker): assert result == [] +# ===================================================================== +# ZenpowerReader +# ===================================================================== + + +class TestZenpowerReader: + def test_available_true(self, mocker): + """available() returns True when zenpower hwmon device exists.""" + mocker.patch( + "glob.glob", + return_value=[ + "/sys/class/hwmon/hwmon0/name", + "/sys/class/hwmon/hwmon1/name", + ], + ) + mocker.patch( + "s_tui.sources.rapl_read.cat", + side_effect=["zenpower-pci-00c3", "k10temp"], + ) + assert ZenpowerReader.available() is True + + def test_available_false_no_zenpower(self, mocker): + """available() returns False when no zenpower device found.""" + mocker.patch( + "glob.glob", + return_value=[ + "/sys/class/hwmon/hwmon0/name", + "/sys/class/hwmon/hwmon1/name", + ], + ) + mocker.patch( + "s_tui.sources.rapl_read.cat", + side_effect=["k10temp", "acpi_thermal"], + ) + assert ZenpowerReader.available() is False + + def test_available_false_empty_list(self, mocker): + """available() returns False when no hwmon devices exist.""" + mocker.patch("glob.glob", return_value=[]) + assert ZenpowerReader.available() is False + + def test_init_discovers_power_inputs(self, mocker): + """__init__ discovers power*_input files for zenpower device.""" + mocker.patch( + "glob.glob", + side_effect=[ + ["/sys/class/hwmon/hwmon0/name"], # first glob in __init__ + ["/sys/class/hwmon/hwmon0/power1_input"], # power files + ], + ) + mocker.patch( + "s_tui.sources.rapl_read.cat", + side_effect=[ + "zenpower-pci-00c3", # name + "package", # power1_label + ], + ) + reader = ZenpowerReader() + assert len(reader.power_inputs) == 1 + assert reader.power_inputs[0][0] == "package" + + def test_read_power_returns_microwatts(self, mocker): + """read_power() returns power in microwatts.""" + mocker.patch( + "glob.glob", + side_effect=[ + ["/sys/class/hwmon/hwmon0/name"], + ["/sys/class/hwmon/hwmon0/power1_input"], + ], + ) + mocker.patch( + "s_tui.sources.rapl_read.cat", + side_effect=[ + "zenpower-pci-00c3", # name + "package", # power1_label + "28500000", # power1_input (28.5W in microwatts) + ], + ) + reader = ZenpowerReader() + result = reader.read_power() + assert len(result) == 1 + assert result[0].label == "package" + assert result[0].current == 28500000.0 + + def test_read_power_multiple_sensors(self, mocker): + """read_power() returns multiple power sensors.""" + mock_glob = mocker.patch("glob.glob") + call_count = [0] # Track call index + + def glob_side_effect(pattern): + call_count[0] += 1 + # First call: available() check + if "hwmon*/name" in pattern: + return ["/sys/class/hwmon/hwmon0/name"] + # Second call: __init__ basenames + elif pattern == f"{ZenpowerReader.ZENPOWER_HWMON_DIR}hwmon*/": + return ["/sys/class/hwmon/hwmon0/", "/sys/class/hwmon/hwmon1/"] + # Third call (hwmon0): power files + elif "hwmon0/power*_input" in pattern: + return ["/sys/class/hwmon/hwmon0/power1_input", "/sys/class/hwmon/hwmon0/power2_input"] + # Fourth call (hwmon1): power files (empty) + elif "hwmon1/power*_input" in pattern: + return [] + # Fifth call: read_power (same as third) + elif "hwmon0/power*_input" in pattern: + return ["/sys/class/hwmon/hwmon0/power1_input", "/sys/class/hwmon/hwmon0/power2_input"] + return [] + + mock_glob.side_effect = glob_side_effect + + def cat_side_effect(path, fallback="", binary=False): + path_str = str(path) + if path_str.endswith("hwmon0/name"): + return "zenpower-pci-00c3" + elif path_str.endswith("hwmon1/name"): + return "" + elif path_str.endswith("hwmon0/power1_label"): + return "package" + elif path_str.endswith("hwmon0/power2_label"): + return "core" + elif path_str.endswith("hwmon0/power1_input"): + return "28500000" + elif path_str.endswith("hwmon0/power2_input"): + return "15000000" + return fallback + + mocker.patch("s_tui.sources.rapl_read.cat", side_effect=cat_side_effect) + reader = ZenpowerReader() + result = reader.read_power() + assert len(result) == 2 + assert result[0].label == "package" + assert result[0].current == 28500000.0 + assert result[1].label == "core" + assert result[1].current == 15000000.0 + + def test_read_power_skips_on_error(self, mocker): + """read_power() skips entries that raise errors.""" + mocker.patch( + "glob.glob", + side_effect=[ + ["/sys/class/hwmon/hwmon0/name"], + ["/sys/class/hwmon/hwmon0/power1_input"], + ], + ) + mocker.patch( + "s_tui.sources.rapl_read.cat", + side_effect=[ + "zenpower-pci-00c3", # name + "package", # power1_label + OSError("read failed"), # power1_input + ], + ) + reader = ZenpowerReader() + result = reader.read_power() + assert len(result) == 0 + + # ===================================================================== # AMDEnergyReader # ===================================================================== @@ -202,6 +360,17 @@ def test_available_cpuinfo_not_found(self, mocker): class TestGetPowerReader: + def test_returns_zenpower_reader_when_available(self, mocker): + """get_power_reader prefers ZenpowerReader when available.""" + mocker.patch.object(ZenpowerReader, "available", return_value=True) + mocker.patch("glob.glob", return_value=[]) + mocker.patch( + "s_tui.sources.rapl_read.cat", + side_effect=["zenpower-pci-00c3"], + ) + reader = get_power_reader() + assert isinstance(reader, ZenpowerReader) + def test_returns_rapl_reader_when_available(self, mocker): """get_power_reader prefers RaplReader when available.""" mocker.patch.object(RaplReader, "available", return_value=True) @@ -219,6 +388,20 @@ def test_returns_amd_energy_reader_second(self, mocker): reader = get_power_reader() assert isinstance(reader, AMDEnergyReader) + def test_returns_zenpower_reader_second(self, mocker): + """get_power_reader checks ZenpowerReader after RaplReader.""" + mocker.patch.object(RaplReader, "available", return_value=False) + mocker.patch.object(ZenpowerReader, "available", return_value=True) + mocker.patch("glob.glob", return_value=[]) + mocker.patch( + "s_tui.sources.rapl_read.cat", + side_effect=[ + "zenpower-pci-00c3", + ], + ) + reader = get_power_reader() + assert isinstance(reader, ZenpowerReader) + def test_returns_none_when_nothing_available(self, mocker): """get_power_reader returns None when no reader is available.""" mocker.patch.object(RaplReader, "available", return_value=False)