handwritten: Run nps fix

Author: bkeryan

src/handwritten/__init__.py

@@ -1,4 +1,10 @@
-from nidaqmx.errors import DaqError, DaqReadError, DaqWriteError, DaqWarning, DaqResourceWarning
+from nidaqmx.errors import (
+    DaqError,
+    DaqReadError,
+    DaqResourceWarning,
+    DaqWarning,
+    DaqWriteError,
+)
 from nidaqmx.grpc_session_options import *
 from nidaqmx.scale import Scale
 from nidaqmx.task import Task

src/handwritten/_bitfield_utils.py

@@ -1,5 +1,4 @@
-def enum_bitfield_to_list(bitfield_value, bitfield_enum_type,
-                          actual_enum_type):
+def enum_bitfield_to_list(bitfield_value, bitfield_enum_type, actual_enum_type):
     """
     Converts a bitfield value to a list of enums.
 
@@ -14,8 +13,7 @@ def enum_bitfield_to_list(bitfield_value, bitfield_enum_type,
     supported_values = []
     for bitfield_mask in bitfield_enum_type:
         if bitfield_value & bitfield_mask.value:
-            enum_value = next(
-                e for e in actual_enum_type if e.name == bitfield_mask.name)
+            enum_value = next(e for e in actual_enum_type if e.name == bitfield_mask.name)
             supported_values.append(enum_value)
 
     return supported_values
@@ -34,8 +32,7 @@ def enum_list_to_bitfield(enum_list, bitfield_enum_type):
     """
     bitfield_value = 0
     for enum_value in enum_list:
-        bitfield_mask = next(
-            b for b in bitfield_enum_type if b.name == enum_value.name)
+        bitfield_mask = next(b for b in bitfield_enum_type if b.name == enum_value.name)
         bitfield_value |= bitfield_mask.value
 
-    return bitfield_value
+    return bitfield_value

src/handwritten/_dotenvpath.py

@@ -70,4 +70,4 @@ def _get_package_path() -> Path:
     """Get the path of this package."""
     module = sys.modules[__package__]
     assert module.__file__ and module.__file__.endswith("__init__.py")
-    return Path(module.__file__).parent
+    return Path(module.__file__).parent

src/handwritten/_feature_toggles.py

@@ -4,9 +4,11 @@
 
 import functools
 import sys
-from decouple import AutoConfig, Undefined, undefined
 from enum import Enum
 from typing import TYPE_CHECKING, Callable, TypeVar
+
+from decouple import AutoConfig, Undefined, undefined
+
 from nidaqmx._dotenvpath import get_dotenv_search_path
 from nidaqmx.errors import FeatureNotSupportedError
 
@@ -36,6 +38,7 @@ def _config(
         cast: Callable[[str], _T] | Undefined = undefined,
     ) -> _T: ...
 
+
 # Based on the recipe at https://docs.python.org/3/howto/enum.html
 class _OrderedEnum(Enum):
     def __ge__(self, other: Self) -> bool:
@@ -156,4 +159,4 @@ def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _T:
 # Define feature toggle constants here:
 # --------------------------------------
 
-WAVEFORM_SUPPORT = FeatureToggle("WAVEFORM_SUPPORT", CodeReadiness.INCOMPLETE)
+WAVEFORM_SUPPORT = FeatureToggle("WAVEFORM_SUPPORT", CodeReadiness.INCOMPLETE)

src/handwritten/_grpc_time.py

@@ -1,14 +1,12 @@
 from __future__ import annotations
 
-from datetime import timezone
-from datetime import datetime as std_datetime
-from datetime import tzinfo as dt_tzinfo
-from hightime import datetime as ht_datetime
-from hightime import timedelta as ht_timedelta
+from datetime import datetime as std_datetime, timezone, tzinfo as dt_tzinfo
 from typing import Optional, Union
-from nidaqmx._time import _convert_to_desired_timezone
 
 from google.protobuf.timestamp_pb2 import Timestamp as GrpcTimestamp
+from hightime import datetime as ht_datetime, timedelta as ht_timedelta
+
+from nidaqmx._time import _convert_to_desired_timezone
 
 # 66 years, 17 leap days = 24107 days = 2082844800 seconds
 _BIAS_FROM_1970_EPOCH = 2082844800
@@ -22,7 +20,10 @@
 
 _EPOCH_1970 = ht_datetime(1970, 1, 1, tzinfo=timezone.utc)
 
-def convert_time_to_timestamp(dt: std_datetime | ht_datetime, ts: GrpcTimestamp | None = 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:
@@ -44,10 +45,11 @@ def convert_time_to_timestamp(dt: std_datetime | ht_datetime, ts: GrpcTimestamp
     ts.FromNanoseconds(seconds_since_1970 * _NS_PER_S + nanos)
     return ts
 
+
 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.
     total_yoctoseconds = int(round(_YS_PER_NS * nanos))
-    dt = _EPOCH_1970 + ht_timedelta(seconds = seconds) + ht_timedelta(yoctoseconds=total_yoctoseconds)
+    dt = _EPOCH_1970 + ht_timedelta(seconds=seconds) + ht_timedelta(yoctoseconds=total_yoctoseconds)
     return _convert_to_desired_timezone(dt, tzinfo)

src/handwritten/_install_daqmx.py

@@ -13,12 +13,12 @@
 import sys
 import tempfile
 import traceback
-import requests
 import zipfile
 from typing import Generator, List, Optional, Tuple
 from urllib.parse import urlparse
 
 import click
+import requests
 
 if sys.platform.startswith("win"):
     import winreg
@@ -101,7 +101,9 @@ def _get_daqmx_installed_version() -> str | None:
             commands_info = LINUX_COMMANDS[distribution]
             query_command = commands_info.get_daqmx_version
             # Run the package query command defined by _linux_installation_commands.py.
-            query_output = subprocess.run(query_command, stdout=subprocess.PIPE, text=True).stdout  # nosec: B603
+            query_output = subprocess.run(
+                query_command, stdout=subprocess.PIPE, text=True
+            ).stdout  # nosec: B603
 
             if distribution == "ubuntu":
                 version_match = re.search(r"ii\s+ni-daqmx\s+(\d+\.\d+\.\d+)", query_output)
@@ -127,9 +129,7 @@ def _get_daqmx_installed_version() -> str | None:
 
 # Creating a temp file that we then close and yield - this is used to allow subprocesses to access
 @contextlib.contextmanager
-def _multi_access_temp_file(
-    *, suffix: str = ".exe", delete: bool = True
-) -> Generator[str]:
+def _multi_access_temp_file(*, suffix: str = ".exe", delete: bool = True) -> Generator[str]:
     """
     Context manager for creating a temporary file.
 
@@ -244,7 +244,7 @@ def _install_daqmx_driver_windows_core(download_url: str) -> None:
             _logger.info("Downloading Driver to %s", temp_file)
             response = requests.get(download_url, timeout=_NETWORK_TIMEOUT_IN_SECONDS)
             response.raise_for_status()
-            with open(temp_file, 'wb') as f:
+            with open(temp_file, "wb") as f:
                 f.write(response.content)
             _logger.info("Installing NI-DAQmx")
             # Run the installer that we just downloaded from https://download.ni.com.
@@ -274,7 +274,7 @@ def _install_daqmx_driver_linux_core(download_url: str, release: str) -> None:
                 _logger.info("Downloading Driver to %s", temp_file)
                 response = requests.get(download_url, timeout=_NETWORK_TIMEOUT_IN_SECONDS)
                 response.raise_for_status()
-                with open(temp_file, 'wb') as f:
+                with open(temp_file, "wb") as f:
                     f.write(response.content)
 
                 with tempfile.TemporaryDirectory() as temp_folder:
@@ -443,9 +443,7 @@ def _install_daqmx_driver():
                 installed_version, latest_version, release
             )
         else:
-            user_response = _fresh_install_daqmx_user_confirmation(
-                latest_version, release
-            )
+            user_response = _fresh_install_daqmx_user_confirmation(latest_version, release)
 
         if user_response:
             if platform == "Linux":

src/handwritten/_lib.py

@@ -1,29 +1,40 @@
 from __future__ import annotations
 
-from ctypes.util import find_library
 import ctypes
-from numpy.ctypeslib import ndpointer
+import locale
 import platform
 import sys
 import threading
-import locale
+from ctypes.util import find_library
+from typing import TYPE_CHECKING, cast
+
 from decouple import config
-from typing import cast, TYPE_CHECKING
+from numpy.ctypeslib import ndpointer
 
-from nidaqmx.errors import DaqNotFoundError, DaqNotSupportedError, DaqFunctionNotSupportedError
+from nidaqmx.errors import (
+    DaqFunctionNotSupportedError,
+    DaqNotFoundError,
+    DaqNotSupportedError,
+)
 
 if TYPE_CHECKING:
     from typing_extensions import TypeAlias
 
 
-_DAQ_NOT_FOUND_MESSAGE = "Could not find an installation of NI-DAQmx. Please ensure that NI-DAQmx " \
-                         "is installed on this machine or contact National Instruments for support."
+_DAQ_NOT_FOUND_MESSAGE = (
+    "Could not find an installation of NI-DAQmx. Please ensure that NI-DAQmx "
+    "is installed on this machine or contact National Instruments for support."
+)
 
-_DAQ_NOT_SUPPORTED_MESSAGE = "NI-DAQmx Python is not supported on this platform: {0}. Please " \
-                             "direct any questions or feedback to National Instruments."
+_DAQ_NOT_SUPPORTED_MESSAGE = (
+    "NI-DAQmx Python is not supported on this platform: {0}. Please "
+    "direct any questions or feedback to National Instruments."
+)
 
-_FUNCTION_NOT_SUPPORTED_MESSAGE = "The NI-DAQmx function \"{0}\" is not supported in this version " \
-                                  "of NI-DAQmx. Visit ni.com/downloads to upgrade."
+_FUNCTION_NOT_SUPPORTED_MESSAGE = (
+    'The NI-DAQmx function "{0}" is not supported in this version '
+    "of NI-DAQmx. Visit ni.com/downloads to upgrade."
+)
 
 
 class c_bool32(ctypes.c_uint):
@@ -48,9 +59,10 @@ def _setter(self, val):
 
 class CtypesByteString:
     """
-    Custom argtype that automatically converts unicode strings to encoding 
+    Custom argtype that automatically converts unicode strings to encoding
     used by the DAQmx C API DLL in Python 3.
     """
+
     @classmethod
     def from_param(cls, param):
         if isinstance(param, str):
@@ -75,8 +87,7 @@ def from_param(cls, obj):
             return obj
         return base.from_param(obj)
 
-    return type(base.__name__, (base,),
-                {'from_param': classmethod(from_param)})
+    return type(base.__name__, (base,), {"from_param": classmethod(from_param)})
 
 
 class DaqFunctionImporter:
@@ -94,9 +105,9 @@ def __init__(self, library):
     def __getattr__(self, function):
         try:
             cfunc = getattr(self._library, function)
-            if not hasattr(cfunc, 'arglock'):
+            if not hasattr(cfunc, "arglock"):
                 with self._lib_lock:
-                    if not hasattr(cfunc, 'arglock'):
+                    if not hasattr(cfunc, "arglock"):
                         cfunc.arglock = threading.Lock()
             return cfunc
         except AttributeError:
@@ -125,7 +136,7 @@ def get_encoding_from_locale() -> str:
     Gets the current locale encoding handling cases where it is unset.
     """
     _, encoding = locale.getlocale()
-    return encoding or 'ascii'
+    return encoding or "ascii"
 
 
 class DaqLibImporter:
@@ -159,7 +170,7 @@ def task_handle(self) -> type:
     @property
     def cal_handle(self) -> type:
         return CalHandle
-    
+
     @property
     def encoding(self):
         if self._encoding is None:
@@ -178,42 +189,44 @@ def _import_lib(self):
         cdll = None
         encoding = None
 
-        if sys.platform.startswith('win'):
+        if sys.platform.startswith("win"):
 
             def _load_lib(libname: str):
                 windll = ctypes.windll.LoadLibrary(libname)
                 cdll = ctypes.cdll.LoadLibrary(libname)
-                return windll, cdll  
-            
+                return windll, cdll
+
             # Feature Toggle to load nicaiu.dll or nicai_utf8.dll
             # The Feature Toggle can be set in the .env file
-            nidaqmx_c_library = config('NIDAQMX_C_LIBRARY', default=None) 
-  
+            nidaqmx_c_library = config("NIDAQMX_C_LIBRARY", default=None)
+
             if nidaqmx_c_library is not None:
-                try: 
-                    if nidaqmx_c_library=="nicaiu":
+                try:
+                    if nidaqmx_c_library == "nicaiu":
                         windll, cdll = _load_lib("nicaiu")
                         encoding = get_encoding_from_locale()
-                    elif nidaqmx_c_library=="nicai_utf8":
+                    elif nidaqmx_c_library == "nicai_utf8":
                         windll, cdll = _load_lib("nicai_utf8")
-                        encoding = 'utf-8'  
+                        encoding = "utf-8"
                     else:
-                        raise ValueError(f"Unsupported NIDAQMX_C_LIBRARY value: {nidaqmx_c_library}")
+                        raise ValueError(
+                            f"Unsupported NIDAQMX_C_LIBRARY value: {nidaqmx_c_library}"
+                        )
                 except OSError as e:
-                    raise DaqNotFoundError(_DAQ_NOT_FOUND_MESSAGE) from e         
+                    raise DaqNotFoundError(_DAQ_NOT_FOUND_MESSAGE) from e
             else:
                 try:
                     windll, cdll = _load_lib("nicai_utf8")
-                    encoding = 'utf-8'  
+                    encoding = "utf-8"
                 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 as e:
-                        raise DaqNotFoundError(_DAQ_NOT_FOUND_MESSAGE) from e       
-        elif sys.platform.startswith('linux'):
-            library_path = find_library('nidaqmx')
+                        raise DaqNotFoundError(_DAQ_NOT_FOUND_MESSAGE) from e
+        elif sys.platform.startswith("linux"):
+            library_path = find_library("nidaqmx")
             if library_path is not None:
                 cdll = ctypes.cdll.LoadLibrary(library_path)
                 windll = cdll