[DO NOT MERGE UNTIL 25C2] Upgrade to Python 3.9 syntax

examples/nidaqmx_warnings.py

@@ -29,7 +29,7 @@
             assert issubclass(w[-1].category, nidaqmx.DaqWarning)
 
         if w:
-            print("DaqWarning caught: {0}\n".format(w[-1].message))
+            print("DaqWarning caught: {}\n".format(w[-1].message))
 
 
 # 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: {0}\n".format(e))
+        print("DaqWarning caught as exception: {}\n".format(e))
         assert e.error_code == DAQmxWarnings.STOPPED_BEFORE_DONE
 
 

examples/system_properties.py

@@ -6,7 +6,7 @@
 driver_version = local_system.driver_version
 
 print(
-    "DAQmx {0}.{1}.{2}".format(
+    "DAQmx {}.{}.{}".format(
         driver_version.major_version,
         driver_version.minor_version,
         driver_version.update_version,
@@ -15,7 +15,7 @@
 
 for device in local_system.devices:
     print(
-        "Device Name: {0}, Product Category: {1}, Product Type: {2}".format(
+        "Device Name: {}, Product Category: {}, Product Type: {}".format(
             device.name, device.product_category, device.product_type
         )
     )

generated/nidaqmx/__main__.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import logging
 from typing import Optional
 

generated/nidaqmx/_grpc_interpreter.py

@@ -5,7 +5,7 @@
 import threading
 import typing
 import warnings
-from typing import Callable, Generic, Optional, TypeVar
+from typing import Callable, Generic, TypeVar
 
 import google.protobuf.message
 from google.protobuf.timestamp_pb2 import Timestamp as GrpcTimestamp
@@ -52,7 +52,7 @@ def __init__(
         self._interpreter = interpreter
         self._event_stream = event_stream
         self._event_callback = event_callback
-        self._event_stream_exception: Optional[Exception] = None
+        self._event_stream_exception: Exception | None = None
         self._thread = threading.Thread(target=self._thread_main, name=f"nidaqmx {event_name} thread")
 
         self._thread.start()

generated/nidaqmx/_grpc_time.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from datetime import timezone
 from datetime import datetime as std_datetime
 from datetime import tzinfo as dt_tzinfo
@@ -20,7 +22,7 @@
 
 _EPOCH_1970 = ht_datetime(1970, 1, 1, tzinfo=timezone.utc)
 
-def convert_time_to_timestamp(dt: Union[std_datetime, ht_datetime], ts: Optional[GrpcTimestamp] = None) -> GrpcTimestamp:
+def convert_time_to_timestamp(dt: std_datetime | ht_datetime, ts: GrpcTimestamp | None = None) -> GrpcTimestamp:
     seconds_since_1970 = 0
 
     if ts is None:
@@ -42,7 +44,7 @@ def convert_time_to_timestamp(dt: Union[std_datetime, ht_datetime], ts: Optional
     ts.FromNanoseconds(seconds_since_1970 * _NS_PER_S + nanos)
     return ts
 
-def convert_timestamp_to_time(ts: GrpcTimestamp, tzinfo: Optional[dt_tzinfo] = None) -> ht_datetime:
+def convert_timestamp_to_time(ts: GrpcTimestamp, tzinfo: dt_tzinfo | None = None) -> ht_datetime:
     total_nanos = ts.ToNanoseconds()
     seconds, nanos = divmod(total_nanos, _NS_PER_S)
     # Convert the nanoseconds to yoctoseconds.

generated/nidaqmx/_install_daqmx.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import contextlib
 import errno
 import importlib.resources as pkg_resources
@@ -34,7 +36,7 @@
 _NETWORK_TIMEOUT_IN_SECONDS = 60
 
 
-def _parse_version(version: str) -> Tuple[int, ...]:
+def _parse_version(version: str) -> tuple[int, ...]:
     """
     Split the version string into a tuple of integers.
 
@@ -54,7 +56,7 @@ def _parse_version(version: str) -> Tuple[int, ...]:
         raise click.ClickException(f"Invalid version number '{version}'.") from e
 
 
-def _get_daqmx_installed_version() -> Optional[str]:
+def _get_daqmx_installed_version() -> str | None:
     """
     Check for existing installation of NI-DAQmx.
 
@@ -127,7 +129,7 @@ def _get_daqmx_installed_version() -> Optional[str]:
 @contextlib.contextmanager
 def _multi_access_temp_file(
     *, suffix: str = ".exe", delete: bool = True
-) -> Generator[str, None, None]:
+) -> Generator[str]:
     """
     Context manager for creating a temporary file.
 
@@ -158,7 +160,7 @@ def _multi_access_temp_file(
 
 def _load_data(
     json_data: str, platform: str
-) -> Tuple[Optional[str], Optional[str], Optional[str], Optional[List[str]]]:
+) -> tuple[str | None, str | None, str | None, list[str] | None]:
     """
     Load data from JSON string and extract Windows metadata.
 
@@ -199,10 +201,10 @@ def _load_data(
         raise click.ClickException(f"Failed to parse the driver metadata.\nDetails: {e}") from e
 
     for metadata_entry in metadata:
-        location: Optional[str] = metadata_entry.get("Location")
-        version: Optional[str] = metadata_entry.get("Version")
-        release: Optional[str] = metadata_entry.get("Release")
-        supported_os: Optional[List[str]] = metadata_entry.get("supportedOS")
+        location: str | None = metadata_entry.get("Location")
+        version: str | None = metadata_entry.get("Version")
+        release: str | None = metadata_entry.get("Release")
+        supported_os: list[str] | None = metadata_entry.get("supportedOS")
         _logger.debug("From metadata file found location %s and version %s.", location, version)
         if location and version:
             return location, version, release, supported_os
@@ -211,7 +213,7 @@ def _load_data(
 
 def _get_driver_details(
     platform: str,
-) -> Tuple[Optional[str], Optional[str], Optional[str], Optional[List[str]]]:
+) -> tuple[str | None, str | None, str | None, list[str] | None]:
     """
     Parse the JSON data and retrieve the download link and version information.
 

generated/nidaqmx/_lib.py

@@ -199,18 +199,18 @@ def _load_lib(libname: str):
                         encoding = 'utf-8'  
                     else:
                         raise ValueError(f"Unsupported NIDAQMX_C_LIBRARY value: {nidaqmx_c_library}")
-                except (OSError, WindowsError) as e:
+                except OSError as e:
                     raise DaqNotFoundError(_DAQ_NOT_FOUND_MESSAGE) from e         
             else:
                 try:
                     windll, cdll = _load_lib("nicai_utf8")
                     encoding = 'utf-8'  
-                except (OSError, WindowsError):
+                except OSError:
                     # Fallback to nicaiu.dll if nicai_utf8.dll cannot be loaded
                     try:
                         windll, cdll = _load_lib("nicaiu")
                         encoding = get_encoding_from_locale()
-                    except (OSError, WindowsError) as e:
+                    except OSError as e:
                         raise DaqNotFoundError(_DAQ_NOT_FOUND_MESSAGE) from e       
         elif sys.platform.startswith('linux'):
             library_path = find_library('nidaqmx')

generated/nidaqmx/_lib_time.py

@@ -35,7 +35,7 @@ class AbsoluteTime(ctypes.Structure):
 
 
     @classmethod
-    def from_datetime(cls, dt: Union[std_datetime, ht_datetime]) -> AbsoluteTime:
+    def from_datetime(cls, dt: std_datetime | ht_datetime) -> AbsoluteTime:
         seconds_since_1904 = 0
 
         # Convert the subseconds.
@@ -55,7 +55,7 @@ def from_datetime(cls, dt: Union[std_datetime, ht_datetime]) -> AbsoluteTime:
 
         return AbsoluteTime(lsb=lsb, msb=seconds_since_1904)
 
-    def to_datetime(self, tzinfo: Optional[dt_tzinfo] = None) -> ht_datetime:
+    def to_datetime(self, tzinfo: dt_tzinfo | None = None) -> ht_datetime:
         total_yoctoseconds = int(
             round(AbsoluteTime._YS_PER_S * self.lsb / AbsoluteTime._NUM_SUBSECONDS)
         )

generated/nidaqmx/_linux_installation_commands.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from dataclasses import dataclass
 from typing import Callable, Dict, List, Tuple
 
@@ -58,8 +60,8 @@ def _get_version_rhel(dist_version: str) -> str:
 @dataclass
 class DistroInfo:
     get_distro_version: Callable[[str], str]
-    get_daqmx_version: List[str]
-    install_commands: List[List[str]]
+    get_daqmx_version: list[str]
+    install_commands: list[list[str]]
 
 
 # Mapping of distros to their command templates and version handlers
@@ -74,7 +76,7 @@ class DistroInfo:
 
 def get_linux_installation_commands(
     _directory_to_extract_to: str, dist_name: str, dist_version: str, _release_string: str
-) -> List[List[str]]:
+) -> list[list[str]]:
     """
     Get the installation commands for Linux based on the distribution.
 

generated/nidaqmx/_time.py

@@ -9,7 +9,7 @@
 from zoneinfo import ZoneInfo
 
 # theoretically the same as astimezone(), but with support for dates before 1970
-def _convert_to_desired_timezone(expected_time_utc: Union[std_datetime, ht_datetime], tzinfo: Optional[dt_tzinfo] = None) -> Union[std_datetime, ht_datetime]:
+def _convert_to_desired_timezone(expected_time_utc: std_datetime | ht_datetime, tzinfo: dt_tzinfo | None = None) -> std_datetime | ht_datetime:
     # if timezone matches, no need to do conversion
     if expected_time_utc.tzinfo is tzinfo:
         return expected_time_utc

generated/nidaqmx/stream_readers.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from typing import Optional, Tuple
 import numpy
 from nidaqmx import DaqError
@@ -69,7 +71,7 @@ def _verify_array(self, data, number_of_samples_per_channel,
         channels_to_read = self._in_stream.channels_to_read
         number_of_channels = len(channels_to_read.channel_names)
 
-        array_shape: Optional[Tuple[int, ...]] = None
+        array_shape: tuple[int, ...] | None = None
         if is_many_chan:
             if is_many_samp:
                 array_shape = (number_of_channels,
@@ -114,7 +116,7 @@ def _verify_array_digital_lines(
         number_of_channels = len(channels_to_read.channel_names)
         number_of_lines = self._in_stream.di_num_booleans_per_chan
 
-        array_shape: Optional[Tuple[int, ...]] = None
+        array_shape: tuple[int, ...] | None = None
         if is_many_chan:
             if is_many_line:
                 array_shape = (number_of_channels, number_of_lines)

generated/nidaqmx/system/storage/persisted_channel.py

@@ -1,4 +1,3 @@
-
 from nidaqmx import utils
 
 __all__ = ['PersistedChannel']

generated/nidaqmx/system/storage/persisted_scale.py

@@ -1,4 +1,3 @@
-
 from nidaqmx import utils
 from nidaqmx.scale import _ScaleAlternateConstructor
 

generated/nidaqmx/system/storage/persisted_task.py

@@ -1,4 +1,3 @@
-
 from nidaqmx import task
 from nidaqmx import utils
 

generated/nidaqmx/task/__init__.py

@@ -1,4 +1,3 @@
-
 from nidaqmx.task._task import (
     Task, _TaskEventType, _TaskAlternateConstructor
 )

generated/nidaqmx/task/_task.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import threading
 import warnings
 from enum import Enum
@@ -601,7 +603,7 @@ def read(self, number_of_samples_per_channel=NUM_SAMPLES_UNSET,
         # Determine the array shape and size to create
         if number_of_channels > 1:
             if not num_samples_not_set:
-                array_shape: Tuple[int, ...] = (
+                array_shape: tuple[int, ...] = (
                     number_of_channels, number_of_samples_per_channel
                 )
             else:
@@ -672,13 +674,13 @@ def read(self, number_of_samples_per_channel=NUM_SAMPLES_UNSET,
 
     def _read_ctr_pulse(
         self,
-        array_shape: Tuple[int, ...],
+        array_shape: tuple[int, ...],
         meas_type: UsageTypeCI,
         number_of_channels: int,
         number_of_samples_per_channel: int,
         num_samples_not_set: bool,
         timeout: float,
-    ) -> Union[CtrFreq, CtrTick, CtrTime, List[CtrFreq], List[CtrTick], List[CtrTime]]:
+    ) -> CtrFreq | CtrTick | CtrTime | list[CtrFreq] | list[CtrTick] | list[CtrTime]:
         if meas_type == UsageTypeCI.PULSE_FREQ:
             frequencies = numpy.zeros(array_shape, dtype=numpy.float64)
             duty_cycles = numpy.zeros(array_shape, dtype=numpy.float64)
@@ -687,7 +689,7 @@ def _read_ctr_pulse(
                 self._handle, number_of_samples_per_channel, timeout,
                 FillMode.GROUP_BY_CHANNEL.value, frequencies, duty_cycles)
 
-            data: Union[List[CtrFreq], List[CtrTick], List[CtrTime]] = [
+            data: list[CtrFreq] | list[CtrTick] | list[CtrTime] = [
                 CtrFreq(freq=f, duty_cycle=d) for f, d in zip(frequencies, duty_cycles)
             ]
 
@@ -727,11 +729,11 @@ def _read_ctr_pulse(
 
     def _read_power(
         self,
-        array_shape: Tuple[int, ...],
+        array_shape: tuple[int, ...],
         number_of_channels: int,
         number_of_samples_per_channel: int,
         timeout: float,
-    ) -> Union[PowerMeasurement, List[PowerMeasurement], List[List[PowerMeasurement]]]:
+    ) -> PowerMeasurement | list[PowerMeasurement] | list[list[PowerMeasurement]]:
         voltages = numpy.zeros(array_shape, dtype=numpy.float64)
         currents = numpy.zeros(array_shape, dtype=numpy.float64)
 

generated/nidaqmx/task/channels/__init__.py

@@ -1,4 +1,3 @@
-
 from nidaqmx.task.channels._channel import Channel
 from nidaqmx.task.channels._ai_channel import AIChannel
 from nidaqmx.task.channels._ao_channel import AOChannel

generated/nidaqmx/task/collections/__init__.py

@@ -1,4 +1,3 @@
-
 from nidaqmx.task.collections._channel_collection import ChannelCollection
 from nidaqmx.task.collections._ai_channel_collection import AIChannelCollection
 from nidaqmx.task.collections._ao_channel_collection import AOChannelCollection

generated/nidaqmx/task/triggering/__init__.py

@@ -1,4 +1,3 @@
-
 from nidaqmx.task.triggering._triggers import Triggers
 from nidaqmx.task.triggering._arm_start_trigger import ArmStartTrigger
 from nidaqmx.task.triggering._handshake_trigger import HandshakeTrigger

generated/nidaqmx/utils.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import re
 from dataclasses import dataclass
 from typing import List, Optional
@@ -37,7 +39,7 @@ def to_flattened_name(self) -> str:
             return f"{self.base_name}{self.start_index_str}:{self.end_index_str}"
 
 
-def flatten_channel_string(channel_names: List[str]) -> str:
+def flatten_channel_string(channel_names: list[str]) -> str:
     """
     Converts a list of channel names to a comma-delimited list of names.
 
@@ -109,7 +111,7 @@ def flatten_channel_string(channel_names: List[str]) -> str:
     return ','.join([_f for _f in flattened_channel_list if _f]).strip()
 
 
-def unflatten_channel_string(channel_names: str) -> List[str]:
+def unflatten_channel_string(channel_names: str) -> list[str]:
     """
     Converts a comma-delimited list of channel names to a list of names.
 
@@ -197,8 +199,8 @@ def unflatten_channel_string(channel_names: str) -> List[str]:
 
 
 def _select_interpreter(
-    grpc_options: Optional[GrpcSessionOptions] = None,
-    interpreter: Optional[BaseInterpreter] = None
+    grpc_options: GrpcSessionOptions | None = None,
+    interpreter: BaseInterpreter | None = None
 ) -> BaseInterpreter:
     if interpreter:
         return interpreter

src/codegen/templates/_grpc_interpreter.py.mako

@@ -22,7 +22,7 @@ import logging
 import threading
 import typing
 import warnings
-from typing import Callable, Generic, Optional, TypeVar
+from typing import Callable, Generic, TypeVar
 
 import google.protobuf.message
 from google.protobuf.timestamp_pb2 import Timestamp as GrpcTimestamp
@@ -69,7 +69,7 @@ class GrpcEventHandler(BaseEventHandler, Generic[TEventResponse]):
         self._interpreter = interpreter
         self._event_stream = event_stream
         self._event_callback = event_callback
-        self._event_stream_exception: Optional[Exception] = None
+        self._event_stream_exception: Exception | None = None
         self._thread = threading.Thread(target=self._thread_main, name=f"nidaqmx {event_name} thread")
 
         self._thread.start()

src/handwritten/__main__.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import logging
 from typing import Optional