From 3bd77d898ae41205de267ffad1da3137a1aa9973 Mon Sep 17 00:00:00 2001 From: Ashwola Date: Fri, 22 May 2026 10:10:03 +0200 Subject: [PATCH] adding pyvisa/pyserial cache utils --- pyproject.toml | 1 + .../hardware/__init__.py | 33 +++ src/pymodaq_plugins_utils/hardware/base.py | 59 +++++ .../hardware/serial_ports.py | 63 +++++ src/pymodaq_plugins_utils/hardware/visa.py | 84 ++++++ tests/hardware_test.py | 249 ++++++++++++++++++ 6 files changed, 489 insertions(+) create mode 100644 src/pymodaq_plugins_utils/hardware/base.py create mode 100644 src/pymodaq_plugins_utils/hardware/serial_ports.py create mode 100644 src/pymodaq_plugins_utils/hardware/visa.py create mode 100644 tests/hardware_test.py diff --git a/pyproject.toml b/pyproject.toml index 56f16ee..888910c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,6 +11,7 @@ package-url = 'https://github.com/PyMoDAQ/pymodaq_plugins_utils' [project.optional-dependencies] serial = [ "pyvisa", + "pyserial", ] [project] diff --git a/src/pymodaq_plugins_utils/hardware/__init__.py b/src/pymodaq_plugins_utils/hardware/__init__.py index e69de29..808fbd0 100644 --- a/src/pymodaq_plugins_utils/hardware/__init__.py +++ b/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() diff --git a/src/pymodaq_plugins_utils/hardware/base.py b/src/pymodaq_plugins_utils/hardware/base.py new file mode 100644 index 0000000..3e88a73 --- /dev/null +++ b/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 diff --git a/src/pymodaq_plugins_utils/hardware/serial_ports.py b/src/pymodaq_plugins_utils/hardware/serial_ports.py new file mode 100644 index 0000000..004f754 --- /dev/null +++ b/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() diff --git a/src/pymodaq_plugins_utils/hardware/visa.py b/src/pymodaq_plugins_utils/hardware/visa.py new file mode 100644 index 0000000..b0a10e5 --- /dev/null +++ b/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() diff --git a/tests/hardware_test.py b/tests/hardware_test.py new file mode 100644 index 0000000..f48c1ac --- /dev/null +++ b/tests/hardware_test.py @@ -0,0 +1,249 @@ +import pytest + +from pymodaq_plugins_utils.hardware.base import HardwareCache +from pymodaq_plugins_utils.hardware import visa, serial_ports, invalidate_all_caches + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +class _CountingCache(HardwareCache): + """Minimal concrete subclass used to test base-class caching logic.""" + _cache = None + fetch_count = 0 + + @classmethod + def _fetch(cls): + cls.fetch_count += 1 + return ['resource_a', 'resource_b'] + + @classmethod + def list_resources(cls): + return list(cls._get_cache()) + + @classmethod + def reset(cls): + cls._cache = None + cls.fetch_count = 0 + + +# --------------------------------------------------------------------------- +# HardwareCache base behaviour +# --------------------------------------------------------------------------- + +class TestHardwareCacheBase: + + def setup_method(self): + _CountingCache.reset() + + def test_fetch_called_once(self): + # Core guarantee: no matter how many times list_resources() is called, + # the expensive OS-level _fetch() runs exactly once per process lifetime. + _CountingCache.list_resources() + _CountingCache.list_resources() + assert _CountingCache.fetch_count == 1 + + def test_returns_correct_data(self): + assert _CountingCache.list_resources() == ['resource_a', 'resource_b'] + + def test_invalidate_triggers_refetch(self): + # After invalidation the cache is empty, so the next call must + # re-run _fetch() — this is the hot-plug refresh path. + _CountingCache.list_resources() + _CountingCache.invalidate_cache() + _CountingCache.list_resources() + assert _CountingCache.fetch_count == 2 + + def test_cache_is_none_after_invalidate(self): + # Validates the internal reset so _get_cache() knows to call _fetch() + # again on the next access (tested separately in test_invalidate_triggers_refetch). + _CountingCache.list_resources() + _CountingCache.invalidate_cache() + assert _CountingCache._cache is None + + def test_subclasses_have_independent_caches(self): + # Each subclass stores its result in its own _cache class variable. + # Invalidating one must not affect the other — otherwise a Newport + # plugin resetting its cache would silently clear an Arduino plugin's cache. + class CacheA(HardwareCache): + _cache = None + + @classmethod + def _fetch(cls): + return ['a'] + + @classmethod + def list_resources(cls): + return list(cls._get_cache()) + + class CacheB(HardwareCache): + _cache = None + + @classmethod + def _fetch(cls): + return ['b'] + + @classmethod + def list_resources(cls): + return list(cls._get_cache()) + + assert CacheA.list_resources() == ['a'] + assert CacheB.list_resources() == ['b'] + CacheA.invalidate_cache() + assert CacheA._cache is None + assert CacheB._cache is not None # CacheB untouched + + +# --------------------------------------------------------------------------- +# visa module +# --------------------------------------------------------------------------- + +class TestVisaModule: + + def setup_method(self): + # Start each test with a clean cache so tests are isolated. + visa.VisaCache.invalidate_cache() + + def test_list_resources_returns_list(self): + # The module must never raise, even when pyvisa is not installed. + assert isinstance(visa.list_resources(), list) + + def test_warns_when_pyvisa_absent(self, monkeypatch): + # When pyvisa is missing the user should get an explicit warning, + # not a silent empty list that looks like "no devices connected". + import builtins + real_import = builtins.__import__ + + def block_pyvisa(name, *args, **kwargs): + if name == 'pyvisa': + raise ImportError('pyvisa blocked for test') + return real_import(name, *args, **kwargs) + + warned = [] + monkeypatch.setattr(builtins, '__import__', block_pyvisa) + monkeypatch.setattr(visa.logger, 'warning', lambda msg: warned.append(msg)) + visa.VisaCache.invalidate_cache() + + visa.list_resources() + assert any('pyvisa' in msg for msg in warned) + + def test_list_serial_resources_returns_list(self): + assert isinstance(visa.list_serial_resources(), list) + + def test_list_resource_aliases_returns_list(self): + assert isinstance(visa.list_resource_aliases(), list) + + def test_serial_resources_are_subset_of_all(self): + # list_serial_resources() is a filtered view of list_resources(); + # every ASRL entry must also appear in the full resource list. + # Skipped when pyvisa is absent: an empty list would make this pass + # vacuously without testing anything. + pytest.importorskip('pyvisa') + all_r = visa.list_resources() + serial_r = visa.list_serial_resources() + assert all(r in all_r for r in serial_r) + + def test_serial_resources_start_with_asrl(self): + # VISA serial resources always begin with 'ASRL' by the VISA standard. + # Skipped when pyvisa is absent for the same reason as above. + pytest.importorskip('pyvisa') + for r in visa.list_serial_resources(): + assert r.startswith('ASRL') + + def test_fetch_called_once_across_multiple_functions(self, monkeypatch): + # Calling list_resources(), list_serial_resources(), and + # list_resource_aliases() in sequence must trigger only one backend + # query — the whole point of this module. + call_count = [] + original_fetch = visa.VisaCache._fetch.__func__ + + @classmethod + def counting_fetch(cls): + call_count.append(1) + return original_fetch(cls) + + monkeypatch.setattr(visa.VisaCache, '_fetch', counting_fetch) + visa.VisaCache.invalidate_cache() + + visa.list_resources() + visa.list_serial_resources() + visa.list_resource_aliases() + + assert len(call_count) == 1 + + +# --------------------------------------------------------------------------- +# serial_ports module +# --------------------------------------------------------------------------- + +class TestSerialPortsModule: + + def setup_method(self): + serial_ports.SerialPortsCache.invalidate_cache() + + def test_list_resources_returns_list(self): + # The module must never raise, even when pyserial is not installed. + assert isinstance(serial_ports.list_resources(), list) + + def test_warns_when_pyserial_absent(self, monkeypatch): + # Same contract as the visa equivalent: missing backend → warning, not silent empty list. + import builtins + real_import = builtins.__import__ + + def block_serial(name, *args, **kwargs): + if name == 'serial.tools.list_ports': + raise ImportError('pyserial blocked for test') + return real_import(name, *args, **kwargs) + + warned = [] + monkeypatch.setattr(builtins, '__import__', block_serial) + monkeypatch.setattr(serial_ports.logger, 'warning', lambda msg: warned.append(msg)) + serial_ports.SerialPortsCache.invalidate_cache() + + serial_ports.list_resources() + assert any('pyserial' in msg for msg in warned) + + def test_list_port_descriptions_returns_list(self): + assert isinstance(serial_ports.list_port_descriptions(), list) + + def test_resources_and_descriptions_same_length(self): + # Both lists are derived from the same cached port objects, so they + # must always be parallel (index N in one matches index N in the other). + # Skipped when pyserial is absent: both would be empty and the assertion + # passes vacuously without testing anything. + pytest.importorskip('serial') + assert len(serial_ports.list_resources()) == len(serial_ports.list_port_descriptions()) + + def test_fetch_called_once_across_multiple_functions(self, monkeypatch): + # Same single-fetch guarantee as the visa module. + call_count = [] + original_fetch = serial_ports.SerialPortsCache._fetch.__func__ + + @classmethod + def counting_fetch(cls): + call_count.append(1) + return original_fetch(cls) + + monkeypatch.setattr(serial_ports.SerialPortsCache, '_fetch', counting_fetch) + serial_ports.SerialPortsCache.invalidate_cache() + + serial_ports.list_resources() + serial_ports.list_port_descriptions() + + assert len(call_count) == 1 + + +# --------------------------------------------------------------------------- +# Package-level helper +# --------------------------------------------------------------------------- + +def test_invalidate_all_caches_clears_both(): + # Populate both caches first so the assertion is meaningful. + visa.list_resources() + serial_ports.list_resources() + + invalidate_all_caches() + + assert visa.VisaCache._cache is None + assert serial_ports.SerialPortsCache._cache is None