Built-in Elements¶

Quick reference for all elements shipped with SGN.

Sources¶

Element	Description	Key Args
`NullSource`	Produces empty frames. Useful for testing.	`num_frames`, `wait`, `frame_factory`
`IterSource`	Produces frames from iterables (lists, generators, etc).	`iters`, `eos_on_empty`, `frame_factory`
`DequeSource`	Like `IterSource` but backed by `deque` objects for dynamic appending.	`iters`, `eos_on_empty`
`StatsSource`	Produces system/process statistics via psutil.	`interval`, `include_process_stats`, `include_system_stats`, `wait`

NullSource¶

from sgn import NullSource

src = NullSource(name="src", source_pad_names=["H1"], num_frames=5)

Produces num_frames empty frames then sends EOS. Inherits SignalEOS for Ctrl+C handling. Set wait=1.0 to slow down frame production.

IterSource¶

from sgn import IterSource

src = IterSource(
    name="src",
    source_pad_names=["H1", "L1"],
    iters={"H1": [1, 2, 3], "L1": [4, 5, 6]},
)

One iterable per pad. Sends EOS when the iterator is exhausted (controlled by eos_on_empty, default True). Iterables are coerced to iterators internally.

DequeSource¶

from collections import deque
from sgn import DequeSource

src = DequeSource(
    name="src",
    source_pad_names=["H1"],
    iters={"H1": deque([1, 2, 3])},
    eos_on_empty=False,  # Keep running even when empty
)
# Append data at runtime:
src.deques["H1"].appendleft(4)

Useful for streaming applications where data arrives dynamically.

StatsSource¶

from sgn.sources import StatsSource

stats = StatsSource(
    name="stats",
    source_pad_names=["metrics"],
    interval=2.0,
)

Requires pip install psutil. See Monitor Stats.

Transforms¶

Element	Description	Key Args
`CallableTransform`	Wraps callables as transform elements.	`callmap`, `depmap`

CallableTransform¶

Create from a single callable:

import functools
from sgn import CallableTransform

def scale(frame, factor):
    return None if frame.data is None else frame.data * factor

t = CallableTransform.from_callable(
    name="t1",
    sink_pad_names=["H1"],
    callable=functools.partial(scale, factor=10),
    output_pad_name="H1",
)

Create from multiple input/output combinations:

from sgn import CallableTransform

t = CallableTransform.from_combinations(
    name="t1",
    combos=[
        (("H1",), lambda f: f.data * 2 if f.data is not None else None, "doubled"),
        (("H1", "L1"), lambda f1, f2: (f1.data or 0) + (f2.data or 0), "summed"),
    ],
)
# Sink pads: H1, L1 (inferred)
# Source pads: doubled, summed (inferred)

Sinks¶

Element	Description	Key Args
`NullSink`	Discards all frames. Useful for testing.	`verbose`
`CollectSink`	Collects frame data into lists.	`extract_data`, `skip_empty`
`DequeSink`	Collects frame data into deques.	`extract_data`, `skip_empty`

NullSink¶

from sgn import NullSink

snk = NullSink(name="snk", sink_pad_names=["H1"], verbose=True)

Set verbose=True to print each frame as it arrives.

CollectSink¶

from sgn import CollectSink

snk = CollectSink(name="snk", sink_pad_names=["H1", "L1"])

# After pipeline.run():
h1_data = list(snk.collects["H1"])  # List of data values
l1_data = list(snk.collects["L1"])

By default, extract_data=True stores frame.data rather than the whole frame, and skip_empty=True skips gap frames.

DequeSink¶

from sgn import DequeSink

snk = DequeSink(name="snk", sink_pad_names=["H1"])

# After pipeline.run():
data = snk.deques["H1"]  # deque of data values

Same as CollectSink but uses deque internally for efficient appending and popping from both ends.