nidaqmx: Upgrade to Python 3.10 syntax

examples/analog_out/cont_gen_voltage_wfm_int_clk.py

@@ -4,8 +4,6 @@
 waveform using an internal sample clock.
 """
 
-from typing import Tuple
-
 import numpy as np
 import numpy.typing
 
@@ -19,7 +17,7 @@ def generate_sine_wave(
     sampling_rate: float,
     number_of_samples: int,
     phase_in: float = 0.0,
-) -> Tuple[numpy.typing.NDArray[numpy.double], float]:
+) -> tuple[numpy.typing.NDArray[numpy.double], float]:
     """Generates a sine wave with a specified phase.
 
     Args:

examples/analog_out/cont_gen_voltage_wfm_int_clk_every_n_samples_event.py

@@ -6,8 +6,6 @@
 the specified number of samples generation is complete.
 """
 
-from typing import Tuple
-
 import numpy as np
 import numpy.typing
 
@@ -21,7 +19,7 @@ def generate_sine_wave(
     sampling_rate: float,
     number_of_samples: int,
     phase_in: float = 0.0,
-) -> Tuple[numpy.typing.NDArray[numpy.double], float]:
+) -> tuple[numpy.typing.NDArray[numpy.double], float]:
     """Generates a sine wave with a specified phase.
 
     Args:

examples/analog_out/cont_gen_voltage_wfm_int_clk_non_regen.py

@@ -12,8 +12,6 @@
 of your sample clock.
 """
 
-from typing import Tuple
-
 import numpy as np
 import numpy.typing
 
@@ -27,7 +25,7 @@ def generate_sine_wave(
     sampling_rate: float,
     number_of_samples: int,
     phase_in: float = 0.0,
-) -> Tuple[numpy.typing.NDArray[numpy.double], float]:
+) -> tuple[numpy.typing.NDArray[numpy.double], float]:
     """Generates a sine wave with a specified phase.
 
     Args:

examples/nidaqmx_warnings.py

@@ -29,7 +29,7 @@
             assert issubclass(w[-1].category, nidaqmx.DaqWarning)
 
         if w:
-            print("DaqWarning caught: {}\n".format(w[-1].message))
+            print(f"DaqWarning caught: {w[-1].message}\n")
 
 
 # Raising warnings as exceptions.
@@ -43,7 +43,7 @@
         task.write([1.1, 2.2, 3.3, 4.4, 5.5], auto_start=True)
         task.stop()
     except nidaqmx.DaqWarning as e:
-        print("DaqWarning caught as exception: {}\n".format(e))
+        print(f"DaqWarning caught as exception: {e}\n")
         assert e.error_code == DAQmxWarnings.STOPPED_BEFORE_DONE
 
 

examples/synchronization/multi_function/ai_ao_sync.py

@@ -4,8 +4,6 @@
 generate data at the same time, synchronized with one another.
 """
 
-from typing import Tuple
-
 import numpy as np
 import numpy.typing
 
@@ -39,7 +37,7 @@ def generate_sine_wave(
     sampling_rate: float,
     number_of_samples: int,
     phase_in: float = 0.0,
-) -> Tuple[numpy.typing.NDArray[numpy.double], float]:
+) -> tuple[numpy.typing.NDArray[numpy.double], float]:
     """Generates a sine wave with a specified phase.
 
     Args:

generated/nidaqmx/_base_interpreter.py

@@ -4,7 +4,9 @@
 import abc
 import numpy
 from nitypes.waveform import AnalogWaveform, DigitalWaveform
-from typing import Any, Sequence
+from typing import Any
+
+from collections.abc import Sequence
 from nidaqmx.constants import WaveformAttributeMode
 
 

generated/nidaqmx/_dotenvpath.py

@@ -45,27 +45,12 @@ def _get_caller_path() -> Path | None:
     for frame, _ in traceback.walk_stack(inspect.currentframe()):
         if frame.f_code.co_filename:
             module_path = Path(frame.f_code.co_filename)
-            if _exists(module_path) and not module_path.is_relative_to(package_path):
+            if module_path.exists() and not module_path.is_relative_to(package_path):
                 return module_path
 
     return None
 
 
-# Path.exists() throws OSError when the path has invalid file characters.
-# https://github.com/python/cpython/issues/79487
-if sys.version_info >= (3, 10):
-
-    def _exists(path: Path) -> bool:
-        return path.exists()
-
-else:
-
-    def _exists(path: Path) -> bool:
-        import os
-
-        return os.path.exists(path)
-
-
 def _get_package_path() -> Path:
     """Get the path of this package."""
     module = sys.modules[__package__]

generated/nidaqmx/_feature_toggles.py

@@ -4,19 +4,17 @@
 
 import functools
 import sys
+from collections.abc import Callable
 from enum import Enum
-from typing import TYPE_CHECKING, Callable, TypeVar
+from typing import TYPE_CHECKING, TypeVar
 
 from decouple import AutoConfig, Undefined, undefined
 
 from nidaqmx._dotenvpath import get_dotenv_search_path
 from nidaqmx.errors import FeatureNotSupportedError
 
 if TYPE_CHECKING:
-    if sys.version_info >= (3, 10):
-        from typing import ParamSpec
-    else:
-        from typing_extensions import ParamSpec
+    from typing import ParamSpec
 
     if sys.version_info >= (3, 11):
         from typing import Self

generated/nidaqmx/_grpc_interpreter.py

@@ -6,10 +6,11 @@
 import typing
 import warnings
 from nitypes.waveform import AnalogWaveform, DigitalWaveform
-from typing import Any, Callable, Generic, Sequence, TypeVar
+from typing import Any, Generic, TypeVar
+
+from collections.abc import Callable, Sequence
 
 import google.protobuf.message
-from google.protobuf.timestamp_pb2 import Timestamp as GrpcTimestamp
 import grpc
 import numpy
 

generated/nidaqmx/_install_daqmx.py

@@ -10,7 +10,7 @@
 import sys
 import tempfile
 import zipfile
-from typing import Generator
+from collections.abc import Generator
 from urllib.parse import urlparse
 
 import click

generated/nidaqmx/_lib.py

@@ -17,7 +17,7 @@
 )
 
 if TYPE_CHECKING:
-    from typing_extensions import TypeAlias
+    from typing import TypeAlias
 
 
 _DAQ_NOT_FOUND_MESSAGE = (

generated/nidaqmx/_library_interpreter.py

@@ -6,12 +6,13 @@
 import numpy
 import platform
 import warnings
-import sys
 from enum import Enum
 from datetime import timezone
 from hightime import datetime as ht_datetime
 from hightime import timedelta as ht_timedelta
-from typing import Any, Callable, List, Sequence, Tuple, TYPE_CHECKING, Union
+from typing import Any, TYPE_CHECKING
+
+from collections.abc import Callable, Sequence
 
 from nidaqmx._base_interpreter import BaseEventHandler, BaseInterpreter
 from nidaqmx._lib import lib_importer, ctypes_byte_str, c_bool32, wrapped_ndpointer, TaskHandle
@@ -23,10 +24,7 @@
 from nitypes.waveform import AnalogWaveform, DigitalWaveform, SampleIntervalMode, Timing, ExtendedPropertyDictionary
 
 if TYPE_CHECKING:
-    if sys.version_info >= (3, 10):
-        from typing import TypeAlias
-    else:
-        from typing_extensions import TypeAlias
+    from typing import TypeAlias
 
 _logger = logging.getLogger(__name__)
 _was_runtime_environment_set = None
@@ -2777,7 +2775,7 @@ def get_digital_power_up_states(self, device_name, channel_name):
         state = []
 
         args = [device_name]
-        argtypes: List[type] = [ctypes_byte_str]
+        argtypes: list[type] = [ctypes_byte_str]
 
         for index in range(len(channel_name)):
             state_element = ctypes.c_int32()
@@ -2803,7 +2801,7 @@ def get_digital_pull_up_pull_down_states(self, device_name, channel_name):
         state = []
 
         args = [device_name]
-        argtypes: List[type] = [ctypes_byte_str]
+        argtypes: list[type] = [ctypes_byte_str]
 
         for index in range(len(channel_name)):
             state_element = ctypes.c_int32()
@@ -5011,7 +5009,7 @@ def self_test_device(self, device_name):
     def set_analog_power_up_states(
             self, device_name, channel_names, state, channel_type):
         args = [device_name]
-        argtypes: List[type] = [ctypes_byte_str]
+        argtypes: list[type] = [ctypes_byte_str]
 
         for index in range(len(channel_names)):
 
@@ -5204,7 +5202,7 @@ def set_digital_logic_family_power_up_state(
 
     def set_digital_power_up_states(self, device_name, channel_names, state):
         args = [device_name]
-        argtypes: List[type] = [ctypes_byte_str]
+        argtypes: list[type] = [ctypes_byte_str]
 
         for index in range(len(channel_names)):
 
@@ -5226,7 +5224,7 @@ def set_digital_power_up_states(self, device_name, channel_names, state):
     def set_digital_pull_up_pull_down_states(
             self, device_name, channel_names, state):
         args = [device_name]
-        argtypes: List[type] = [ctypes_byte_str]
+        argtypes: list[type] = [ctypes_byte_str]
 
         for index in range(len(channel_names)):
 
@@ -6481,7 +6479,7 @@ def _internal_read_analog_waveform_ex(
         properties: Sequence[ExtendedPropertyDictionary] | None,
         t0_array: numpy.typing.NDArray[numpy.int64] | None,
         dt_array: numpy.typing.NDArray[numpy.int64] | None,
-    ) -> Tuple[
+    ) -> tuple[
         int, # error code
         int, # The number of samples per channel that were read
     ]:
@@ -6535,7 +6533,7 @@ def _internal_read_analog_waveform_per_chan(
         properties: Sequence[ExtendedPropertyDictionary] | None,
         t0_array: numpy.typing.NDArray[numpy.int64] | None,
         dt_array: numpy.typing.NDArray[numpy.int64] | None,
-    ) -> Tuple[
+    ) -> tuple[
         int, # error code
         int, # The number of samples per channel that were read
     ]:
@@ -6651,7 +6649,7 @@ def _invoke_callback(
 
     def _set_waveform_timings(
         self, 
-        waveforms: Sequence[Union[AnalogWaveform[numpy.float64], DigitalWaveform[numpy.uint8]]], 
+        waveforms: Sequence[AnalogWaveform[numpy.float64] | DigitalWaveform[numpy.uint8]], 
         t0_array: numpy.typing.NDArray[numpy.int64], 
         dt_array: numpy.typing.NDArray[numpy.int64]
     ) -> None:
@@ -6835,7 +6833,7 @@ def _internal_read_digital_waveform(
         t0_array: numpy.typing.NDArray[numpy.int64] | None,
         dt_array: numpy.typing.NDArray[numpy.int64] | None,
         bytes_per_chan_array: numpy.typing.NDArray[numpy.uint32] | None = None,
-    ) -> Tuple[
+    ) -> tuple[
         int, # error code
         int, # The number of samples per channel that were read
     ]:
@@ -7049,7 +7047,7 @@ def _internal_write_analog_waveform_per_chan(
         auto_start: bool,
         timeout: float,
         write_arrays: Sequence[numpy.typing.NDArray[numpy.float64]],
-    ) -> Tuple[
+    ) -> tuple[
         int, # error code
         int, # The number of samples per channel that were written
     ]:
@@ -7176,7 +7174,7 @@ def _internal_write_digital_waveform(
         data_layout: int,
         write_array: numpy.typing.NDArray[numpy.uint8],
         bytes_per_chan_array: numpy.typing.NDArray[numpy.uint32] | None = None,
-    ) -> Tuple[
+    ) -> tuple[
         int, # error code
         int, # The number of samples per channel that were written
     ]:

generated/nidaqmx/_linux_installation_commands.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
+from collections.abc import Callable
 from dataclasses import dataclass
-from typing import Callable
 
 
 def _get_version_ubuntu(dist_version: str) -> str:

generated/nidaqmx/stream_writers/_digital_multi_channel_writer.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
-from typing import Any, Sequence
+from collections.abc import Sequence
+from typing import Any
 
 from nitypes.waveform import DigitalWaveform
 

generated/nidaqmx/system/_collections/device_collection.py

@@ -67,7 +67,7 @@ def __getitem__(self, index):
             return [_DeviceAlternateConstructor(name, self._interpreter) for name in device_names]
         else:
             raise DaqError(
-                'Invalid index type "{}" used to access collection.'.format(type(index)),
+                f'Invalid index type "{type(index)}" used to access collection.',
                 DAQmxErrors.UNKNOWN,
             )
 

generated/nidaqmx/system/_collections/persisted_channel_collection.py

@@ -74,7 +74,7 @@ def __getitem__(self, index):
             ]
         else:
             raise DaqError(
-                'Invalid index type "{}" used to access collection.'.format(type(index)),
+                f'Invalid index type "{type(index)}" used to access collection.',
                 DAQmxErrors.UNKNOWN,
             )
 

generated/nidaqmx/system/_collections/persisted_scale_collection.py

@@ -70,7 +70,7 @@ def __getitem__(self, index):
             return [_PersistedScaleAlternateConstructor(name, self._interpreter) for name in names]
         else:
             raise DaqError(
-                'Invalid index type "{}" used to access collection.'.format(type(index)),
+                f'Invalid index type "{type(index)}" used to access collection.',
                 DAQmxErrors.UNKNOWN,
             )
 

generated/nidaqmx/system/_collections/persisted_task_collection.py

@@ -70,7 +70,7 @@ def __getitem__(self, index):
             return [_PersistedTaskAlternateConstructor(name, self._interpreter) for name in names]
         else:
             raise DaqError(
-                'Invalid index type "{}" used to access collection.'.format(type(index)),
+                f'Invalid index type "{type(index)}" used to access collection.',
                 DAQmxErrors.UNKNOWN,
             )
 

generated/nidaqmx/system/_collections/physical_channel_collection.py

@@ -87,7 +87,7 @@ def __getitem__(self, index):
             ]
         else:
             raise DaqError(
-                'Invalid index type "{}" used to access collection.'.format(type(index)),
+                f'Invalid index type "{type(index)}" used to access collection.',
                 DAQmxErrors.UNKNOWN,
             )
 

generated/nidaqmx/system/_watchdog_modules/expiration_states_collection.py

@@ -41,5 +41,5 @@ def __getitem__(self, index):
             return ExpirationState(self._handle, index, self._interpreter)
         else:
             raise DaqError(
-                'Invalid index type "{}" used to access expiration states.'.format(type(index)), -1
+                f'Invalid index type "{type(index)}" used to access expiration states.', -1
             )