feature/connection_utils

pyproject.toml

@@ -11,6 +11,7 @@ package-url = 'https://github.com/PyMoDAQ/pymodaq_plugins_utils'
 [project.optional-dependencies]
 serial = [
     "pyvisa",
+    "pyserial",
 ]
 
 [project]

src/pymodaq_plugins_utils/hardware/__init__.py

@@ -0,0 +1,33 @@
+"""Hardware discovery caches for PyMoDAQ plugins.
+
+Each backend (:mod:`~pymodaq_utils.hardware.visa`,
+:mod:`~pymodaq_utils.hardware.serial_ports`) queries the OS exactly once per
+process. Subsequent calls reuse the cached result, so plugin startup cost is
+paid at most once regardless of how many plugins share the same backend.
+
+Quick reference::
+
+    # VISA-based plugin (Newport, Thorlabs, PI, ...)
+    from pymodaq_utils.hardware.visa import list_serial_resources
+    ports = list_serial_resources()
+
+    # pyserial-based plugin (Arduino, Ocean Optics, ...)
+    from pymodaq_utils.hardware.serial_ports import list_resources
+    ports = list_resources()
+
+    # After hot-plugging a device
+    from pymodaq_utils.hardware import invalidate_all_caches
+    invalidate_all_caches()
+"""
+from .visa import invalidate_cache as _invalidate_visa
+from .serial_ports import invalidate_cache as _invalidate_serial
+
+
+def invalidate_all_caches() -> None:
+    """Clear both the VISA and serial discovery caches.
+
+    Call this after hot-plugging a device so the next call to any
+    ``list_*`` function re-discovers the current set of instruments.
+    """
+    _invalidate_visa()
+    _invalidate_serial()

src/pymodaq_plugins_utils/hardware/base.py

@@ -0,0 +1,59 @@
+
+class HardwareCache:
+    """Base class for process-lifetime hardware discovery caches.
+
+    Each subclass calls its backend (pyvisa, pyserial, …) exactly once per
+    process. The result is stored as a class variable and reused by every
+    caller, regardless of which plugin package triggered the first call.
+
+    Subclasses must override :meth:`_fetch` and :meth:`list_resources`.
+    Call :meth:`invalidate_cache` to force re-discovery, for example after
+    hot-plugging a device.
+
+    Example — defining a new backend::
+
+        class MyCache(HardwareCache):
+            _cache = None
+
+            @classmethod
+            def _fetch(cls):
+                return some_expensive_os_call()
+
+            @classmethod
+            def list_resources(cls) -> list[str]:
+                return [item.id for item in cls._get_cache()]
+    """
+
+    _cache = None
+
+    @classmethod
+    def _fetch(cls):
+        """Perform the actual hardware discovery.
+
+        Called at most once per process. Must return a value that can be
+        stored and reused (list, dict, …). Should catch all exceptions and
+        return an empty container so that callers never need to guard against
+        missing backends.
+        """
+        raise NotImplementedError
+
+    @classmethod
+    def _get_cache(cls):
+        """Return the cached discovery result, populating it on first call."""
+        if cls._cache is None:
+            cls._cache = cls._fetch()
+        return cls._cache
+
+    @classmethod
+    def invalidate_cache(cls) -> None:
+        """Clear the cache so the next call to any list_* method re-discovers.
+
+        Use this after hot-plugging a device or when the set of available
+        instruments may have changed since process startup.
+        """
+        cls._cache = None
+
+    @classmethod
+    def list_resources(cls) -> list[str]:
+        """Return a list of connectable resource strings for this backend."""
+        raise NotImplementedError

src/pymodaq_plugins_utils/hardware/serial_ports.py

@@ -0,0 +1,63 @@
+"""pyserial hardware discovery cache.
+
+Wraps :mod:`serial.tools.list_ports` to enumerate available serial ports
+exactly once per process. ``pyserial`` is an optional dependency: if it is
+not installed, all functions return empty lists and a warning is logged.
+
+Typical usage in a plugin::
+
+    from pymodaq_utils.hardware.serial_ports import list_resources
+
+    ports = list_resources()  # e.g. ['/dev/ttyUSB0', 'COM3']
+
+After hot-plugging a device, refresh the cache with::
+
+    from pymodaq_utils.hardware.serial_ports import invalidate_cache
+    invalidate_cache()
+"""
+import pymodaq_utils.logger as logger_module
+from .base import HardwareCache
+
+logger = logger_module.set_logger(logger_module.get_module_name(__file__))
+
+
+class SerialPortsCache(HardwareCache):
+    _cache = None
+
+    @classmethod
+    def _fetch(cls) -> list:
+        try:
+            from serial.tools.list_ports import comports
+            return list(comports())
+        except ImportError:
+            logger.warning('pyserial is not installed — serial port discovery unavailable. '
+                           'Install it with: pip install pyserial')
+            return []
+        except Exception as e:
+            logger.warning(f'Serial port discovery failed: {e}')
+            return []
+
+    @classmethod
+    def list_resources(cls) -> list[str]:
+        """Serial port device strings (e.g. '/dev/ttyUSB0', 'COM3')."""
+        return [p.device for p in cls._get_cache()]
+
+    @classmethod
+    def list_port_descriptions(cls) -> list[str]:
+        """Human-readable descriptions for each serial port."""
+        return [p.description for p in cls._get_cache()]
+
+
+def list_resources() -> list[str]:
+    """Serial port device strings (e.g. '/dev/ttyUSB0', 'COM3')."""
+    return SerialPortsCache.list_resources()
+
+
+def list_port_descriptions() -> list[str]:
+    """Human-readable descriptions for each serial port."""
+    return SerialPortsCache.list_port_descriptions()
+
+
+def invalidate_cache() -> None:
+    """Clear the serial port cache so the next call re-discovers."""
+    SerialPortsCache.invalidate_cache()

src/pymodaq_plugins_utils/hardware/visa.py

@@ -0,0 +1,84 @@
+"""VISA hardware discovery cache.
+
+Wraps :mod:`pyvisa` to enumerate available VISA resources exactly once per
+process. ``pyvisa`` is an optional dependency: if it is not installed, or no
+VISA backend is found, all functions return empty lists and a warning is logged.
+
+Typical usage in a plugin::
+
+    from pymodaq_utils.hardware.visa import list_serial_resources
+
+    ports = list_serial_resources()  # e.g. ['ASRL/dev/ttyUSB0::INSTR']
+
+After hot-plugging a device, refresh the cache with::
+
+    from pymodaq_utils.hardware.visa import invalidate_cache
+    invalidate_cache()
+"""
+import pymodaq_utils.logger as logger_module
+from .base import HardwareCache
+
+logger = logger_module.set_logger(logger_module.get_module_name(__file__))
+
+
+class VisaCache(HardwareCache):
+    _cache = None
+
+    @classmethod
+    def _fetch(cls) -> dict:
+        try:
+            import pyvisa
+            rm = pyvisa.ResourceManager()
+            info = dict(rm.list_resources_info())
+            rm.close()
+            return info
+        except ImportError:
+            logger.warning('pyvisa is not installed — VISA resource discovery unavailable. '
+                           'Install it with: pip install pyvisa pyvisa-py')
+            return {}
+        except Exception as e:
+            logger.warning(f'VISA resource discovery failed: {e}')
+            return {}
+
+    @classmethod
+    def list_resources(cls) -> list[str]:
+        """All available VISA resource strings (e.g. 'GPIB0::5::INSTR')."""
+        return list(cls._get_cache().keys())
+
+    @classmethod
+    def list_serial_resources(cls) -> list[str]:
+        """ASRL (serial-over-VISA) resource strings only.
+
+        Linux:   'ASRL/dev/ttyUSB0::INSTR'
+        Windows: 'ASRL3::INSTR'
+        """
+        return [r for r in cls._get_cache() if r.startswith('ASRL')]
+
+    @classmethod
+    def list_resource_aliases(cls) -> list[str]:
+        """Human-readable aliases where available (e.g. 'COM3' on Windows)."""
+        return [i.alias for i in cls._get_cache().values() if i.alias]
+
+
+def list_resources() -> list[str]:
+    """All available VISA resource strings (e.g. 'GPIB0::5::INSTR', 'TCPIP0::...')."""
+    return VisaCache.list_resources()
+
+
+def list_serial_resources() -> list[str]:
+    """ASRL (serial-over-VISA) resource strings only.
+
+    Linux:   ``'ASRL/dev/ttyUSB0::INSTR'``
+    Windows: ``'ASRL3::INSTR'``
+    """
+    return VisaCache.list_serial_resources()
+
+
+def list_resource_aliases() -> list[str]:
+    """Human-readable aliases where available (e.g. ``'COM3'`` on Windows)."""
+    return VisaCache.list_resource_aliases()
+
+
+def invalidate_cache() -> None:
+    """Clear the VISA resource cache so the next call re-discovers."""
+    VisaCache.invalidate_cache()