feature: Add more APIs to L0 Sysman python binding

aviralni · aviralni · commit 1ae486cec5d3 · 2026-06-03T17:21:34.000+05:30
Related-To: NEO-18602

Added following APIs to python binding:

1. zesDevicePciGetProperties
2. zesDevicePciGetStats
3. zesDeviceEccAvailable
4. zesDeviceEccConfigurable
5. zesDeviceGetEccState
6. zesDeviceSetEccState
7. zesPowerGetLimitsExt
8. zesPowerSetLimitsExt
9. zesFrequencyGetAvailableClocks
10. zesFrequencyGetRange
11. zesFrequencySetRange
12. zesFrequencyGetThrottleTime

Signed-off-by: Aviral Nigam &lt;aviral.nigam@intel.com&gt;
diff --git a/bindings/sysman/python/source/examples/pyzes_black_box_test.py b/bindings/sysman/python/source/examples/pyzes_black_box_test.py
@@ -139,6 +139,41 @@ def get_device_type_string(device_type):
     return type_map.get(device_type, f"UNKNOWN_DEVICE_TYPE_{device_type}")
 
 
+def get_pci_link_status_string(link_status):
+    """Convert PCI link status enum to string"""
+    status_map = {
+        pz.ZES_PCI_LINK_STATUS_UNKNOWN: "ZES_PCI_LINK_STATUS_UNKNOWN",
+        pz.ZES_PCI_LINK_STATUS_GOOD: "ZES_PCI_LINK_STATUS_GOOD",
+        pz.ZES_PCI_LINK_STATUS_QUALITY_ISSUES: "ZES_PCI_LINK_STATUS_QUALITY_ISSUES",
+        pz.ZES_PCI_LINK_STATUS_STABILITY_ISSUES: "ZES_PCI_LINK_STATUS_STABILITY_ISSUES",
+    }
+    return status_map.get(link_status, f"UNKNOWN_PCI_LINK_STATUS_{link_status}")
+
+
+def get_pci_quality_issues_string(quality_issues):
+    """Convert PCI quality issue flags to string"""
+    if quality_issues == 0:
+        return "None"
+
+    issues = []
+    if quality_issues & pz.ZES_PCI_LINK_QUAL_ISSUE_FLAG_REPLAYS:
+        issues.append("REPLAYS")
+    if quality_issues & pz.ZES_PCI_LINK_QUAL_ISSUE_FLAG_SPEED:
+        issues.append("SPEED")
+    return " | ".join(issues)
+
+
+def get_pci_stability_issues_string(stability_issues):
+    """Convert PCI stability issue flags to string"""
+    if stability_issues == 0:
+        return "None"
+
+    issues = []
+    if stability_issues & pz.ZES_PCI_LINK_STAB_ISSUE_FLAG_RETRAINING:
+        issues.append("RETRAINING")
+    return " | ".join(issues)
+
+
 def get_frequency_domain_string(freq_domain):
     """Convert frequency domain enum to string"""
     domain_map = {
@@ -440,9 +475,9 @@ def test_global_operation(driver_handle, device_handle, device_index):
 
                 # Initialize each structure
                 for i in range(subdevice_count.value):
-                    subdevice_props[i].stype = (
-                        pz.ZES_STRUCTURE_TYPE_SUBDEVICE_EXP_PROPERTIES
-                    )
+                    subdevice_props[
+                        i
+                    ].stype = pz.ZES_STRUCTURE_TYPE_SUBDEVICE_EXP_PROPERTIES
                     subdevice_props[i].pNext = None
 
                 # Second call to get properties
@@ -537,7 +572,7 @@ def test_global_operation(driver_handle, device_handle, device_index):
 
 
 def test_pci_module(device_handle, device_index):
-    """Test PCI properties and stats operations"""
+    """Test PCI properties, state, and stats operations"""
     print(f"\n---- Device {device_index} PCI Test ----")
 
     properties = pz.zes_pci_properties_t()
@@ -562,6 +597,26 @@ def test_pci_module(device_handle, device_index):
     print_verbose(f"  Have Packet Counters: {bool(properties.havePacketCounters)}")
     print_verbose(f"  Have Replay Counters: {bool(properties.haveReplayCounters)}")
 
+    state = pz.zes_pci_state_t()
+    state.stype = pz.ZES_STRUCTURE_TYPE_PCI_STATE
+    state.pNext = None
+
+    rc = pz.zesDevicePciGetState(device_handle, byref(state))
+    if not check_rc(f"zesDevicePciGetState(device {device_index})", rc):
+        return False
+
+    print_verbose("PCI State:")
+    print_verbose(f"  Link Status: {get_pci_link_status_string(state.status)}")
+    print_verbose(
+        f"  Quality Issues: {get_pci_quality_issues_string(state.qualityIssues)}"
+    )
+    print_verbose(
+        f"  Stability Issues: {get_pci_stability_issues_string(state.stabilityIssues)}"
+    )
+    print_verbose(f"  Current Gen: {state.speed.gen}")
+    print_verbose(f"  Current Width: {state.speed.width}")
+    print_verbose(f"  Current Max Bandwidth: {state.speed.maxBandwidth}")
+
     stats = pz.zes_pci_stats_t()
     rc = pz.zesDevicePciGetStats(device_handle, byref(stats))
     if not check_rc(f"zesDevicePciGetStats(device {device_index})", rc):
@@ -897,9 +952,9 @@ def test_power_module(device_handle, device_index):
             limit_descs = PowerLimitArray()
 
             for limit_index in range(limit_count.value):
-                limit_descs[limit_index].stype = (
-                    pz.ZES_STRUCTURE_TYPE_POWER_LIMIT_EXT_DESC
-                )
+                limit_descs[
+                    limit_index
+                ].stype = pz.ZES_STRUCTURE_TYPE_POWER_LIMIT_EXT_DESC
                 limit_descs[limit_index].pNext = None
 
             rc = pz.zesPowerGetLimitsExt(
diff --git a/bindings/sysman/python/source/pyzes.py b/bindings/sysman/python/source/pyzes.py
@@ -355,6 +355,7 @@ class zes_engine_handle_t(c_void_p):
 # Structure type enum values
 ZES_STRUCTURE_TYPE_DEVICE_PROPERTIES = 0x1
 ZES_STRUCTURE_TYPE_PCI_PROPERTIES = 0x2
+ZES_STRUCTURE_TYPE_PCI_STATE = 0x17
 ZES_STRUCTURE_TYPE_DEVICE_ECC_DESC = 0x25
 ZES_STRUCTURE_TYPE_DEVICE_ECC_PROPERTIES = 0x26
 ZES_STRUCTURE_TYPE_POWER_LIMIT_EXT_DESC = 0x27
@@ -462,6 +463,35 @@ class zes_pci_properties_t(_PrintableStructure):
     ]
 
 
+zes_pci_link_status_t = c_int32
+ZES_PCI_LINK_STATUS_UNKNOWN = 0
+ZES_PCI_LINK_STATUS_GOOD = 1
+ZES_PCI_LINK_STATUS_QUALITY_ISSUES = 2
+ZES_PCI_LINK_STATUS_STABILITY_ISSUES = 3
+ZES_PCI_LINK_STATUS_FORCE_UINT32 = 0x7FFFFFFF
+
+zes_pci_link_qual_issue_flags_t = c_uint32
+ZES_PCI_LINK_QUAL_ISSUE_FLAG_REPLAYS = 1 << 0
+ZES_PCI_LINK_QUAL_ISSUE_FLAG_SPEED = 1 << 1
+ZES_PCI_LINK_QUAL_ISSUE_FLAG_FORCE_UINT32 = 0x7FFFFFFF
+
+zes_pci_link_stab_issue_flags_t = c_uint32
+ZES_PCI_LINK_STAB_ISSUE_FLAG_RETRAINING = 1 << 0
+ZES_PCI_LINK_STAB_ISSUE_FLAG_FORCE_UINT32 = 0x7FFFFFFF
+
+
+class zes_pci_state_t(_PrintableStructure):
+    _fields_ = [
+        ("stype", c_int32),
+        ("pNext", c_void_p),
+        ("status", zes_pci_link_status_t),
+        ("qualityIssues", zes_pci_link_qual_issue_flags_t),
+        ("stabilityIssues", zes_pci_link_stab_issue_flags_t),
+        ("speed", zes_pci_speed_t),
+    ]
+    _fmt_ = {"qualityIssues": "0x%08X", "stabilityIssues": "0x%08X"}
+
+
 class zes_pci_stats_t(_PrintableStructure):
     _fields_ = [
         ("timestamp", c_uint64),
@@ -801,6 +831,23 @@ def zesDevicePciGetProperties(hDevice, pProperties):
     return retVal
 
 
+def zesDevicePciGetState(hDevice, pState):
+    """Wraps API:
+    ze_result_t zesDevicePciGetState(zes_device_handle_t hDevice, zes_pci_state_t* pState)
+
+    Parameters:
+        hDevice: device handle
+        pState: POINTER(zes_pci_state_t) - PCI state structure to fill
+    Returns:
+        ze_result_t - return code only, state is filled into pState
+    """
+    funcPtr = getFunctionPointerList("zesDevicePciGetState")
+    funcPtr.argtypes = [zes_device_handle_t, POINTER(zes_pci_state_t)]
+    funcPtr.restype = ze_result_t
+    retVal = funcPtr(hDevice, pState)
+    return retVal
+
+
 def zesDevicePciGetStats(hDevice, pStats):
     """Wraps API:
     ze_result_t zesDevicePciGetStats(zes_device_handle_t hDevice, zes_pci_stats_t* pStats)
diff --git a/bindings/sysman/python/test/unit_tests/test_ecc.py b/bindings/sysman/python/test/unit_tests/test_ecc.py
@@ -0,0 +1,138 @@
+##
+# Copyright (C) 2026 Intel Corporation
+#
+# SPDX-License-Identifier: MIT
+#
+##
+
+import os
+import sys
+import unittest
+from ctypes import *
+from unittest.mock import MagicMock, patch
+
+# Add the source directory to Python path so we can import pyzes
+script_dir = os.path.dirname(os.path.abspath(__file__))
+source_dir = os.path.join(script_dir, "..", "..", "source")
+source_dir = os.path.abspath(source_dir)
+if source_dir not in sys.path:
+    sys.path.insert(0, source_dir)
+
+
+@patch("pyzes.getFunctionPointerList")
+class TestEccFunctions(unittest.TestCase):
+    def setUp(self):
+        import pyzes
+
+        self.pyzes = pyzes
+
+    def test_GivenValidDeviceHandleWhenCallingZesDeviceEccAvailableThenCallSucceedsWithAvailability(
+        self, mock_get_func
+    ):
+        mock_available = 1
+
+        def mock_ecc_available(device_handle, available_ptr):
+            available_ptr._obj.value = mock_available
+            return self.pyzes.ZE_RESULT_SUCCESS
+
+        mock_func = MagicMock(side_effect=mock_ecc_available)
+        mock_get_func.return_value = mock_func
+
+        device_handle = self.pyzes.zes_device_handle_t()
+        available = self.pyzes.ze_bool_t(0)
+
+        result = self.pyzes.zesDeviceEccAvailable(device_handle, byref(available))
+
+        self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS)
+        self.assertEqual(available.value, mock_available)
+        mock_get_func.assert_called_with("zesDeviceEccAvailable")
+        mock_func.assert_called_once()
+
+    def test_GivenValidDeviceHandleWhenCallingZesDeviceEccConfigurableThenCallSucceedsWithConfigurability(
+        self, mock_get_func
+    ):
+        mock_configurable = 1
+
+        def mock_ecc_configurable(device_handle, configurable_ptr):
+            configurable_ptr._obj.value = mock_configurable
+            return self.pyzes.ZE_RESULT_SUCCESS
+
+        mock_func = MagicMock(side_effect=mock_ecc_configurable)
+        mock_get_func.return_value = mock_func
+
+        device_handle = self.pyzes.zes_device_handle_t()
+        configurable = self.pyzes.ze_bool_t(0)
+
+        result = self.pyzes.zesDeviceEccConfigurable(device_handle, byref(configurable))
+
+        self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS)
+        self.assertEqual(configurable.value, mock_configurable)
+        mock_get_func.assert_called_with("zesDeviceEccConfigurable")
+        mock_func.assert_called_once()
+
+    def test_GivenValidDeviceHandleWhenCallingZesDeviceGetEccStateThenCallSucceedsWithState(
+        self, mock_get_func
+    ):
+        mock_current_state = self.pyzes.ZES_DEVICE_ECC_STATE_ENABLED
+        mock_pending_state = self.pyzes.ZES_DEVICE_ECC_STATE_DISABLED
+        mock_pending_action = self.pyzes.ZES_DEVICE_ACTION_WARM_CARD_RESET
+
+        def mock_get_ecc_state(device_handle, state_ptr):
+            state_ptr._obj.currentState = mock_current_state
+            state_ptr._obj.pendingState = mock_pending_state
+            state_ptr._obj.pendingAction = mock_pending_action
+            return self.pyzes.ZE_RESULT_SUCCESS
+
+        mock_func = MagicMock(side_effect=mock_get_ecc_state)
+        mock_get_func.return_value = mock_func
+
+        device_handle = self.pyzes.zes_device_handle_t()
+        ecc_state = self.pyzes.zes_device_ecc_properties_t()
+
+        result = self.pyzes.zesDeviceGetEccState(device_handle, byref(ecc_state))
+
+        self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS)
+        self.assertEqual(ecc_state.currentState, mock_current_state)
+        self.assertEqual(ecc_state.pendingState, mock_pending_state)
+        self.assertEqual(ecc_state.pendingAction, mock_pending_action)
+        mock_get_func.assert_called_with("zesDeviceGetEccState")
+        mock_func.assert_called_once()
+
+    def test_GivenValidDeviceHandleWhenCallingZesDeviceSetEccStateThenCallSucceedsWithUpdatedState(
+        self, mock_get_func
+    ):
+        requested_state = self.pyzes.ZES_DEVICE_ECC_STATE_ENABLED
+        mock_current_state = self.pyzes.ZES_DEVICE_ECC_STATE_DISABLED
+        mock_pending_state = self.pyzes.ZES_DEVICE_ECC_STATE_ENABLED
+        mock_pending_action = self.pyzes.ZES_DEVICE_ACTION_COLD_CARD_RESET
+
+        def mock_set_ecc_state(device_handle, new_state_ptr, state_ptr):
+            self.assertEqual(new_state_ptr._obj.state, requested_state)
+            state_ptr._obj.currentState = mock_current_state
+            state_ptr._obj.pendingState = mock_pending_state
+            state_ptr._obj.pendingAction = mock_pending_action
+            return self.pyzes.ZE_RESULT_SUCCESS
+
+        mock_func = MagicMock(side_effect=mock_set_ecc_state)
+        mock_get_func.return_value = mock_func
+
+        device_handle = self.pyzes.zes_device_handle_t()
+        new_state = self.pyzes.zes_device_ecc_desc_t()
+        new_state.stype = self.pyzes.ZES_STRUCTURE_TYPE_DEVICE_ECC_DESC
+        new_state.state = requested_state
+        ecc_state = self.pyzes.zes_device_ecc_properties_t()
+
+        result = self.pyzes.zesDeviceSetEccState(
+            device_handle, byref(new_state), byref(ecc_state)
+        )
+
+        self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS)
+        self.assertEqual(ecc_state.currentState, mock_current_state)
+        self.assertEqual(ecc_state.pendingState, mock_pending_state)
+        self.assertEqual(ecc_state.pendingAction, mock_pending_action)
+        mock_get_func.assert_called_with("zesDeviceSetEccState")
+        mock_func.assert_called_once()
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/bindings/sysman/python/test/unit_tests/test_global_operations.py b/bindings/sysman/python/test/unit_tests/test_global_operations.py
@@ -195,6 +195,47 @@ def mock_get_pci_stats(device_handle, stats_ptr):
         mock_get_func.assert_called_with("zesDevicePciGetStats")
         mock_func.assert_called_once()
 
+    def test_GivenValidDeviceHandleWhenCallingZesDevicePciGetStateThenCallSucceedsWithValidState(
+        self, mock_get_func
+    ):
+        mock_status = self.pyzes.ZES_PCI_LINK_STATUS_QUALITY_ISSUES
+        mock_quality_issues = (
+            self.pyzes.ZES_PCI_LINK_QUAL_ISSUE_FLAG_REPLAYS
+            | self.pyzes.ZES_PCI_LINK_QUAL_ISSUE_FLAG_SPEED
+        )
+        mock_stability_issues = self.pyzes.ZES_PCI_LINK_STAB_ISSUE_FLAG_RETRAINING
+        mock_gen = 4
+        mock_width = 8
+        mock_max_bandwidth = 16_000_000_000
+
+        def mock_get_pci_state(device_handle, state_ptr):
+            state_ptr._obj.status = mock_status
+            state_ptr._obj.qualityIssues = mock_quality_issues
+            state_ptr._obj.stabilityIssues = mock_stability_issues
+            state_ptr._obj.speed.gen = mock_gen
+            state_ptr._obj.speed.width = mock_width
+            state_ptr._obj.speed.maxBandwidth = mock_max_bandwidth
+            return self.pyzes.ZE_RESULT_SUCCESS
+
+        mock_func = MagicMock(side_effect=mock_get_pci_state)
+        mock_get_func.return_value = mock_func
+
+        device_handle = self.pyzes.zes_device_handle_t()
+        pci_state = self.pyzes.zes_pci_state_t()
+
+        result = self.pyzes.zesDevicePciGetState(device_handle, byref(pci_state))
+
+        self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS)
+        self.assertEqual(pci_state.status, mock_status)
+        self.assertEqual(pci_state.qualityIssues, mock_quality_issues)
+        self.assertEqual(pci_state.stabilityIssues, mock_stability_issues)
+        self.assertEqual(pci_state.speed.gen, mock_gen)
+        self.assertEqual(pci_state.speed.width, mock_width)
+        self.assertEqual(pci_state.speed.maxBandwidth, mock_max_bandwidth)
+
+        mock_get_func.assert_called_with("zesDevicePciGetState")
+        mock_func.assert_called_once()
+
     def test_GivenValidDeviceHandleWhenCallingZesDeviceProcessesGetStateThenCallSucceedsWithProcessCount(
         self, mock_get_func
     ):
