Using it as a Python library¶

The library is the primary door - the CLI and the MCP server are thin shells over it. Anything shown here works in any Python application, notebook or script.

The public API¶

Everything you need is importable straight from spectro_kernel:

from spectro_kernel import (
    Spectrum1D, WorkContext,             # data types
    list_algorithms, describe_algorithm, # discovery
    run_algorithm,                       # run one algorithm
    PipelineBuilder,                     # compose many
    read_fits, read_ascii_spectrum,      # I/O
)

Loading a spectrum¶

from spectro_kernel.io import read_fits, read_ascii_spectrum, read_votable_spectrum

spec = read_fits("obs.fits")               # local path …
spec = read_fits("https://example/obs.fits")  # … or an http(s) URL
spec = read_ascii_spectrum("obs.csv")

read_fits handles both common layouts - a 1D flux array with a WCS header, and a binary table with explicit columns.

URLs and the trust boundary¶

Remote downloads are streamed with a 60 s timeout and a 1 GiB size ceiling (timeout= / max_bytes= to adjust). By default the library applies no SSRF policy — like astropy or requests, it cannot know whether http://10.0.0.5/ is your own MinIO or an attacker's pivot; only the application knows its trust boundary.

If your URLs come from untrusted users (a web backend, a relay service), opt into the kernel's canonical guard:

from spectro_kernel.io import read_fits
from spectro_kernel.url_safety import validate_safe_url

spec = read_fits(user_url, url_validator=validate_safe_url)

With a validator set, private / loopback / link-local targets are refused and so are HTTP redirects (each hop would evade the check). Pass your own callable for a custom policy, or allow_remote=False to guarantee the call never touches the network.

Running a single algorithm¶

from spectro_kernel import WorkContext, run_algorithm

ctx = WorkContext(spectrum=spec)
output = run_algorithm("normalize_polynomial", ctx, {"order": 3})

output.success        # True / False
output.metrics        # scalar summary, e.g. {"continuum_median": 1.2e4}
ctx.spectrum          # the normalised spectrum (algorithms write back into ctx)

If you prefer a failed run to return an AlgorithmOutput instead of raising:

output = run_algorithm("snr_der", ctx, raise_on_error=False)
if not output.success:
    print(output.error)

Discovering what exists¶

from spectro_kernel import list_algorithms, describe_algorithm

for meta in list_algorithms("line_fitting"):
    print(meta["name"], "-", meta["description"])

info = describe_algorithm("fit_gaussian_line")
info["default_params"]   # {'line_center_angstrom': None, 'window_angstrom': 20.0, ...}
info["required_params"]  # ['line_center_angstrom']
info["backend"]          # 'scipy'
info["references"]       # ['scipy.optimize.curve_fit - …']

Composing a pipeline¶

from spectro_kernel import PipelineBuilder

pipeline = (
    PipelineBuilder()
    .add("normalize_polynomial", order=3)
    .add("snr_der")
    .add("fit_gaussian_line", line_center_angstrom=6562.79, window_angstrom=30)
    .build()
)
result = pipeline.execute(ctx)

See Concepts: pipelines for presets and reproducibility.

Reading the results¶

After a run, everything is on the WorkContext:

ctx.spectrum            # the current (possibly transformed) spectrum
ctx.metrics             # {"snr_der": 142.3, ...}
ctx.line_fits           # {"H-alpha": LineFitResult(...)}
ctx.periodograms        # {"lomb_scargle": Periodogram(...)}
ctx.exports             # {"csv": b"...", "fits": b"..."}
ctx.history             # list[ProcessingStep] - the audit trail
ctx.summary()           # compact JSON-safe overview

Writing your own algorithm¶

Because the kernel is a normal library, an algorithm you register in your own codebase joins the catalogue just like a built-in one:

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


@register_algorithm("my_metric", category=AlgorithmCategory.QUALITY, version="1.0.0")
class MyMetric(BaseAlgorithm):
    """A project-specific quality metric."""

    input_requirements = ["spectrum"]

    def run(self, ctx: WorkContext, params: dict) -> AlgorithmOutput:
        value = float(ctx.spectrum.flux.mean())
        ctx.metrics["my_metric"] = value
        return AlgorithmOutput.ok(metrics={"my_metric": value})

Once the module is imported, run_algorithm("my_metric", ctx) works - and so does the CLI and the MCP server. The full walkthrough is in Tutorial: add an algorithm.