Creating Estimator Replacement Plugins

Estimator replacement plugins allow you to replace standard Luna SDK estimators with custom implementations without modifying the core service code. This enables integration of custom neural networks, alternative algorithms, or optimized implementations for specific use cases.

Overview

An estimator replacement plugin consists of two main components:

  1. Custom estimator class - Implements the EstimatorBase interface with your custom logic

  2. Plugin class - Extends EstimatorReplacementPlugin and configures the replacement

The plugin system handles all the complexity of integrating your custom estimator into the SDK engine and replacing it across all workers.

Architecture

The estimator replacement plugin system uses an inheritance pattern from the base class EstimatorReplacementPlugin, which provides infrastructure for replacing any estimator in Luna Remote SDK. The base class handles:

  • Application context and configuration access

  • Device type determination (CPU/GPU)

  • SDK engine integration

  • Direct estimator replacement in all workers

This allows you to focus solely on implementing the custom estimator logic.

Development setup

Before developing an estimator replacement plugin, it’s highly recommended to install the luna-plugins package with development dependencies. This provides access to the source code of libraries like lunavl and sdk_loop, which is essential for:

  • Understanding the interfaces you need to implement (e.g., EstimatorBase)

  • Checking return types of original estimators (e.g., PeopleCount from lunavl)

  • Exploring available classes and methods in EstimatorsCollection

  • Reading implementation details and documentation

Installation options

Using pip:

# Install luna-plugins with remote_sdk extra dependencies
pip install luna-plugins[remote_sdk]

This installs luna-plugins along with lunavl, sdk_loop, and other dependencies needed for Luna Remote SDK plugin development.

Using poetry:

# Add luna-plugins with remote_sdk extras to your project
poetry add luna-plugins[remote_sdk]

# Or add to pyproject.toml:
# luna-plugins = {extras = ["remote_sdk"], version = "^X.X.X"}

Benefits:

With these packages installed, your IDE will provide:

  • Autocomplete for Luna SDK classes and methods

  • Type hints and inline documentation

  • “Go to definition” navigation to source code

  • Better understanding of expected interfaces and return types

Tip

When implementing estimateBatch(), you can navigate to the original estimator’s source code (e.g., lunavl.sdk.estimators.image_estimators.people_count.PeopleCountEstimatorV2) to see exactly what it returns and how it handles different input formats.

Step 1: Implement the custom estimator class

Your custom estimator must implement the EstimatorBase interface with two required methods:

from luna_plugins.estimator_plugins.estimator_replacement_plugin import EstimatorBase

class CustomEstimator(EstimatorBase):
    def __init__(self, modelPath, useGpu=False, gpuDeviceId=0):
        # Initialize your model, load weights, etc.
        pass

    def close(self):
        """Release resources held by the estimator."""
        # Clean up model, free memory, close connections
        pass

    async def estimateBatch(self, images, estimationTargets, asyncEstimate=True):
        """Process a batch of images and return estimation results."""
        # Your estimation logic here
        # Must return objects compatible with the original estimator
        pass

Required methods

close()

This method should release all resources held by the estimator:

  • Model weights and computational graphs

  • GPU memory allocations

  • File handles and network connections

  • Any other resources

estimateBatch()

This is the main estimation method with specific requirements:

Signature requirements:

  • Must be defined as async (coroutine by default)

  • Must accept parameters matching the original estimator’s interface

  • Must support asynchronous execution (asyncEstimate=True by default)

Return value requirements:

The return value must be compatible with the original estimator being replaced. To find the correct return type:

  1. Navigate to the lunavl library: lunavl.sdk.estimators

  2. Find the estimator class you’re replacing

  3. Check what estimateBatch() returns

Example for people count estimator:

The original estimator is lunavl.sdk.estimators.image_estimators.people_count.PeopleCountEstimatorV2. Its estimateBatch() method returns a list of PeopleCount objects.

Therefore, your custom estimator must also return PeopleCount objects:

from lunavl.sdk.estimators.image_estimators.people_count import PeopleCount

async def estimateBatch(self, images, estimationTargets, asyncEstimate=True) -> list[PeopleCount]:
    # Your implementation
    results = []
    for image in images:
        # Process image
        count, coordinates = self._process_image(image)
        # Create PeopleCount object matching the original estimator's output
        results.append(PeopleCount(...))
    return results

Step 2: Create the plugin class

The plugin class configures how your custom estimator replaces the original one:

from pathlib import Path
from typing import Literal
from luna_plugins.estimator_plugins.estimator_replacement_plugin import EstimatorReplacementPlugin

class CustomEstimatorPlugin(EstimatorReplacementPlugin):
    @property
    def estimatorBrokerName(self) -> str:
        """Return the system name of the estimator to replace."""
        return "peopleCountEstimator"

    @property
    def deviceClass(self) -> Literal["cpu", "gpu"]:
        """Determine device to use based on configuration.

        It's highly recommended to implement this for each plugin to respect
        estimator-specific configuration from config.conf or configurator.
        """
        # Access configuration for the specific estimator being replaced
        # Replace 'peopleCounter' with your estimator's config name
        estimatorConfig = self.appConfig.estimatorsSettings.peopleCounter

        # Respect estimator-specific setting, fall back to global if "global"
        if estimatorConfig.deviceClass == "global":
            return self.globalDeviceClass
        return estimatorConfig.deviceClass

    def createReplacementEstimator(self) -> CustomEstimator:
        """Create and return an instance of your custom estimator."""
        return CustomEstimator(
            modelPath=Path(__file__).parent / "model.onnx",
            useGpu=True if self.deviceClass == "gpu" else False,
            gpuDeviceId=self.defaultGpuDevice,
        )

    def close(self) -> None:
        """Cleanup plugin resources if needed."""
        pass

Required properties and methods

estimatorBrokerName (property)

Purpose: Specifies which estimator in the SDK to replace.

Where to find the name:

The required name can be found in the attributes of the EstimatorsCollection class:

  • Path in library: sdk_loop.estimators.estimators_collection.EstimatorsCollection

  • Each estimator has a corresponding attribute name in this class

Example:

@property
def estimatorBrokerName(self) -> str:
    return "peopleCountEstimator"

deviceClass (property)

Purpose: Determines whether to use CPU or GPU for inference.

Implementation recommendation: It is highly recommended to implement this property for each plugin individually, as Luna Remote SDK configuration supports both global device settings and estimator-specific device settings. Implementing custom logic allows your plugin to respect the specific configuration for the estimator being replaced.

Default behavior: If not overridden, returns self.globalDeviceClass from application runtime settings.

Recommended implementation pattern:

@property
def deviceClass(self) -> Literal["cpu", "gpu"]:
    # Access specific estimator configuration
    # Replace 'peopleCounter' with the name of your estimator
    estimatorConfig = self.appConfig.estimatorsSettings.peopleCounter

    # If set to "global", use the global device class
    if estimatorConfig.deviceClass == "global":
        return self.globalDeviceClass

    # Otherwise use the estimator-specific setting
    return estimatorConfig.deviceClass

How to find the estimator configuration name:

The configuration name corresponds to the estimator you’re replacing. To find it:

  1. Check the service configuration file (config.conf)

  2. Look in the [XXXX_SETTINGS.estimator_settings] sections

  3. Or use the Luna Configurator service to browse available estimator settings

createReplacementEstimator() (method)

Purpose: Factory method that creates an instance of your custom estimator.

This method is called once during plugin initialization to create the estimator that will replace the original one.

Implementation guidelines:

  • Create and return an instance of your custom estimator class

  • Pass necessary configuration (model paths, device settings, etc.)

  • Use self.deviceClass to determine CPU/GPU usage

  • Use self.defaultGpuDevice for the GPU device index

Example:

def createReplacementEstimator(self) -> CustomEstimator:
    return CustomEstimator(
        modelPath=Path(__file__).parent / "model.onnx",
        useGpu=True if self.deviceClass == "gpu" else False,
        gpuDeviceId=self.defaultGpuDevice,
        maxWorkers=4,
    )

close() (method)

Purpose: Cleanup method called when the plugin is being shut down.

Use this to release any plugin-level resources (not estimator resources - those are handled by the estimator’s own close() method).

Plugin structure

A complete estimator replacement plugin typically has this structure:

my_estimator_plugin/
├── __init__.py                # Plugin module initialization
├── plugin.py                  # Main plugin implementation
├── model.onnx                 # Model file(s)
├── requirements.txt           # Python dependencies
└── README.md                  # Plugin documentation

requirements.txt example:

onnxruntime>=1.22.0
onnxruntime-gpu>=1.16.0
numpy>=1.24.0

Important

The requirements.txt file lists all Python dependencies needed by the plugin. These dependencies must be installed in the service environment before activating the plugin. Failure to install dependencies will cause import errors and plugin activation failures.

Integration with Luna Remote SDK

Installing the plugin in a container

Important

Critical step: Plugin dependencies from requirements.txt must be installed in the container before the service starts. Without proper dependencies, the plugin will fail to load.

Option 1: Build custom Docker image (Recommended)

FROM my-registry/luna-remote-sdk:v.x.x.x

# Copy plugin directory
COPY --chown=1001:0 my_estimator_plugin ./luna_remote_sdk/plugins/my_estimator_plugin

# Install plugin dependencies (REQUIRED if plugin has requirements.txt)
RUN pip install --no-cache-dir -r ./luna_remote_sdk/plugins/my_estimator_plugin/requirements.txt

Option 2: Mount plugin directory

docker run -d \
  --mount type=bind,source=/path/to/my_estimator_plugin,target=/srv/luna_remote_sdk/plugins/my_estimator_plugin \
  my-registry/luna-remote-sdk:v.x.x.x

# Install plugin dependencies (REQUIRED before enabling the plugin)
docker exec <container_id> pip install -r /srv/luna_remote_sdk/plugins/my_estimator_plugin/requirements.txt

Option 3: Copy to running container

docker cp my_estimator_plugin <container_id>:/srv/luna_remote_sdk/plugins/
docker exec <container_id> pip install -r /srv/luna_remote_sdk/plugins/my_estimator_plugin/requirements.txt
docker commit <container_id> my-registry/luna-remote-sdk-custom:v.x.x.x

Enabling the plugin

Add your plugin module name to the LUNA_REMOTE_SDK_ACTIVE_PLUGINS configuration:

Via environment variable:

export LUNA_REMOTE_SDK_ACTIVE_PLUGINS="my_estimator_plugin.plugin"

Via configuration file:

{
  "LUNA_REMOTE_SDK_ACTIVE_PLUGINS": "my_estimator_plugin.plugin"
}

Restart the service:

docker restart <container_id>

Verification

Check the service logs to verify the plugin was loaded successfully:

docker logs <container_id>

Look for messages indicating:

  • Plugin initialization

  • Estimator replacement

  • No error messages

Best practices

Development:

  • Install luna-plugins with dev dependencies: Use pip install luna-plugins[remote_sdk] or poetry add luna-plugins[remote_sdk] to get access to source code of lunavl, sdk_loop, and other libraries. This is crucial for understanding interfaces and exploring available classes.

  • Implement deviceClass property: Always implement custom deviceClass logic for your plugin to respect estimator-specific configuration settings. Check config.conf or use the configurator service to find the correct configuration path for your estimator.

  • Start with the example plugin as a template

  • Use your IDE to navigate to source code of original estimators to understand their interfaces

  • Test your estimator independently before integrating it as a plugin

  • Ensure return types match the original estimator exactly

  • Implement proper error handling and logging

Performance:

  • Optimize for batch processing when possible

  • Consider memory management for large models

  • Profile your estimator to identify bottlenecks

  • Use appropriate device (CPU/GPU) based on model requirements

Deployment:

  • Version your custom Docker images for easy rollback

  • Test thoroughly in a staging environment

  • Monitor resource usage after deployment

  • Keep dependencies up to date for security

Maintenance:

  • Document your plugin’s purpose and behavior

  • Include examples of input/output formats

  • Specify compatible Luna SDK versions

  • Provide troubleshooting guidelines

Common pitfalls

Incorrect return types:

Your estimator must return objects compatible with the original estimator. Check the lunavl library to find the correct types.

Blocking operations:

The estimateBatch() method must be async and should not block the event loop. Use async I/O operations or offload blocking work to threads/processes.

Resource leaks:

Always implement the close() method to properly release resources. This is especially important for GPU memory and file handles.

Import issues:

Avoid importing abstract base classes or other plugins that might cause circular dependencies or duplicate activations.

Next steps

See Estimator Replacement Plugin Example for a detailed walkthrough of a complete estimator replacement plugin implementation using ONNX Runtime for people counting.