Automate Driver Install Experience within nidaqmx Module for Linux OS

generated/nidaqmx/_install_daqmx.py

@@ -5,17 +5,27 @@
 import logging
 import os
 import pathlib
+import re
+import shutil
 import subprocess
 import sys
 import tempfile
 import traceback
 import urllib.request
+import zipfile
 from typing import Generator, List, Optional, Tuple
 
 import click
 
 if sys.platform.startswith("win"):
     import winreg
+elif sys.platform.startswith("linux"):
+    import distro
+
+    from nidaqmx._linux_installation_commands import (
+        get_linux_installation_commands,
+        LINUX_COMMANDS,
+    )
 
 _logger = logging.getLogger(__name__)
 
@@ -80,8 +90,31 @@ def _get_daqmx_installed_version() -> Optional[str]:
             raise click.ClickException(
                 f"An OS error occurred while getting the installed NI-DAQmx version.\nDetails: {e}"
             ) from e
+    elif sys.platform.startswith("linux"):
+        try:
+            _logger.debug("Checking for installed NI-DAQmx version")
+            commands_info = LINUX_COMMANDS[distro.id()]
+            query_command = commands_info["get_daqmx_version"]
+            query_output = subprocess.run(query_command, stdout=subprocess.PIPE, text=True).stdout
+
+            if distro.id() == "ubuntu":
+                version_match = re.search(r"ii\s+ni-daqmx\s+(\d+\.\d+\.\d+)", query_output)
+            elif distro.id() == "opensuse" or distro.id() == "rhel":
+                version_match = re.search(r"ni-daqmx-(\d+\.\d+\.\d+)", query_output)
+            else:
+                raise click.ClickException(f"Unsupported distribution '{distro.id()}'")
+
+            installed_version = version_match.group(1)
+            _logger.info("Found installed NI-DAQmx version: %s", installed_version)
+            return installed_version
+
+        except subprocess.CalledProcessError as e:
+            _logger.info("Failed to get installed NI-DAQmx version.", exc_info=True)
+            raise click.ClickException(
+                f"An error occurred while getting the installed NI-DAQmx version.\nCommand returned non-zero exit status '{e.returncode}'."
+            ) from e
     else:
-        raise NotImplementedError("This function is only supported on Windows.")
+        raise NotImplementedError("This function is only supported on Windows and Linux.")
 
 
 # Creating a temp file that we then close and yield - this is used to allow subprocesses to access
@@ -117,49 +150,71 @@ def _multi_access_temp_file(
                 ) from e
 
 
-def _load_data(json_data: str) -> Tuple[Optional[str], Optional[str]]:
+def _load_data(
+    json_data: str, platform: str
+) -> Tuple[Optional[str], Optional[str], Optional[str], Optional[List[str]]]:
     """
     Load data from JSON string and extract Windows metadata.
 
-    >>> _load_data('{"Windows": [{"Location": "path/to/driver", "Version": "1.0"}]}')
-    ('path/to/driver', '1.0')
+    >>> json_data = '{"Windows": [{"Location": "path/to/windows/driver", "Version": "24.0", "Release": "2024Q1", "supportedOS": ["Windows 11"]}], "Linux": []}'
+    >>> _load_data(json_data, "Windows")
+    ('path/to/windows/driver', '24.0', '2024Q1', ['Windows 11'])
+
+    >>> json_data = '{"Windows": [], "Linux": [{"Location": "path/to/linux/driver", "Version": "24.0", "Release": "2024Q1", "supportedOS": ["ubuntu 20.04 ", "rhel 9"]}]}'
+    >>> _load_data(json_data, "Linux")
+    ('path/to/linux/driver', '24.0', '2024Q1', ['ubuntu 20.04', 'rhel 9'])
 
-    >>> _load_data('{"Windows": [{"Location": "path/to/driver"}]}')
+    >>> json_data = '{"Windows": [{"Location": "path/to/windows/driver", "Version": "24.0", "Release": "2024Q1", "supportedOS": ["Windows 11"]}], "Linux": []}'
+    >>> _load_data(json_data, "Linux")
     Traceback (most recent call last):
-    ...
-    click.exceptions.ClickException: Unable to fetch driver details.
+    click.exceptions.ClickException: Unable to fetch driver details
 
-    >>> _load_data('{"Linux": [{"Location": "path/to/driver", "Version": "1.0"}]}')
+    >>> json_data = 'invalid json'
+    >>> _load_data(json_data, "Windows")
     Traceback (most recent call last):
-    ...
-    click.exceptions.ClickException: Unable to fetch driver details.
+    click.exceptions.ClickException: Failed to parse the driver metadata.
+    Details: Expecting value: line 1 column 1 (char 0)
+
+    >>> json_data = '{"Windows": [{"Location": "path/to/windows/driver", "Version": "24.0", "Release": "2024Q1", "supportedOS": ["Windows 11"]}], "Linux": []}'
+    >>> _load_data(json_data, "macOS")
+    Traceback (most recent call last):
+    click.exceptions.ClickException: Unsupported os 'macOS'
 
     """
     try:
-        metadata = json.loads(json_data).get("Windows", [])
+        if platform == "Windows":
+            metadata = json.loads(json_data).get("Windows", [])
+        elif platform == "Linux":
+            metadata = json.loads(json_data).get("Linux", [])
+        else:
+            raise click.ClickException(f"Unsupported os '{platform}'")
     except json.JSONDecodeError as e:
         _logger.info("Failed to parse the json data.", exc_info=True)
         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")
         _logger.debug("From metadata file found location %s and version %s.", location, version)
         if location and version:
-            return location, version
+            return location, version, release, supported_os
     raise click.ClickException(f"Unable to fetch driver details")
 
 
-def _get_driver_details() -> Tuple[Optional[str], Optional[str]]:
+def _get_driver_details(
+    platform: str,
+) -> Tuple[Optional[str], Optional[str], Optional[str], Optional[List[str]]]:
     """
     Parse the JSON data and retrieve the download link and version information.
 
     """
     try:
         with pkg_resources.open_text(__package__, METADATA_FILE) as json_file:
             _logger.debug("Opening the metadata file %s.", METADATA_FILE)
-            location, version = _load_data(json_file.read())
-        return location, version
+            location, version, release, supported_os = _load_data(json_file.read(), platform)
+        return location, version, release, supported_os
 
     except click.ClickException:
         raise
@@ -170,7 +225,7 @@ def _get_driver_details() -> Tuple[Optional[str], Optional[str]]:
         ) from e
 
 
-def _install_daqmx_driver(download_url: str) -> None:
+def _install_daqmx_driver_windows_core(download_url: str) -> None:
     """
     Download and launch NI-DAQmx Driver installation in an interactive mode
 
@@ -183,7 +238,7 @@ def _install_daqmx_driver(download_url: str) -> None:
             _logger.info("Installing NI-DAQmx")
             subprocess.run([temp_file], shell=True, check=True)
     except subprocess.CalledProcessError as e:
-        _logger.info("Failed to installed NI-DAQmx driver.", exc_info=True)
+        _logger.info("Failed to install NI-DAQmx driver.", exc_info=True)
         raise click.ClickException(
             f"An error occurred while installing the NI-DAQmx driver. Command returned non-zero exit status '{e.returncode}'."
         ) from e
@@ -195,6 +250,59 @@ def _install_daqmx_driver(download_url: str) -> None:
         raise click.ClickException(f"Failed to install the NI-DAQmx driver.\nDetails: {e}") from e
 
 
+def _install_daqmx_driver_linux_core(download_url: str, release: str) -> None:
+    """
+    Download NI Linux Device Drivers and install NI-DAQmx on Linux OS
+
+    """
+    if sys.platform.startswith("linux"):
+        try:
+            with _multi_access_temp_file(suffix=".zip") as temp_file:
+                _logger.info("Downloading Driver to %s", temp_file)
+                urllib.request.urlretrieve(download_url, temp_file)
+
+                with tempfile.TemporaryDirectory() as temp_folder:
+                    directory_to_extract_to = temp_folder
+
+                    _logger.info("Unzipping the downloaded file to %s", directory_to_extract_to)
+
+                    with zipfile.ZipFile(temp_file, "r") as zip_ref:
+                        zip_ref.extractall(directory_to_extract_to)
+
+                    _logger.info("Installing NI-DAQmx")
+                    for command in get_linux_installation_commands(
+                        directory_to_extract_to, distro.id(), distro.version(), release
+                    ):
+                        print("\nRunning command:", " ".join(command))
+                        subprocess.run(command, check=True)
+
+            # Check if the installation was successful
+            if not _get_daqmx_installed_version():
+                raise click.ClickException(
+                    "Failed to install NI-DAQmx driver. All installation commands ran successfully but the driver is not installed."
+                )
+            else:
+                print("NI-DAQmx driver installed successfully. Please reboot the system.")
+
+        except subprocess.CalledProcessError as e:
+            _logger.info("Failed to install NI-DAQmx driver.", exc_info=True)
+            raise click.ClickException(
+                f"An error occurred while installing the NI-DAQmx driver. Command returned non-zero exit status '{e.returncode}'."
+            ) from e
+        except urllib.error.URLError as e:
+            _logger.info("Failed to download NI-DAQmx driver.", exc_info=True)
+            raise click.ClickException(
+                f"Failed to download the NI-DAQmx driver.\nDetails: {e}"
+            ) from e
+        except Exception as e:
+            _logger.info("Failed to install NI-DAQmx driver.", exc_info=True)
+            raise click.ClickException(
+                f"Failed to install the NI-DAQmx driver.\nDetails: {e}"
+            ) from e
+    else:
+        raise NotImplementedError("This function is only supported on Linux.")
+
+
 def _ask_user_confirmation(user_message: str) -> bool:
     """
     Prompt for user confirmation
@@ -211,7 +319,10 @@ def _ask_user_confirmation(user_message: str) -> bool:
 
 
 def _confirm_and_upgrade_daqmx_driver(
-    latest_version: str, installed_version: str, download_url: str
+    latest_version: str,
+    installed_version: str,
+    download_url: str,
+    release: str,
 ) -> None:
     """
     Confirm with the user and upgrade the NI-DAQmx driver if necessary.
@@ -229,23 +340,81 @@ def _confirm_and_upgrade_daqmx_driver(
         f"Installed NI-DAQmx version is {installed_version}. Latest version available is {latest_version}. Do you want to upgrade?"
     )
     if is_upgrade:
-        _install_daqmx_driver(download_url)
+        if sys.platform.startswith("win"):
+            _install_daqmx_driver_windows_core(download_url)
+        elif sys.platform.startswith("linux"):
+            _install_daqmx_driver_linux_core(download_url, release)
 
 
-def _install_daqmx_windows_driver() -> None:
+def _install_daqmx_driver_windows() -> None:
     """
     Install the NI-DAQmx driver on Windows.
 
     """
     installed_version = _get_daqmx_installed_version()
-    download_url, latest_version = _get_driver_details()
+    download_url, latest_version, release, supported_os = _get_driver_details("Windows")
+    if not download_url:
+        raise click.ClickException(f"Failed to fetch the download url.")
+    if not release:
+        raise click.ClickException(f"Failed to fetch the release version string.")
+    else:
+        if installed_version and latest_version:
+            _confirm_and_upgrade_daqmx_driver(
+                latest_version, installed_version, download_url, release
+            )
+        else:
+            _install_daqmx_driver_windows_core(download_url)
+
+
+def _is_distribution_supported() -> None:
+    """
+    Raises an exception if the linux distribution and its version are not supported.
+
+    """
+    if sys.platform.startswith("linux"):
+        dist_name = distro.id()
+        dist_version = distro.version()
+
+        # For rhel, we only need the major version
+        if dist_name == "rhel":
+            dist_version = dist_version.split(".")[0]
+        dist_name_and_version = dist_name + " " + dist_version
+
+        download_url, latest_version, release, supported_os = _get_driver_details("Linux")
+
+        # Check if the platform is one of the supported ones
+        if dist_name_and_version in supported_os:
+            _logger.info(f"The platform is supported: {dist_name_and_version}")
+        else:
+            raise click.ClickException(f"The platform {dist_name_and_version} is not supported.")
+    else:
+        raise NotImplementedError("This function is only supported on Linux.")
+
+
+def _install_daqmx_driver_linux() -> None:
+    """
+    Install the NI-DAQmx driver on Linux.
+
+    """
+    try:
+        _is_distribution_supported()
+    except Exception as e:
+        raise click.ClickException(f"Distribution not supported.\nDetails: {e}") from e
+
+    installed_version = _get_daqmx_installed_version()
+    download_url, latest_version, release, supported_os = _get_driver_details("Linux")
+
     if not download_url:
         raise click.ClickException(f"Failed to fetch the download url.")
+    if not release:
+        raise click.ClickException(f"Failed to fetch the release version string.")
     else:
         if installed_version and latest_version:
-            _confirm_and_upgrade_daqmx_driver(latest_version, installed_version, download_url)
+            _confirm_and_upgrade_daqmx_driver(
+                latest_version, installed_version, download_url, release
+            )
         else:
-            _install_daqmx_driver(download_url)
+            _install_daqmx_driver_linux_core(download_url, release)
 
 
 def installdriver() -> None:
@@ -256,9 +425,14 @@ def installdriver() -> None:
     try:
         if sys.platform.startswith("win"):
             _logger.info("Windows platform detected")
-            _install_daqmx_windows_driver()
+            _install_daqmx_driver_windows()
+        elif sys.platform.startswith("linux"):
+            _logger.info("Linux platform detected")
+            _install_daqmx_driver_linux()
         else:
-            raise click.ClickException(f"The 'installdriver' command is supported only on Windows.")
+            raise click.ClickException(
+                f"The 'installdriver' command is supported only on Windows and Linux."
+            )
     except click.ClickException:
         raise
     except Exception as e:

generated/nidaqmx/_installer_metadata.json

@@ -1,8 +1,9 @@
 {
     "Windows": [
         {
-            "Version": "24.0.0",
-            "Location": "https://download.ni.com/support/nipkg/products/ni-d/ni-daqmx/24.0/online/ni-daqmx_24.0_online.exe",
+            "Version": "24.3.0",
+            "Release": "2024Q2",
+            "Location": "https://download.ni.com/support/nipkg/products/ni-d/ni-daqmx/24.3/online/ni-daqmx_24.3_online.exe",
             "supportedOS": [
                 "Windows 11",
                 "Windows Server 2022 64-bit",
@@ -14,11 +15,13 @@
     ],
     "Linux": [
         {
-            "Version": "24.0.0",
-            "Location": "https://download.ni.com/support/softlib/MasterRepository/LinuxDrivers2024Q1/NILinux2024Q1DeviceDrivers.zip",
+            "Version": "24.3.0",
+            "Release": "2024Q2",
+            "_comment": "Location url must be of the format 'https://download.ni.com/support/softlib/MasterRepository/LinuxDrivers<Release>/NILinux<Release>DeviceDrivers.zip'. Any change to the format will require a change in the code.",
+            "Location": "https://download.ni.com/support/softlib/MasterRepository/LinuxDrivers2024Q2/NILinux2024Q2DeviceDrivers.zip",
             "supportedOS": [
                 "ubuntu 20.04",
-                "ubuntu 22.00",
+                "ubuntu 22.04",
                 "rhel 8",
                 "rhel 9",
                 "opensuse 15.4",