mypy: Disallow untyped defs (#379)

* service: Fix grpc_servicer type hints

ni_measurementlink_service\_internal\grpc_servicer.py:168: error: Function is missing a return type annotation  [no-untyped-def]
ni_measurementlink_service\_internal\grpc_servicer.py:311: error: Function is missing a return type annotation  [no-untyped-def]

* service: Fix a serializer type hint

ni_measurementlink_service\_internal\parameter\serializer.py:113: error: Function is missing a return type annotation  [no-untyped-def]

* service: Add a missing cast

* service: More grpc_servicer type hints

* service: More type hints

* pyproject.toml: Enable mypy --disable-untyped-defs

* generator: Add missing type hints

ni_measurementlink_generator\template.py:14: error: Function is missing a type annotation for one or more arguments  [no-untyped-def]
ni_measurementlink_generator\template.py:27: error: Function is missing a return type annotation  [no-untyped-def]
ni_measurementlink_generator\template.py:27: error: Function is missing a type annotation for one or more arguments  [no-untyped-def]

* generator: Disallow untyped defs

* examples: Add missing _helpers.py type hints

_helpers.py:42: error: Function is missing a type annotation for one or more arguments  [no-untyped-def]
_helpers.py:184: error: Function is missing a return type annotation  [no-untyped-def]

* examples: Enable mypy --disallow-untyped-defs and fix errors

* tests: Allow untyped defs :(

* generator: Fix lint errors

* generator: Fix test errors caused by rerunning black

* examples: Only type alias outputs when needed for yield + return

Author: bkeryan

examples/nidaqmx_analog_input/_helpers.py

@@ -39,7 +39,7 @@ class ServiceOptions(NamedTuple):
     use_simulation: bool = False
 
 
-def get_service_options(**kwargs) -> ServiceOptions:
+def get_service_options(**kwargs: Any) -> ServiceOptions:
     """Get service options from keyword arguments."""
     return ServiceOptions(
         use_grpc_device=kwargs.get("use_grpc_device", False),
@@ -81,7 +81,7 @@ def update_pin_map(self, pin_map_path: str) -> str:
 class GrpcChannelPoolHelper(GrpcChannelPool):
     """Class that manages gRPC channel lifetimes."""
 
-    def __init__(self):
+    def __init__(self) -> None:
         """Initialize the GrpcChannelPool object."""
         super().__init__()
         self._discovery_client = DiscoveryClient()
@@ -181,7 +181,7 @@ def resolve_file_path(self, file_path: str) -> str:
         return absolute_path
 
 
-def configure_logging(verbosity: int):
+def configure_logging(verbosity: int) -> None:
     """Configure logging for this process."""
     if verbosity > 1:
         level = logging.DEBUG

examples/nidaqmx_analog_input/_nidaqmx_helpers.py

@@ -11,7 +11,7 @@
 def create_task(
     session_info: nims.session_management.SessionInformation,
     session_grpc_channel: Optional[grpc.Channel] = None,
-    initialization_behavior=nidaqmx.SessionInitializationBehavior.AUTO,
+    initialization_behavior: nidaqmx.SessionInitializationBehavior = nidaqmx.SessionInitializationBehavior.AUTO,
 ) -> nidaqmx.Task:
     """Create daqmx task based on reserved session and grpc channel."""
     session_kwargs = {}

examples/nidaqmx_analog_input/measurement.py

@@ -3,7 +3,7 @@
 import logging
 import pathlib
 import sys
-from typing import List, Optional, Tuple
+from typing import Any, List, Optional, Tuple
 
 import click
 import nidaqmx
@@ -95,7 +95,7 @@ def cancel_callback() -> None:
     return (voltage_values,)
 
 
-def _log_measured_values(samples, max_samples_to_display=5):
+def _log_measured_values(samples: List[float], max_samples_to_display: int = 5) -> None:
     """Log the measured values."""
     if len(samples) > max_samples_to_display:
         for index, value in enumerate(samples[0 : max_samples_to_display - 1]):
@@ -110,7 +110,7 @@ def _log_measured_values(samples, max_samples_to_display=5):
 @click.command
 @verbosity_option
 @grpc_device_options
-def main(verbosity: int, **kwargs):
+def main(verbosity: int, **kwargs: Any) -> None:
     """Perform a finite analog input measurement with NI-DAQmx."""
     configure_logging(verbosity)
     global service_options

examples/nidaqmx_analog_input/pyproject.toml

@@ -21,6 +21,9 @@ grpc-stubs = "^1.53"
 requires = ["poetry-core>=1.2.0"]
 build-backend = "poetry.core.masonry.api"
 
+[tool.mypy]
+disallow_untyped_defs = true
+
 [[tool.mypy.overrides]]
 module = [
     "nidaqmx.*",

examples/nidcpower_source_dc_voltage/_helpers.py

@@ -39,7 +39,7 @@ class ServiceOptions(NamedTuple):
     use_simulation: bool = False
 
 
-def get_service_options(**kwargs) -> ServiceOptions:
+def get_service_options(**kwargs: Any) -> ServiceOptions:
     """Get service options from keyword arguments."""
     return ServiceOptions(
         use_grpc_device=kwargs.get("use_grpc_device", False),
@@ -81,7 +81,7 @@ def update_pin_map(self, pin_map_path: str) -> str:
 class GrpcChannelPoolHelper(GrpcChannelPool):
     """Class that manages gRPC channel lifetimes."""
 
-    def __init__(self):
+    def __init__(self) -> None:
         """Initialize the GrpcChannelPool object."""
         super().__init__()
         self._discovery_client = DiscoveryClient()
@@ -181,7 +181,7 @@ def resolve_file_path(self, file_path: str) -> str:
         return absolute_path
 
 
-def configure_logging(verbosity: int):
+def configure_logging(verbosity: int) -> None:
     """Configure logging for this process."""
     if verbosity > 1:
         level = logging.DEBUG

examples/nidcpower_source_dc_voltage/_nidcpower_helpers.py

@@ -12,7 +12,7 @@
 def create_session(
     session_info: nims.session_management.SessionInformation,
     session_grpc_channel: Optional[grpc.Channel] = None,
-    initialization_behavior=nidcpower.SessionInitializationBehavior.AUTO,
+    initialization_behavior: nidcpower.SessionInitializationBehavior = nidcpower.SessionInitializationBehavior.AUTO,
 ) -> nidcpower.Session:
     """Create driver session based on reserved session and grpc channel."""
     options: Dict[str, Any] = {}

examples/nidcpower_source_dc_voltage/measurement.py

@@ -5,7 +5,7 @@
 import sys
 import threading
 import time
-from typing import Iterable, List, Tuple
+from typing import Any, Iterable, List, Tuple
 
 import click
 import grpc
@@ -170,7 +170,7 @@ def _log_measured_values(
 @verbosity_option
 @grpc_device_options
 @use_simulation_option(default=USE_SIMULATION)
-def main(verbosity: int, **kwargs) -> None:
+def main(verbosity: int, **kwargs: Any) -> None:
     """Source and measure a DC voltage with an NI SMU."""
     configure_logging(verbosity)
 

examples/nidcpower_source_dc_voltage/pyproject.toml

@@ -22,6 +22,9 @@ grpc-stubs = "^1.53"
 requires = ["poetry-core>=1.2.0"]
 build-backend = "poetry.core.masonry.api"
 
+[tool.mypy]
+disallow_untyped_defs = true
+
 [[tool.mypy.overrides]]
 module = [
     "hightime.*",

examples/nidigital_spi/_helpers.py

@@ -39,7 +39,7 @@ class ServiceOptions(NamedTuple):
     use_simulation: bool = False
 
 
-def get_service_options(**kwargs) -> ServiceOptions:
+def get_service_options(**kwargs: Any) -> ServiceOptions:
     """Get service options from keyword arguments."""
     return ServiceOptions(
         use_grpc_device=kwargs.get("use_grpc_device", False),
@@ -81,7 +81,7 @@ def update_pin_map(self, pin_map_path: str) -> str:
 class GrpcChannelPoolHelper(GrpcChannelPool):
     """Class that manages gRPC channel lifetimes."""
 
-    def __init__(self):
+    def __init__(self) -> None:
         """Initialize the GrpcChannelPool object."""
         super().__init__()
         self._discovery_client = DiscoveryClient()
@@ -181,7 +181,7 @@ def resolve_file_path(self, file_path: str) -> str:
         return absolute_path
 
 
-def configure_logging(verbosity: int):
+def configure_logging(verbosity: int) -> None:
     """Configure logging for this process."""
     if verbosity > 1:
         level = logging.DEBUG

examples/nidigital_spi/_nidigital_helpers.py

@@ -12,7 +12,7 @@
 def create_session(
     session_info: nims.session_management.SessionInformation,
     session_grpc_channel: Optional[grpc.Channel] = None,
-    initialization_behavior=nidigital.SessionInitializationBehavior.AUTO,
+    initialization_behavior: nidigital.SessionInitializationBehavior = nidigital.SessionInitializationBehavior.AUTO,
 ) -> nidigital.Session:
     """Create driver session based on reserved session and grpc channel."""
     options: Dict[str, Any] = {}