"""Acceptance checks for PyMoDAQ instrument plugins (``pymodaq_plugins_*``), without any dependency on pytest.
No hardware is needed: the checks are static / import-level. Use them in a plugin repository with::
from pymodaq_utils.plugin_testing import PluginPackageChecks
class TestMyPlugin(PluginPackageChecks):
package_name = 'pymodaq_plugins_myinstrument' # optional, read from the nearest pyproject.toml if omitted
Every check is also exposed as a plain function returning a list of problems (empty if all is fine), see
``check_move_class``, ``check_viewer_class`` and ``check_package_layout``.
To get a report without pytest, for any installed plugin package::
from pymodaq_utils.plugin_checks import check_plugin_package
print(check_plugin_package('pymodaq_plugins_mock'))
.. versionadded:: 5.3.1
"""
from __future__ import annotations
import importlib
import importlib.util
import inspect
import pkgutil
import re
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
import toml
from pymodaq_utils.utils import get_entrypoints
from pymodaq_utils.plugin_rules import (PLACEHOLDER_NAMES_RE, Finding, Severity, check_file_names, check_leftovers,
check_plugin_source, check_pyproject, colorize, is_valid_unit, package_root,
project_root, use_color)
VIEWER_DIMS = ('0D', '1D', '2D', 'ND')
ENTRYPOINT_GROUPS = ('pymodaq.plugins', 'pymodaq.instruments')
OPTIONAL_ENTRYPOINT_GROUPS = ('pymodaq.extensions', 'pymodaq.models', 'pymodaq.pid_models',
'pymodaq.scanners', 'pymodaq.h5exporters')
# methods that have to be overridden by the plugin (they are abstract in the base classes, but these are not
# enforced as the bases are QObjects)
MANDATORY_MOVE_METHODS = ('ini_stage', 'get_actuator_value', 'close') # stop_motion has a default that does nothing
MANDATORY_VIEWER_METHODS = ('ini_detector', 'grab_data', 'stop', 'close')
# files generated by parser generators (pycparser/ply) that can end up within the plugin packages
IGNORED_MODULES = ('lextab', 'yacctab')
# findings of which severities make the checks fail, for each ``fail_on`` level
FAIL_LEVELS = {'error': (Severity.ERROR,),
'warning': (Severity.ERROR, Severity.WARNING),
'todo': (Severity.ERROR, Severity.WARNING, Severity.TODO)}
PACKAGE_NAME_RE = re.compile(r'^pymodaq_plugins_\w+$')
[docs]
@dataclass(frozen=True)
class PluginModule:
"""A plugin module found in a plugin package."""
package: str
module_name: str # e.g. daq_move_Mock
kind: str # 'move' or '0D', '1D', '2D', 'ND' for viewers
@property
def import_path(self) -> str:
if self.kind == 'move':
return f'{self.package}.daq_move_plugins.{self.module_name}'
return f'{self.package}.daq_viewer_plugins.plugins_{self.kind}.{self.module_name}'
@property
def prefix(self) -> str:
return 'daq_move_' if self.kind == 'move' else f'daq_{self.kind}viewer_'
@property
def name(self) -> str:
return self.module_name[len(self.prefix):]
@property
def class_name(self) -> str:
return f'DAQ_Move_{self.name}' if self.kind == 'move' else f'DAQ_{self.kind}Viewer_{self.name}'
def __str__(self):
return self.module_name
def _submodule_names(package_path: str) -> list[str]:
"""Names of the modules (not packages) found in a package that can be imported, without importing them."""
return sorted(m.name for m in pkgutil.iter_modules([package_path]) if not m.ispkg and m.name.isidentifier())
[docs]
def find_plugin_modules(package: str) -> list[PluginModule]:
"""Find all plugin modules of a plugin package from the file system (no plugin module is imported).
Modules that don't follow the naming convention (``daq_move_*``, ``daq_{N}viewer_*``) are reported by
:func:`check_package_layout`.
"""
found = []
try:
move_pkg = importlib.import_module(f'{package}.daq_move_plugins')
found.extend(PluginModule(package, name, 'move') for name in _submodule_names(list(move_pkg.__path__)[0])
if name.startswith('daq_move_'))
except ModuleNotFoundError:
pass
for dim in VIEWER_DIMS:
try:
viewer_pkg = importlib.import_module(f'{package}.daq_viewer_plugins.plugins_{dim}')
except ModuleNotFoundError:
continue
found.extend(PluginModule(package, name, dim) for name in _submodule_names(list(viewer_pkg.__path__)[0])
if name.startswith(f'daq_{dim}viewer_'))
return found
[docs]
def guess_package_name(start: Path) -> str:
"""Get the plugin package name from the nearest pyproject.toml above ``start``
The name of the (single) ``pymodaq_plugins_*`` folder of the project is used if there is one, as the project name
of the pyproject.toml may not have been modified yet, the project name otherwise.
"""
for folder in [start, *start.parents]:
pyproject = folder / 'pyproject.toml'
if pyproject.is_file():
for base in (folder / 'src', folder):
packages = [p for p in base.glob('pymodaq_plugins_*') if (p / '__init__.py').is_file()]
if len(packages) == 1:
return packages[0].name
return toml.load(pyproject)['project']['name'].replace('-', '_')
raise FileNotFoundError(f'No pyproject.toml found above {start}, define package_name explicitly')
def _is_overridden(klass: type, base: type, method: str) -> bool:
return getattr(klass, method, None) is not None and getattr(klass, method) is not getattr(base, method, None)
[docs]
def is_installed(package: str) -> bool:
"""Whether an entry point of the instrument groups points to the package (so the plugin is installed)"""
return package in {ep.value.split(':')[0] for group in ENTRYPOINT_GROUPS for ep in get_entrypoints(group)}
[docs]
def check_package_layout(package: str, check_entry_points: bool = True) -> list[str]:
"""Package name, entry points, mandatory sub-modules and naming of the plugin modules
The entry points only exist once the plugin is installed: set ``check_entry_points`` to False before.
"""
problems = []
if not PACKAGE_NAME_RE.match(package):
problems.append(f"Package name '{package}' should be of the form pymodaq_plugins_<name>")
try:
pkg = importlib.import_module(package)
except Exception as e:
return problems + [f'Package {package} cannot be imported: {e!r}']
for attr in ('config', '__version__'):
if not hasattr(pkg, attr):
problems.append(f"{package}/__init__.py should define '{attr}'")
try:
importlib.import_module(f'{package}.utils')
except Exception as e:
problems.append(f'{package}.utils cannot be imported: {e!r}')
if check_entry_points and not is_installed(package):
problems.append(f'No entry point of groups {ENTRYPOINT_GROUPS} points to {package}: is the plugin installed, '
f'and does its pyproject.toml declare the entry points?')
for group in OPTIONAL_ENTRYPOINT_GROUPS:
for ep in get_entrypoints(group):
if ep.value.split(':')[0].split('.')[0] == package:
try:
ep.load()
except Exception as e:
problems.append(f'Entry point {group}:{ep.name} cannot be loaded: {e!r}')
# module naming: everything in daq_move_plugins / plugins_{N}D should follow the convention
plugin_pkgs = [(f'{package}.daq_move_plugins', 'daq_move_')] + \
[(f'{package}.daq_viewer_plugins.plugins_{dim}', f'daq_{dim}viewer_') for dim in VIEWER_DIMS]
for pkg_path, prefix in plugin_pkgs:
try:
sub = importlib.import_module(pkg_path)
except ModuleNotFoundError:
continue
for name in _submodule_names(list(sub.__path__)[0]):
if not name.startswith(prefix) and name not in IGNORED_MODULES:
problems.append(f"Module {pkg_path}.{name} should be named '{prefix}<Name>'")
return problems
def _check_params(klass: type) -> list[str]:
problems = []
params = klass.params
if not isinstance(params, list) or not all(isinstance(p, dict) for p in params):
return [f'{klass.__name__}.params should be a list of dict']
try:
from pyqtgraph.parametertree import Parameter
except ImportError: # only pymodaq_utils is installed
return problems
try:
tree = Parameter.create(name='settings', type='group', children=params)
except Exception as e:
return [f'{klass.__name__}.params cannot build a Parameter tree: {e!r}']
names = [child.name() for child in tree.children()]
duplicates = {n for n in names if names.count(n) > 1}
if duplicates:
problems.append(f'{klass.__name__}.params has duplicated top level names: {sorted(duplicates)}')
return problems
def _check_methods(klass: type, base: type, mandatory) -> list[str]:
problems = [f'{klass.__name__} should override {m}()' for m in mandatory if not _is_overridden(klass, base, m)]
if inspect.isabstract(klass): # in case a base ever becomes a real ABC
problems.append(f'{klass.__name__} is abstract: {sorted(klass.__abstractmethods__)}')
return problems
[docs]
def check_move_class(klass: type) -> list[str]:
"""Checks on an actuator plugin class"""
from pymodaq.control_modules.move_utility_classes import DAQ_Move_base # only needed when checking the classes
if not (inspect.isclass(klass) and issubclass(klass, DAQ_Move_base)):
return [f'{klass} should derive from DAQ_Move_base']
problems = _check_methods(klass, DAQ_Move_base, MANDATORY_MOVE_METHODS)
problems += _check_params(klass)
axis_names = klass.get_class_axis_names()
units = klass._controller_units
if isinstance(units, str):
unit_list = [units]
elif isinstance(units, (list, dict)):
unit_list = list(units.values()) if isinstance(units, dict) else units
if not isinstance(units, type(axis_names)):
problems.append(f'_controller_units ({type(units).__name__}) should be of the same type as '
f'_axis_names ({type(axis_names).__name__}), or a single str')
elif len(units) != len(axis_names):
problems.append('_controller_units should define one unit per axis')
else:
return problems + [f'_controller_units has an invalid type: {type(units).__name__}']
for unit in unit_list:
if not is_valid_unit(unit):
problems.append(f"Unit '{unit}' in _controller_units is unknown from pint")
epsilons = klass._epsilons
if epsilons is not None and not isinstance(epsilons, (int, float)):
if not isinstance(epsilons, type(axis_names)) or len(epsilons) != len(axis_names):
problems.append('_epsilons should be a number or have the same type and length as _axis_names')
if isinstance(axis_names, list) and len(set(axis_names)) != len(axis_names):
problems.append('_axis_names contains duplicates')
if klass.is_multiaxes is False and len(axis_names) > 1:
problems.append('_axis_names defines several axes but is_multiaxes is False')
return problems
[docs]
def check_viewer_class(klass: type) -> list[str]:
"""Checks on a detector plugin class"""
from pymodaq.control_modules.viewer_utility_classes import DAQ_Viewer_base
if not (inspect.isclass(klass) and issubclass(klass, DAQ_Viewer_base)):
return [f'{klass} should derive from DAQ_Viewer_base']
problems = _check_methods(klass, DAQ_Viewer_base, MANDATORY_VIEWER_METHODS)
problems += _check_params(klass)
return problems
[docs]
class PluginLoadError(Exception):
"""A plugin module cannot be imported or does not define the expected class
Attributes
----------
environmental: bool
True if the failure comes from the environment rather than from the plugin code: a third party module or
vendor SDK/driver that is not installed or not available on this OS (``clr``, ``pyvisa``, ``windll``...).
The plugin classes cannot be checked then, but this is not necessarily a defect of the plugin.
syntax_error: bool
True if the module has a syntax error: it is located and reported by the static rules (PMQ300)
placeholder: bool
True if the module imports the fake wrapper of the template, that was not replaced yet: this is a todo
"""
def __init__(self, message: str, environmental: bool = False, placeholder: bool = False,
syntax_error: bool = False):
super().__init__(message)
self.environmental = environmental
self.placeholder = placeholder
self.syntax_error = syntax_error
def _is_environmental(error: Exception, package: str) -> bool:
"""Whether an import error is due to the environment (third party module, SDK, OS) and not the plugin/pymodaq"""
if isinstance(error, (SyntaxError, NameError, AttributeError, TypeError)):
return False
if isinstance(error, ImportError):
name = getattr(error, 'name', None) or ''
return not (name == 'pymodaq' or name.startswith('pymodaq.') or name.split('.')[0] == package)
return isinstance(error, (OSError, KeyError))
[docs]
def load_plugin_class(plugin_module: PluginModule) -> type:
"""Import a plugin module and return its plugin class
Raises
------
PluginLoadError
with an explicit message if the module cannot be imported or lacks the expected class
"""
try:
module = importlib.import_module(plugin_module.import_path)
except Exception as e:
placeholder = isinstance(e, ImportError) and bool(PLACEHOLDER_NAMES_RE.search(getattr(e, 'name', None) or ''))
raise PluginLoadError(f'{plugin_module.import_path} cannot be imported: {e!r}',
not placeholder and _is_environmental(e, plugin_module.package), placeholder,
isinstance(e, SyntaxError)) from e
klass = getattr(module, plugin_module.class_name, None)
if klass is None:
raise PluginLoadError(f'{plugin_module.import_path} should define a class named {plugin_module.class_name}')
return klass
[docs]
@dataclass
class CheckResult:
"""Outcome of the checks on one item (the package itself or one of its plugin modules)
``problems`` are defects of the plugin found on the imported classes. ``warnings`` mean the item could not be
fully checked, for instance because a third party module or a vendor SDK is not available in the current
environment. ``findings`` come from the static rules of :mod:`pymodaq_utils.plugin_rules`.
"""
item: str
problems: list[str] = field(default_factory=list)
warnings: list[str] = field(default_factory=list)
findings: list[Finding] = field(default_factory=list)
[docs]
def failing(self, fail_on: str = 'error', strict_imports: bool = False) -> list[str]:
"""What makes this item fail
Parameters
----------
fail_on: str
the findings of this level and above fail, see :data:`FAIL_LEVELS`
strict_imports: bool
if True the modules that could not be imported because of the environment also fail
"""
severities = FAIL_LEVELS[fail_on]
return (self.problems + (self.warnings if strict_imports else []) +
[str(f) for f in self.findings if f.severity in severities])
@property
def ok(self) -> bool:
return not self.failing()
[docs]
def check_plugin_module(plugin_module: PluginModule) -> CheckResult:
"""Load a plugin module and run the checks relevant for its kind, plus the static rules on its source"""
result = CheckResult(str(plugin_module))
spec = importlib.util.find_spec(plugin_module.import_path)
path = Path(spec.origin) if spec is not None and spec.origin else None
loaded = False
try:
klass = load_plugin_class(plugin_module)
loaded = True
except PluginLoadError as e:
if e.syntax_error: # located and explained by the static rules below (PMQ300)
pass
elif e.placeholder: # not finished: reported as a todo, controlled by fail_on
result.findings.append(Finding('PMQ203', Severity.TODO, str(e), 'replace the import of the fake wrapper '
'of the template with the one of your instrument',
Path(spec.origin) if spec is not None and spec.origin else None))
else:
(result.warnings if e.environmental else result.problems).append(str(e))
if loaded:
result.problems = check_move_class(klass) if plugin_module.kind == 'move' else check_viewer_class(klass)
if plugin_module.kind == 'move':
from pymodaq.control_modules.move_utility_classes import DAQ_Move_base
if not _is_overridden(klass, DAQ_Move_base, 'stop_motion'):
result.findings.append(Finding(
'PMQ311', Severity.WARNING,
'stop_motion() is not overridden: the Stop of the actuator does nothing',
'implement it with the stop command of your controller (or ignore this if it cannot be stopped)',
path, None))
if path is not None:
result.findings.extend(check_plugin_source(path, plugin_module.kind, plugin_module.class_name,
static_fallback=not loaded))
return result
[docs]
@dataclass
class PluginReport:
"""Result of all the checks on a plugin package, see :func:`check_plugin_package`
``fail_on`` sets what makes the report fail, see :data:`FAIL_LEVELS`. The other findings are just listed.
"""
package: str
results: list[CheckResult] = field(default_factory=list)
fail_on: str = 'error'
strict_imports: bool = False
@property
def ok(self) -> bool:
return not self.failures
@property
def failures(self) -> list[CheckResult]:
return [res for res in self.results if res.failing(self.fail_on, self.strict_imports)]
@property
def todos(self) -> list[Finding]:
return [f for res in self.results for f in res.findings if f.severity == Severity.TODO]
[docs]
def to_dict(self) -> dict:
return {'package': self.package, 'ok': self.ok, 'fail_on': self.fail_on,
'results': {res.item: {'problems': res.problems, 'warnings': res.warnings,
'findings': [str(f) for f in res.findings]}
for res in self.results}}
def __str__(self) -> str:
return self.format()
[docs]
def check_package_sources(package: str, project: Optional[Path] = None) -> CheckResult:
"""Static rules on the packaging (pyproject.toml...) and on the leftovers of the template
``project`` is the folder of the pyproject.toml, found from the package if not given (only possible if its name
matches the package one).
"""
result = CheckResult('package sources')
root = package_root(package)
if root is None:
result.problems.append(f'package {package} not found')
return result
project = project or project_root(package) # None if installed from a wheel: no pyproject.toml to check
if project is not None:
result.findings.extend(check_pyproject(package, project))
result.findings.extend(check_file_names(root))
result.findings.extend(check_leftovers(package, root, project))
return result
[docs]
def check_plugin_package(package: str, fail_on: str = 'error', strict_imports: bool = False,
project: Optional[Path] = None, check_entry_points: bool = True) -> PluginReport:
"""Run all the checks on an installed plugin package and return a report, without using pytest
Parameters
----------
package: str
name of the plugin package, for instance pymodaq_plugins_mock
fail_on: str
'error' (default), 'warning' (also fails on warnings) or 'todo' (also on the unfinished parts: TODO comments,
placeholders...), see :data:`FAIL_LEVELS`
strict_imports: bool
if True the modules that cannot be imported because of the environment (missing third party module, SDK or
OS specific error) also fail instead of being reported as warnings
project: Path
folder of the pyproject.toml of the plugin, if not found from the package
check_entry_points: bool
set to False for a plugin that is not installed yet (see :func:`resolve_target`)
Examples
--------
>>> report = check_plugin_package('pymodaq_plugins_mock')
>>> print(report)
>>> report.ok
"""
if fail_on not in FAIL_LEVELS:
raise ValueError(f'fail_on should be one of {list(FAIL_LEVELS)}')
layout = CheckResult('package layout', check_package_layout(package, check_entry_points))
if not check_entry_points:
layout.warnings.append('the plugin is not installed: its entry points were not checked, run pip install -e . '
'in the plugin folder to check them')
report = PluginReport(package, [layout, check_package_sources(package, project)], fail_on, strict_imports)
try:
modules = find_plugin_modules(package)
except Exception as e:
report.results.append(CheckResult('plugin discovery', [f'Cannot list the plugin modules: {e!r}']))
return report
report.results.extend(check_plugin_module(mod) for mod in modules)
located = {(f.path, f.line) for res in report.results for f in res.findings if f.code == 'PMQ300'}
for res in report.results: # a syntax error of a plugin module is reported once, with the module
res.findings = [f for f in res.findings if not (f.code == 'PMQ200' and (f.path, f.line) in located)]
return report
[docs]
def resolve_target(target: Optional[str] = None) -> tuple[str, Optional[Path]]:
"""Find the plugin package to check from a package name or from a folder, without needing it to be installed
Parameters
----------
target: str
either the name of an importable package, or a folder: the plugin repository (holding the pyproject.toml and
the package, in ``src`` or not) or the package folder itself. The current folder if not given.
Returns
-------
str: the package name, as found in the folder name (so it does not depend on the project name in the
pyproject.toml, that may not have been modified yet)
Path: the folder of the pyproject.toml, if any. The folder holding the package is added to ``sys.path``
Raises
------
ValueError
if no, or several, ``pymodaq_plugins_*`` package are found in the folder
"""
path = Path(target) if target is not None else Path.cwd()
if target is not None and not path.is_dir():
return target, None # a package name
path = path.resolve()
def packages_in(base: Path) -> list[Path]:
return sorted(p for p in base.glob('pymodaq_plugins_*') if (p / '__init__.py').is_file())
if path.name.startswith('pymodaq_plugins_') and (path / '__init__.py').is_file():
folders = [path]
else:
folders = packages_in(path / 'src') or packages_in(path)
if len(folders) != 1:
raise ValueError(f'{len(folders)} pymodaq_plugins_* packages found in {path}: expected one. Give the folder of '
f'the plugin repository or of the package, or the name of an installed package')
folder = folders[0]
sys.path.insert(0, str(folder.parent))
project = next((d for d in (folder.parent, *folder.parent.parents) if (d / 'pyproject.toml').is_file()), None)
return folder.name, project
[docs]
def resolve_file(path: str | Path) -> PluginModule:
"""Find the plugin module of a python file of a plugin package, without importing it
The folder holding the package is added to ``sys.path``.
Raises
------
ValueError
if the file is not in a ``pymodaq_plugins_*`` package, or not at the location, or does not have the name, of
an instrument module: ``daq_move_plugins/daq_move_<Name>.py`` or
``daq_viewer_plugins/plugins_<N>D/daq_<N>viewer_<Name>.py``
"""
path = Path(path).resolve()
if path.suffix != '.py':
raise ValueError(f'{path.name} is not a python file')
folder = next((p for p in path.parents if p.name.startswith('pymodaq_plugins_') and (p / '__init__.py').is_file()),
None)
if folder is None:
raise ValueError(f'{path} is not in a pymodaq_plugins_* package')
parts = path.relative_to(folder).parts
module_name = path.stem
kind = None
if len(parts) == 2 and parts[0] == 'daq_move_plugins' and module_name.startswith('daq_move_'):
kind = 'move'
elif len(parts) == 3 and parts[0] == 'daq_viewer_plugins':
dim = parts[1][len('plugins_'):] if parts[1].startswith('plugins_') else None
if dim in VIEWER_DIMS and module_name.startswith(f'daq_{dim}viewer_'):
kind = dim
if kind is None or not module_name.isidentifier():
raise ValueError(f'{path.name} is not an instrument module: it should be named daq_move_<Name>.py in the '
f'daq_move_plugins folder or daq_<N>viewer_<Name>.py in the daq_viewer_plugins/plugins_<N>D '
f'folder (N = 0, 1, 2 or N) and be a valid python module name')
sys.path.insert(0, str(folder.parent))
return PluginModule(folder.name, module_name, kind)
[docs]
def check_plugin_file(path: str | Path, fail_on: str = 'error', strict_imports: bool = False) -> PluginReport:
"""Run the checks on a single instrument module of a plugin package (and the todos of this file)
The checks on the package (``pyproject.toml``, entry points...) are not done, see :func:`check_plugin_package`.
The package does not need to be installed.
"""
if fail_on not in FAIL_LEVELS:
raise ValueError(f'fail_on should be one of {list(FAIL_LEVELS)}')
module = resolve_file(path)
result = check_plugin_module(module)
file = Path(path).resolve()
result.findings.extend(f for f in check_leftovers(module.package, package_root(module.package))
if f.path is not None and f.path.resolve() == file and f.code != 'PMQ200')
return PluginReport(module.package, [result], fail_on, strict_imports)
[docs]
def main(argv=None) -> int:
"""Print the report of a plugin package, for the developer: the ``check_plugin`` command (or
``python -m pymodaq_utils.plugin_checks``)
Returns 0 if the checks pass, 1 otherwise (so that it can also be used in a script or a CI).
"""
import argparse
parser = argparse.ArgumentParser(prog='check_plugin',
description='Check a PyMoDAQ plugin package and print a report')
parser.add_argument('package', nargs='?', default=None,
help='name of an installed plugin package, or folder of the plugin repository or package '
'(that does not need to be installed), or a single instrument module (python file) of '
'a plugin package. Default: the current folder')
parser.add_argument('--fail-on', choices=list(FAIL_LEVELS), default='error',
help="findings of this level and above give a non zero exit code ('todo': everything)")
parser.add_argument('--strict-imports', action='store_true',
help='fail on the modules that cannot be imported (missing dependency or SDK)')
parser.add_argument('-v', '--verbose', action='store_true', help='list every todo')
parser.add_argument('--color', choices=['auto', 'always', 'never'], default='auto',
help='color the report: auto (default) if the output is a terminal and NO_COLOR is not set')
args = parser.parse_args(argv)
if args.package is not None and Path(args.package).is_file(): # a single instrument module
try:
report = check_plugin_file(args.package, args.fail_on, args.strict_imports)
except ValueError as e:
parser.error(str(e))
print(report.format(verbose=args.verbose, color=use_color(args.color)))
return 0 if report.ok else 1
try:
package, project = resolve_target(args.package)
except ValueError as e:
parser.error(str(e))
report = check_plugin_package(package, args.fail_on, args.strict_imports, project,
check_entry_points=is_installed(package) or project is None)
print(report.format(verbose=args.verbose, color=use_color(args.color)))
return 0 if report.ok else 1
if __name__ == '__main__':
sys.exit(main())