From 143c65283671d411322ed21612270498ddba183f Mon Sep 17 00:00:00 2001 From: Aviral Nigam Date: Wed, 23 Sep 2026 11:13:42 +0530 Subject: [PATCH] feature: Add more power APIs to L0 Sysman python binding Added following power APIs to python binding: 1. zesPowerGetUsage 2. zesPowerGetLimitsExt2 3. zesPowerSetLimitsExt2 4. zesPowerGetEnergyThreshold 5. zesPowerSetEnergyThreshold Along with the zes_energy_threshold_t structure, unit tests, power black box test coverage and README updates. Related-To: SYSM-430 Signed-off-by: Aviral Nigam --- bindings/sysman/python/README.md | 5 + .../source/examples/pyzes_black_box_test.py | 171 +++++++++++++++++- bindings/sysman/python/source/pyzes.py | 111 ++++++++++++ .../python/test/unit_tests/test_power.py | 118 ++++++++++++ 4 files changed, 403 insertions(+), 2 deletions(-) diff --git a/bindings/sysman/python/README.md b/bindings/sysman/python/README.md index 8bafe655..4b580056 100644 --- a/bindings/sysman/python/README.md +++ b/bindings/sysman/python/README.md @@ -169,6 +169,11 @@ SUPPORTED APIs | `zesPowerGetProperties` | Power | 0.1.2 | None | | `zesPowerGetLimitsExt` | Power | 0.1.2 | Extended power limits API | | `zesPowerSetLimitsExt` | Power | 0.1.2 | Linux: Requires superuser or write permissions for telem nodes | +| `zesPowerGetUsage` | Power | 0.1.3 | None | +| `zesPowerGetLimitsExt2` | Power | 0.1.3 | None | +| `zesPowerSetLimitsExt2` | Power | 0.1.3 | Linux: Requires superuser | +| `zesPowerGetEnergyThreshold` | Power | 0.1.3 | None | +| `zesPowerSetEnergyThreshold` | Power | 0.1.3 | Linux: Requires superuser | | **Frequency Management** |-|-|-| | `zesDeviceEnumFrequencyDomains` | Frequency | 0.1.0 | None | | `zesFrequencyGetState` | Frequency | 0.1.0 | None | diff --git a/bindings/sysman/python/source/examples/pyzes_black_box_test.py b/bindings/sysman/python/source/examples/pyzes_black_box_test.py index 135191c3..9de4ab8f 100755 --- a/bindings/sysman/python/source/examples/pyzes_black_box_test.py +++ b/bindings/sysman/python/source/examples/pyzes_black_box_test.py @@ -929,7 +929,9 @@ def test_memory_modules(device_handle, device_index): return True -def test_power_module(device_handle, device_index): +def test_power_module( + device_handle, device_index, set_global_limit=None, set_energy_threshold=None +): """Test power domain enumeration, properties, energy-derived power, and power limit extension operations""" print(f"\n---- Device {device_index} Power Domains Test ----") @@ -1056,6 +1058,140 @@ def test_power_module(device_handle, device_index): else: print_verbose(" Current Power: unavailable due to zero delta time") + instant_power = c_uint32(0) + average_power = c_uint32(0) + rc = pz.zesPowerGetUsage( + power_handles[i], byref(instant_power), byref(average_power) + ) + if rc == pz.ZE_RESULT_SUCCESS: + print_verbose(" Power Usage:") + print_verbose(f" Instant Power: {instant_power.value} mW") + print_verbose(f" Average Power: {average_power.value} mW") + else: + print_verbose(f" Power Usage: Not available ({get_result_string(rc)})") + + limit_ext2 = c_uint32(0) + rc = pz.zesPowerGetLimitsExt2(power_handles[i], byref(limit_ext2)) + limit_ext2_available = rc == pz.ZE_RESULT_SUCCESS + if limit_ext2_available: + print_verbose(f" Power Limit (Ext2): {limit_ext2.value} mW") + else: + print_verbose( + f" Power Limit (Ext2): Not available ({get_result_string(rc)})" + ) + + energy_threshold = None + if properties.isEnergyThresholdSupported: + energy_threshold = pz.zes_energy_threshold_t() + rc = pz.zesPowerGetEnergyThreshold( + power_handles[i], byref(energy_threshold) + ) + if rc == pz.ZE_RESULT_SUCCESS: + print_verbose(" Energy Threshold:") + print_verbose(f" Enabled: {bool(energy_threshold.enable)}") + print_verbose(f" Threshold: {energy_threshold.threshold} J") + print_verbose(f" Process ID: 0x{energy_threshold.processId:X}") + else: + print_verbose( + f" Energy Threshold: Not available ({get_result_string(rc)})" + ) + energy_threshold = None + + # Requested set operations only apply to the root power domain of the selected device + requested_limit = None + if set_global_limit is not None and set_global_limit[0] == device_index: + requested_limit = set_global_limit[1] + requested_threshold = None + if ( + set_energy_threshold is not None + and int(set_energy_threshold[0]) == device_index + ): + requested_threshold = set_energy_threshold[1] + if properties.onSubdevice: + requested_limit = None + requested_threshold = None + + if not is_root_user(): + if requested_limit is not None or requested_threshold is not None: + print_verbose( + " Skipping power set operations due to insufficient permissions" + ) + elif properties.canControl: + if limit_ext2_available and requested_limit is not None: + rc = pz.zesPowerSetLimitsExt2(power_handles[i], requested_limit) + if check_rc(f"zesPowerSetLimitsExt2(power {i}, {requested_limit})", rc): + print_verbose(f" Power limit (Ext2) set to {requested_limit} mW") + read_back = c_uint32(0) + rc = pz.zesPowerGetLimitsExt2(power_handles[i], byref(read_back)) + if check_rc(f"zesPowerGetLimitsExt2(power {i}, verify)", rc): + if read_back.value == requested_limit: + print_verbose( + f" Read back power limit (Ext2): {read_back.value} mW (OK)" + ) + else: + print_verbose( + f" Warning: requested power limit {requested_limit} mW " + f"does not match applied limit {read_back.value} mW" + ) + + # Restore the limit read before the test + rc = pz.zesPowerSetLimitsExt2(power_handles[i], limit_ext2.value) + if check_rc(f"zesPowerSetLimitsExt2(power {i}, restore)", rc): + print_verbose( + f" Restored power limit (Ext2) to {limit_ext2.value} mW" + ) + elif limit_ext2_available: + # Write back the limit just read so the device configuration is unchanged + rc = pz.zesPowerSetLimitsExt2(power_handles[i], limit_ext2.value) + if check_rc(f"zesPowerSetLimitsExt2(power {i})", rc): + print_verbose(" Set power limit (Ext2) successfully") + + if energy_threshold is not None and requested_threshold is not None: + rc = pz.zesPowerSetEnergyThreshold( + power_handles[i], requested_threshold + ) + if check_rc( + f"zesPowerSetEnergyThreshold(power {i}, {requested_threshold})", rc + ): + print_verbose( + f" Energy threshold set to {requested_threshold} J" + ) + read_back_threshold = pz.zes_energy_threshold_t() + rc = pz.zesPowerGetEnergyThreshold( + power_handles[i], byref(read_back_threshold) + ) + if check_rc(f"zesPowerGetEnergyThreshold(power {i}, verify)", rc): + print_verbose( + f" Read back energy threshold: {read_back_threshold.threshold} J " + f"(enabled: {bool(read_back_threshold.enable)}, " + f"process ID: 0x{read_back_threshold.processId:X})" + ) + + # An energy threshold cannot be disabled, so only an enabled one can be restored + if energy_threshold.enable: + rc = pz.zesPowerSetEnergyThreshold( + power_handles[i], energy_threshold.threshold + ) + if check_rc(f"zesPowerSetEnergyThreshold(power {i}, restore)", rc): + print_verbose( + f" Restored energy threshold to {energy_threshold.threshold} J" + ) + else: + print_verbose( + " Energy threshold was not enabled before the test and cannot be disabled" + ) + elif energy_threshold is not None and energy_threshold.enable: + # Only re-apply an already enabled threshold since it cannot be unset afterwards + rc = pz.zesPowerSetEnergyThreshold( + power_handles[i], energy_threshold.threshold + ) + if check_rc(f"zesPowerSetEnergyThreshold(power {i})", rc): + print_verbose(" Set energy threshold successfully") + elif requested_limit is not None or requested_threshold is not None: + print_verbose( + " Skipping power set operations since the domain cannot be controlled" + ) + if properties.onSubdevice or limit_descs is None: continue @@ -1410,6 +1546,8 @@ def main(): %(prog)s -p # PCI tests only %(prog)s -C # ECC tests only %(prog)s -o # Power tests only + %(prog)s -o --set-global-limit 0 150000 # Power tests, set then restore device 0 limit (root) + %(prog)s -o --set-energy-threshold 0 5000.0 # Power tests, set device 0 energy threshold (root) %(prog)s -f # Frequency tests only %(prog)s -t # Temperature tests only %(prog)s -e # Engine tests only @@ -1430,6 +1568,20 @@ def main(): parser.add_argument( "-o", "--power", action="store_true", help="Run only power-related tests" ) + parser.add_argument( + "--set-global-limit", + nargs=2, + type=int, + metavar=("DEVICE", "LIMIT_MW"), + help="With -o, set the power limit (Ext2) of a device in mW, verify it and restore the original (requires root)", + ) + parser.add_argument( + "--set-energy-threshold", + nargs=2, + type=float, + metavar=("DEVICE", "JOULES"), + help="With -o, set the energy threshold of a device in Joules and read it back (requires root)", + ) parser.add_argument("-p", "--pci", action="store_true", help="Run only PCI tests") parser.add_argument("-C", "--ecc", action="store_true", help="Run only ECC tests") parser.add_argument( @@ -1452,6 +1604,16 @@ def main(): parser.add_argument("-e", "--engine", action="store_true", help="Run engine tests ") args = parser.parse_args() + if args.set_global_limit is not None and min(args.set_global_limit) < 0: + parser.error("--set-global-limit values must be non-negative integers") + if args.set_energy_threshold is not None and ( + args.set_energy_threshold[0] < 0 + or not args.set_energy_threshold[0].is_integer() + or args.set_energy_threshold[1] < 0 + ): + parser.error( + "--set-energy-threshold needs a device index and a non-negative threshold" + ) # Check if any specific test is requested specific_test = ( @@ -1507,7 +1669,12 @@ def main(): test_engine_modules(devices[device_idx], device_idx) if args.power: - test_power_module(devices[device_idx], device_idx) + test_power_module( + devices[device_idx], + device_idx, + set_global_limit=args.set_global_limit, + set_energy_threshold=args.set_energy_threshold, + ) if args.frequency: test_frequency_domains(devices[device_idx], device_idx) diff --git a/bindings/sysman/python/source/pyzes.py b/bindings/sysman/python/source/pyzes.py index 25ab2e3d..c2d8c1a4 100644 --- a/bindings/sysman/python/source/pyzes.py +++ b/bindings/sysman/python/source/pyzes.py @@ -669,6 +669,15 @@ class zes_power_energy_counter_t(_PrintableStructure): _fmt_ = {"energy": "%d", "timestamp": "%d microseconds"} +class zes_energy_threshold_t(_PrintableStructure): + _fields_ = [ + ("enable", ze_bool_t), # is the energy threshold enabled + ("threshold", c_double), # energy threshold in Joules, 0.0 if not set + ("processId", c_uint32), # host process ID that set the threshold + ] + _fmt_ = {"threshold": "%.3f J"} + + ## Frequency structures ## class zes_freq_properties_t(_PrintableStructure): _fields_ = [ @@ -1262,6 +1271,108 @@ def zesPowerSetLimitsExt(hPower, pCount, pSustained): return retVal +def zesPowerGetUsage(hPower, pInstantPower, pAveragePower): + """Wraps API: + ze_result_t zesPowerGetUsage( + zes_pwr_handle_t hPower, + uint32_t* pInstantPower, + uint32_t* pAveragePower) + + Parameters: + hPower: power handle + pInstantPower: POINTER(c_uint32) or None - instant power usage in milliwatts + pAveragePower: POINTER(c_uint32) or None - average power usage in milliwatts + Returns: + ze_result_t - return code only, power usage is filled into pInstantPower and pAveragePower + """ + funcPtr = getFunctionPointerList("zesPowerGetUsage") + funcPtr.argtypes = [zes_pwr_handle_t, POINTER(c_uint32), POINTER(c_uint32)] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hPower, pInstantPower, pAveragePower) + return retVal + + +def zesPowerGetLimitsExt2(hPower, pLimit): + """Wraps API: + ze_result_t zesPowerGetLimitsExt2( + zes_pwr_handle_t hPower, + uint32_t* pLimit) + + Parameters: + hPower: power handle + pLimit: POINTER(c_uint32) - power limit in milliwatts to fill + Returns: + ze_result_t - return code only, power limit is filled into pLimit + """ + funcPtr = getFunctionPointerList("zesPowerGetLimitsExt2") + funcPtr.argtypes = [zes_pwr_handle_t, POINTER(c_uint32)] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hPower, pLimit) + return retVal + + +def zesPowerSetLimitsExt2(hPower, limit): + """Wraps API: + ze_result_t zesPowerSetLimitsExt2( + zes_pwr_handle_t hPower, + const uint32_t limit) + + Parameters: + hPower: power handle + limit: power limit in milliwatts to set + Returns: + ze_result_t - return code only + """ + funcPtr = getFunctionPointerList("zesPowerSetLimitsExt2") + funcPtr.argtypes = [zes_pwr_handle_t, c_uint32] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hPower, limit) + return retVal + + +def zesPowerGetEnergyThreshold(hPower, pThreshold): + """Wraps API: + ze_result_t zesPowerGetEnergyThreshold( + zes_pwr_handle_t hPower, + zes_energy_threshold_t* pThreshold) + + Parameters: + hPower: power handle + pThreshold: POINTER(zes_energy_threshold_t) - energy threshold structure to fill + Returns: + ze_result_t - return code only, energy threshold is filled into pThreshold + """ + funcPtr = getFunctionPointerList("zesPowerGetEnergyThreshold") + funcPtr.argtypes = [zes_pwr_handle_t, POINTER(zes_energy_threshold_t)] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hPower, pThreshold) + return retVal + + +def zesPowerSetEnergyThreshold(hPower, threshold): + """Wraps API: + ze_result_t zesPowerSetEnergyThreshold( + zes_pwr_handle_t hPower, + double threshold) + + Parameters: + hPower: power handle + threshold: energy threshold in Joules to set + Returns: + ze_result_t - return code only + """ + funcPtr = getFunctionPointerList("zesPowerSetEnergyThreshold") + funcPtr.argtypes = [zes_pwr_handle_t, c_double] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hPower, threshold) + return retVal + + ## Frequency module functions ## def zesDeviceEnumFrequencyDomains(hDevice, pCount, phFrequency): """Wraps API: diff --git a/bindings/sysman/python/test/unit_tests/test_power.py b/bindings/sysman/python/test/unit_tests/test_power.py index a9094205..cd2967bc 100644 --- a/bindings/sysman/python/test/unit_tests/test_power.py +++ b/bindings/sysman/python/test/unit_tests/test_power.py @@ -202,6 +202,124 @@ def mock_set_limits_ext(power_handle, count_ptr, limits_ptr): mock_get_func.assert_called_with("zesPowerSetLimitsExt") mock_func.assert_called_once() + def test_GivenValidPowerHandleWhenCallingZesPowerGetUsageThenCallSucceedsWithUsageData( + self, mock_get_func + ): + def mock_get_usage(power_handle, instant_ptr, average_ptr): + instant_ptr._obj.value = 45000 + average_ptr._obj.value = 40000 + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_get_usage) + mock_get_func.return_value = mock_func + + power_handle = self.pyzes.zes_pwr_handle_t() + instant_power = c_uint32(0) + average_power = c_uint32(0) + + result = self.pyzes.zesPowerGetUsage( + power_handle, byref(instant_power), byref(average_power) + ) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(instant_power.value, 45000) + self.assertEqual(average_power.value, 40000) + mock_get_func.assert_called_with("zesPowerGetUsage") + mock_func.assert_called_once() + + def test_GivenValidPowerHandleWhenCallingZesPowerGetUsageWithOnlyInstantPowerThenCallSucceeds( + self, mock_get_func + ): + def mock_get_usage(power_handle, instant_ptr, average_ptr): + self.assertIsNone(average_ptr) + instant_ptr._obj.value = 45000 + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_get_usage) + mock_get_func.return_value = mock_func + + power_handle = self.pyzes.zes_pwr_handle_t() + instant_power = c_uint32(0) + + result = self.pyzes.zesPowerGetUsage(power_handle, byref(instant_power), None) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(instant_power.value, 45000) + mock_get_func.assert_called_with("zesPowerGetUsage") + mock_func.assert_called_once() + + def test_GivenValidPowerHandleWhenCallingZesPowerGetLimitsExt2ThenCallSucceedsWithLimit( + self, mock_get_func + ): + def mock_get_limits_ext2(power_handle, limit_ptr): + limit_ptr._obj.value = 250000 + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_get_limits_ext2) + mock_get_func.return_value = mock_func + + power_handle = self.pyzes.zes_pwr_handle_t() + limit = c_uint32(0) + + result = self.pyzes.zesPowerGetLimitsExt2(power_handle, byref(limit)) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(limit.value, 250000) + mock_get_func.assert_called_with("zesPowerGetLimitsExt2") + mock_func.assert_called_once() + + def test_GivenValidPowerHandleWhenCallingZesPowerSetLimitsExt2ThenCallSucceeds( + self, mock_get_func + ): + mock_func = MagicMock(return_value=self.pyzes.ZE_RESULT_SUCCESS) + mock_get_func.return_value = mock_func + + power_handle = self.pyzes.zes_pwr_handle_t() + + result = self.pyzes.zesPowerSetLimitsExt2(power_handle, 275000) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + mock_get_func.assert_called_with("zesPowerSetLimitsExt2") + mock_func.assert_called_once_with(power_handle, 275000) + + def test_GivenValidPowerHandleWhenCallingZesPowerGetEnergyThresholdThenCallSucceedsWithThresholdData( + self, mock_get_func + ): + def mock_get_energy_threshold(power_handle, threshold_ptr): + threshold_ptr._obj.enable = 1 + threshold_ptr._obj.threshold = 1500.5 + threshold_ptr._obj.processId = 4321 + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_get_energy_threshold) + mock_get_func.return_value = mock_func + + power_handle = self.pyzes.zes_pwr_handle_t() + threshold = self.pyzes.zes_energy_threshold_t() + + result = self.pyzes.zesPowerGetEnergyThreshold(power_handle, byref(threshold)) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(threshold.enable, 1) + self.assertEqual(threshold.threshold, 1500.5) + self.assertEqual(threshold.processId, 4321) + mock_get_func.assert_called_with("zesPowerGetEnergyThreshold") + mock_func.assert_called_once() + + def test_GivenValidPowerHandleWhenCallingZesPowerSetEnergyThresholdThenCallSucceeds( + self, mock_get_func + ): + mock_func = MagicMock(return_value=self.pyzes.ZE_RESULT_SUCCESS) + mock_get_func.return_value = mock_func + + power_handle = self.pyzes.zes_pwr_handle_t() + + result = self.pyzes.zesPowerSetEnergyThreshold(power_handle, 1500.5) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + mock_get_func.assert_called_with("zesPowerSetEnergyThreshold") + mock_func.assert_called_once_with(power_handle, 1500.5) + if __name__ == "__main__": unittest.main()