Developing EDD pipelines and plugins

New observing capabilities belong in a plugin collection, not in the EDD core or a site repository. Put only reusable framework functionality in the core and only observatory-specific configuration or adapters in a site repository.

Developing deployment behaviour

Use edd-tool make-devenv to create the persistent collection search path needed to iterate on site playbooks or the Ansible deployment behaviour of a plugin. The same workflow exposes the installed EDD.core collection for changes to shared core roles and playbooks. It is intentionally separate from edd-tool deploy, which always starts from a clean repository checkout for a reproducible operational deployment.

See Development environments with make-devenv for setup commands, collection locations, refresh behaviour, and the generated per-environment README.md.

Plugin layout

A typical plugin contains:

galaxy.yml
pyproject.toml
README.md
edd/<plugin>/              Python control package
cpp/                       optional C++/CUDA processing
roles/<pipeline>/          observation product role
roles/installer/           collection installation/build entry point
tests/                     unit tests
test_execution/            lifecycle/integration tests

Keep the Galaxy collection version, Python package version, Debian package, container image, and release tag traceable to the same source revision.

Minimal measuring pipeline

This is a runnable control-plane skeleton. A real pipeline must add resource management, validation, sensors, output registration, and tests.

import asyncio

from edd.core.edd_pipeline_aio import EDDPipeline, launchPipelineServer
from edd.core.pipeline_statemodel import PipelineFlavor


DEFAULT_CONFIG = {
    "id": "example_pipeline",
    "type": "ExamplePipeline",
    "input_data_streams": [],
    "output_data_streams": [],
    "gain": 1.0,
}


class ExamplePipeline(EDDPipeline):
    _PIPELINE_VERSION = "0.1.0"

    def __init__(self, ip, port, loop=None):
        super().__init__(
            ip,
            port,
            flavor=PipelineFlavor.MEASURING,
            default_config=DEFAULT_CONFIG,
            loop=loop,
        )
        self._process = None

    async def configure(self):
        if self._config["gain"] <= 0:
            raise ValueError("gain must be positive")
        # Allocate buffers and prepare subprocess configuration here.

    async def capture_start(self):
        # Start persistent ingest and wait until it is ready.
        pass

    async def measurement_prepare(self, config=None):
        # Validate and store scan-specific metadata.
        pass

    async def measurement_start(self):
        # Start writing or scan-specific processing.
        pass

    async def measurement_stop(self):
        # Stop the scan and register every completed output file.
        pass

    async def deconfigure(self):
        # Idempotently stop processes and release every acquired resource.
        self._process = None


if __name__ == "__main__":
    asyncio.run(launchPipelineServer(ExamplePipeline))

The base class supplies standard --host, --port, --redis-ip, --redis-port, --register-id, --log-level, and --version options.

Configuration design

The current base class validates that incoming keys exist in the default configuration and warns on type changes. It does not validate ranges, units, cross-field relationships, or scientific meaning. Each pipeline must perform that validation before allocating resources.

Classify settings by audience:

  • normal observing parameters, documented with units and constraints;

  • site or hardware parameters, derived from inventory and stream descriptors;

  • performance parameters, changed only after benchmarking; and

  • debug parameters, safe by default and clearly excluded from production use.

Avoid embedding a site hostname, Redis server, multicast address, filesystem path, antenna list, or GPU assumption in a reusable default configuration.

Stream interfaces

Create a versioned stream format for every externally consumed data product. Document dimensions, ordering, units, timestamp convention, metadata items, transport requirements, and compatibility policy. Producers must finalize their output descriptors during configure; consumers must reject incompatible formats rather than guessing.

Resource and error handling

Use core helpers for NUMA selection, CPU allocation, DADA buffers, subprocess monitoring, and SPEAD I/O. Deconfigure must be idempotent and work after a partially completed configure. Preserve the original exception and enough context to identify the failed process or resource.

Use a stalled outcome only when the next scan can reasonably recover without a new configure. Use an error for a failed control loop and panic only when safe cleanup is not possible.

Ansible role

Each product role should include EDD.core.common with a unique default container name, pipeline command, image registry, dashboard panels, and provision_test_imports. See Common role contract for the current contract. Non-pipeline services must set register_as_pipeline: false and explicitly select the monitoring components they need.

Tests

Required coverage includes:

  • configuration validation and stream-model unit tests;

  • configure failure followed by successful deconfigure;

  • complete measuring or streaming lifecycle;

  • missing data, slow consumer, subprocess exit, and full-buffer behaviour;

  • output metadata and independent data-format validation;

  • role build, launch, registration, dashboard creation, and stop; and

  • a complete provision test on representative hardware.

Build suites

EDD builds against named Ubuntu/CUDA suites defined by the core common role. At the time this page was verified, the default suite was Ubuntu 24.04 with CUDA 12.6.1, with Ubuntu 22.04 and 20.04 legacy suites. A plugin must support the current default or explicitly select and document a legacy suite.