Add zenpower rapl source

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)
 

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

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)