Source code for nexusLIMS.extractors.plugins.profiles

"""Instrument profile modules for customizing extraction behavior.

This package contains instrument-specific profiles that customize metadata
extraction without modifying core extractor code. Profiles are automatically
discovered and registered during plugin initialization.

Each profile module should:
1. Import InstrumentProfile and get_profile_registry
2. Define parser/transformation functions
3. Create an InstrumentProfile instance
4. Register it via `get_profile_registry().register()`

Profile modules are loaded automatically - just add a new .py file to this
directory and it will be discovered during plugin initialization.

Examples
--------
Creating a new instrument profile (in profiles/my_instrument.py):

>>> from nexusLIMS.extractors.base import InstrumentProfile
>>> from nexusLIMS.extractors.profiles import get_profile_registry
>>>
>>> def custom_parser(metadata: dict, context) -> dict:
...     # Custom parsing logic
...     return metadata
>>>
>>> my_profile = InstrumentProfile(
...     instrument_id="My-Instrument-12345",
...     parsers={"custom": custom_parser},
... )
>>> get_profile_registry().register(my_profile)
"""

from __future__ import annotations

import importlib
import importlib.util
import logging
import pkgutil
from pathlib import Path

_logger = logging.getLogger(__name__)

__all__ = [
    "register_all_profiles",
]


[docs] def register_all_profiles() -> None: """ Auto-discover and register all instrument profiles. Loads profiles from two sources: 1. Built-in profiles (nexusLIMS/extractors/plugins/profiles/) 2. Local profiles (from NX_LOCAL_PROFILES_PATH env var, if set) Each profile module should register itself by calling get_profile_registry().register() at module level. This function is called automatically during extractor plugin discovery. Examples -------- >>> from nexusLIMS.extractors.plugins.profiles import register_all_profiles >>> register_all_profiles() >>> # All built-in and local profiles are now registered """ _logger.info("Discovering instrument profiles...") # Load built-in profiles package_path = Path(__file__).parent profile_count = _load_profiles_from_directory(package_path, __name__) # Load local profiles if configured try: from nexusLIMS import config # noqa: PLC0415 if config.settings.NX_LOCAL_PROFILES_PATH: local_path_obj = config.settings.NX_LOCAL_PROFILES_PATH _logger.info("Loading local profiles from: %s", local_path_obj) local_count = _load_profiles_from_directory( local_path_obj, module_prefix=None ) profile_count += local_count except Exception: _logger.debug( "NexusLIMS config unavailable; skipping local profile loading " "(built-in profiles still active)" ) _logger.info("Loaded %d total instrument profile modules", profile_count)
def _load_profiles_from_directory(directory: Path, module_prefix: str | None) -> int: """ Load all profile modules from a directory. Parameters ---------- directory Directory containing profile modules module_prefix Module name prefix for package-based imports (built-in profiles). If None, profiles are loaded as standalone files (local profiles). Returns ------- int Number of profiles successfully loaded Notes ----- Built-in profiles are loaded using Python's standard import system (pkgutil.walk_packages), while local profiles are loaded directly from files using importlib.util. This allows local profiles to exist outside the package structure without needing to be installed. """ profile_count = 0 if module_prefix is None: # Load local profiles as standalone Python files for profile_file in directory.glob("*.py"): # Skip private modules if profile_file.name.startswith("_"): continue try: # Create a unique module name for this local profile module_name = f"nexuslims_local_profile_{profile_file.stem}" # Load the profile file as a module spec = importlib.util.spec_from_file_location(module_name, profile_file) if spec is None or spec.loader is None: _logger.warning( "Failed to create module spec for local profile: %s", profile_file, ) continue module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) profile_count += 1 _logger.debug("Loaded local profile: %s", profile_file.name) except Exception as e: _logger.warning( "Failed to load local profile '%s': %s", profile_file, e, exc_info=True, ) else: # Load built-in profiles as package modules for _finder, module_name, _ispkg in pkgutil.walk_packages( [str(directory)], prefix=f"{module_prefix}.", ): # Skip __pycache__ and this __init__ module if "__pycache__" in module_name or module_name == module_prefix: continue try: # Import the module - this triggers profile registration importlib.import_module(module_name) profile_count += 1 _logger.debug("Loaded built-in profile module: %s", module_name) except Exception as e: _logger.warning( "Failed to load built-in profile module '%s': %s", module_name, e, exc_info=True, ) return profile_count