Plugins

The service supports the system of plugins. Plugins must be written in the Python programming language.

Plugin types

There are three sorts of plugins:

• On event plugin. The plugin is triggered when an event occurs. The plugin should implement a callback function. This function is called on each event of the corresponding type. The set of event types is defined by the service developers.

  event type	description
  monitoring_event	Event contains monitoring points for sending to a custom monitoring system

  Monitoring plugin example:

  Module request monitoring plugin example

  class cow.plugins.plugin_examples.request_monitoring_plugin_example.BaseRequestMonitoringPlugin(app)

  Base class for requests monitoring.

  abstractmethod async flushPointToMonitoring(point, logger)

  All plugins must realize this method.

  This function call after end of request

  Parameters:

  • point (TypeVar(MonitoringPoint)) – point for monitoring

  • logger – logger

  Return type:

  None

  async handleEvent(points, logger)

  Handle event

  Parameters:

  • *args – positional arg for event handler function

  • **kwargs – named arg for event handler function

  class cow.plugins.plugin_examples.request_monitoring_plugin_example.RequestMonitoringPlugin(app)

  Example plugin sends a request data for monitoring to third-party source. Only one instance of this class exist during the program execution.

  async close()

  Stop plugin.

  Close all open connections and ect

  async flushPointToMonitoring(point, logger)

  Callback for sending a request monitoring data.

  Parameters:

  • point (TypeVar(MonitoringPoint)) – point for monitoring

  • logger – logger

  Return type:

  None

  async initialize()

  Initialize plugin.

  Close all open connections and ect

  This plugin demonstrates the sending of a request monitoring data to another service. All monitoring plugins must implement the BaseRequestMonitoringPlugin abstract class.

• Background plugin. This sort of plugin is intended for background work.

  The background plugin can implement:

  • custom route

  • background monitoring of service resources

  • collaboration of an event plugin and a background plugin (batching monitoring points)

  • connection to other data sources (Redis, RabbitMQ) and their data processing

    Module realizes background plugin example

    class cow.plugins.plugin_examples.background_plugin_example.BackgroundPluginExample(app)

    Background plugin example.

    Create background task and add a route.

    async close()

    Stop background process Returns:

    async initialize()

    Initialize plugin

    async start()

    Run background process

    Warning

    The function suppose that the process is handle in this coroutine. The coroutine must start the process only without awaiting a end of the process

    async usefulJob()

    Some useful async work

    class cow.plugins.plugin_examples.background_plugin_example.HandlerExample

    Handler example

    async get(request)

    Method get example.

    Returns:

    response

    cow.plugins.plugin_examples.background_plugin_example.anotherHandlerExample(request)

    Standalone handler example

  This plugin demonstrates background work and implements a route. All background plugins must implement the BaseRequestMonitoringPlugin abstract class.

• Estimator replacement plugin. This type of plugin allows replacing standard SDK estimators (the per-frame estimators run by sdk_loop, e.g. the people count estimator) 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.

  class luna_video_agent.plugins.plugin_examples.onnx_people_counter.people_count_onnx_plugin.Estimation(count, coordinates)

  Container for people counting estimation results.

  Holds the count of detected people and their coordinate positions.

  count

  The number of people detected in the image.

  points

  Points object containing coordinate information.

  class Points(coordinates)

  Container for point coordinates in an estimation.

  coordinates

  List of (x, y) coordinate tuples representing people positions.

  getPoints()

  Convert coordinates to Vec2D objects.

  Returns:

  List of Vec2D objects representing the point coordinates.

  class luna_video_agent.plugins.plugin_examples.onnx_people_counter.people_count_onnx_plugin.ONNXPeopleCountEstimator(modelPath, useGpu=False, gpuDeviceId=0, maxWorkers=4)

  ONNX-based people counting estimator using density map models.

  This estimator uses ONNX Runtime to perform inference with density map models like CSRNet or MCNN for counting people in images.

  onnxWrapper

  Async ONNX Runtime wrapper.

  session

  ONNX Runtime inference session.

  input_name

  Name of the model’s input node.

  output_name

  Name of the model’s output node.

  close()

  Close the estimator and release resources.

  async estimateBatch(images, estimationTargets=EstimationTargets.T1, asyncEstimate=True)

  Estimate people count in multiple images asynchronously.

  Parameters:

  • images (list[VLImage | ImageForPeopleEstimation | tuple[VLImage, Rect]]) – List of input images in various supported formats.

  • estimationTargets (EstimationTargets) – Target type (T1 with coordinates, T2 count only). Defaults to T1.

  • asyncEstimate (Literal[True]) – Must be True (only async mode supported).

  Return type:

  list[PeopleCount]

  Returns:

  List of PeopleCount objects with estimation results.

  class luna_video_agent.plugins.plugin_examples.onnx_people_counter.people_count_onnx_plugin.OnnxruntimeWrapper(onnxSession)

  Wrapper for asynchronous ONNX Runtime execution.

  async forward(output_names, input_feed)

  Execute inference asynchronously.

  Parameters:

  • output_names – List of output names

  • input_feed – Dict with numpy array inputs

  Returns:

  List of numpy arrays (outputs)

  class luna_video_agent.plugins.plugin_examples.onnx_people_counter.people_count_onnx_plugin.PeopleCountReplacementPlugin(app)

  Plugin for replacing the default people count estimator with ONNX implementation.

  This plugin integrates the ONNXPeopleCountEstimator into the Luna SDK, replacing the built-in estimator with a custom ONNX-based implementation.

  close()

  Stop plugin.

  Close all open connections, etc

  Return type:

  None

  createReplacementEstimator()

  Create a new ONNX people count estimator instance.

  Return type:

  ONNXPeopleCountEstimator

  Returns:

  Configured ONNXPeopleCountEstimator instance.

  property deviceClass: Literal['cpu', 'gpu']

  Get the device class for this estimator.

  Checks the configuration for people counter device settings, falling back to global device class if set to “global”.

  Returns:

  The device class, either “cpu” or “gpu”.

  property estimatorBrokerName: str

  Get the system name of the estimator to replace.

  Returns:

  The name “peopleCountEstimator”.

  property modelDirPath: Path

  Get the directory containing the model file.

  Returns:

  Path to the directory containing this plugin file.

  property modelName: str

  Get the model filename.

  Returns:

  The ONNX model filename.

  property modelPath: Path

  Get the full path to the model file.

  Returns:

  Complete path to the ONNX model file.

  luna_video_agent.plugins.plugin_examples.onnx_people_counter.people_count_onnx_plugin.adjustCoordsToOriginal(coords, cropArea)

  Adjust coordinates from cropped space back to original image space.

  Parameters:

  • coords (list[tuple[int, int]]) – List of (x, y) coordinates in cropped image space.

  • cropArea (Rect) – Rectangle defining the crop region offset.

  Return type:

  list[tuple[int, int]]

  Returns:

  List of (x, y) coordinates adjusted to original image space.

  async luna_video_agent.plugins.plugin_examples.onnx_people_counter.people_count_onnx_plugin.execute(output_names, input_feed, session, loop)

  Async execute onnx prediction using run_async.

  Parameters:

  • output_names – name of the outputs

  • input_feed – dictionary { input_name: numpy_array }

  • session (InferenceSession) – ort session

  • loop (AbstractEventLoop) – current asyncio event loop

  Returns:

  prediction result

  luna_video_agent.plugins.plugin_examples.onnx_people_counter.people_count_onnx_plugin.extractCoordinatesFromDensityMap(densityMap, imgHeight, imgWidth, count)

  Extract person coordinates from a density map.

  Finds the top peak locations in a density map and converts them to image coordinates scaled to the original image dimensions.

  Parameters:

  • densityMap (ndarray) – The density map output from the model.

  • imgHeight (int) – Height of the original image.

  • imgWidth (int) – Width of the original image.

  • count (int) – Number of people to extract coordinates for.

  Return type:

  list[tuple[int, int]]

  Returns:

  List of (x, y) tuples representing person locations in image coordinates.

  luna_video_agent.plugins.plugin_examples.onnx_people_counter.people_count_onnx_plugin.ortCallback(result, waiter, error)

  Callback for onnx thread.

  luna_video_agent.plugins.plugin_examples.onnx_people_counter.people_count_onnx_plugin.setFutureResult(future, result, error)

  Set result to asyncio future.

  This plugin demonstrates replacement of the people count estimator with a custom ONNX Runtime implementation. All estimator replacement plugins must extend the EstimatorReplacementPlugin abstract class.

  For detailed information about creating and integrating estimator replacement plugins, see:

  • Creating Estimator Replacement Plugins
    • Overview
    • Architecture
    • Development setup
    • Step 1: Implement the custom estimator class
    • Step 2: Create the plugin class
    • Plugin structure
    • Integration with the service
    • Best practices
    • Common pitfalls
    • Next steps
  • Estimator Replacement Plugin Example
    • Overview
    • Plugin architecture
    • Asynchronous ONNX Runtime execution
    • Custom estimator implementation
    • Plugin class implementation
    • Usage in the service
    • Advanced topics
    • Summary
    • Complete source files

Enable plugin

If the user implements a plugin, the file with the plugin should be added to the luna_video_agent/plugins directory of the service. The plugin filename should be added to the LUNA_VIDEO_AGENT_ACTIVE_PLUGINS configuration setting.

Warning

If the plugin has custom dependencies (listed in a requirements.txt), they must be installed in the service environment before the plugin is activated, e.g. pip install -r luna_video_agent/plugins/my_plugin/requirements.txt. Plugins with unmet dependencies will fail to load and may cause service startup errors. This is especially relevant for estimator replacement plugins that pull in inference libraries such as onnxruntime.