Base & Interfaces

biopro_sdk.plugin.base

Base plugin class for BioPro SDK.

Provides the main PluginBase class that all plugins should inherit from, with integrated state management and undo/redo support.

PluginBase

Bases: QWidget

Abstract base class for all BioPro plugins.

This class implements the BioProPlugin Protocol.

Source code in src/biopro_sdk/plugin/base.py

class PluginBase(QWidget):
    """Abstract base class for all BioPro plugins.

    This class implements the BioProPlugin Protocol.
    """

    """Abstract base class for all BioPro plugins.

    Provides:
    - Standard signals (status, state_changed, analysis_*, etc)
    - History management integration for undo/redo
    - State serialization/deserialization
    - Consistent plugin interface

    All plugins must inherit from this class and implement get_state() and set_state().

    Example:
        >>> class MyPlugin(PluginBase):
        ...     def __init__(self, plugin_id: str):
        ...         super().__init__(plugin_id)
        ...         self.state = MyState()
        ...         self.analyzer = MyAnalyzer(plugin_id)
        ...         # Build UI...
        ...
        ...     def get_state(self) -> PluginState:
        ...         return self.state
        ...
        ...     def set_state(self, state: PluginState) -> None:
        ...         self.state = state
        ...         self.update_ui()
    """

    def __init__(self, plugin_id: str, parent=None):
        """Initialize the plugin.

        Args:
            plugin_id: Unique identifier for this plugin
            parent: Parent QWidget (usually None for top-level plugins)
        """
        super().__init__(parent)
        self.signals = PluginSignals()

        self.plugin_id = plugin_id

        # Initialize context-aware logger
        from .logging import get_logger

        self.logger = get_logger(f"plugin.{plugin_id}", plugin_id)

        self._history = None
        self._current_state = None

        # Connect to global theme engine
        theme_manager.theme_changed.connect(self._apply_theme_styles)

    @property
    def history(self):
        """Lazy-loaded HistoryManager to avoid circular dependencies."""
        if not hasattr(self, "_history") or self._history is None:
            try:
                from biopro.core.history_manager import HistoryManager

                self._history = HistoryManager()
            except ImportError:

                class MockHistoryManager:
                    def get_module_history(self, *args, **kwargs):
                        class MockHistory:
                            def push(self, *args):
                                pass

                            def undo(self):
                                return None

                            def redo(self):
                                return None

                            @property
                            def undo_stack(self):
                                return [1, 2]

                            @property
                            def redo_stack(self):
                                return []

                        return MockHistory()

                self._history = MockHistoryManager()
        return self._history

    @history.setter
    def history(self, value):
        self._history = value

    def publish_event(self, topic: str, data: Any = None) -> None:
        """Publish an event to the Central Event Bus."""
        CentralEventBus.publish(topic, data)

    def subscribe_event(self, topic: str, callback: Callable[[Any], None]) -> None:
        """Subscribe to an event on the Central Event Bus."""
        CentralEventBus.subscribe(topic, callback)

    def __getattr__(self, name: str):
        """Proxy signal access to self.signals for convenience.

        Allows using `self.state_changed.emit()` instead of
        `self.signals.state_changed.emit()`.
        """
        if hasattr(self, "signals") and hasattr(self.signals, name):
            return getattr(self.signals, name)
        raise AttributeError(f"'{type(self).__name__}' object has no attribute '{name}'")

    @abstractmethod
    def get_state(self) -> PluginState:
        """Return the current analysis state.

        Must be implemented by subclasses. Called by BioPro core to capture
        plugin state for undo/redo and workflow persistence.

        Returns:
            Current PluginState instance
        """
        pass

    @abstractmethod
    def set_state(self, state: PluginState) -> None:
        """Set the plugin state and update UI accordingly.

        Must be implemented by subclasses. Called by BioPro core to restore
        plugin state during undo/redo and workflow loading.

        Args:
            state: PluginState instance to restore
        """
        pass

    def push_state(self) -> None:
        """Save current state to undo history.

        Call this whenever the user makes a destructive edit (e.g., changing
        a parameter, drawing on an image). BioPro will emit state_changed signal
        and automatically capture this state for undo/redo.
        """
        state_dict = self.get_state().to_dict()
        self.history.get_module_history(self.plugin_id).push(state_dict)
        self.state_changed.emit()

    def undo(self) -> None:
        """Undo to previous state."""
        history = self.history.get_module_history(self.plugin_id)
        prev_state_dict = history.undo()
        if prev_state_dict:
            state = self.get_state().__class__.from_dict(prev_state_dict)
            self.set_state(state)
            self.state_changed.emit()

    def redo(self) -> None:
        """Redo to next state."""
        history = self.history.get_module_history(self.plugin_id)
        next_state_dict = history.redo()
        if next_state_dict:
            state = self.get_state().__class__.from_dict(next_state_dict)
            self.set_state(state)
            self.state_changed.emit()

    def can_undo(self) -> bool:
        """Check if undo is available.

        Returns:
            True if there are states to undo to
        """
        history = self.history.get_module_history(self.plugin_id)
        return len(history.undo_stack) > 1

    def can_redo(self) -> bool:
        """Check if redo is available.

        Returns:
            True if there are states to redo to
        """
        history = self.history.get_module_history(self.plugin_id)
        return len(history.redo_stack) > 0

    # ── Resource Lifecycle (RAII) ────────────────────────────────────

    def cleanup(self) -> None:
        """Automatic Resource Cleansing.

        Uses ResourceInspector to break references to heavy objects in both
        the plugin instance and its state. This helps the GC reclaim memory
        immediately when a tab is closed.
        """
        try:
            from biopro.core.resource_inspector import ResourceInspector
        except ImportError:

            class ResourceInspector:
                @staticmethod
                def get_heavy_resources(*args, **kwargs):
                    return []

        # 1. Clean PluginState
        state = self.get_state()
        if state:
            heavy_in_state = ResourceInspector.get_heavy_resources(state)
            for name, _ in heavy_in_state:
                setattr(state, name, None)

        # 2. Clean Plugin Instance attributes
        heavy_in_instance = ResourceInspector.get_heavy_resources(self)
        for name, _ in heavy_in_instance:
            if name != "state":  # Don't wipe the state container itself
                setattr(self, name, None)

        self.state_changed.emit()
        self.status_message.emit("Resources released.")

    def shutdown(self) -> None:
        """Default shutdown. Subclasses should override if managing GPU models."""
        pass

    def _apply_theme_styles(self) -> None:
        """Re-applies theme-aware styles to the plugin.

        Subclasses should override this if they have complex custom styling.
        """
        # Force a re-evaluation of the base stylesheet
        self.setStyleSheet(f"background: {Colors.BG_DARKEST}; color: {Colors.FG_PRIMARY};")

        # Propagate to children if they have their own theme handlers
        from PyQt6.QtWidgets import QWidget

        for child in self.findChildren(QWidget):
            if hasattr(child, "_apply_theme_styles") and child is not self:
                child._apply_theme_styles()
            elif hasattr(child, "refresh_styles"):
                child.refresh_styles()

            # Re-evaluate local stylesheets to pick up {Colors.VAR} changes
            if child.styleSheet():
                child.setStyleSheet(child.styleSheet())
            child.update()

    def closeEvent(self, event):
        """Triggers automatic cleanup when the plugin widget is closed."""
        self.cleanup()
        super().closeEvent(event)

    # ── Protocol Compliance ───────────────────────────────────────────

    @property
    def __version__(self) -> str:
        return "1.0.0"  # Default for base plugins

    @property
    def __plugin_id__(self) -> str:
        return self.plugin_id

    @classmethod
    def get_panel_class(cls):
        """Standard protocol requirement: return the class itself."""
        return cls

history property writable

Lazy-loaded HistoryManager to avoid circular dependencies.

__getattr__(name)

Proxy signal access to self.signals for convenience.

Allows using self.state_changed.emit() instead of self.signals.state_changed.emit().

Source code in src/biopro_sdk/plugin/base.py

def __getattr__(self, name: str):
    """Proxy signal access to self.signals for convenience.

    Allows using `self.state_changed.emit()` instead of
    `self.signals.state_changed.emit()`.
    """
    if hasattr(self, "signals") and hasattr(self.signals, name):
        return getattr(self.signals, name)
    raise AttributeError(f"'{type(self).__name__}' object has no attribute '{name}'")

__init__(plugin_id, parent=None)

Initialize the plugin.

Parameters:

Name	Type	Description	Default
plugin_id	str	Unique identifier for this plugin	required
parent		Parent QWidget (usually None for top-level plugins)	None

Source code in src/biopro_sdk/plugin/base.py

def __init__(self, plugin_id: str, parent=None):
    """Initialize the plugin.

    Args:
        plugin_id: Unique identifier for this plugin
        parent: Parent QWidget (usually None for top-level plugins)
    """
    super().__init__(parent)
    self.signals = PluginSignals()

    self.plugin_id = plugin_id

    # Initialize context-aware logger
    from .logging import get_logger

    self.logger = get_logger(f"plugin.{plugin_id}", plugin_id)

    self._history = None
    self._current_state = None

    # Connect to global theme engine
    theme_manager.theme_changed.connect(self._apply_theme_styles)

can_redo()

Check if redo is available.

Returns:

Type	Description
bool	True if there are states to redo to

Source code in src/biopro_sdk/plugin/base.py

def can_redo(self) -> bool:
    """Check if redo is available.

    Returns:
        True if there are states to redo to
    """
    history = self.history.get_module_history(self.plugin_id)
    return len(history.redo_stack) > 0

can_undo()

Check if undo is available.

Returns:

Type	Description
bool	True if there are states to undo to

Source code in src/biopro_sdk/plugin/base.py

def can_undo(self) -> bool:
    """Check if undo is available.

    Returns:
        True if there are states to undo to
    """
    history = self.history.get_module_history(self.plugin_id)
    return len(history.undo_stack) > 1

cleanup()

Automatic Resource Cleansing.

Uses ResourceInspector to break references to heavy objects in both the plugin instance and its state. This helps the GC reclaim memory immediately when a tab is closed.

Source code in src/biopro_sdk/plugin/base.py

def cleanup(self) -> None:
    """Automatic Resource Cleansing.

    Uses ResourceInspector to break references to heavy objects in both
    the plugin instance and its state. This helps the GC reclaim memory
    immediately when a tab is closed.
    """
    try:
        from biopro.core.resource_inspector import ResourceInspector
    except ImportError:

        class ResourceInspector:
            @staticmethod
            def get_heavy_resources(*args, **kwargs):
                return []

    # 1. Clean PluginState
    state = self.get_state()
    if state:
        heavy_in_state = ResourceInspector.get_heavy_resources(state)
        for name, _ in heavy_in_state:
            setattr(state, name, None)

    # 2. Clean Plugin Instance attributes
    heavy_in_instance = ResourceInspector.get_heavy_resources(self)
    for name, _ in heavy_in_instance:
        if name != "state":  # Don't wipe the state container itself
            setattr(self, name, None)

    self.state_changed.emit()
    self.status_message.emit("Resources released.")

closeEvent(event)

Triggers automatic cleanup when the plugin widget is closed.

Source code in src/biopro_sdk/plugin/base.py

def closeEvent(self, event):
    """Triggers automatic cleanup when the plugin widget is closed."""
    self.cleanup()
    super().closeEvent(event)

get_panel_class() classmethod

Standard protocol requirement: return the class itself.

Source code in src/biopro_sdk/plugin/base.py

@classmethod
def get_panel_class(cls):
    """Standard protocol requirement: return the class itself."""
    return cls

get_state() abstractmethod

Return the current analysis state.

Must be implemented by subclasses. Called by BioPro core to capture plugin state for undo/redo and workflow persistence.

Returns:

Type	Description
PluginState	Current PluginState instance

Source code in src/biopro_sdk/plugin/base.py

@abstractmethod
def get_state(self) -> PluginState:
    """Return the current analysis state.

    Must be implemented by subclasses. Called by BioPro core to capture
    plugin state for undo/redo and workflow persistence.

    Returns:
        Current PluginState instance
    """
    pass

publish_event(topic, data=None)

Publish an event to the Central Event Bus.

Source code in src/biopro_sdk/plugin/base.py

def publish_event(self, topic: str, data: Any = None) -> None:
    """Publish an event to the Central Event Bus."""
    CentralEventBus.publish(topic, data)

push_state()

Save current state to undo history.

Call this whenever the user makes a destructive edit (e.g., changing a parameter, drawing on an image). BioPro will emit state_changed signal and automatically capture this state for undo/redo.

Source code in src/biopro_sdk/plugin/base.py

def push_state(self) -> None:
    """Save current state to undo history.

    Call this whenever the user makes a destructive edit (e.g., changing
    a parameter, drawing on an image). BioPro will emit state_changed signal
    and automatically capture this state for undo/redo.
    """
    state_dict = self.get_state().to_dict()
    self.history.get_module_history(self.plugin_id).push(state_dict)
    self.state_changed.emit()

redo()

Redo to next state.

Source code in src/biopro_sdk/plugin/base.py

def redo(self) -> None:
    """Redo to next state."""
    history = self.history.get_module_history(self.plugin_id)
    next_state_dict = history.redo()
    if next_state_dict:
        state = self.get_state().__class__.from_dict(next_state_dict)
        self.set_state(state)
        self.state_changed.emit()

set_state(state) abstractmethod

Set the plugin state and update UI accordingly.

Must be implemented by subclasses. Called by BioPro core to restore plugin state during undo/redo and workflow loading.

Parameters:

Name	Type	Description	Default
state	PluginState	PluginState instance to restore	required

Source code in src/biopro_sdk/plugin/base.py

@abstractmethod
def set_state(self, state: PluginState) -> None:
    """Set the plugin state and update UI accordingly.

    Must be implemented by subclasses. Called by BioPro core to restore
    plugin state during undo/redo and workflow loading.

    Args:
        state: PluginState instance to restore
    """
    pass

shutdown()

Default shutdown. Subclasses should override if managing GPU models.

Source code in src/biopro_sdk/plugin/base.py

def shutdown(self) -> None:
    """Default shutdown. Subclasses should override if managing GPU models."""
    pass

subscribe_event(topic, callback)

Subscribe to an event on the Central Event Bus.

Source code in src/biopro_sdk/plugin/base.py

def subscribe_event(self, topic: str, callback: Callable[[Any], None]) -> None:
    """Subscribe to an event on the Central Event Bus."""
    CentralEventBus.subscribe(topic, callback)

undo()

Undo to previous state.

Source code in src/biopro_sdk/plugin/base.py

def undo(self) -> None:
    """Undo to previous state."""
    history = self.history.get_module_history(self.plugin_id)
    prev_state_dict = history.undo()
    if prev_state_dict:
        state = self.get_state().__class__.from_dict(prev_state_dict)
        self.set_state(state)
        self.state_changed.emit()

biopro_sdk.plugin.interfaces

BioPro Plugin Interfaces.

Defines the formal structural requirements for BioPro analysis modules using PEP 544 Protocols.

BioProPlugin

Bases: Protocol

Structural protocol for a valid BioPro Plugin module.

Any Python module or class that implements this protocol can be loaded as a BioPro plugin.

Source code in src/biopro_sdk/plugin/interfaces.py

@runtime_checkable
class BioProPlugin(Protocol):
    """Structural protocol for a valid BioPro Plugin module.

    Any Python module or class that implements this protocol can be
    loaded as a BioPro plugin.
    """

    def get_panel_class(self) -> type[QWidget]:
        """Returns the main QWidget class for the plugin's UI."""
        ...

    @property
    def __version__(self) -> str:
        """The version string of the plugin (e.g. '1.2.3')."""
        ...

    @property
    def __plugin_id__(self) -> str:
        """The unique identifier for the plugin (matches manifest.json id)."""
        ...

    def cleanup(self) -> None:
        """Release instance-specific resources (e.g. caches, local memory).

        Called when a specific instance of the plugin panel is closed.
        """
        ...

    def shutdown(self) -> None:
        """Release global/module resources (e.g. GPU models, open files).

        Called when the application exists or the module is unloaded.
        """
        ...

__plugin_id__ property

The unique identifier for the plugin (matches manifest.json id).

__version__ property

The version string of the plugin (e.g. '1.2.3').

cleanup()

Release instance-specific resources (e.g. caches, local memory).

Called when a specific instance of the plugin panel is closed.

Source code in src/biopro_sdk/plugin/interfaces.py

def cleanup(self) -> None:
    """Release instance-specific resources (e.g. caches, local memory).

    Called when a specific instance of the plugin panel is closed.
    """
    ...

get_panel_class()

Returns the main QWidget class for the plugin's UI.

Source code in src/biopro_sdk/plugin/interfaces.py

def get_panel_class(self) -> type[QWidget]:
    """Returns the main QWidget class for the plugin's UI."""
    ...

shutdown()

Release global/module resources (e.g. GPU models, open files).

Called when the application exists or the module is unloaded.

Source code in src/biopro_sdk/plugin/interfaces.py

def shutdown(self) -> None:
    """Release global/module resources (e.g. GPU models, open files).

    Called when the application exists or the module is unloaded.
    """
    ...