Add mechanism to allow subscribing to Driver Warning events

CHANGELOG.md

@@ -49,6 +49,7 @@
 
 #### [nidcpower] Unreleased
 - Added
+  - (Common) Driver warning subscription mechanism (fixes [#2088](https://github.com/ni/nimi-python/issues/2088))
 - Changed
 - Removed
 
@@ -522,6 +523,7 @@
 
 #### [nidigital] Unreleased
 - Added
+  - (Common) Driver warning subscription mechanism (fixes [#2088](https://github.com/ni/nimi-python/issues/2088))
 - Changed
 - Removed
 
@@ -755,6 +757,7 @@
 
 #### [nidmm] Unreleased
 - Added
+  - (Common) Driver warning subscription mechanism (fixes [#2088](https://github.com/ni/nimi-python/issues/2088))
 - Changed
 - Removed
 
@@ -1070,6 +1073,7 @@
 
 #### [nifgen] Unreleased
 - Added
+  - (Common) Driver warning subscription mechanism (fixes [#2088](https://github.com/ni/nimi-python/issues/2088))
 - Changed
 - Removed
 
@@ -1446,6 +1450,7 @@
 
 #### [nimodinst] Unreleased
 - Added
+  - (Common) Driver warning subscription mechanism (fixes [#2088](https://github.com/ni/nimi-python/issues/2088))
 - Changed
 - Removed
 
@@ -1670,6 +1675,7 @@
   - Enabled selected public APIs
   - Basic example
   - Documentation for APIs (not final) 
+  - (Common) Driver warning subscription mechanism (fixes [#2088](https://github.com/ni/nimi-python/issues/2088))
 - Changed
 - Removed
 
@@ -1703,6 +1709,7 @@
 
 #### [niscope] Unreleased
 - Added
+  - (Common) Driver warning subscription mechanism (fixes [#2088](https://github.com/ni/nimi-python/issues/2088))
 - Changed
 - Removed
 
@@ -2142,6 +2149,7 @@
 
 #### [nise] Unreleased
 - Added
+  - (Common) Driver warning subscription mechanism (fixes [#2088](https://github.com/ni/nimi-python/issues/2088))
 - Changed
 - Removed
 
@@ -2298,6 +2306,7 @@
 
 #### [niswitch] Unreleased
 - Added
+  - (Common) Driver warning subscription mechanism (fixes [#2088](https://github.com/ni/nimi-python/issues/2088))
 - Changed
 - Removed
 
@@ -2549,6 +2558,7 @@
 
 #### [nitclk] Unreleased
 - Added
+  - (Common) Driver warning subscription mechanism (fixes [#2088](https://github.com/ni/nimi-python/issues/2088))
 - Changed
 - Removed
 

build/templates/_grpc_stub_interpreter.py.mako

@@ -31,10 +31,11 @@ from . import ${c['file_name']} as ${c['file_name']}  # noqa: F401
 class GrpcStubInterpreter(object):
     '''Interpreter for interacting with a gRPC Stub class'''
 
-    def __init__(self, grpc_options):
+    def __init__(self, grpc_options, warning_event_handler: errors.DriverWarningEvent):
         self._grpc_options = grpc_options
         self._lock = threading.RLock()
         self._client = ${module_name}_grpc.${service_class_prefix}Stub(grpc_options.grpc_channel)
+        self._warning_event_handler = warning_event_handler
         self.set_session_handle()
 
     def set_session_handle(self, value=session_grpc_types.Session()):
@@ -82,7 +83,10 @@ class GrpcStubInterpreter(object):
                     error_message = self.error_message(error_code)
                 except errors.Error:
                     error_message = 'Failed to retrieve error description.'
-            warnings.warn(errors.DriverWarning(error_code, error_message))
+            driver_warning = errors.DriverWarning(error_code, error_message)
+            if self._warning_event_handler is not None:
+                self._warning_event_handler.notify(driver_warning)
+            warnings.warn(driver_warning)
         return response
 % for func_name in sorted(functions):
 % for method_template in functions[func_name]['method_templates']:

build/templates/_library_interpreter.py.mako

@@ -81,9 +81,10 @@ class LibraryInterpreter(object):
     * Converting errors returned by Library into Python exceptions.
     '''
 
-    def __init__(self, encoding):
+    def __init__(self, encoding, warning_event_handler: errors.DriverWarningEvent):
         self._encoding = encoding
         self._library = _library_singleton.get()
+        self._warning_event_handler = warning_event_handler
         % if 'SetRuntimeEnvironment' in functions:
         global _was_runtime_environment_set
         if _was_runtime_environment_set is None:
@@ -112,6 +113,14 @@ class LibraryInterpreter(object):
     def get_session_handle(self):
         return self._${config['session_handle_parameter_name']}
 
+    def generate_driver_warning_event(self, driverwarning: errors.DriverWarning):
+        '''generate_driver_warning_event
+
+        Generates a driver warning event.
+        '''
+        if self._warning_event_handler is not None:
+            self._warning_event_handler.notify(driverwarning)
+
 <%include file="/_library_interpreter.py/_get_error_description.py.mako" args="config=config" />\
 % for func_name in sorted(functions):
 % for method_template in functions[func_name]['method_templates']:

build/templates/errors.py.mako

@@ -120,6 +120,28 @@ class SelfTestError(Error):
 
 
 % endif
+class DriverWarningEvent:
+    '''Event handler for driver warnings.'''
+
+    def __init__(self):
+        self.subscribers = []
+
+    def subscribe(self, callback):
+        """Subscribe to warning events."""
+        if callback not in self.subscribers:
+            self.subscribers.append(callback)
+
+    def unsubscribe(self, callback):
+        """Unsubscribe from warning events."""
+        if callback in self.subscribers:
+            self.subscribers.remove(callback)
+
+    def notify(self, driver_warning: DriverWarning):
+        """Notify all subscribers about the warning."""
+        for callback in self.subscribers:
+            callback(driver_warning)
+
+
 def handle_error(library_interpreter, code, ignore_warnings, is_error_handling):
     '''handle_error
 
@@ -142,4 +164,6 @@ def handle_error(library_interpreter, code, ignore_warnings, is_error_handling):
         raise DriverError(code, description)
 
     assert _is_warning(code)
-    warnings.warn(DriverWarning(code, description))
+    driver_warning = DriverWarning(code, description)
+    library_interpreter.generate_driver_warning_event(driver_warning)
+    warnings.warn(driver_warning)

build/templates/session.py.mako

@@ -227,9 +227,9 @@ class Session(_SessionBase):
 
 <%
 ctor_for_docs = init_function
+import copy
+ctor_for_docs = copy.deepcopy(ctor_for_docs)
 if grpc_supported:
-    import copy
-    ctor_for_docs = copy.deepcopy(ctor_for_docs)
     ctor_for_docs['parameters'].append(
         {
             'default_value': None,
@@ -245,17 +245,34 @@ if grpc_supported:
             'use_in_python_api': False,
         },
     )
+
+ctor_for_docs['parameters'].append(
+    {
+        'default_value': None,
+        'direction': 'in',
+        'documentation': { 'description': 'Driver warning event which can be subscribed to, with a callback method.\nSample callback method:\n\ndef sample_callback_method(driver_warning: ' + module_name + '.DriverWarning):\n    print(str(driver_warning))\n' },
+        'enum': None,
+        'is_repeated_capability': False,
+        'is_session_handle': False,
+        'python_name': 'driver_warning_event',
+        'size': {'mechanism': 'fixed', 'value': 1},
+        'type_in_documentation': module_name + '.DriverWarningEvent',
+        'type_in_documentation_was_calculated': False,
+        'use_in_python_api': False,
+    },
+)
 %>\
         ${helper.get_function_docstring(ctor_for_docs, False, config, indent=8)}
         '''
+        driver_warning_event = errors.DriverWarningEvent()
 % if grpc_supported:
         if grpc_options:
             import ${module_name}._grpc_stub_interpreter as _grpc_stub_interpreter
-            interpreter = _grpc_stub_interpreter.GrpcStubInterpreter(grpc_options)
+            interpreter = _grpc_stub_interpreter.GrpcStubInterpreter(grpc_options, warning_event_handler=driver_warning_event)
         else:
-            interpreter = _library_interpreter.LibraryInterpreter(encoding='windows-1251')
+            interpreter = _library_interpreter.LibraryInterpreter(encoding='windows-1251', warning_event_handler=driver_warning_event)
 % else:
-        interpreter = _library_interpreter.LibraryInterpreter(encoding='windows-1251')
+        interpreter = _library_interpreter.LibraryInterpreter(encoding='windows-1251', warning_event_handler=driver_warning_event)
 % endif
 
         # Initialize the superclass with default values first, populate them later

generated/nidcpower/nidcpower/_grpc_stub_interpreter.py

@@ -20,10 +20,11 @@
 class GrpcStubInterpreter(object):
     '''Interpreter for interacting with a gRPC Stub class'''
 
-    def __init__(self, grpc_options):
+    def __init__(self, grpc_options, warning_event_handler: errors.DriverWarningEvent):
         self._grpc_options = grpc_options
         self._lock = threading.RLock()
         self._client = nidcpower_grpc.NiDCPowerStub(grpc_options.grpc_channel)
+        self._warning_event_handler = warning_event_handler
         self.set_session_handle()
 
     def set_session_handle(self, value=session_grpc_types.Session()):
@@ -71,7 +72,10 @@ def _invoke(self, func, request, metadata=None):
                     error_message = self.error_message(error_code)
                 except errors.Error:
                     error_message = 'Failed to retrieve error description.'
-            warnings.warn(errors.DriverWarning(error_code, error_message))
+            driver_warning = errors.DriverWarning(error_code, error_message)
+            if self._warning_event_handler is not None:
+                self._warning_event_handler.notify(driver_warning)
+            warnings.warn(driver_warning)
         return response
 
     def abort(self, channel_name):  # noqa: N802

generated/nidcpower/nidcpower/_library_interpreter.py

@@ -62,9 +62,10 @@ class LibraryInterpreter(object):
     * Converting errors returned by Library into Python exceptions.
     '''
 
-    def __init__(self, encoding):
+    def __init__(self, encoding, warning_event_handler: errors.DriverWarningEvent):
         self._encoding = encoding
         self._library = _library_singleton.get()
+        self._warning_event_handler = warning_event_handler
         global _was_runtime_environment_set
         if _was_runtime_environment_set is None:
             try:
@@ -91,6 +92,14 @@ def set_session_handle(self, value=0):
     def get_session_handle(self):
         return self._vi
 
+    def generate_driver_warning_event(self, driverwarning: errors.DriverWarning):
+        '''generate_driver_warning_event
+
+        Generates a driver warning event.
+        '''
+        if self._warning_event_handler is not None:
+            self._warning_event_handler.notify(driverwarning)
+
     def get_error_description(self, error_code):
         '''get_error_description
 

generated/nidcpower/nidcpower/errors.py

@@ -101,6 +101,28 @@ def __init__(self, code, msg):
         super(SelfTestError, self).__init__('Self-test failed with code {}: {}'.format(code, msg))
 
 
+class DriverWarningEvent:
+    '''Event handler for driver warnings.'''
+
+    def __init__(self):
+        self.subscribers = []
+
+    def subscribe(self, callback):
+        """Subscribe to warning events."""
+        if callback not in self.subscribers:
+            self.subscribers.append(callback)
+
+    def unsubscribe(self, callback):
+        """Unsubscribe from warning events."""
+        if callback in self.subscribers:
+            self.subscribers.remove(callback)
+
+    def notify(self, driver_warning: DriverWarning):
+        """Notify all subscribers about the warning."""
+        for callback in self.subscribers:
+            callback(driver_warning)
+
+
 def handle_error(library_interpreter, code, ignore_warnings, is_error_handling):
     '''handle_error
 
@@ -123,4 +145,6 @@ def handle_error(library_interpreter, code, ignore_warnings, is_error_handling):
         raise DriverError(code, description)
 
     assert _is_warning(code)
-    warnings.warn(DriverWarning(code, description))
+    driver_warning = DriverWarning(code, description)
+    library_interpreter.generate_driver_warning_event(driver_warning)
+    warnings.warn(driver_warning)

generated/nidcpower/nidcpower/session.py

@@ -7498,16 +7498,23 @@ def __init__(self, resource_name, channels=None, reset=False, options={}, indepe
 
             grpc_options (nidcpower.grpc_session_options.GrpcSessionOptions): MeasurementLink gRPC session options
 
+            driver_warning_event (nidcpower.DriverWarningEvent): Driver warning event which can be subscribed to, with a callback method.
+                Sample callback method:
+
+                def sample_callback_method(driver_warning: nidcpower.DriverWarning):
+                    print(str(driver_warning))
+
 
         Returns:
             session (nidcpower.Session): A session object representing the device.
 
         '''
+        driver_warning_event = errors.DriverWarningEvent()
         if grpc_options:
             import nidcpower._grpc_stub_interpreter as _grpc_stub_interpreter
-            interpreter = _grpc_stub_interpreter.GrpcStubInterpreter(grpc_options)
+            interpreter = _grpc_stub_interpreter.GrpcStubInterpreter(grpc_options, warning_event_handler=driver_warning_event)
         else:
-            interpreter = _library_interpreter.LibraryInterpreter(encoding='windows-1251')
+            interpreter = _library_interpreter.LibraryInterpreter(encoding='windows-1251', warning_event_handler=driver_warning_event)
 
         # Initialize the superclass with default values first, populate them later
         super(Session, self).__init__(

generated/nidigital/nidigital/_grpc_stub_interpreter.py

@@ -18,10 +18,11 @@
 class GrpcStubInterpreter(object):
     '''Interpreter for interacting with a gRPC Stub class'''
 
-    def __init__(self, grpc_options):
+    def __init__(self, grpc_options, warning_event_handler: errors.DriverWarningEvent):
         self._grpc_options = grpc_options
         self._lock = threading.RLock()
         self._client = nidigital_grpc.NiDigitalStub(grpc_options.grpc_channel)
+        self._warning_event_handler = warning_event_handler
         self.set_session_handle()
 
     def set_session_handle(self, value=session_grpc_types.Session()):
@@ -69,7 +70,10 @@ def _invoke(self, func, request, metadata=None):
                     error_message = self.error_message(error_code)
                 except errors.Error:
                     error_message = 'Failed to retrieve error description.'
-            warnings.warn(errors.DriverWarning(error_code, error_message))
+            driver_warning = errors.DriverWarning(error_code, error_message)
+            if self._warning_event_handler is not None:
+                self._warning_event_handler.notify(driver_warning)
+            warnings.warn(driver_warning)
         return response
 
     def abort(self):  # noqa: N802

generated/nidigital/nidigital/_library_interpreter.py

@@ -60,9 +60,10 @@ class LibraryInterpreter(object):
     * Converting errors returned by Library into Python exceptions.
     '''
 
-    def __init__(self, encoding):
+    def __init__(self, encoding, warning_event_handler: errors.DriverWarningEvent):
         self._encoding = encoding
         self._library = _library_singleton.get()
+        self._warning_event_handler = warning_event_handler
         global _was_runtime_environment_set
         if _was_runtime_environment_set is None:
             try:
@@ -89,6 +90,14 @@ def set_session_handle(self, value=0):
     def get_session_handle(self):
         return self._vi
 
+    def generate_driver_warning_event(self, driverwarning: errors.DriverWarning):
+        '''generate_driver_warning_event
+
+        Generates a driver warning event.
+        '''
+        if self._warning_event_handler is not None:
+            self._warning_event_handler.notify(driverwarning)
+
     def get_error_description(self, error_code):
         '''get_error_description