Source code for pymodaq_utils.plugin_rules

"""Static rules to guide the development of a PyMoDAQ instrument plugin (``pymodaq_plugins_*``).

The rules read the *source files* of a plugin package (``pyproject.toml``, python modules): nothing is imported, so
they also work when the instrument SDK or other dependencies are not installed. Each rule produces
:class:`Finding` objects with a stable code, a severity, a location and a hint about how to fix it.

Severities:

* ``error``: the plugin will not work or will not be seen by PyMoDAQ
* ``warning``: deprecated or suspicious, should be fixed
* ``todo``: the plugin is not finished (leftover of the template: TODO comments, placeholders, ``NotImplementedError``)

They are used by :mod:`pymodaq_utils.plugin_checks` but can be called directly::

    from pymodaq_utils.plugin_rules import check_package_sources
    for finding in check_package_sources('pymodaq_plugins_xxxx'):
        print(finding)

.. versionadded:: 5.3.1
"""
from __future__ import annotations

import ast
import importlib.util
import os
import re
import sys
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Optional

import toml

DOC = 'https://pymodaq.cnrs.fr/en/latest/developer_folder/instrument_plugins.html'
TEMPLATE_NAME = 'pymodaq_plugins_template'

IGNORED_MODULES = ('lextab', 'yacctab')  # files generated by parser generators (pycparser/ply)

PLACEHOLDER_DESCRIPTION = 'some word about your plugin'
PLACEHOLDER_AUTHORS = ('Name Surname', 'myname@test.fr')
PLACEHOLDER_NAMES_RE = re.compile(r'python_wrapper_file_of_your_instrument|PythonWrapperObjectOfYourInstrument|'
                                  r'your_method_to_\w+|a_method_or_atttribute_to_check_if_init')

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')

# features of [features] in pyproject.toml -> folders (relative to the package) holding the corresponding code
FEATURE_FOLDERS = {'instruments': ('daq_move_plugins', 'daq_viewer_plugins'),
                   'extensions': ('extensions',),
                   'models': ('models',),
                   'h5exporters': ('exporters',),
                   'scanners': ('scanners',)}


ANSI_STYLES = {'bold': '1', 'dim': '2', 'red': '31', 'green': '32', 'yellow': '33', 'cyan': '36'}


[docs] def colorize(text: str, style: str, enabled: bool = True) -> str: """Wrap a text in the ANSI escape codes of a style (bold, dim, red, green, yellow, cyan) if enabled""" if not enabled or not text: return text return f'\033[{ANSI_STYLES[style]}m{text}\033[0m'
[docs] def use_color(mode: str = 'auto', stream=None) -> bool: """Whether to color the output: mode is 'always', 'never' or 'auto' ``auto`` colors if the stream (stdout by default) is a terminal and the NO_COLOR environment variable is not set, or if FORCE_COLOR is set (see https://no-color.org). """ if mode == 'never': return False if mode == 'auto': if os.environ.get('NO_COLOR'): return False stream = stream or sys.stdout if not (os.environ.get('FORCE_COLOR') or (hasattr(stream, 'isatty') and stream.isatty())): return False if sys.platform == 'win32': os.system('') # enables the processing of the escape codes in the console return True
[docs] class Severity(str, Enum): ERROR = 'error' WARNING = 'warning' TODO = 'todo'
[docs] @dataclass(frozen=True) class Finding: """One thing found by a rule Attributes ---------- code: str stable identifier of the rule, PMQ1xx: packaging, PMQ2xx: unfinished plugin, PMQ3xx: plugin classes severity: Severity message: str hint: str how to fix it path: Path file concerned line: int line concerned column: int column concerned """ code: str severity: Severity message: str hint: str = '' path: Optional[Path] = None line: Optional[int] = None column: Optional[int] = None @property def location(self) -> str: if self.path is None: return '' if not self.line: return self.path.name return f'{self.path.name}:{self.line}' + (f':{self.column}' if self.column else '')
[docs] def format(self, color: bool = False) -> str: """Text of the finding, with colors for a terminal if ``color``""" style = {Severity.ERROR: 'red', Severity.WARNING: 'yellow', Severity.TODO: 'cyan'}[self.severity] where = f' {self.location}' if self.location else '' hint = colorize(f' -> {self.hint}', 'dim', color) if self.hint else '' return (f'{colorize(self.code, "bold", color)} {colorize(f"[{self.severity.value}]", style, color)}' f'{where}: {self.message}{hint}')
def __str__(self) -> str: return self.format()
[docs] def is_valid_unit(unit: str) -> bool: """Whether a unit is known from pint (with the units of PyMoDAQ if pymodaq_data is installed)""" try: from pymodaq_data import Unit except ImportError: # only pymodaq_utils is installed from pint import UnitRegistry Unit = UnitRegistry().Unit try: Unit(unit) except Exception: return False return True
[docs] def package_root(package: str) -> Optional[Path]: """Folder of the python package, found without importing it""" spec = importlib.util.find_spec(package) if spec is None or not spec.submodule_search_locations: return None return Path(list(spec.submodule_search_locations)[0])
[docs] def project_root(package: str) -> Optional[Path]: """Folder holding the pyproject.toml of the plugin: only exists for a source tree or an editable install""" root = package_root(package) for folder in (root.parents if root is not None else []): pyproject = folder / 'pyproject.toml' if pyproject.is_file(): try: name = toml.load(pyproject).get('project', {}).get('name', '') except toml.TomlDecodeError: return folder return folder if name.replace('-', '_').lower() == package else None return None
[docs] def syntax_error_finding(code: str, path: Path, error: SyntaxError) -> Finding: """Finding with the location (line and column) and the offending line of a syntax error""" text = (error.text or '').strip() hint = (f'"{text}" ' if text else '') + '(the other checks on this file are skipped until it is fixed)' return Finding(code, Severity.ERROR, f'syntax error: {error.msg}', hint, path, error.lineno, error.offset or None)
def _python_files(root: Path): """The python modules of the package: the generated files and the files that cannot be imported are left out""" for path in sorted(root.rglob('*.py')): if path.stem not in IGNORED_MODULES and '__pycache__' not in path.parts and path.stem.isidentifier(): yield path
[docs] def check_file_names(root: Path) -> list[Finding]: """Python files whose name is not a valid module name: they cannot be imported and should not be shipped""" return [Finding('PMQ112', Severity.WARNING, f"'{path.name}' is not a valid python module name", 'delete or rename it: it is often a conflicted copy made by a file synchronization tool ' '(Dropbox, OneDrive, Syncthing...) or an editor backup', path) for path in sorted(root.rglob('*.py')) if '__pycache__' not in path.parts and path.stem not in IGNORED_MODULES and not path.stem.isidentifier()]
# ------------------------------------------------------------------------------------------------------------------- # PMQ1xx: packaging # -------------------------------------------------------------------------------------------------------------------
[docs] def check_pyproject(package: str, project: Path) -> list[Finding]: """Rules on the ``pyproject.toml`` and the layout of the project""" path = project / 'pyproject.toml' try: content = toml.load(path) except toml.TomlDecodeError as e: return [Finding('PMQ100', Severity.ERROR, f'pyproject.toml cannot be parsed: {e}', path=path)] proj = content.get('project', {}) findings = [] def add(code, severity, message, hint='', needle=None): line = None if needle: for i, text in enumerate(path.read_text().splitlines(), 1): if needle in text: line = i break findings.append(Finding(code, severity, message, hint, path, line)) name = proj.get('name', '') normalized = name.replace('-', '_').lower() if not re.match(r'^pymodaq_plugins_\w+$', normalized): add('PMQ101', Severity.ERROR, f"project name '{name}' does not follow the convention", 'name it pymodaq_plugins_<manufacturer or topic>', 'name =') elif normalized == TEMPLATE_NAME and package != TEMPLATE_NAME: add('PMQ102', Severity.TODO, 'the project is still named after the template', 'set the name of your plugin in [project] name and rename the folder under src/ accordingly', 'name =') if normalized != package: add('PMQ103', Severity.ERROR, f"project name '{name}' and package folder '{package}' differ", 'rename the package folder under src/ (and its imports) or the project name', 'name =') if TEMPLATE_NAME in content.get('urls', {}).get('package-url', '') and package != TEMPLATE_NAME: add('PMQ104', Severity.TODO, '[urls] package-url still points to the template repository', 'use the url of your plugin repository', 'package-url') if proj.get('description', '') == PLACEHOLDER_DESCRIPTION: add('PMQ105', Severity.TODO, 'the description is still the template placeholder', 'describe the instruments of your plugin in a few words', 'description') for key in ('authors', 'maintainers'): if any(a.get('name') == PLACEHOLDER_AUTHORS[0] or a.get('email') == PLACEHOLDER_AUTHORS[1] for a in proj.get(key, [])): add('PMQ106', Severity.TODO, f'{key} still hold the template placeholder', f'fill the {key} with your name and email', f'{key} =') if not any(dep.replace(' ', '').lower().startswith('pymodaq') for dep in proj.get('dependencies', [])): add('PMQ107', Severity.WARNING, 'no pymodaq package in the dependencies', "add for instance 'pymodaq>=5.2.0' to [project] dependencies", 'dependencies') dynamic = proj.get('dynamic', []) if 'entry-points' not in dynamic and not proj.get('entry-points'): add('PMQ108', Severity.ERROR, 'no entry points are defined: PyMoDAQ will not find the plugin', "keep dynamic = ['version', 'urls', 'entry-points'] and the [tool.hatch.metadata.hooks.custom] table " "of the template", 'dynamic') features = content.get('features', {}) pkg_root = project / 'src' / package if (project / 'src' / package).is_dir() else project / package for feature, folders in FEATURE_FOLDERS.items(): has_code = any(f for folder in folders for f in _code_files(pkg_root / folder)) declared = bool(features.get(feature, False)) if has_code and not declared: add('PMQ109', Severity.WARNING, f"[features] {feature} is false but the package has code in {'/'.join(folders)}", f'set {feature} = true or its entry points will not be generated and PyMoDAQ will not see it', f'{feature} =') elif declared and not has_code: add('PMQ110', Severity.WARNING, f'[features] {feature} is true but nothing was found in {"/".join(folders)}', f'set {feature} = false if the plugin has no such feature', f'{feature} =') for filename in ('README.rst', 'LICENSE'): if not (project / filename).is_file(): findings.append(Finding('PMQ111', Severity.WARNING, f'{filename} is missing', 'add it at the project root', project)) return findings
def _code_files(folder: Path): if not folder.is_dir(): return [] return [p for p in folder.rglob('*.py') if p.stem != '__init__' and p.stem not in IGNORED_MODULES] # ------------------------------------------------------------------------------------------------------------------- # PMQ2xx: unfinished plugin (leftovers of the template) # ------------------------------------------------------------------------------------------------------------------- _TODO_RE = re.compile(r'(?i:#\s*todo\b)|\bTODO\b')
[docs] def check_leftovers(package: str, root: Path, project: Optional[Path] = None) -> list[Finding]: """Template leftovers: TODO comments, placeholders, ``NotImplementedError``, example modules""" findings = [] files = list(_python_files(root)) if project is not None: files.append(project / 'pyproject.toml') for path in files: if not path.is_file(): continue for i, text in enumerate(path.read_text(errors='replace').splitlines(), 1): if path.suffix == '.py' and _TODO_RE.search(text) and not text.strip().startswith('#!'): comment = text.strip().lstrip('#').strip() findings.append(Finding('PMQ201', Severity.TODO, comment[:90], 'complete it, then remove the TODO', path, i)) elif path.suffix == '.toml' and re.search(r'#\s*todo\b', text, re.IGNORECASE): findings.append(Finding('PMQ201', Severity.TODO, text.split('#', 1)[1].strip()[:90], 'complete it, then remove the TODO', path, i)) if path.suffix == '.py' and PLACEHOLDER_NAMES_RE.search(text) and package != TEMPLATE_NAME: findings.append(Finding('PMQ203', Severity.TODO, f'placeholder left from the template: ' f'{PLACEHOLDER_NAMES_RE.search(text).group(0)}', 'replace it with the wrapper and methods of your instrument', path, i)) if path.suffix == '.py': try: tree = ast.parse(path.read_text(errors='replace')) except SyntaxError as e: findings.append(syntax_error_finding('PMQ200', path, e)) continue for node in ast.walk(tree): if isinstance(node, ast.Raise) and _is_name(node.exc, 'NotImplementedError'): findings.append(Finding('PMQ202', Severity.TODO, 'a method still raises NotImplementedError', 'implement the method and remove this line', path, node.lineno)) if package != TEMPLATE_NAME: for path in _code_files(root): if path.stem.endswith('_Template') and re.match(r'daq_(move|\dDviewer|NDviewer)_Template$', path.stem): findings.append(Finding('PMQ204', Severity.TODO, 'example module of the template still present', 'rename it to your instrument or delete it', path)) config = root / 'resources' / 'config_template.toml' if config.is_file() and 'XXX' in config.read_text(): findings.append(Finding('PMQ205', Severity.TODO, 'config_template.toml still has the template title', 'describe the plugin settings and set a proper title', config, 1)) return findings
def _is_name(node: Optional[ast.AST], name: str) -> bool: if isinstance(node, ast.Call): node = node.func return isinstance(node, ast.Name) and node.id == name # ------------------------------------------------------------------------------------------------------------------- # PMQ3xx: plugin classes, from the source # ------------------------------------------------------------------------------------------------------------------- def _class_attrs(cls: ast.ClassDef) -> dict[str, ast.AST]: attrs = {} for node in cls.body: if isinstance(node, ast.Assign): for target in node.targets: if isinstance(target, ast.Name): attrs[target.id] = node.value elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name) and node.value is not None: attrs[node.target.id] = node.value return attrs def _literal(node: ast.AST): try: return ast.literal_eval(node) except (ValueError, SyntaxError, TypeError): return None def _base_names(cls: ast.ClassDef) -> list[str]: return [b.attr if isinstance(b, ast.Attribute) else getattr(b, 'id', '?') for b in cls.bases]
[docs] def check_plugin_source(path: Path, kind: str, class_name: str, static_fallback: bool = False) -> list[Finding]: """Rules on the source of a plugin module, without importing it Parameters ---------- path: Path the python file kind: str 'move' or the dimensionality of a viewer ('0D', '1D', '2D', 'ND') class_name: str name of the plugin class that is expected in the module static_fallback: bool if True also apply the rules that duplicate the checks done on the imported class (use it when the module cannot be imported because of a missing dependency) """ findings = [] try: tree = ast.parse(path.read_text(errors='replace')) except SyntaxError as e: return [syntax_error_finding('PMQ300', path, e)] for node in ast.walk(tree): if isinstance(node, (ast.Import, ast.ImportFrom)): module = node.module if isinstance(node, ast.ImportFrom) else ' '.join(a.name for a in node.names) if module and (module == 'pymodaq.daq_utils' or module.startswith('pymodaq.daq_utils.')): findings.append(Finding('PMQ307', Severity.ERROR, f"'{module}' does not exist anymore", 'import from pymodaq.utils (or pymodaq_utils / pymodaq_data / pymodaq_gui)', path, node.lineno)) cls = next((n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == class_name), None) if cls is None: # reported when loading the module, see plugin_testing return findings if ast.get_docstring(cls) is None: findings.append(Finding('PMQ310', Severity.WARNING, f'{class_name} has no docstring', 'document the instrument, the tested models and the needed drivers', path, cls.lineno)) if not any(isinstance(n, ast.If) and 'main' in ast.dump(n) and '__main__' in ast.dump(n) for n in tree.body): findings.append(Finding('PMQ308', Severity.WARNING, "no 'if __name__ == \"__main__\": main(__file__)' block", 'add it to be able to run and debug the plugin standalone', path)) attrs = _class_attrs(cls) methods = {n.name: n for n in cls.body if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef))} if kind == 'move': findings.extend(_check_move_source(path, cls, attrs, static_fallback)) mandatory = MANDATORY_MOVE_METHODS else: mandatory = MANDATORY_VIEWER_METHODS simple_bases = all(b in ('DAQ_Move_base', 'DAQ_Viewer_base') for b in _base_names(cls)) if static_fallback and simple_bases: for meth in mandatory: if meth not in methods: findings.append(Finding('PMQ306', Severity.ERROR, f'{class_name} should define {meth}()', 'implement it (see the template)', path, cls.lineno)) init = methods.get('ini_stage' if kind == 'move' else 'ini_detector') if init is not None: returns = [n for n in ast.walk(init) if isinstance(n, ast.Return)] if not returns or any(not isinstance(r.value, ast.Tuple) or len(r.value.elts) != 2 for r in returns): findings.append(Finding('PMQ309', Severity.WARNING, f'{init.name}() should return a tuple (info: str, initialized: bool)', 'return info, initialized', path, init.lineno)) return findings
def _check_move_source(path: Path, cls: ast.ClassDef, attrs: dict, static_fallback: bool) -> list[Finding]: findings = [] line = cls.lineno if '_epsilon' in attrs and '_epsilons' not in attrs: findings.append(Finding('PMQ302', Severity.WARNING, "'_epsilon' is deprecated", "use '_epsilons'", path, attrs['_epsilon'].lineno)) if 'stage_names' in attrs: findings.append(Finding('PMQ302', Severity.WARNING, "'stage_names' is deprecated", "use '_axis_names'", path, attrs['stage_names'].lineno)) if 'axis_names' in attrs and '_axis_names' not in attrs: findings.append(Finding('PMQ302', Severity.ERROR, "the axes are declared as 'axis_names', that is ignored", "use '_axis_names'", path, attrs['axis_names'].lineno)) dat = attrs.get('data_actuator_type') if dat is None or 'DataActuator' not in ast.dump(dat): findings.append(Finding('PMQ303', Severity.WARNING, 'data_actuator_type is not DataActuatorType.DataActuator', 'new plugins should exchange DataActuator objects: set ' 'data_actuator_type = DataActuatorType.DataActuator', path, dat.lineno if dat is not None else line)) if static_fallback: axes = _literal(attrs['_axis_names']) if '_axis_names' in attrs else None units = _literal(attrs['_controller_units']) if '_controller_units' in attrs else None if isinstance(units, str): units = [units] elif isinstance(units, dict): if isinstance(axes, list): findings.append(Finding('PMQ304', Severity.ERROR, '_controller_units is a dict but _axis_names a list', 'define them with the same type', path, attrs['_controller_units'].lineno)) units = list(units.values()) for unit in units if isinstance(units, list) else []: if not is_valid_unit(unit): findings.append(Finding('PMQ305', Severity.ERROR, f"unit '{unit}' is unknown from pint", 'use a unit known from pint, for instance mm or degree', path, attrs['_controller_units'].lineno)) return findings
[docs] def check_package_sources(package: str) -> list[Finding]: """All the static rules on an installed (or editable) plugin package: packaging, leftovers and plugin classes""" from pymodaq_utils.plugin_checks import find_plugin_modules # no import cycle at module level root = package_root(package) if root is None: return [Finding('PMQ100', Severity.ERROR, f'package {package} not found', 'install the plugin')] project = project_root(package) findings = check_pyproject(package, project) if project is not None else [] findings.extend(check_file_names(root)) findings.extend(check_leftovers(package, root, project)) for mod in find_plugin_modules(package): file = Path(importlib.util.find_spec(mod.import_path).origin) findings.extend(check_plugin_source(file, mod.kind, mod.class_name)) return findings