Sources¶

All sources are imported from sgnts.sources.

FakeSeriesSource¶

Generates synthetic time-series data. Useful for testing and prototyping.

from sgnts.sources import FakeSeriesSource

# White noise at 2048 Hz for 2 seconds
src = FakeSeriesSource(
    name="src",
    source_pad_names=["out"],
    signal_type="white",
    rate=2048,
    duration=2,
)

Signal types:

Type	Description
`"white"`	Gaussian white noise
`"sin"`	Sine wave (frequency set by `fsin`, default 5 Hz)
`"impulse"`	Single non-zero sample (position set by `impulse_position`)
`"const"`	Constant value (set by `const`, default 1)

Multi-pad with per-pad configuration via the signals dictionary:

src = FakeSeriesSource(
    name="src",
    source_pad_names=["H1", "L1"],
    signals={
        "H1": {"signal_type": "sin", "rate": 4096, "fsin": 10},
        "L1": {"signal_type": "white", "rate": 2048},
    },
    duration=5,
)

Parameters not specified per-pad fall back to the top-level defaults (signal_type, rate, sample_shape, etc.).

Multi-dimensional data with sample_shape:

# 3-channel data: shape (3, 2048) per buffer
src = FakeSeriesSource(
    name="src",
    source_pad_names=["out"],
    rate=2048,
    sample_shape=(3,),
    duration=1,
)

Gap generation with ngap:

# Generate a gap buffer every 5th buffer
src = FakeSeriesSource(name="src", source_pad_names=["out"], ngap=5, duration=10)

Other parameters: start (start time in seconds), end (end time), random_seed, real_time (when True, buffers are emitted at wall-clock rate — useful for simulating live data streams).

SegmentSource¶

Produces non-gap data only within specified time segments. Outside those segments, gap buffers are produced.

from sgnts.sources import SegmentSource
from sgnts.base import Time

seg = SegmentSource(
    name="seg",
    source_pad_names=["out"],
    rate=2048,
    segments=(
        (0, 2 * Time.SECONDS),
        (5 * Time.SECONDS, 7 * Time.SECONDS),
    ),
    start=0,
    duration=10,
)

Custom values per segment with values:

seg = SegmentSource(
    name="seg",
    source_pad_names=["out"],
    rate=2048,
    segments=((0, 2 * Time.SECONDS),),
    values=(5,),  # Fill with 5 instead of default 1
    start=0,
    duration=3,
)

Segments must be non-overlapping. Segments partially outside the start/end range are clipped automatically.

TSIterSource¶

Iterates through a provided list of pre-built TSFrames. After the last frame, an EOS frame is sent automatically.

import numpy as np
from sgnts.base import SeriesBuffer, TSFrame
from sgnts.sources import TSIterSource

frames = [
    TSFrame(buffers=[SeriesBuffer(offset=0, sample_rate=2048, data=np.ones(2048))]),
    TSFrame(buffers=[SeriesBuffer(offset=16384, sample_rate=2048, data=np.zeros(2048))]),
]

src = TSIterSource(name="src", source_pad_names=["out"], frames=frames)

Useful for testing pipelines with specific data patterns.