feature: Add pyzes support for driver APIs

bindings/sysman/python/README.md

@@ -158,6 +158,13 @@ SUPPORTED APIs
 | `zesDeviceGetProperties` | Device | 0.1.0 | None |
 | `zesDriverGetDeviceByUuidExp` | Device | 0.1.0 | Experimental API |
 | `zesDeviceProcessesGetState` | Device | 0.1.0 | None |
+| **Driver Management** |-|-|-|
+| `zesDriverGetProperties` | Driver | 0.1.3 | None |
+| `zesDriverGetExtensionProperties` | Driver | 0.1.3 | None |
+| `zesDriverGetExtensionFunctionAddress` | Driver | 0.1.3 | Returns a raw function address |
+| `zesDriverEventListenEx` | Driver | 0.1.3 | None |
+| `zesDriverEventRegisterExt` | Driver | 0.1.3 | None |
+| `zesDriverEventListenExt` | Driver | 0.1.3 | None |
 | **Memory Management** |-|-|-|
 | `zesDeviceEnumMemoryModules` | Memory | 0.1.0 | None |
 | `zesMemoryGetProperties` | Memory | 0.1.0 | None |

bindings/sysman/python/source/examples/pyzes_black_box_test.py

@@ -321,6 +321,33 @@ def get_device_action_string(action):
     return action_map.get(action, f"UNKNOWN_DEVICE_ACTION_{action}")
 
 
+def get_event_type_flags_string(events):
+    """Convert event type flags to a string of flag names"""
+    flag_map = {
+        pz.ZES_EVENT_TYPE_FLAG_DEVICE_DETACH: "DEVICE_DETACH",
+        pz.ZES_EVENT_TYPE_FLAG_DEVICE_ATTACH: "DEVICE_ATTACH",
+        pz.ZES_EVENT_TYPE_FLAG_DEVICE_SLEEP_STATE_ENTER: "DEVICE_SLEEP_STATE_ENTER",
+        pz.ZES_EVENT_TYPE_FLAG_DEVICE_SLEEP_STATE_EXIT: "DEVICE_SLEEP_STATE_EXIT",
+        pz.ZES_EVENT_TYPE_FLAG_FREQ_THROTTLED: "FREQ_THROTTLED",
+        pz.ZES_EVENT_TYPE_FLAG_ENERGY_THRESHOLD_CROSSED: "ENERGY_THRESHOLD_CROSSED",
+        pz.ZES_EVENT_TYPE_FLAG_TEMP_CRITICAL: "TEMP_CRITICAL",
+        pz.ZES_EVENT_TYPE_FLAG_TEMP_THRESHOLD1: "TEMP_THRESHOLD1",
+        pz.ZES_EVENT_TYPE_FLAG_TEMP_THRESHOLD2: "TEMP_THRESHOLD2",
+        pz.ZES_EVENT_TYPE_FLAG_MEM_HEALTH: "MEM_HEALTH",
+        pz.ZES_EVENT_TYPE_FLAG_FABRIC_PORT_HEALTH: "FABRIC_PORT_HEALTH",
+        pz.ZES_EVENT_TYPE_FLAG_PCI_LINK_HEALTH: "PCI_LINK_HEALTH",
+        pz.ZES_EVENT_TYPE_FLAG_RAS_CORRECTABLE_ERRORS: "RAS_CORRECTABLE_ERRORS",
+        pz.ZES_EVENT_TYPE_FLAG_RAS_UNCORRECTABLE_ERRORS: "RAS_UNCORRECTABLE_ERRORS",
+        pz.ZES_EVENT_TYPE_FLAG_DEVICE_RESET_REQUIRED: "DEVICE_RESET_REQUIRED",
+        pz.ZES_EVENT_TYPE_FLAG_SURVIVABILITY_MODE_DETECTED: "SURVIVABILITY_MODE_DETECTED",
+        pz.ZES_EVENT_TYPE_FLAG_INFO_LOG_CPER_DATA_AVAILABLE_EXT: "INFO_LOG_CPER_DATA_AVAILABLE_EXT",
+    }
+    if events == 0:
+        return "None"
+    names = [name for flag, name in flag_map.items() if events & flag]
+    return " | ".join(names) if names else f"UNKNOWN_EVENTS_{events:#x}"
+
+
 def is_root_user():
     """Return whether the current user has root privileges on platforms that support it"""
     geteuid = getattr(os, "geteuid", None)
@@ -840,6 +867,159 @@ def test_engine_modules(device_handle, device_index):
     return True
 
 
+def print_received_events(device_count, num_device_events, events, driver_events=None):
+    """Print the events reported by an event listen call"""
+    print_verbose(f"  Devices With Events: {num_device_events}")
+    for i in range(device_count):
+        if events[i] == 0:
+            continue
+        for bit in range(32):
+            flag = events[i] & (1 << bit)
+            if flag:
+                print_verbose(
+                    f"    Device {i} got {get_event_type_flags_string(flag)} event"
+                )
+
+    if driver_events is None:
+        return
+    if driver_events == 0:
+        print_verbose("  Driver Events: none")
+    if driver_events & pz.ZES_EVENT_TYPE_FLAG_INFO_LOG_CPER_DATA_AVAILABLE_EXT:
+        print_verbose("    Driver got INFO_LOG_CPER_DATA_AVAILABLE_EXT event")
+        print_verbose("    CPER records are available to read from the device info log")
+    other_events = (
+        driver_events & ~pz.ZES_EVENT_TYPE_FLAG_INFO_LOG_CPER_DATA_AVAILABLE_EXT
+    )
+    if other_events:
+        print_verbose(
+            f"    Driver got unexpected events: {get_event_type_flags_string(other_events)}"
+        )
+
+
+def test_driver_properties_module(driver_handle, driver_index):
+    """Test driver properties, extensions, and the extension function lookup"""
+    print(f"\n---- Driver {driver_index} Driver Properties Test ----")
+
+    props = pz.zes_driver_properties_t()
+    props.stype = pz.ZES_STRUCTURE_TYPE_DRIVER_PROPERTIES
+    props.pNext = None
+
+    rc = pz.zesDriverGetProperties(driver_handle, byref(props))
+    if check_rc(f"zesDriverGetProperties(driver {driver_index})", rc):
+        print_verbose("  Properties:")
+        print_verbose(
+            f"    UUID: {'-'.join(f'{props.uuid.id[i]:02x}' for i in range(16))}"
+        )
+        print_verbose(f"    Driver Version: {props.driverVersion}")
+
+    ext_count = c_uint32(0)
+    rc = pz.zesDriverGetExtensionProperties(driver_handle, byref(ext_count), None)
+    if check_rc(f"zesDriverGetExtensionProperties(driver {driver_index}, count)", rc):
+        print_verbose(f"  Found {ext_count.value} extension(s)")
+        if ext_count.value > 0:
+            ExtensionArray = pz.zes_driver_extension_properties_t * ext_count.value
+            extensions = ExtensionArray()
+            rc = pz.zesDriverGetExtensionProperties(
+                driver_handle, byref(ext_count), extensions
+            )
+            if check_rc(
+                f"zesDriverGetExtensionProperties(driver {driver_index}, data)", rc
+            ):
+                for i in range(ext_count.value):
+                    version = extensions[i].version
+                    print_verbose(
+                        f"    {extensions[i].name.decode('utf-8', errors='replace')}: "
+                        f"v{version >> 16}.{version & 0xFFFF}"
+                    )
+
+    # Only the address of this extension function is queried, the function is not called
+    function_name = b"zesIntelDriverRescanDevicesExp"
+    function_address = c_void_p()
+    rc = pz.zesDriverGetExtensionFunctionAddress(
+        driver_handle, function_name, byref(function_address)
+    )
+    if rc == pz.ZE_RESULT_SUCCESS:
+        print_verbose(
+            f"  Extension Function Address ({function_name.decode()}): "
+            f"{function_address.value:#x}"
+        )
+    else:
+        print_verbose(
+            f"  Extension Function Address ({function_name.decode()}): "
+            f"Not available ({get_result_string(rc)})"
+        )
+
+    return True
+
+
+def test_event_module(driver_handle, driver_index, listen_timeout=0):
+    """Test driver event registration and listening"""
+    print(f"\n---- Driver {driver_index} Event Test ----")
+
+    device_info = get_devices(driver_handle)
+    if device_info is None:
+        print_verbose("  Skipping event listen tests as no devices were found")
+        return True
+
+    devices, device_count = device_info
+    EventsArray = pz.zes_event_type_flags_t * device_count
+
+    # A timeout of 0 checks for pending events and returns immediately
+    print_verbose(
+        f"  Listening for device events on {device_count} device(s) with a {listen_timeout} ms timeout"
+    )
+    num_device_events = c_uint32(0)
+    events = EventsArray()
+    rc = pz.zesDriverEventListenEx(
+        driver_handle,
+        listen_timeout,
+        device_count,
+        devices,
+        byref(num_device_events),
+        events,
+    )
+    if check_rc(f"zesDriverEventListenEx(driver {driver_index})", rc):
+        print_received_events(device_count, num_device_events.value, events)
+
+    driver_events_to_register = pz.ZES_EVENT_TYPE_FLAG_INFO_LOG_CPER_DATA_AVAILABLE_EXT
+    rc = pz.zesDriverEventRegisterExt(driver_handle, driver_events_to_register)
+    if rc != pz.ZE_RESULT_SUCCESS:
+        print_verbose(
+            f"  Driver Event Registration: Not available ({get_result_string(rc)})"
+        )
+        return True
+
+    print_verbose(
+        f"  Registered driver events: {get_event_type_flags_string(driver_events_to_register)}"
+    )
+
+    print_verbose(
+        f"  Listening for device and driver events with a {listen_timeout} ms timeout"
+    )
+    num_device_events = c_uint32(0)
+    events = EventsArray()
+    driver_events = pz.zes_event_type_flags_t(0)
+    rc = pz.zesDriverEventListenExt(
+        driver_handle,
+        listen_timeout,
+        device_count,
+        devices,
+        byref(num_device_events),
+        events,
+        byref(driver_events),
+    )
+    if check_rc(f"zesDriverEventListenExt(driver {driver_index})", rc):
+        print_received_events(
+            device_count, num_device_events.value, events, driver_events.value
+        )
+
+    # Clear the driver event registration made by this test
+    rc = pz.zesDriverEventRegisterExt(driver_handle, 0)
+    check_rc(f"zesDriverEventRegisterExt(driver {driver_index}, clear)", rc)
+
+    return True
+
+
 def test_memory_modules(device_handle, device_index):
     """Test memory module enumeration and operations"""
     print(f"\n---- Device {device_index} Memory Modules Test ----")
@@ -1359,11 +1539,11 @@ def run_all_tests():
     for driver_idx in range(driver_count):
         print(f"\n=== Driver {driver_idx} Tests ===")
 
-        # Get devices for this driver
+        # Get devices for this driver, driver scoped tests still run without them
         device_info = get_devices(drivers[driver_idx])
         if device_info is None:
             print(f"No devices found for driver {driver_idx}")
-            continue
+            device_info = (None, 0)
 
         devices, device_count = device_info
 
@@ -1395,6 +1575,10 @@ def run_all_tests():
             # Test engine modules
             test_engine_modules(devices[device_idx], device_idx)
 
+        # Test driver properties and events (driver scoped)
+        test_driver_properties_module(drivers[driver_idx], driver_idx)
+        test_event_module(drivers[driver_idx], driver_idx)
+
     print("\n=== Test Completed ===")
     return True
 
@@ -1413,6 +1597,9 @@ def main():
   %(prog)s -f                 # Frequency tests only
   %(prog)s -t                 # Temperature tests only
   %(prog)s -e                 # Engine tests only
+  %(prog)s -D                 # Driver properties tests only
+  %(prog)s -E                 # Event tests only
+  %(prog)s -E --listen-timeout 0       # Event tests, return from event listen immediately
   %(prog)s -h                 # Show help message""",
         formatter_class=argparse.RawDescriptionHelpFormatter,
     )
@@ -1450,8 +1637,26 @@ def main():
         version="Python Level Zero Sysman Black Box Test v1.0",
     )
     parser.add_argument("-e", "--engine", action="store_true", help="Run engine tests ")
+    parser.add_argument(
+        "-D",
+        "--driverproperties",
+        action="store_true",
+        help="Run only driver properties tests",
+    )
+    parser.add_argument(
+        "-E", "--event", action="store_true", help="Run only event tests"
+    )
+    parser.add_argument(
+        "--listen-timeout",
+        type=int,
+        default=10000,
+        metavar="MS",
+        help="With -E, time in milliseconds to wait for events, default 10000, 0 returns immediately",
+    )
 
     args = parser.parse_args()
+    if args.listen_timeout < 0:
+        parser.error("--listen-timeout must be a non-negative integer")
 
     # Check if any specific test is requested
     specific_test = (
@@ -1463,6 +1668,8 @@ def main():
         or args.frequency
         or args.temperature
         or args.engine
+        or args.driverproperties
+        or args.event
         or args.all
     )
 
@@ -1485,6 +1692,18 @@ def main():
                 print("No drivers available for testing")
                 return 1
 
+            # Driver tests are driver scoped, so run them once per driver before the device check
+            for driver_idx in range(driver_count):
+                if args.driverproperties:
+                    test_driver_properties_module(drivers[driver_idx], driver_idx)
+
+                if args.event:
+                    test_event_module(
+                        drivers[driver_idx],
+                        driver_idx,
+                        listen_timeout=args.listen_timeout,
+                    )
+
             if not devices or device_count == 0:
                 print("No devices available for testing")
                 return 1