examples: Add type hints, use cancellation event, name MeasurementService consistently (#372)

* examples: Standardize measurement_service variable name

* examples: Add missing type hints

* examples: Use event to signal cancellation

Author: bkeryan

examples/nidaqmx_analog_input/measurement.py

@@ -3,7 +3,7 @@
 import logging
 import pathlib
 import sys
-from typing import Optional
+from typing import List, Optional, Tuple
 
 import click
 import nidaqmx
@@ -49,7 +49,7 @@
 @measurement_service.configuration("sample_rate", nims.DataType.Double, 1000.0)
 @measurement_service.configuration("number_of_samples", nims.DataType.UInt64, 100)
 @measurement_service.output("acquired_samples", nims.DataType.DoubleArray1D)
-def measure(pin_name, sample_rate, number_of_samples):
+def measure(pin_name: str, sample_rate: float, number_of_samples: int) -> Tuple[List[float]]:
     """Perform a finite analog input measurement with NI-DAQmx."""
     logging.info(
         "Executing measurement: pin_name=%s sample_rate=%g number_of_samples=%d",
@@ -66,7 +66,7 @@ def measure(pin_name, sample_rate, number_of_samples):
     ) as reservation:
         task: Optional[nidaqmx.Task] = None
 
-        def cancel_callback():
+        def cancel_callback() -> None:
             logging.info("Canceling measurement")
             task_to_abort = task
             if task_to_abort is not None:

examples/nidcpower_source_dc_voltage/measurement.py

@@ -3,8 +3,9 @@
 import logging
 import pathlib
 import sys
+import threading
 import time
-from typing import Iterable
+from typing import Iterable, List, Tuple
 
 import click
 import grpc
@@ -72,7 +73,7 @@ def measure(
     current_limit: float,
     current_limit_range: float,
     source_delay: float,
-):
+) -> Tuple[List[int], List[str], List[float], List[float], List[bool]]:
     """Source and measure a DC voltage with an NI SMU."""
     logging.info("Executing measurement: pin_names=%s voltage_level=%g", pin_names, voltage_level)
 
@@ -91,17 +92,8 @@ def measure(
             channels = session.channels[reservation.session_info.channel_list]
             channel_mappings = reservation.session_info.channel_mappings
 
-            pending_cancellation = False
-
-            def cancel_callback():
-                logging.info("Canceling measurement")
-                session_to_abort = session
-                if session_to_abort is not None:
-                    nonlocal pending_cancellation
-                    pending_cancellation = True
-                    session_to_abort.abort()
-
-            measurement_service.context.add_cancel_callback(cancel_callback)
+            cancellation_event = threading.Event()
+            measurement_service.context.add_cancel_callback(cancellation_event.set)
             time_remaining = measurement_service.context.time_remaining
 
             channels.source_mode = nidcpower.SourceMode.SINGLE_POINT
@@ -121,7 +113,7 @@ def cancel_callback():
                         measurement_service.context.abort(
                             grpc.StatusCode.DEADLINE_EXCEEDED, "deadline exceeded"
                         )
-                    if pending_cancellation:
+                    if cancellation_event.is_set():
                         measurement_service.context.abort(
                             grpc.StatusCode.CANCELLED, "client requested cancellation"
                         )
@@ -165,7 +157,7 @@ def cancel_callback():
 def _log_measured_values(
     channel_mappings: Iterable[nims.session_management.ChannelMapping],
     measured_values: Iterable,
-):
+) -> None:
     """Log the measured values."""
     for mapping, measurement in zip(channel_mappings, measured_values):
         logging.info("site%s/%s:", mapping.site, mapping.pin_or_relay_name)

examples/nidmm_measurement/measurement.py

@@ -82,7 +82,7 @@ def measure(
     measurement_type: Function,
     range: float,
     resolution_digits: float,
-) -> Tuple:
+) -> Tuple[float, bool, float]:
     """Perform a measurement using an NI DMM."""
     logging.info(
         "Starting measurement: pin_name=%s measurement_type=%s range=%g resolution_digits=%g",

examples/nifgen_standard_function/measurement.py

@@ -4,6 +4,7 @@
 import logging
 import pathlib
 import sys
+import threading
 import time
 from enum import Enum
 from typing import Tuple
@@ -81,7 +82,7 @@ def measure(
     frequency: float,
     amplitude: float,
     duration: float,
-) -> Tuple:
+) -> Tuple[()]:
     """Generate a standard function waveform using an NI waveform generator."""
     logging.info(
         "Starting generation: pin_name=%s waveform_type=%s frequency=%g amplitude=%g",
@@ -91,14 +92,8 @@ def measure(
         amplitude,
     )
 
-    pending_cancellation = False
-
-    def cancel_callback():
-        logging.info("Canceling generation")
-        nonlocal pending_cancellation
-        pending_cancellation = True
-
-    measurement_service.context.add_cancel_callback(cancel_callback)
+    cancellation_event = threading.Event()
+    measurement_service.context.add_cancel_callback(cancellation_event.set)
 
     session_management_client = create_session_management_client(measurement_service)
 
@@ -139,7 +134,7 @@ def cancel_callback():
                 measurement_service.context.abort(
                     grpc.StatusCode.DEADLINE_EXCEEDED, "Deadline exceeded."
                 )
-            if pending_cancellation:
+            if cancellation_event.is_set():
                 measurement_service.context.abort(
                     grpc.StatusCode.CANCELLED, "Client requested cancellation."
                 )

examples/niscope_acquire_waveform/measurement.py

@@ -3,8 +3,9 @@
 import logging
 import pathlib
 import sys
+import threading
 import time
-from typing import Tuple
+from typing import List, Tuple
 
 import click
 import grpc
@@ -96,7 +97,7 @@ def measure(
     auto_trigger: bool,
     trigger_coupling: str,
     timeout: float,
-) -> Tuple:
+) -> Tuple[List[float], ...]:
     """Acquire a waveform using an NI oscilloscope."""
     logging.info(
         "Starting acquisition: pin_or_relay_names=%s vertical_range=%g trigger_source=%s trigger_level=%g",
@@ -106,14 +107,8 @@ def measure(
         trigger_level,
     )
 
-    pending_cancellation = False
-
-    def cancel_callback():
-        logging.info("Canceling acquisition")
-        nonlocal pending_cancellation
-        pending_cancellation = True
-
-    measurement_service.context.add_cancel_callback(cancel_callback)
+    cancellation_event = threading.Event()
+    measurement_service.context.add_cancel_callback(cancellation_event.set)
 
     session_management_client = create_session_management_client(measurement_service)
 
@@ -167,7 +162,7 @@ def cancel_callback():
                         measurement_service.context.abort(
                             grpc.StatusCode.DEADLINE_EXCEEDED, "Deadline exceeded."
                         )
-                    if pending_cancellation:
+                    if cancellation_event.is_set():
                         measurement_service.context.abort(
                             grpc.StatusCode.CANCELLED, "Client requested cancellation."
                         )

examples/niswitch_control_relays/measurement.py

@@ -46,7 +46,7 @@
 def measure(
     relay_names: str,
     close_relays: bool,
-) -> Tuple:
+) -> Tuple[()]:
     """Control relays using an NI relay driver (e.g. PXI-2567)."""
     logging.info(
         "Controlling relays: relay_names=%s close_relay=%s",

examples/nivisa_dmm_measurement/measurement.py

@@ -75,7 +75,7 @@ def measure(
     measurement_type: Function,
     range: float,
     resolution_digits: float,
-) -> Tuple:
+) -> Tuple[float]:
     """Perform a DMM measurement using NI-VISA and an NI Instrument Simulator v2.0."""
     logging.info(
         "Starting measurement: pin_name=%s measurement_type=%s range=%g resolution_digits=%g",

examples/output_voltage_measurement/measurement.py

@@ -5,6 +5,7 @@
 import threading
 import time
 from enum import Enum
+from typing import Tuple
 
 import _nidcpower_helpers
 import _visa_helpers
@@ -100,7 +101,7 @@ def measure(
     range: float,
     resolution_digits: float,
     output_pin: str,
-):
+) -> Tuple[float]:
     """Source DC voltage as input with an NI SMU and measure output using NI-VISA DMM."""
     logging.info(
         "Executing measurement: pin_names=%s voltage_level=%g measurement_type=%s range=%g resolution_digits=%g",

examples/sample_measurement/measurement.py

@@ -3,6 +3,7 @@
 import pathlib
 import sys
 from enum import Enum
+from typing import Iterable, Tuple
 
 import click
 from _helpers import configure_logging, verbosity_option
@@ -16,7 +17,7 @@
 
 script_or_exe = sys.executable if getattr(sys, "frozen", False) else __file__
 service_directory = pathlib.Path(script_or_exe).resolve().parent
-sample_measurement_service = nims.MeasurementService(
+measurement_service = nims.MeasurementService(
     service_config_path=service_directory / "SampleMeasurement.serviceconfig",
     version="0.1.0.0",
     ui_file_paths=[
@@ -36,50 +37,48 @@ class Color(Enum):
     BLUE = 3
 
 
-@sample_measurement_service.register_measurement
-@sample_measurement_service.configuration("Float In", nims.DataType.Float, 0.06)
-@sample_measurement_service.configuration(
-    "Double Array In", nims.DataType.DoubleArray1D, [0.1, 0.2, 0.3]
-)
-@sample_measurement_service.configuration("Bool In", nims.DataType.Boolean, False)
-@sample_measurement_service.configuration("String In", nims.DataType.String, "sample string")
-@sample_measurement_service.configuration(
-    "Enum In", nims.DataType.Enum, Color.BLUE, enum_type=Color
-)
-@sample_measurement_service.configuration(
+@measurement_service.register_measurement
+@measurement_service.configuration("Float In", nims.DataType.Float, 0.06)
+@measurement_service.configuration("Double Array In", nims.DataType.DoubleArray1D, [0.1, 0.2, 0.3])
+@measurement_service.configuration("Bool In", nims.DataType.Boolean, False)
+@measurement_service.configuration("String In", nims.DataType.String, "sample string")
+@measurement_service.configuration("Enum In", nims.DataType.Enum, Color.BLUE, enum_type=Color)
+@measurement_service.configuration(
     "Protobuf Enum In",
     nims.DataType.Enum,
     color_pb2.ProtobufColor.BLACK,
     enum_type=color_pb2.ProtobufColor,
 )
-@sample_measurement_service.configuration(
+@measurement_service.configuration(
     "String Array In", nims.DataType.StringArray1D, ["String1", "String2"]
 )
-@sample_measurement_service.output("Float out", nims.DataType.Float)
-@sample_measurement_service.output("Double Array out", nims.DataType.DoubleArray1D)
-@sample_measurement_service.output("Bool out", nims.DataType.Boolean)
-@sample_measurement_service.output("String out", nims.DataType.String)
-@sample_measurement_service.output("Enum out", nims.DataType.Enum, enum_type=Color)
-@sample_measurement_service.output(
+@measurement_service.output("Float out", nims.DataType.Float)
+@measurement_service.output("Double Array out", nims.DataType.DoubleArray1D)
+@measurement_service.output("Bool out", nims.DataType.Boolean)
+@measurement_service.output("String out", nims.DataType.String)
+@measurement_service.output("Enum out", nims.DataType.Enum, enum_type=Color)
+@measurement_service.output(
     "Protobuf Enum out", nims.DataType.Enum, enum_type=color_pb2.ProtobufColor
 )
-@sample_measurement_service.output("String Array out", nims.DataType.StringArray1D)
+@measurement_service.output("String Array out", nims.DataType.StringArray1D)
 def measure(
-    float_input,
-    double_array_input,
-    bool_input,
-    string_input,
-    enum_input,
-    protobuf_enum_input,
-    string_array_in,
-):
+    float_input: float,
+    double_array_input: Iterable[float],
+    bool_input: bool,
+    string_input: str,
+    enum_input: Color,
+    protobuf_enum_input: color_pb2.ProtobufColor.ValueType,
+    string_array_in: Iterable[str],
+) -> Tuple[
+    float, Iterable[float], bool, str, Color, color_pb2.ProtobufColor.ValueType, Iterable[str]
+]:
     """Perform a loopback measurement with various data types."""
     logging.info("Executing measurement")
 
-    def cancel_callback():
+    def cancel_callback() -> None:
         logging.info("Canceling measurement")
 
-    sample_measurement_service.context.add_cancel_callback(cancel_callback)
+    measurement_service.context.add_cancel_callback(cancel_callback)
 
     float_output = float_input
     float_array_output = double_array_input
@@ -107,7 +106,7 @@ def main(verbosity: int) -> None:
     """Perform a loopback measurement with various data types."""
     configure_logging(verbosity)
 
-    with sample_measurement_service.host_service():
+    with measurement_service.host_service():
         input("Press enter to close the measurement service.\n")
 
 

tests/utilities/loopback_measurement/__init__.py

@@ -1,6 +1,7 @@
 """Contains utility functions to test loopback measurement service. """
 import pathlib
 from enum import Enum
+from typing import Iterable, Tuple
 
 from examples.sample_measurement._stubs import color_pb2
 
@@ -51,14 +52,16 @@ class Color(Enum):
 )
 @measurement_service.output("String Array out", nims.DataType.StringArray1D)
 def measure(
-    float_input,
-    double_array_input,
-    bool_input,
-    string_input,
-    enum_input,
-    protobuf_enum_input,
-    string_array_in,
-):
+    float_input: float,
+    double_array_input: Iterable[float],
+    bool_input: bool,
+    string_input: str,
+    enum_input: Color,
+    protobuf_enum_input: color_pb2.ProtobufColor.ValueType,
+    string_array_in: Iterable[str],
+) -> Tuple[
+    float, Iterable[float], bool, str, Color, color_pb2.ProtobufColor.ValueType, Iterable[str]
+]:
     """Perform a loopback measurement with various data types."""
     float_output = float_input
     float_array_output = double_array_input