Writing Custom Elements

This tutorial walks through writing custom time-series transforms and sinks. It assumes you've completed the first TS pipeline tutorial and understand buffers, frames, and pipeline connections.

Backends

The examples below call NumPy directly, which is the default: an element accepts only NumPy data unless it declares otherwise. To accept PyTorch (or write backend-agnostic xp code), and to declare a dtype-changing output, see Array Backends and the Backends and dtype section of the element development guide.

A Simple Transform

The simplest transform takes one input and produces one output. Here's an element that squares every sample:

from dataclasses import dataclass

from sgn import validator
from sgnts.base import TSCollectFrame, TSFrame, TSTransform
from sgnts.decorators import transform


@dataclass
class Square(TSTransform):

    @validator.one_to_one
    def validate(self) -> None:
        pass

    @transform.one_to_one
    def process(self, input_frame: TSFrame, output_frame: TSCollectFrame) -> None:
        for buf in input_frame:
            if not buf.is_gap:
                buf = buf.copy(data=buf.data ** 2)
            output_frame.append(buf)

Key points:

• @dataclass — all elements are dataclasses.
• validate() — called at construction. The @validator.one_to_one decorator checks that the element has exactly one sink pad and one source pad.
• process() — called each iteration with aligned input and output. The @transform.one_to_one decorator unwraps the single input and output from their dictionaries, so you work with a single TSFrame and TSCollectFrame directly.
• buf.copy(data=...) — creates a new buffer with modified data but the same offset, sample rate, and shape.
• output_frame.append(buf) — adds the buffer to the output collector. The collector is closed automatically after process() returns.

Use it in a pipeline:

from sgn import Pipeline
from sgnts.sources import FakeSeriesSource
from sgnts.sinks import TSFrameCollectSink

src = FakeSeriesSource(
    name="src",
    source_pad_names=["out"],
    signal_type="const",
    const=3.0,
    rate=2048,
    duration=1,
)
sq = Square(name="sq", sink_pad_names=["in"], source_pad_names=["out"])
snk = TSFrameCollectSink(name="snk", sink_pad_names=["out"])

Pipeline().connect(src, sq).connect(sq, snk).run()

frames = snk.out_frames()
for buf in frames["out"]:
    if not buf.is_gap:
        print(buf.data[:3])  # [9. 9. 9.]

A Simple Sink

Sinks consume data without producing output. Here's a sink that prints summary statistics:

from dataclasses import dataclass

import numpy as np
from sgn import validator
from sgnts.base import TSFrame, TSSink
from sgnts.decorators import sink


@dataclass
class StatsSink(TSSink):

    @validator.single_pad
    def validate(self) -> None:
        pass

    @sink.single_pad
    def process(self, input_frame: TSFrame) -> None:
        if input_frame.EOS:
            self.mark_eos(self.sink_pads[0])
            return
        for buf in input_frame:
            if not buf.is_gap:
                print(f"offset={buf.offset} mean={np.mean(buf.data):.4f}")

The @sink.single_pad decorator unwraps the single input pad, so process() receives a TSFrame directly. Call self.mark_eos() when EOS arrives to signal the pipeline to stop.

Adding Parameters

Elements are dataclasses, so adding parameters is just adding fields:

@dataclass
class Scale(TSTransform):
    factor: float = 1.0
    offset: float = 0.0

    @validator.one_to_one
    def validate(self) -> None:
        pass

    @transform.one_to_one
    def process(self, input_frame: TSFrame, output_frame: TSCollectFrame) -> None:
        for buf in input_frame:
            if not buf.is_gap:
                buf = buf.copy(data=buf.data * self.factor + self.offset)
            output_frame.append(buf)

scale = Scale(name="scale", factor=2.0, offset=-1.0, sink_pad_names=["in"], source_pad_names=["out"])

Using configure()

Override configure() for one-time setup that depends on pad configuration. It runs after pads are created but before the pipeline starts:

@dataclass
class Normalizer(TSTransform):
    """Normalize each channel independently."""

    def configure(self) -> None:
        # Pad references are available here
        self._running_sum = {}
        self._count = {}
        for pad in self.sink_pads:
            self._running_sum[pad] = 0.0
            self._count[pad] = 0

    @validator.one_to_one
    def validate(self) -> None:
        pass

    @transform.one_to_one
    def process(self, input_frame: TSFrame, output_frame: TSCollectFrame) -> None:
        for buf in input_frame:
            if not buf.is_gap:
                self._running_sum[self.sink_pads[0]] += np.sum(buf.data)
                self._count[self.sink_pads[0]] += buf.data.size
                mean = self._running_sum[self.sink_pads[0]] / self._count[self.sink_pads[0]]
                buf = buf.copy(data=buf.data - mean)
            output_frame.append(buf)

Next Steps

For more advanced patterns — many-to-one transforms, custom internal() overrides, adapter configuration for streaming, and make_ts_element() — see the Element Development guide.