Sinks¶

All sinks are imported from sgnts.sinks.

TSFrameCollectSink¶

Collects all incoming buffers during pipeline execution. After the pipeline finishes, retrieve results with out_frames().

from sgnts.sinks import TSFrameCollectSink

sink = TSFrameCollectSink(name="sink", sink_pad_names=["H1", "L1"])

After running the pipeline:

frames = sink.out_frames()  # dict[str, TSFrame]
h1_frame = frames["H1"]

This is the most common sink for testing and prototyping. Multi-pad.

DumpSeriesSink¶

Writes time-series data to a text file. Each row contains a timestamp followed by the sample values.

from sgnts.sinks import DumpSeriesSink

dump = DumpSeriesSink(name="dump", sink_pad_names=["out"], fname="output.txt")

Verbose mode prints each frame as it arrives:

dump = DumpSeriesSink(name="dump", sink_pad_names=["out"], fname="output.txt", verbose=True)

Gap buffers are skipped (only non-gap data is written). Single-pad.

NullSeriesSink¶

Discards all incoming data. Useful as a pipeline terminator when you only care about side effects in upstream elements.

from sgnts.sinks import NullSeriesSink

null = NullSeriesSink(name="null", sink_pad_names=["out"])

Verbose mode prints each frame and its latency relative to the current GPS time:

null = NullSeriesSink(name="null", sink_pad_names=["out"], verbose=True)

Multi-pad.

TSAppSink¶

Executes user-defined callbacks on each buffer. Similar to GStreamer's appsink.

from sgnts.sinks import TSAppSink

def on_buffer(pad, buf):
    print(f"{pad.name}: offset={buf.offset}, samples={len(buf.data)}")

app = TSAppSink(
    name="app",
    sink_pad_names=["out"],
    callbacks={"out": on_buffer},
)

Callbacks can also be registered after construction:

app = TSAppSink(name="app", sink_pad_names=["H1", "L1"])
app.set_callback("H1", on_buffer)
app.set_callback("L1", on_buffer)

Skip gaps by setting emit_gaps=False (default is True, meaning gap buffers are also passed to callbacks):

app = TSAppSink(name="app", sink_pad_names=["out"], callbacks={"out": on_buffer}, emit_gaps=False)

Multi-pad.

TSPlotSink¶

Extends TSFrameCollectSink with built-in plotting. Collects data during pipeline execution and provides a plot() method for visualization.

from sgnts.sinks import TSPlotSink

sink = TSPlotSink(name="detector_data", sink_pad_names=["H1", "L1"])

After running the pipeline:

# Overlay all pads on one axes
fig, ax = sink.plot(time_unit="s")

# Separate subplots per pad
fig, axes = sink.plot(layout="subplots")

Plot specific pads with custom labels:

fig, ax = sink.plot(pads=["H1"], labels={"H1": "Hanford"})

Gap visualization is enabled by default — gap regions appear as shaded areas. Disable with show_gaps=False.

Requires matplotlib (pip install sgn-ts[plot]). Multi-pad.