diff --git a/CHANGELOG.md b/CHANGELOG.md index c3563df429..adfbe1d2b1 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/build/templates/_grpc_stub_interpreter.py.mako b/build/templates/_grpc_stub_interpreter.py.mako index 0a5f0e4ef1..1ec73a93aa 100644 --- a/build/templates/_grpc_stub_interpreter.py.mako +++ b/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']: diff --git a/build/templates/_library_interpreter.py.mako b/build/templates/_library_interpreter.py.mako index cf1aea315a..14d3f129eb 100644 --- a/build/templates/_library_interpreter.py.mako +++ b/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']: diff --git a/build/templates/errors.py.mako b/build/templates/errors.py.mako index 15069f184e..4b036b4748 100644 --- a/build/templates/errors.py.mako +++ b/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) diff --git a/build/templates/session.py.mako b/build/templates/session.py.mako index df755677b3..87a81d3465 100644 --- a/build/templates/session.py.mako +++ b/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 diff --git a/generated/nidcpower/nidcpower/_grpc_stub_interpreter.py b/generated/nidcpower/nidcpower/_grpc_stub_interpreter.py index a46fc64c13..76df28cc55 100644 --- a/generated/nidcpower/nidcpower/_grpc_stub_interpreter.py +++ b/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 diff --git a/generated/nidcpower/nidcpower/_library_interpreter.py b/generated/nidcpower/nidcpower/_library_interpreter.py index 473680f188..a6e5dd3713 100644 --- a/generated/nidcpower/nidcpower/_library_interpreter.py +++ b/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 diff --git a/generated/nidcpower/nidcpower/errors.py b/generated/nidcpower/nidcpower/errors.py index 2f4a964e68..50f0ffbd21 100644 --- a/generated/nidcpower/nidcpower/errors.py +++ b/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) diff --git a/generated/nidcpower/nidcpower/session.py b/generated/nidcpower/nidcpower/session.py index 21742df537..299d559d4b 100644 --- a/generated/nidcpower/nidcpower/session.py +++ b/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__( diff --git a/generated/nidigital/nidigital/_grpc_stub_interpreter.py b/generated/nidigital/nidigital/_grpc_stub_interpreter.py index 2891d110dc..246cdf5cf4 100644 --- a/generated/nidigital/nidigital/_grpc_stub_interpreter.py +++ b/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 diff --git a/generated/nidigital/nidigital/_library_interpreter.py b/generated/nidigital/nidigital/_library_interpreter.py index e830756ab8..5ecfba87e1 100644 --- a/generated/nidigital/nidigital/_library_interpreter.py +++ b/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 diff --git a/generated/nidigital/nidigital/errors.py b/generated/nidigital/nidigital/errors.py index dafbadc9a9..e54940da33 100644 --- a/generated/nidigital/nidigital/errors.py +++ b/generated/nidigital/nidigital/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) diff --git a/generated/nidigital/nidigital/session.py b/generated/nidigital/nidigital/session.py index 3678e5dc3f..dffab48bea 100644 --- a/generated/nidigital/nidigital/session.py +++ b/generated/nidigital/nidigital/session.py @@ -3255,16 +3255,23 @@ def __init__(self, resource_name, id_query=False, reset_device=False, options={} grpc_options (nidigital.grpc_session_options.GrpcSessionOptions): MeasurementLink gRPC session options + driver_warning_event (nidigital.DriverWarningEvent): Driver warning event which can be subscribed to, with a callback method. + Sample callback method: + + def sample_callback_method(driver_warning: nidigital.DriverWarning): + print(str(driver_warning)) + Returns: new_vi (int): The returned instrument session. ''' + driver_warning_event = errors.DriverWarningEvent() if grpc_options: import nidigital._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__( diff --git a/generated/nidmm/nidmm/_grpc_stub_interpreter.py b/generated/nidmm/nidmm/_grpc_stub_interpreter.py index e67ea42f9b..deae6d0054 100644 --- a/generated/nidmm/nidmm/_grpc_stub_interpreter.py +++ b/generated/nidmm/nidmm/_grpc_stub_interpreter.py @@ -16,10 +16,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 = nidmm_grpc.NiDmmStub(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()): @@ -67,7 +68,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 diff --git a/generated/nidmm/nidmm/_library_interpreter.py b/generated/nidmm/nidmm/_library_interpreter.py index 4e87b4274b..107f10bc1a 100644 --- a/generated/nidmm/nidmm/_library_interpreter.py +++ b/generated/nidmm/nidmm/_library_interpreter.py @@ -58,9 +58,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: @@ -87,6 +88,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 diff --git a/generated/nidmm/nidmm/errors.py b/generated/nidmm/nidmm/errors.py index fba2f79427..a0ecc6a16b 100644 --- a/generated/nidmm/nidmm/errors.py +++ b/generated/nidmm/nidmm/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) diff --git a/generated/nidmm/nidmm/session.py b/generated/nidmm/nidmm/session.py index 2d04e204e2..24adf446d4 100644 --- a/generated/nidmm/nidmm/session.py +++ b/generated/nidmm/nidmm/session.py @@ -1024,16 +1024,23 @@ def __init__(self, resource_name, id_query=False, reset_device=False, options={} grpc_options (nidmm.grpc_session_options.GrpcSessionOptions): MeasurementLink gRPC session options + driver_warning_event (nidmm.DriverWarningEvent): Driver warning event which can be subscribed to, with a callback method. + Sample callback method: + + def sample_callback_method(driver_warning: nidmm.DriverWarning): + print(str(driver_warning)) + Returns: session (nidmm.Session): A session object representing the device. ''' + driver_warning_event = errors.DriverWarningEvent() if grpc_options: import nidmm._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__( diff --git a/generated/nifake/nifake/_grpc_stub_interpreter.py b/generated/nifake/nifake/_grpc_stub_interpreter.py index ee8d29c047..7adc075299 100644 --- a/generated/nifake/nifake/_grpc_stub_interpreter.py +++ b/generated/nifake/nifake/_grpc_stub_interpreter.py @@ -22,10 +22,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 = nifake_grpc.NiFakeStub(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()): @@ -73,7 +74,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 diff --git a/generated/nifake/nifake/_library_interpreter.py b/generated/nifake/nifake/_library_interpreter.py index d2715505a5..074d447ec8 100644 --- a/generated/nifake/nifake/_library_interpreter.py +++ b/generated/nifake/nifake/_library_interpreter.py @@ -64,9 +64,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: @@ -93,6 +94,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 diff --git a/generated/nifake/nifake/errors.py b/generated/nifake/nifake/errors.py index e36463b1a7..8dc1909dc2 100644 --- a/generated/nifake/nifake/errors.py +++ b/generated/nifake/nifake/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) diff --git a/generated/nifake/nifake/session.py b/generated/nifake/nifake/session.py index 6f80d48585..00435a2f7b 100644 --- a/generated/nifake/nifake/session.py +++ b/generated/nifake/nifake/session.py @@ -693,16 +693,23 @@ def __init__(self, resource_name, options={}, id_query=False, reset_device=False grpc_options (nifake.grpc_session_options.GrpcSessionOptions): MeasurementLink gRPC session options + driver_warning_event (nifake.DriverWarningEvent): Driver warning event which can be subscribed to, with a callback method. + Sample callback method: + + def sample_callback_method(driver_warning: nifake.DriverWarning): + print(str(driver_warning)) + Returns: session (nifake.Session): A session object representing the device. ''' + driver_warning_event = errors.DriverWarningEvent() if grpc_options: import nifake._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__( diff --git a/generated/nifake/nifake/unit_tests/test_grpc.py b/generated/nifake/nifake/unit_tests/test_grpc.py index 9fae9c8160..385fe36150 100644 --- a/generated/nifake/nifake/unit_tests/test_grpc.py +++ b/generated/nifake/nifake/unit_tests/test_grpc.py @@ -77,13 +77,15 @@ def setup_method(self, method): self.get_ctypes_pointer_for_buffer_side_effect_count = 0 self.get_ctypes_pointer_for_buffer_side_effect_items = [] + self.driver_warning_event = nifake.errors.DriverWarningEvent() + def teardown_method(self, method): self.grpc_stub_patch.stop() self.grpc_types_patch.stop() def _get_initialized_stub_interpreter(self, grpc_channel=object()): session_options = nifake.GrpcSessionOptions(grpc_channel, '', initialization_behavior=nifake.SessionInitializationBehavior.AUTO) - interpreter = nifake._grpc_stub_interpreter.GrpcStubInterpreter(session_options) + interpreter = nifake._grpc_stub_interpreter.GrpcStubInterpreter(session_options, self.driver_warning_event) assert interpreter._client is self.patched_grpc_stub assert interpreter.get_session_handle().id == 0 assert interpreter.get_session_handle().name == "" @@ -174,7 +176,7 @@ def test_server_unavailable(self): expected_error_message = 'Failed to connect to server' self._set_side_effect(library_func, side_effect=MyRpcError(None, '', grpc_error=grpc_error)) grpc_options = nifake.GrpcSessionOptions(object(), '', initialization_behavior=nifake.SessionInitializationBehavior.AUTO) - interpreter = nifake._grpc_stub_interpreter.GrpcStubInterpreter(grpc_options) + interpreter = nifake._grpc_stub_interpreter.GrpcStubInterpreter(grpc_options, self.driver_warning_event) try: interpreter.init_with_options('dev1', False, False, '') assert False @@ -207,7 +209,7 @@ def test_api_key_sent_to_init(self): response_object = self._set_side_effect(library_func, new_session_initialized=True, vi=grpc_session_object) init_behavior = nifake.SessionInitializationBehavior.AUTO grpc_options = nifake.GrpcSessionOptions(object(), '', initialization_behavior=init_behavior) - interpreter = nifake._grpc_stub_interpreter.GrpcStubInterpreter(grpc_options) + interpreter = nifake._grpc_stub_interpreter.GrpcStubInterpreter(grpc_options, self.driver_warning_event) interpreter.init_with_options('dev1', False, False, '') self._assert_call(library_func, response_object, metadata=expected_metadata).assert_called_once_with( resource_name='dev1', id_query=False, reset_device=False, option_string='', session_name='', initialization_behavior=init_behavior, @@ -221,7 +223,7 @@ def test_new_session_already_exists(self): self._set_side_effect(library_func, side_effect=MyRpcError(None, error_message, grpc_error=grpc_error)) init_behavior = nifake.SessionInitializationBehavior.INITIALIZE_SERVER_SESSION grpc_options = nifake.GrpcSessionOptions(object(), session_name, initialization_behavior=init_behavior) - interpreter = nifake._grpc_stub_interpreter.GrpcStubInterpreter(grpc_options) + interpreter = nifake._grpc_stub_interpreter.GrpcStubInterpreter(grpc_options, self.driver_warning_event) try: interpreter.init_with_options('dev1', False, False, '') assert False @@ -238,7 +240,7 @@ def test_attach_to_non_existent_session(self): self._set_side_effect(library_func, side_effect=MyRpcError(None, error_message, grpc_error=grpc_error)) init_behavior = nifake.SessionInitializationBehavior.ATTACH_TO_SERVER_SESSION grpc_options = nifake.GrpcSessionOptions(object(), session_name, initialization_behavior=init_behavior) - interpreter = nifake._grpc_stub_interpreter.GrpcStubInterpreter(grpc_options) + interpreter = nifake._grpc_stub_interpreter.GrpcStubInterpreter(grpc_options, self.driver_warning_event) try: interpreter.init_with_options('dev1', False, False, '') assert False diff --git a/generated/nifake/nifake/unit_tests/test_library_interpreter.py b/generated/nifake/nifake/unit_tests/test_library_interpreter.py index 4298bcb56d..fc1cf99cae 100644 --- a/generated/nifake/nifake/unit_tests/test_library_interpreter.py +++ b/generated/nifake/nifake/unit_tests/test_library_interpreter.py @@ -39,11 +39,14 @@ def setup_method(self, method): self.get_ctypes_pointer_for_buffer_side_effect_count = 0 self.get_ctypes_pointer_for_buffer_side_effect_items = [] + self.driver_warning_event = nifake.errors.DriverWarningEvent() + self.warning_at_callback = None + def teardown_method(self, method): self.patched_library_singleton_get.stop() def get_initialized_library_interpreter(self): - interpreter = nifake._library_interpreter.LibraryInterpreter('windows-1251') + interpreter = nifake._library_interpreter.LibraryInterpreter('windows-1251', self.driver_warning_event) interpreter._vi = SESSION_NUM_FOR_TEST return interpreter @@ -56,6 +59,9 @@ def get_ctypes_pointer_for_buffer_side_effect(self, value, library_type=None): self.get_ctypes_pointer_for_buffer_side_effect_count += 1 return ret_val + def warning_event_callback(self, driver_warning: nifake.DriverWarning): + self.warning_at_callback = driver_warning + # Methods def test_simple_function(self): @@ -389,13 +395,36 @@ def test_method_with_warning(self): assert test_error_desc in str(w[0].message) assert f'Warning {test_error_code} occurred.' in str(w[0].message) + def test_method_with_warningevent(self): + # We want to ignore warnings.warn and validate only warning event + warnings.filterwarnings("ignore", category=nifake.DriverWarning) + + # Register the callback to the warning event + self.driver_warning_event.subscribe(self.warning_event_callback) + + test_error_code = 43 + test_error_desc = "The answer to the ultimate question, only positive" + self.patched_library.niFake_PoorlyNamedSimpleFunction.side_effect = self.side_effects_helper.niFake_PoorlyNamedSimpleFunction + self.side_effects_helper['PoorlyNamedSimpleFunction']['return'] = test_error_code + self.patched_library.niFake_GetError.side_effect = self.side_effects_helper.niFake_GetError + self.side_effects_helper['GetError']['errorCode'] = test_error_code + self.side_effects_helper['GetError']['description'] = test_error_desc + interpreter = self.get_initialized_library_interpreter() + interpreter.simple_function() + assert test_error_desc in str(self.warning_at_callback) + assert f'Warning {test_error_code} occurred.' in str(self.warning_at_callback) + + # Unregister the callback to the warning event and clear warning + self.driver_warning_event.unsubscribe(self.warning_event_callback) + self.warning_at_callback = None + def test_read_with_warning(self): # We want to capture all of our warnings, not just the first one warnings.filterwarnings("always", category=nifake.DriverWarning) test_maximum_time_s = 10.0 test_reading = float('nan') - test_error_code = 42 + test_error_code = 44 test_error_desc = "The answer to the ultimate question, only positive" self.patched_library.niFake_Read.side_effect = self.niFake_read_warning self.error_code_return = test_error_code diff --git a/generated/nifake/nifake/unit_tests/test_session.py b/generated/nifake/nifake/unit_tests/test_session.py index 7f243dcf8a..69c82f8c45 100644 --- a/generated/nifake/nifake/unit_tests/test_session.py +++ b/generated/nifake/nifake/unit_tests/test_session.py @@ -17,13 +17,14 @@ class TestSession: class PatchedLibraryInterpreter(nifake._library_interpreter.LibraryInterpreter): - def __init__(self, encoding): + def __init__(self, encoding, driver_warning_event: nifake.errors.DriverWarningEvent()): for f in dir(self): - if not f.startswith("_") and f not in {'get_session_handle', 'set_session_handle'}: + if not f.startswith("_") and f not in {'get_session_handle', 'set_session_handle', 'generate_driver_warning_event'}: setattr(self, f, MagicMock(spec_set=getattr(self, f), side_effect=_mock_helper.MockFunctionCallError(f))) def setup_method(self, method): - self.patched_library_interpreter = self.PatchedLibraryInterpreter(None) + self.driver_warning_event = nifake.errors.DriverWarningEvent() + self.patched_library_interpreter = self.PatchedLibraryInterpreter(None, self.driver_warning_event) self.patched_library_interpreter_ctor = patch('nifake.session._library_interpreter.LibraryInterpreter', return_value=self.patched_library_interpreter) self.patched_library_interpreter_ctor.start() @@ -841,13 +842,14 @@ def test_return_timedeltas(self): class TestGrpcSession: class PatchedGrpcInterpreter(nifake._grpc_stub_interpreter.GrpcStubInterpreter): - def __init__(self, grpc_options): + def __init__(self, grpc_options, driver_warning_event: nifake.errors.DriverWarningEvent): for f in dir(self): if not f.startswith("_") and f not in {'get_session_handle', 'set_session_handle'}: setattr(self, f, MagicMock(spec_set=getattr(self, f), side_effect=_mock_helper.MockFunctionCallError(f))) def setup_method(self, method): - self.patched_grpc_interpreter = self.PatchedGrpcInterpreter(None) + self.driver_warning_event = nifake.errors.DriverWarningEvent() + self.patched_grpc_interpreter = self.PatchedGrpcInterpreter(None, self.driver_warning_event) self.patched_grpc_constructor = patch('nifake._grpc_stub_interpreter.GrpcStubInterpreter', return_value=self.patched_grpc_interpreter) self.patched_grpc_constructor.start() diff --git a/generated/nifgen/nifgen/_grpc_stub_interpreter.py b/generated/nifgen/nifgen/_grpc_stub_interpreter.py index 4ce5fefa39..51a6868dde 100644 --- a/generated/nifgen/nifgen/_grpc_stub_interpreter.py +++ b/generated/nifgen/nifgen/_grpc_stub_interpreter.py @@ -16,10 +16,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 = nifgen_grpc.NiFgenStub(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()): @@ -67,7 +68,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 diff --git a/generated/nifgen/nifgen/_library_interpreter.py b/generated/nifgen/nifgen/_library_interpreter.py index 6370353a26..6a13633cf8 100644 --- a/generated/nifgen/nifgen/_library_interpreter.py +++ b/generated/nifgen/nifgen/_library_interpreter.py @@ -58,9 +58,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: @@ -87,6 +88,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 diff --git a/generated/nifgen/nifgen/errors.py b/generated/nifgen/nifgen/errors.py index f799868c53..ce71552696 100644 --- a/generated/nifgen/nifgen/errors.py +++ b/generated/nifgen/nifgen/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) diff --git a/generated/nifgen/nifgen/session.py b/generated/nifgen/nifgen/session.py index 4c381e760b..3a13409c8d 100644 --- a/generated/nifgen/nifgen/session.py +++ b/generated/nifgen/nifgen/session.py @@ -3128,16 +3128,23 @@ def __init__(self, resource_name, channel_name=None, reset_device=False, options grpc_options (nifgen.grpc_session_options.GrpcSessionOptions): MeasurementLink gRPC session options + driver_warning_event (nifgen.DriverWarningEvent): Driver warning event which can be subscribed to, with a callback method. + Sample callback method: + + def sample_callback_method(driver_warning: nifgen.DriverWarning): + print(str(driver_warning)) + Returns: session (nifgen.Session): A session object representing the device. ''' + driver_warning_event = errors.DriverWarningEvent() if grpc_options: import nifgen._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__( diff --git a/generated/nimodinst/nimodinst/_library_interpreter.py b/generated/nimodinst/nimodinst/_library_interpreter.py index b51ce157e1..545081c8b5 100644 --- a/generated/nimodinst/nimodinst/_library_interpreter.py +++ b/generated/nimodinst/nimodinst/_library_interpreter.py @@ -52,9 +52,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 # Initialize _handle to 0 for now. # Session will directly update it once the driver runtime init function has been called and # we have a valid session handle. @@ -66,6 +67,14 @@ def set_session_handle(self, value=0): def get_session_handle(self): return self._handle + 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 diff --git a/generated/nimodinst/nimodinst/errors.py b/generated/nimodinst/nimodinst/errors.py index de0698a79b..a98a597c97 100644 --- a/generated/nimodinst/nimodinst/errors.py +++ b/generated/nimodinst/nimodinst/errors.py @@ -71,6 +71,28 @@ def __init__(self): super(DriverTooNewError, self).__init__('The NI-ModInst runtime returned an unexpected value. This can occur if it is too new for the nimodinst Python module. Upgrade the nimodinst Python module.') +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 @@ -93,4 +115,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) diff --git a/generated/nimodinst/nimodinst/session.py b/generated/nimodinst/nimodinst/session.py index f657fc03d5..36488ad49b 100644 --- a/generated/nimodinst/nimodinst/session.py +++ b/generated/nimodinst/nimodinst/session.py @@ -142,10 +142,12 @@ class Session(object): # This is needed during __init__. Without it, __setattr__ raises an exception _is_frozen = False + driver_warning_event = errors.DriverWarningEvent() + def __init__(self, driver): self._item_count = 0 self._current_item = 0 - self._interpreter = _library_interpreter.LibraryInterpreter('windows-1251') + self._interpreter = _library_interpreter.LibraryInterpreter('windows-1251', self.driver_warning_event) # Note that _library_interpreter clears the session handle in its constructor, so that if # _open_installed_devices_session fails, the error handler can reference it. # And then once _open_installed_devices_session succeeds, we can call this again with the diff --git a/generated/nirfsg/nirfsg/_library_interpreter.py b/generated/nirfsg/nirfsg/_library_interpreter.py index 0646b26fd7..b86945d10c 100644 --- a/generated/nirfsg/nirfsg/_library_interpreter.py +++ b/generated/nirfsg/nirfsg/_library_interpreter.py @@ -53,9 +53,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 # Initialize _vi to 0 for now. # Session will directly update it once the driver runtime init function has been called and # we have a valid session handle. @@ -67,6 +68,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 diff --git a/generated/nirfsg/nirfsg/errors.py b/generated/nirfsg/nirfsg/errors.py index 3212823321..23d542e49c 100644 --- a/generated/nirfsg/nirfsg/errors.py +++ b/generated/nirfsg/nirfsg/errors.py @@ -87,6 +87,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 @@ -109,4 +131,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) diff --git a/generated/nirfsg/nirfsg/session.py b/generated/nirfsg/nirfsg/session.py index c81c41bad1..c96f454f92 100644 --- a/generated/nirfsg/nirfsg/session.py +++ b/generated/nirfsg/nirfsg/session.py @@ -6455,12 +6455,19 @@ def __init__(self, resource_name, id_query, reset_device, options={}): | driver_setup | {} | +-------------------------+---------+ + driver_warning_event (nirfsg.DriverWarningEvent): Driver warning event which can be subscribed to, with a callback method. + Sample callback method: + + def sample_callback_method(driver_warning: nirfsg.DriverWarning): + print(str(driver_warning)) + Returns: new_vi (int): Returns a ViSession handle that you use to identify the NI-RFSG device in all subsequent NI-RFSG method calls. ''' - interpreter = _library_interpreter.LibraryInterpreter(encoding='windows-1251') + driver_warning_event = errors.DriverWarningEvent() + 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__( diff --git a/generated/niscope/niscope/_grpc_stub_interpreter.py b/generated/niscope/niscope/_grpc_stub_interpreter.py index 058387eb2f..7d8b3f1946 100644 --- a/generated/niscope/niscope/_grpc_stub_interpreter.py +++ b/generated/niscope/niscope/_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 = niscope_grpc.NiScopeStub(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): # noqa: N802 diff --git a/generated/niscope/niscope/_library_interpreter.py b/generated/niscope/niscope/_library_interpreter.py index 1cb9a7d26a..dd29b6b1c0 100644 --- a/generated/niscope/niscope/_library_interpreter.py +++ b/generated/niscope/niscope/_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 diff --git a/generated/niscope/niscope/errors.py b/generated/niscope/niscope/errors.py index 6390cd1e31..56e73f6aeb 100644 --- a/generated/niscope/niscope/errors.py +++ b/generated/niscope/niscope/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) diff --git a/generated/niscope/niscope/session.py b/generated/niscope/niscope/session.py index b17b43b900..c920d87178 100644 --- a/generated/niscope/niscope/session.py +++ b/generated/niscope/niscope/session.py @@ -4010,16 +4010,23 @@ def __init__(self, resource_name, id_query=False, reset_device=False, options={} grpc_options (niscope.grpc_session_options.GrpcSessionOptions): MeasurementLink gRPC session options + driver_warning_event (niscope.DriverWarningEvent): Driver warning event which can be subscribed to, with a callback method. + Sample callback method: + + def sample_callback_method(driver_warning: niscope.DriverWarning): + print(str(driver_warning)) + Returns: session (niscope.Session): A session object representing the device. ''' + driver_warning_event = errors.DriverWarningEvent() if grpc_options: import niscope._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__( diff --git a/generated/nise/nise/_library_interpreter.py b/generated/nise/nise/_library_interpreter.py index ad3f07dd8d..10270ab70c 100644 --- a/generated/nise/nise/_library_interpreter.py +++ b/generated/nise/nise/_library_interpreter.py @@ -53,9 +53,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 # Initialize _vi to 0 for now. # Session will directly update it once the driver runtime init function has been called and # we have a valid session handle. @@ -67,6 +68,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 diff --git a/generated/nise/nise/errors.py b/generated/nise/nise/errors.py index 4a084bfcf7..577fa7639c 100644 --- a/generated/nise/nise/errors.py +++ b/generated/nise/nise/errors.py @@ -78,6 +78,28 @@ def __init__(self, invalid_character, invalid_string): super(InvalidRepeatedCapabilityError, self).__init__('An invalid character ({}) was found in repeated capability string ({})'.format(invalid_character, invalid_string)) +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 @@ -100,4 +122,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) diff --git a/generated/nise/nise/session.py b/generated/nise/nise/session.py index 095ecf57fa..0b1646e715 100644 --- a/generated/nise/nise/session.py +++ b/generated/nise/nise/session.py @@ -99,12 +99,19 @@ def __init__(self, virtual_device_name, options={}): | driver_setup | {} | +-------------------------+---------+ + driver_warning_event (nise.DriverWarningEvent): Driver warning event which can be subscribed to, with a callback method. + Sample callback method: + + def sample_callback_method(driver_warning: nise.DriverWarning): + print(str(driver_warning)) + Returns: session (nise.Session): A session object representing the device. ''' - interpreter = _library_interpreter.LibraryInterpreter(encoding='windows-1251') + driver_warning_event = errors.DriverWarningEvent() + 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__( diff --git a/generated/niswitch/niswitch/_grpc_stub_interpreter.py b/generated/niswitch/niswitch/_grpc_stub_interpreter.py index 143b81ebd8..9355646c4f 100644 --- a/generated/niswitch/niswitch/_grpc_stub_interpreter.py +++ b/generated/niswitch/niswitch/_grpc_stub_interpreter.py @@ -16,10 +16,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 = niswitch_grpc.NiSwitchStub(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()): @@ -67,7 +68,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 diff --git a/generated/niswitch/niswitch/_library_interpreter.py b/generated/niswitch/niswitch/_library_interpreter.py index 4d7dc8ea7c..f940c69f22 100644 --- a/generated/niswitch/niswitch/_library_interpreter.py +++ b/generated/niswitch/niswitch/_library_interpreter.py @@ -58,9 +58,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: @@ -87,6 +88,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 diff --git a/generated/niswitch/niswitch/errors.py b/generated/niswitch/niswitch/errors.py index d67194e48d..4581193cb7 100644 --- a/generated/niswitch/niswitch/errors.py +++ b/generated/niswitch/niswitch/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) diff --git a/generated/niswitch/niswitch/session.py b/generated/niswitch/niswitch/session.py index ad4063da63..6d0e68fbf4 100644 --- a/generated/niswitch/niswitch/session.py +++ b/generated/niswitch/niswitch/session.py @@ -1350,16 +1350,23 @@ def __init__(self, resource_name, topology="Configured Topology", simulate=False grpc_options (niswitch.grpc_session_options.GrpcSessionOptions): MeasurementLink gRPC session options + driver_warning_event (niswitch.DriverWarningEvent): Driver warning event which can be subscribed to, with a callback method. + Sample callback method: + + def sample_callback_method(driver_warning: niswitch.DriverWarning): + print(str(driver_warning)) + Returns: session (niswitch.Session): A session object representing the device. ''' + driver_warning_event = errors.DriverWarningEvent() if grpc_options: import niswitch._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__( diff --git a/generated/nitclk/nitclk/_library_interpreter.py b/generated/nitclk/nitclk/_library_interpreter.py index 8a23273dd6..0a94a6b179 100644 --- a/generated/nitclk/nitclk/_library_interpreter.py +++ b/generated/nitclk/nitclk/_library_interpreter.py @@ -52,9 +52,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 # Initialize _session_number to 0 for now. # Session will directly update it once the driver runtime init function has been called and # we have a valid session handle. @@ -66,6 +67,14 @@ def set_session_handle(self, value=0): def get_session_handle(self): return self._session_number + 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 diff --git a/generated/nitclk/nitclk/errors.py b/generated/nitclk/nitclk/errors.py index f371d699fd..9e13d508ab 100644 --- a/generated/nitclk/nitclk/errors.py +++ b/generated/nitclk/nitclk/errors.py @@ -71,6 +71,28 @@ def __init__(self): super(DriverTooNewError, self).__init__('The NI-TClk runtime returned an unexpected value. This can occur if it is too new for the nitclk Python module. Upgrade the nitclk Python module.') +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 @@ -93,4 +115,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) diff --git a/generated/nitclk/nitclk/session.py b/generated/nitclk/nitclk/session.py index ed859bdd1e..85471bc984 100644 --- a/generated/nitclk/nitclk/session.py +++ b/generated/nitclk/nitclk/session.py @@ -5,6 +5,7 @@ import nitclk._attributes as _attributes import nitclk._converters as _converters import nitclk._library_interpreter as _library_interpreter +import nitclk.errors as errors # Used for __repr__ and __str__ import pprint @@ -20,6 +21,8 @@ class SessionReference(object): # This is needed during __init__. Without it, __setattr__ raises an exception _is_frozen = False + driver_warning_event = errors.DriverWarningEvent() + exported_sync_pulse_output_terminal = _attributes.AttributeViString(2) '''Type: str @@ -130,7 +133,7 @@ class SessionReference(object): ''' def __init__(self, session_number, encoding='windows-1251'): - self._interpreter = _library_interpreter.LibraryInterpreter(encoding) + self._interpreter = _library_interpreter.LibraryInterpreter(encoding, self.driver_warning_event) self._interpreter.set_session_handle(session_number) # We need a self._repeated_capability string for passing down to function calls on the LibraryInterpreter class. We just need to set it to empty string. self._repeated_capability = '' @@ -352,8 +355,10 @@ class _Session(object): indentation. ''' + driver_warning_event = errors.DriverWarningEvent() + def __init__(self): - self._interpreter = _library_interpreter.LibraryInterpreter('windows-1251') + self._interpreter = _library_interpreter.LibraryInterpreter('windows-1251', self.driver_warning_event) # Instantiate any repeated capability objects diff --git a/src/nifake/unit_tests/test_grpc.py b/src/nifake/unit_tests/test_grpc.py index 9fae9c8160..385fe36150 100644 --- a/src/nifake/unit_tests/test_grpc.py +++ b/src/nifake/unit_tests/test_grpc.py @@ -77,13 +77,15 @@ def setup_method(self, method): self.get_ctypes_pointer_for_buffer_side_effect_count = 0 self.get_ctypes_pointer_for_buffer_side_effect_items = [] + self.driver_warning_event = nifake.errors.DriverWarningEvent() + def teardown_method(self, method): self.grpc_stub_patch.stop() self.grpc_types_patch.stop() def _get_initialized_stub_interpreter(self, grpc_channel=object()): session_options = nifake.GrpcSessionOptions(grpc_channel, '', initialization_behavior=nifake.SessionInitializationBehavior.AUTO) - interpreter = nifake._grpc_stub_interpreter.GrpcStubInterpreter(session_options) + interpreter = nifake._grpc_stub_interpreter.GrpcStubInterpreter(session_options, self.driver_warning_event) assert interpreter._client is self.patched_grpc_stub assert interpreter.get_session_handle().id == 0 assert interpreter.get_session_handle().name == "" @@ -174,7 +176,7 @@ def test_server_unavailable(self): expected_error_message = 'Failed to connect to server' self._set_side_effect(library_func, side_effect=MyRpcError(None, '', grpc_error=grpc_error)) grpc_options = nifake.GrpcSessionOptions(object(), '', initialization_behavior=nifake.SessionInitializationBehavior.AUTO) - interpreter = nifake._grpc_stub_interpreter.GrpcStubInterpreter(grpc_options) + interpreter = nifake._grpc_stub_interpreter.GrpcStubInterpreter(grpc_options, self.driver_warning_event) try: interpreter.init_with_options('dev1', False, False, '') assert False @@ -207,7 +209,7 @@ def test_api_key_sent_to_init(self): response_object = self._set_side_effect(library_func, new_session_initialized=True, vi=grpc_session_object) init_behavior = nifake.SessionInitializationBehavior.AUTO grpc_options = nifake.GrpcSessionOptions(object(), '', initialization_behavior=init_behavior) - interpreter = nifake._grpc_stub_interpreter.GrpcStubInterpreter(grpc_options) + interpreter = nifake._grpc_stub_interpreter.GrpcStubInterpreter(grpc_options, self.driver_warning_event) interpreter.init_with_options('dev1', False, False, '') self._assert_call(library_func, response_object, metadata=expected_metadata).assert_called_once_with( resource_name='dev1', id_query=False, reset_device=False, option_string='', session_name='', initialization_behavior=init_behavior, @@ -221,7 +223,7 @@ def test_new_session_already_exists(self): self._set_side_effect(library_func, side_effect=MyRpcError(None, error_message, grpc_error=grpc_error)) init_behavior = nifake.SessionInitializationBehavior.INITIALIZE_SERVER_SESSION grpc_options = nifake.GrpcSessionOptions(object(), session_name, initialization_behavior=init_behavior) - interpreter = nifake._grpc_stub_interpreter.GrpcStubInterpreter(grpc_options) + interpreter = nifake._grpc_stub_interpreter.GrpcStubInterpreter(grpc_options, self.driver_warning_event) try: interpreter.init_with_options('dev1', False, False, '') assert False @@ -238,7 +240,7 @@ def test_attach_to_non_existent_session(self): self._set_side_effect(library_func, side_effect=MyRpcError(None, error_message, grpc_error=grpc_error)) init_behavior = nifake.SessionInitializationBehavior.ATTACH_TO_SERVER_SESSION grpc_options = nifake.GrpcSessionOptions(object(), session_name, initialization_behavior=init_behavior) - interpreter = nifake._grpc_stub_interpreter.GrpcStubInterpreter(grpc_options) + interpreter = nifake._grpc_stub_interpreter.GrpcStubInterpreter(grpc_options, self.driver_warning_event) try: interpreter.init_with_options('dev1', False, False, '') assert False diff --git a/src/nifake/unit_tests/test_library_interpreter.py b/src/nifake/unit_tests/test_library_interpreter.py index 4298bcb56d..fc1cf99cae 100644 --- a/src/nifake/unit_tests/test_library_interpreter.py +++ b/src/nifake/unit_tests/test_library_interpreter.py @@ -39,11 +39,14 @@ def setup_method(self, method): self.get_ctypes_pointer_for_buffer_side_effect_count = 0 self.get_ctypes_pointer_for_buffer_side_effect_items = [] + self.driver_warning_event = nifake.errors.DriverWarningEvent() + self.warning_at_callback = None + def teardown_method(self, method): self.patched_library_singleton_get.stop() def get_initialized_library_interpreter(self): - interpreter = nifake._library_interpreter.LibraryInterpreter('windows-1251') + interpreter = nifake._library_interpreter.LibraryInterpreter('windows-1251', self.driver_warning_event) interpreter._vi = SESSION_NUM_FOR_TEST return interpreter @@ -56,6 +59,9 @@ def get_ctypes_pointer_for_buffer_side_effect(self, value, library_type=None): self.get_ctypes_pointer_for_buffer_side_effect_count += 1 return ret_val + def warning_event_callback(self, driver_warning: nifake.DriverWarning): + self.warning_at_callback = driver_warning + # Methods def test_simple_function(self): @@ -389,13 +395,36 @@ def test_method_with_warning(self): assert test_error_desc in str(w[0].message) assert f'Warning {test_error_code} occurred.' in str(w[0].message) + def test_method_with_warningevent(self): + # We want to ignore warnings.warn and validate only warning event + warnings.filterwarnings("ignore", category=nifake.DriverWarning) + + # Register the callback to the warning event + self.driver_warning_event.subscribe(self.warning_event_callback) + + test_error_code = 43 + test_error_desc = "The answer to the ultimate question, only positive" + self.patched_library.niFake_PoorlyNamedSimpleFunction.side_effect = self.side_effects_helper.niFake_PoorlyNamedSimpleFunction + self.side_effects_helper['PoorlyNamedSimpleFunction']['return'] = test_error_code + self.patched_library.niFake_GetError.side_effect = self.side_effects_helper.niFake_GetError + self.side_effects_helper['GetError']['errorCode'] = test_error_code + self.side_effects_helper['GetError']['description'] = test_error_desc + interpreter = self.get_initialized_library_interpreter() + interpreter.simple_function() + assert test_error_desc in str(self.warning_at_callback) + assert f'Warning {test_error_code} occurred.' in str(self.warning_at_callback) + + # Unregister the callback to the warning event and clear warning + self.driver_warning_event.unsubscribe(self.warning_event_callback) + self.warning_at_callback = None + def test_read_with_warning(self): # We want to capture all of our warnings, not just the first one warnings.filterwarnings("always", category=nifake.DriverWarning) test_maximum_time_s = 10.0 test_reading = float('nan') - test_error_code = 42 + test_error_code = 44 test_error_desc = "The answer to the ultimate question, only positive" self.patched_library.niFake_Read.side_effect = self.niFake_read_warning self.error_code_return = test_error_code diff --git a/src/nifake/unit_tests/test_session.py b/src/nifake/unit_tests/test_session.py index 7f243dcf8a..69c82f8c45 100644 --- a/src/nifake/unit_tests/test_session.py +++ b/src/nifake/unit_tests/test_session.py @@ -17,13 +17,14 @@ class TestSession: class PatchedLibraryInterpreter(nifake._library_interpreter.LibraryInterpreter): - def __init__(self, encoding): + def __init__(self, encoding, driver_warning_event: nifake.errors.DriverWarningEvent()): for f in dir(self): - if not f.startswith("_") and f not in {'get_session_handle', 'set_session_handle'}: + if not f.startswith("_") and f not in {'get_session_handle', 'set_session_handle', 'generate_driver_warning_event'}: setattr(self, f, MagicMock(spec_set=getattr(self, f), side_effect=_mock_helper.MockFunctionCallError(f))) def setup_method(self, method): - self.patched_library_interpreter = self.PatchedLibraryInterpreter(None) + self.driver_warning_event = nifake.errors.DriverWarningEvent() + self.patched_library_interpreter = self.PatchedLibraryInterpreter(None, self.driver_warning_event) self.patched_library_interpreter_ctor = patch('nifake.session._library_interpreter.LibraryInterpreter', return_value=self.patched_library_interpreter) self.patched_library_interpreter_ctor.start() @@ -841,13 +842,14 @@ def test_return_timedeltas(self): class TestGrpcSession: class PatchedGrpcInterpreter(nifake._grpc_stub_interpreter.GrpcStubInterpreter): - def __init__(self, grpc_options): + def __init__(self, grpc_options, driver_warning_event: nifake.errors.DriverWarningEvent): for f in dir(self): if not f.startswith("_") and f not in {'get_session_handle', 'set_session_handle'}: setattr(self, f, MagicMock(spec_set=getattr(self, f), side_effect=_mock_helper.MockFunctionCallError(f))) def setup_method(self, method): - self.patched_grpc_interpreter = self.PatchedGrpcInterpreter(None) + self.driver_warning_event = nifake.errors.DriverWarningEvent() + self.patched_grpc_interpreter = self.PatchedGrpcInterpreter(None, self.driver_warning_event) self.patched_grpc_constructor = patch('nifake._grpc_stub_interpreter.GrpcStubInterpreter', return_value=self.patched_grpc_interpreter) self.patched_grpc_constructor.start() diff --git a/src/nimodinst/templates/session.py.mako b/src/nimodinst/templates/session.py.mako index a0c1d5f942..b8688663d3 100644 --- a/src/nimodinst/templates/session.py.mako +++ b/src/nimodinst/templates/session.py.mako @@ -122,10 +122,12 @@ class Session(object): # This is needed during __init__. Without it, __setattr__ raises an exception _is_frozen = False + driver_warning_event = errors.DriverWarningEvent() + def __init__(self, driver): self._item_count = 0 self._current_item = 0 - self._interpreter = _library_interpreter.LibraryInterpreter('windows-1251') + self._interpreter = _library_interpreter.LibraryInterpreter('windows-1251', self.driver_warning_event) # Note that _library_interpreter clears the session handle in its constructor, so that if # _open_installed_devices_session fails, the error handler can reference it. # And then once _open_installed_devices_session succeeds, we can call this again with the diff --git a/src/nitclk/templates/session.py.mako b/src/nitclk/templates/session.py.mako index f9e4f77534..be01478312 100644 --- a/src/nitclk/templates/session.py.mako +++ b/src/nitclk/templates/session.py.mako @@ -17,6 +17,7 @@ import hightime import ${module_name}._attributes as _attributes import ${module_name}._converters as _converters import ${module_name}._library_interpreter as _library_interpreter +import ${module_name}.errors as errors # Used for __repr__ and __str__ import pprint @@ -32,6 +33,8 @@ class SessionReference(object): # This is needed during __init__. Without it, __setattr__ raises an exception _is_frozen = False + driver_warning_event = errors.DriverWarningEvent() + % for attribute in helper.sorted_attrs(attributes): <% helper.add_attribute_rep_cap_tip(attributes[attribute], config) @@ -50,7 +53,7 @@ helper.add_attribute_rep_cap_tip(attributes[attribute], config) % endfor def __init__(self, ${config['session_handle_parameter_name']}, encoding='windows-1251'): - self._interpreter = _library_interpreter.LibraryInterpreter(encoding) + self._interpreter = _library_interpreter.LibraryInterpreter(encoding, self.driver_warning_event) self._interpreter.set_session_handle(${config['session_handle_parameter_name']}) # We need a self._repeated_capability string for passing down to function calls on the LibraryInterpreter class. We just need to set it to empty string. self._repeated_capability = '' @@ -92,8 +95,10 @@ class _Session(object): indentation. ''' + driver_warning_event = errors.DriverWarningEvent() + def __init__(self): - self._interpreter = _library_interpreter.LibraryInterpreter('windows-1251') + self._interpreter = _library_interpreter.LibraryInterpreter('windows-1251', self.driver_warning_event) # Instantiate any repeated capability objects % for rep_cap in config['repeated_capabilities']: