Custom Function Management

Module: openhcs.processing.custom_functions.manager Status: STABLE

—

Overview

Custom functions are user-defined processing functions that integrate seamlessly with OpenHCS pipelines. The custom function management system provides atomic updates, module injection, and signal-based synchronization to keep UI and CLI in sync.

Quick Reference

from openhcs.processing.custom_functions.manager import CustomFunctionManager

# Create manager
manager = CustomFunctionManager()

# Add custom function
function_code = '''
@numpy_memory
def my_custom_filter(image: np.ndarray, sigma: float = 1.0) -> np.ndarray:
    """Apply custom Gaussian filter."""
    return gaussian_filter(image, sigma=sigma)
'''

manager.add_function('my_custom_filter', function_code)

# Function is now available in pipelines
step = FunctionStep(function='my_custom_filter', parameters={'sigma': 2.0})

End-to-End Flow

The custom function lifecycle involves four stages:

  1. Code Editing: User writes function code in GUI editor

  2. Validation: Function signature and decorators validated

  3. Registration: Function added to global registry

  4. Synchronization: UI and CLI updated via signals

# Stage 1: User edits code in CustomFunctionManagerDialog
dialog = CustomFunctionManagerDialog()
dialog.show()

# Stage 2: Validation on save
# - Check for required decorators (@numpy_memory, @cupy_memory, etc.)
# - Validate function signature (first arg must be image array)
# - Verify return type annotation

# Stage 3: Atomic registration
manager.add_function(name, code)
# - Execute code in isolated namespace
# - Extract function object
# - Inject module name
# - Register in global function registry

# Stage 4: Signal emission
# - function_added signal emitted
# - UI updates function list
# - CLI sees new function immediately

Atomic Update Mechanism

Custom function updates are atomic to prevent partial registration:

def add_function(self, name: str, code: str) -> None:
    """Atomically add or update custom function."""

    # Execute code in isolated namespace
    namespace = {}
    try:
        exec(code, namespace)
    except Exception as e:
        raise ValidationError(f"Code execution failed: {e}")

    # Extract function object
    if name not in namespace:
        raise ValidationError(f"Function '{name}' not found in code")

    func = namespace[name]

    # Validate function
    validation_result = validate_function(func)
    if not validation_result.is_valid:
        raise ValidationError(validation_result.errors)

    # Atomic update: Only register if all validation passes
    register_function(func, backend='openhcs')
    self._save_to_disk(name, code)

    # Emit signal for UI synchronization
    self.function_added.emit(name, func)

Key insight: If any step fails, the entire operation is rolled back. No partial registration.

Module Injection

Custom functions executed via exec() don’t have __module__ set properly. Module injection fixes this:

# Without module injection
exec(code, namespace)
func = namespace['my_function']
print(func.__module__)  # '__main__' or None (WRONG!)

# With module injection
exec(code, namespace)
func = namespace['my_function']

# Inject proper module name
if not hasattr(func, '__module__') or func.__module__ in (None, '__main__'):
    func.__module__ = 'openhcs.processing.custom_functions'

print(func.__module__)  # 'openhcs.processing.custom_functions' (CORRECT!)

Why this matters:

  • Function registry uses __module__ for organization

  • Code generation needs proper module paths

  • Serialization requires valid module names

Signal-Based Synchronization

The manager emits signals to keep UI and CLI synchronized:

class CustomFunctionManager(QObject):
    """Manager with Qt signals for UI synchronization."""

    # Signals
    function_added = pyqtSignal(str, object)    # (name, function)
    function_removed = pyqtSignal(str)          # (name,)
    function_updated = pyqtSignal(str, object)  # (name, function)

    def add_function(self, name, code):
        # ... validation and registration ...

        # Emit signal
        self.function_added.emit(name, func)

    def remove_function(self, name):
        # ... unregistration ...

        # Emit signal
        self.function_removed.emit(name)

UI components connect to these signals:

class CustomFunctionManagerDialog(QDialog):
    def __init__(self):
        self.manager = CustomFunctionManager()

        # Connect signals
        self.manager.function_added.connect(self._on_function_added)
        self.manager.function_removed.connect(self._on_function_removed)

    def _on_function_added(self, name, func):
        """Update function list when function added."""
        self.function_list.addItem(name)

    def _on_function_removed(self, name):
        """Update function list when function removed."""
        items = self.function_list.findItems(name, Qt.MatchExactly)
        for item in items:
            self.function_list.takeItem(self.function_list.row(item))

Key insight: Signals decouple manager from UI, enabling multiple UI components to stay synchronized.

Function Validation

Custom functions must meet specific requirements:

Required Decorators

from openhcs.core.memory.decorators import numpy_memory, cupy_memory

# Valid: Has memory type decorator
@numpy_memory
def valid_function(image: np.ndarray) -> np.ndarray:
    return image

# Invalid: Missing decorator
def invalid_function(image: np.ndarray) -> np.ndarray:
    return image  # ValidationError!

Validation check:

if not hasattr(func, 'input_memory_type'):
    raise ValidationError("Function must have memory type decorator")

Signature Requirements

# Valid: First parameter is image array
def valid_function(image: np.ndarray, sigma: float = 1.0) -> np.ndarray:
    return gaussian_filter(image, sigma=sigma)

# Invalid: First parameter is not image
def invalid_function(sigma: float, image: np.ndarray) -> np.ndarray:
    return gaussian_filter(image, sigma=sigma)  # ValidationError!

Validation check:

sig = inspect.signature(func)
first_param = list(sig.parameters.values())[0]
if first_param.annotation not in (np.ndarray, 'np.ndarray'):
    raise ValidationError("First parameter must be image array")

Return Type Annotation

# Valid: Return type annotated
def valid_function(image: np.ndarray) -> np.ndarray:
    return image

# Invalid: Missing return type
def invalid_function(image: np.ndarray):
    return image  # ValidationError!

Validation check:

if sig.return_annotation == inspect.Signature.empty:
    raise ValidationError("Function must have return type annotation")

Persistence and Loading

Custom functions are persisted to disk for reuse across sessions:

# Save to disk
custom_functions_dir = Path.home() / '.openhcs' / 'custom_functions'
function_file = custom_functions_dir / f'{name}.py'

with open(function_file, 'w') as f:
    f.write(code)

# Load on startup
manager = CustomFunctionManager()
manager.load_all_functions()

# All custom functions available in pipelines

Storage location: ~/.openhcs/custom_functions/

Integration with Function Registry

Custom functions integrate with the global function registry:

from openhcs.processing.func_registry import register_function, get_function

# Register custom function
register_function(func, backend='openhcs')

# Retrieve in pipeline
func = get_function('my_custom_filter', backend='openhcs')

# Use in step
step = FunctionStep(function='my_custom_filter')

Backend organization: Custom functions use backend='openhcs' to distinguish from library functions.

Common Patterns

Adding Function from GUI

# User clicks "Add Function" button
dialog = CustomFunctionManagerDialog()

# User writes code in editor
code = '''
@numpy_memory
def my_filter(image: np.ndarray, threshold: float = 0.5) -> np.ndarray:
    return image > threshold
'''

# User clicks "Save"
dialog.manager.add_function('my_filter', code)

# Function immediately available in pipeline editor

Updating Existing Function

# Load existing function
existing_code = manager.get_function_code('my_filter')

# User edits code
updated_code = existing_code.replace('threshold: float = 0.5', 'threshold: float = 0.3')

# Save update (atomic replacement)
manager.add_function('my_filter', updated_code)

# All pipelines using 'my_filter' now use updated version

Removing Function

# Remove function
manager.remove_function('my_filter')

# Function no longer available in pipelines
# Existing pipelines using 'my_filter' will fail validation

Implementation Notes

🔬 Source Code:

  • Manager: openhcs/processing/custom_functions/manager.py (line 117)

  • Registry integration: openhcs/processing/func_registry.py (line 33)

  • GUI dialog: openhcs/pyqt_gui/dialogs/custom_function_manager_dialog.py (line 1)

🏗️ Architecture:

📊 Performance:

  • Function registration is fast (< 10ms)

  • Disk persistence is asynchronous (doesn’t block UI)

  • Signal emission overhead is negligible

Key Design Decisions

Why atomic updates?

Prevents partial registration that could leave system in inconsistent state. Either function is fully registered or not at all.

Why inject module name?

Functions executed via exec() don’t have proper __module__. Injection ensures proper organization and serialization.

Why use signals instead of callbacks?

Signals decouple manager from UI, enabling multiple components to stay synchronized without tight coupling.

Common Gotchas

  • Don’t forget memory type decorator: Functions without @numpy_memory or similar will fail validation

  • First parameter must be image: Function signature must start with image array parameter

  • Return type annotation required: Functions without return type annotation will fail validation

  • Function names must be unique: Adding function with existing name replaces the old function

  • Removed functions break pipelines: Pipelines using removed functions will fail validation

Debugging Custom Functions

Symptom: Function Not Appearing in Pipeline Editor

Cause: Validation failed during registration

Diagnosis:

# Check validation errors
try:
    manager.add_function(name, code)
except ValidationError as e:
    logger.error(f"Validation failed: {e}")

Fix: Ensure function meets all validation requirements

Symptom: Function Execution Fails

Cause: Missing imports in function code

Diagnosis:

# Function code missing imports
@numpy_memory
def my_filter(image: np.ndarray) -> np.ndarray:
    return gaussian_filter(image, sigma=1.0)  # NameError: gaussian_filter not defined

Fix: Add imports to function code:

from scipy.ndimage import gaussian_filter

@numpy_memory
def my_filter(image: np.ndarray) -> np.ndarray:
    return gaussian_filter(image, sigma=1.0)