Tutorial: add an algorithm¶

The promise of spectro-kernel is that extending the catalogue is one Python file plus one test - no core to modify, no front-end to touch. This tutorial proves it by adding a real algorithm: a band integrator that sums the flux inside a wavelength window.

Step 1 - write the algorithm file¶

Create src/spectro_kernel/algorithms/quality/band_flux.py:

"""Algorithm: integrate the flux inside a wavelength band."""

from __future__ import annotations

from typing import Any

import numpy as np

from ...base import AlgorithmOutput, BaseAlgorithm
from ...registry import register_algorithm
from ...types import AlgorithmCategory, WorkContext

# numpy renamed trapz -> trapezoid in 2.0; support both.
try:
    from numpy import trapezoid as _trapezoid
except ImportError:  # numpy < 2.0
    from numpy import trapz as _trapezoid


@register_algorithm("band_flux", category=AlgorithmCategory.QUALITY, version="1.0.0")
class BandFlux(BaseAlgorithm):
    """Integrate the flux over a wavelength band.

    Returns the area under the spectrum between two wavelengths - a simple proxy for
    the energy emitted in that band.
    """

    backend = "numpy"
    references = ["numpy.trapezoid - trapezoidal integration"]
    default_params = {"wavelength_min": None, "wavelength_max": None}
    required_params = ["wavelength_min", "wavelength_max"]
    param_descriptions = {
        "wavelength_min": "Lower bound of the band (Å).",
        "wavelength_max": "Upper bound of the band (Å).",
    }
    input_requirements = ["spectrum"]
    output_produces = ["metrics.band_flux"]

    def run(self, ctx: WorkContext, params: dict[str, Any]) -> AlgorithmOutput:
        band = ctx.spectrum.select_range(
            float(params["wavelength_min"]), float(params["wavelength_max"])
        )
        value = float(_trapezoid(band.flux, band.wavelength))
        ctx.metrics["band_flux"] = value
        return AlgorithmOutput.ok(
            metrics={"band_flux": value},
            message=f"Integrated flux over the band = {value:.4g}.",
        )

That is the whole algorithm. A few things to notice:

@register_algorithm is all the wiring there is. Automatic discovery imports the file; the decorator runs; the algorithm joins the catalogue.
run(self, ctx, params) is the only method. Read from ctx, write to ctx, return an AlgorithmOutput. params arrives already merged with default_params and validated.
backend and references make the provenance explicit - see Algorithms → Provenance.
input_requirements = ["spectrum"] means the runner fails cleanly if no spectrum is loaded - you do not have to check it yourself.

Step 2 - write a test¶

Create tests/unit/test_band_flux.py:

import numpy as np

from spectro_kernel import WorkContext, run_algorithm
from spectro_kernel.types import Spectrum1D


def test_band_flux_of_a_flat_spectrum():
    # A flat spectrum at flux = 2 over a 100 Å band integrates to ~200.
    wave = np.linspace(5000, 5200, 400)
    ctx = WorkContext(spectrum=Spectrum1D(wave, np.full(wave.size, 2.0)))
    run_algorithm("band_flux", ctx,
                  {"wavelength_min": 5050, "wavelength_max": 5150})
    assert abs(ctx.metrics["band_flux"] - 200.0) < 1.0

Prefer a synthetic input with a known answer - that is what turns the test into a real check of correctness.

Step 3 - it is already everywhere¶

You did not edit a single other file. Verify:

spectro describe band_flux           # the CLI sees it
spectro run band_flux -i obs.fits -p wavelength_min=6500 -p wavelength_max=6620
pytest tests/unit/test_band_flux.py  # the test passes

It is also now an MCP tool, and list_algorithms() returns it, and it can be used as a pipeline step. One file in, available through every door.

Conventions to follow¶

Naming: snake_case, verb-first. Add a suffix when several implementations of one operation coexist (band_flux, band_flux_weighted).
Determinism: no eval/exec, no hidden global state; seed any RNG from params.
Versioning: bump version on any behaviour change so old results stay reproducible.
Provenance: always set backend; cite the literature in references when you wrap a library or implement a published method.
Optional dependencies: if the algorithm needs a package outside the core, import it lazily and register the algorithm conditionally (see algorithms/catalogs/simbad.py for the pattern), so the catalogue still loads without the extra.

See CONTRIBUTING.md for the full checklist.