sgnts.sources.fake_series

FakeSeriesSource dataclass

Bases: TSSource

              flowchart TD
              sgnts.sources.fake_series.FakeSeriesSource[FakeSeriesSource]
              sgnts.base.base.TSSource[TSSource]
              sgnts.base.base._TSSource[_TSSource]

                              sgnts.base.base.TSSource --> sgnts.sources.fake_series.FakeSeriesSource
                                sgnts.base.base._TSSource --> sgnts.base.base.TSSource
                



              click sgnts.sources.fake_series.FakeSeriesSource href "" "sgnts.sources.fake_series.FakeSeriesSource"
              click sgnts.base.base.TSSource href "" "sgnts.base.base.TSSource"
              click sgnts.base.base._TSSource href "" "sgnts.base.base._TSSource"
            

A time-series source that generates fake data in fixed-size buffers.

If start is not specified the current GPS time will be used as the start time.

Parameters:

Name	Type	Description	Default
signals	dict[str, dict[str, Any]] | None	dict, keyed by source pad name, defining the signals to be produced on that pad. The expected values are: signal_type: str, currently supported types: (1) 'white': white noise data. (2) 'sin' or 'sine': sine wave data. (3) 'impulse': creates an impulse data, where the value is one at one sample point, and everywhere else is zero. (4) 'const': constant values as specified by user. (5) 'square' or 'toggle': a 50%-duty square wave alternating between exactly 0 and const at frequency fsin (useful for mocking an on/off gain channel). rate: int, the sample rate of the data sample_shape: tuple[int, ...], the shape of a sample of data, or the shape of the data in each dimension except the last (time) dimension, i.e., sample_shape = data.shape[:-1]. For example, if the data is a multi-dimensional array and has shape=(2, 4, 16) then sample_shape = (2, 4). Note that if data is one dimensional and has shape (16,), sample_shape would be an empty tuple (). fsin: float, wave frequency for 'sin' and 'square' signals. impulse_position: int, impulse position for 'impulse' signals. If -1, then the impulse position will be random. const: int | float, constant value for 'const' signals. dtype: numpy.dtype or str, the dtype to cast the generated data to. If not specified, the native numpy dtype of the generator is used (typically float64, or the dtype of const for 'const' signals). These parameters may be specified directly as keyword arguments during class init, in which case they will be used as the defaults for undefined parameters in the signals dict.	None
ngap	int	int, the frequency to generate gap buffers, will generate a gap buffer every ngap buffers. ngap=0: do not generate gap buffers. ngap=-1: generates gap buffers randomly.	0
random_seed	Optional[int]	int, set the random seed, used for 'white' and 'impulse' signals.	None
real_time	bool	bool, run the source in "real time", such that frames are produced at the rate corresponding to their relative offsets. In real-time mode, start will default to the current GPS time if not otherwise specified.	False
heartbeat_interval	Optional[float]	Optional[float], longest time in seconds this source may stall the pipeline's graph loop waiting for a frame's span to become due in real-time mode. Past that, it emits a zero-length heartbeat frame instead, so the rest of the pipeline keeps ticking and elements that have fallen behind can catch up (at roughly stride / heartbeat_interval times real time). Defaults to a quarter of the stride. Set to float("inf") to always sleep out the full frame, which pins the throughput of every element in the pipeline to real time. Only meaningful with real_time=True.	None

Notes

Thread safety: Marked thread_safe = True. With Pipeline.run(threaded=N) the pad callbacks for this element are dispatched onto worker threads.

Pad layout: N source pads (no sink pads). The N source
pads' ``new`` callbacks CAN run concurrently in the same
wave — that is the per-pad concurrency to keep safe.
``internal`` runs alone.

Per-pad concurrency analysis of ``new``:

- ``self.cnt[pad] += 1`` — per-pad-keyed dict mutation.
  Distinct keys per call → safe under concurrent calls.
- ``np.random.randn`` / ``np.sin`` / ``np.full`` /
  ``np.zeros`` etc. — these release the GIL during the
  bulk array fill, so the work itself parallelizes.
- ``self._next_frame_dict`` / ``self._new_buffer_dict``
  from ``prepare_frame``: per-pad-keyed (the keying lives
  in the parent ``_TSSource``).

**Reproducibility caveat (important):** ``np.random.randn``
and friends draw from NumPy's *global* RNG. The global RNG
is internally thread-safe (no crash), but concurrent
``new`` calls from multiple source pads consume samples
from the shared stream in unspecified order. With threading
enabled, ``random_seed`` no longer determines the per-pad
output. If you need reproducible random output under
threading, either keep this element single-pad, run the
pipeline without ``threaded=...``, or migrate the source
to per-pad ``np.random.default_rng()`` instances.

**Future editors MUST preserve thread safety**: keep all
per-pad state per-pad-keyed. Do NOT call
``np.random.seed(...)`` from ``new``/``internal`` (it
mutates the global RNG state and would race with any
other code in the process that uses it).

Source code in src/sgnts/sources/fake_series.py

@dataclass
class FakeSeriesSource(TSSource):
    """A time-series source that generates fake data in fixed-size buffers.

    If `start` is not specified the current GPS time will be used as the
    start time.

    Args:
        signals:
            dict, keyed by source pad name, defining the signals to be
            produced on that pad.  The expected values are:

            signal_type:
                str, currently supported types: (1) 'white': white
                noise data. (2) 'sin' or 'sine': sine wave data. (3)
                'impulse': creates an impulse data, where the value is
                one at one sample point, and everywhere else is zero.
                (4) 'const': constant values as specified by user.
                (5) 'square' or 'toggle': a 50%-duty square wave
                alternating between exactly 0 and ``const`` at frequency
                ``fsin`` (useful for mocking an on/off gain channel).
            rate:
                int, the sample rate of the data
            sample_shape:
                tuple[int, ...], the shape of a sample of data, or the
                shape of the data in each dimension except the last
                (time) dimension, i.e., sample_shape =
                data.shape[:-1]. For example, if the data is a
                multi-dimensional array and has shape=(2, 4, 16) then
                sample_shape = (2, 4).  Note that if data is one
                dimensional and has shape (16,), sample_shape would be
                an empty tuple ().
            fsin:
                float, wave frequency for 'sin' and 'square' signals.
            impulse_position:
                int, impulse position for 'impulse' signals. If -1,
                then the impulse position will be random.
            const:
                int | float, constant value for 'const' signals.
            dtype:
                numpy.dtype or str, the dtype to cast the generated
                data to.  If not specified, the native numpy dtype of
                the generator is used (typically float64, or the
                dtype of ``const`` for 'const' signals).

            These parameters may be specified directly as keyword
            arguments during class init, in which case they will be
            used as the defaults for undefined parameters in the
            signals dict.
        ngap:
            int, the frequency to generate gap buffers, will generate
            a gap buffer every ngap buffers. ngap=0: do not generate
            gap buffers. ngap=-1: generates gap buffers randomly.
        random_seed:
            int, set the random seed, used for 'white' and 'impulse'
            signals.
        real_time:
            bool, run the source in "real time", such that frames are
            produced at the rate corresponding to their relative
            offsets.  In real-time mode, start will default to the
            current GPS time if not otherwise specified.
        heartbeat_interval:
            Optional[float], longest time in seconds this source may
            stall the pipeline's graph loop waiting for a frame's span
            to become due in real-time mode. Past that, it emits a
            zero-length heartbeat frame instead, so the rest of the
            pipeline keeps ticking and elements that have fallen behind
            can catch up (at roughly stride / heartbeat_interval times
            real time). Defaults to a quarter of the stride. Set to
            float("inf") to always sleep out the full frame, which pins
            the throughput of every element in the pipeline to real
            time. Only meaningful with real_time=True.

    Notes:
        Thread safety:
            Marked ``thread_safe = True``. With
            ``Pipeline.run(threaded=N)`` the pad callbacks for this
            element are dispatched onto worker threads.

            Pad layout: N source pads (no sink pads). The N source
            pads' ``new`` callbacks CAN run concurrently in the same
            wave — that is the per-pad concurrency to keep safe.
            ``internal`` runs alone.

            Per-pad concurrency analysis of ``new``:

            - ``self.cnt[pad] += 1`` — per-pad-keyed dict mutation.
              Distinct keys per call → safe under concurrent calls.
            - ``np.random.randn`` / ``np.sin`` / ``np.full`` /
              ``np.zeros`` etc. — these release the GIL during the
              bulk array fill, so the work itself parallelizes.
            - ``self._next_frame_dict`` / ``self._new_buffer_dict``
              from ``prepare_frame``: per-pad-keyed (the keying lives
              in the parent ``_TSSource``).

            **Reproducibility caveat (important):** ``np.random.randn``
            and friends draw from NumPy's *global* RNG. The global RNG
            is internally thread-safe (no crash), but concurrent
            ``new`` calls from multiple source pads consume samples
            from the shared stream in unspecified order. With threading
            enabled, ``random_seed`` no longer determines the per-pad
            output. If you need reproducible random output under
            threading, either keep this element single-pad, run the
            pipeline without ``threaded=...``, or migrate the source
            to per-pad ``np.random.default_rng()`` instances.

            **Future editors MUST preserve thread safety**: keep all
            per-pad state per-pad-keyed. Do NOT call
            ``np.random.seed(...)`` from ``new``/``internal`` (it
            mutates the global RNG state and would race with any
            other code in the process that uses it).

    """

    thread_safe = True

    signals: dict[str, dict[str, Any]] | None = None
    signal_type: str = "white"
    rate: int = 2048
    sample_shape: tuple[int, ...] = ()
    fsin: float = 5
    impulse_position: int = -1
    const: Union[int, float] = 1
    dtype: Optional[Union[str, np.dtype]] = None
    ngap: int = 0
    random_seed: Optional[int] = None
    real_time: bool = False
    heartbeat_interval: Optional[float] = None

    def __post_init__(self):
        if self.signals is None:
            self.signals = {}

        if self.start is None:
            if self.real_time:
                # FIXME make this aligned to a general buffer (which depends on
                # rate) integers will always be aligned.
                self.start = int(gpsnow())
            else:
                self.start = 0

        # Validate time alignments early as we know the input sample rate
        if self.start is not None:
            Offset.validate_time_alignment(self.start, self.rate, param_name="start")
        if self.end is not None and self.end != np.iinfo(np.int64).max:
            Offset.validate_time_alignment(self.end, self.rate, param_name="end")

        super().__post_init__()

        self.cnt = {p: 0 for p in self.source_pads}

        # setup buffers this gives us the first timestamp / offset too
        for pad in self.source_pads:
            signal = self.signals.get(self.rsrcs[pad], {})
            sample_rate = signal.get("rate", self.rate)
            sample_shape = signal.get("sample_shape", self.sample_shape)
            self.set_pad_buffer_params(
                pad=pad, sample_shape=sample_shape, rate=sample_rate
            )

        # This is guaranteed to be the start time of the element at this point
        self._start_time = self.current_start

        if self.random_seed is not None:
            np.random.seed(self.random_seed)

    def create_data(self, pad: SourcePad, buf: SeriesBuffer) -> Array:
        """Create the fake data.

        Args:
            pad:
                SourcePad, the source pad generating data.
            buf:
                SeriesBuffer, the buffer to create the data for.

        Returns:
            Array, the fake data array.
        """
        offset = buf.offset
        ngap = self.ngap
        cnt = self.cnt[pad]
        metadata: dict[str, int] = {}

        assert (
            self.signals is not None
        ), "Signals dictionary must be initialized before generating data"

        signal = self.signals.get(self.rsrcs[pad], {})
        signal_type = signal.get("signal_type", self.signal_type)

        if (ngap == -1 and np.random.rand(1) > 0.5) or (ngap > 0 and cnt % ngap == 0):
            data = None
        elif signal_type == "white":
            data = np.random.randn(*buf.shape)
        elif signal_type in ["sin", "sine"]:
            data = np.sin(
                2
                * np.pi
                * signal.get("fsin", self.fsin)
                * np.tile(
                    buf.tarr,
                    buf.sample_shape + (1,),
                ),
            )
        elif signal_type == "impulse":
            # return self.create_impulse_data(offset, buf.samples, buf.sample_rate)
            impulse_position = signal.get("impulse_position", self.impulse_position)
            if impulse_position == -1 and self.end is not None:
                impulse_position = np.random.randint(0, int(self.end * buf.sample_rate))
            data = np.zeros(buf.samples)
            current_samples = Offset.tosamples(offset, buf.sample_rate)
            if (
                current_samples <= impulse_position
                and impulse_position < current_samples + buf.samples
            ):
                data[impulse_position - current_samples] = 1
            metadata["impulse_offset"] = Offset.fromsamples(
                impulse_position, buf.sample_rate
            )
        elif signal_type in ["square", "toggle"]:
            # 50%-duty square wave toggling between exactly 0 (off) and
            # `const` (on) at frequency `fsin`. floor(2*f*t) is even on the
            # first half of each 1/f period and odd on the second.
            t = np.tile(buf.tarr, buf.sample_shape + (1,))
            on = np.mod(np.floor(2.0 * signal.get("fsin", self.fsin) * t), 2) == 0
            data = np.where(on, signal.get("const", self.const), 0.0)
        elif signal_type == "const":
            data = np.full(buf.shape, signal.get("const", self.const))
        else:
            msg = f"Unknown signal type '{signal_type}'."
            raise ValueError(msg)

        dtype = signal.get("dtype", self.dtype)
        if data is not None and dtype is not None:
            data = data.astype(np.dtype(dtype), copy=False)

        return data, metadata

    def internal(self):
        super().internal()

        if self.real_time:
            # in real-time mode we want to "release" the data after
            # the time of the last sample in the output frame.
            sleep = self.current_end - gpsnow()
            if sleep < 0:
                if sleep < -1:
                    logger.getChild(self.name).warning(
                        "Warning: FakeSeriesSource falling behind real time (%.2f s)",
                        sleep,
                    )
            else:
                # Never stall the shared graph loop longer than the
                # heartbeat interval: sleeping out the full frame here
                # would hold up *every* element in the pipeline, pinning
                # it at exactly real time so that nothing behind could
                # ever catch up. If the frame is still not due after the
                # bounded sleep, new() emits a heartbeat instead.
                time.sleep(
                    min(sleep, resolve_heartbeat_interval(self.heartbeat_interval))
                )

    def output_prototype(self, pad: SourcePad) -> Array:
        """Declare the numpy spec this pad emits.

        FakeSeriesSource is numpy, but ``ngap`` makes its *first* frame an all-gap
        frame (``ngap > 0`` gaps at ``cnt == 0``; ``ngap == -1`` gaps randomly), so
        the spec can't always be inferred from data -- declare it. Backend/device
        enforcement only checks numpy/cpu, so the dtype here only needs to match
        what gaps should materialize as.
        """
        assert self.signals is not None
        signal = self.signals.get(self.rsrcs[pad], {})
        dtype = signal.get("dtype", self.dtype)
        if dtype is not None:
            return np.zeros(0, dtype=np.dtype(dtype))
        signal_type = signal.get("signal_type", self.signal_type)
        if signal_type == "const":
            return np.zeros(0, dtype=np.asarray(signal.get("const", self.const)).dtype)
        return np.zeros(0)

    def new(self, pad: SourcePad) -> TSFrame:
        """New buffers are created on "pad" with an instance specific count and a name
        derived from the pad name. "EOS" is set if we have surpassed the requested
        end time.

        Args:
            pad:
                SourcePad, the source pad to generate TSFrames.

        Returns:
            TSFrame, the TSFrame that carries the buffers with fake data.
        """
        latest_offset = None
        if self.real_time:
            # A simulated live source only "has" data up to the wall clock:
            # pass it as latest_offset so prepare_frame emits a heartbeat
            # when the frame's span is not yet due (internal() sleeps at
            # most the heartbeat interval, not the full frame). Clamped to
            # the frame offset to allow a start time in the future.
            latest_offset = max(
                Offset.fromsec(gpsnow()),
                self._next_frame_dict[pad].offset,
            )

        metadata = {"name": f"{self.rsrcs[pad]}", "cnt": self.cnt[pad] + 1}

        frame = self.prepare_frame(
            pad, data=None, metadata=metadata, latest_offset=latest_offset
        )
        if frame.offset == frame.end_offset:
            # Zero-length heartbeat: the frame's span is not due yet
            # (real_time) or the source is past its end. Nothing to
            # synthesize and no count consumed.
            return frame
        self.cnt[pad] += 1
        for buf in frame:
            data, _metadata = self.create_data(pad, buf)
            buf.set_data(data)
            metadata.update(_metadata)

        # Update the frame attrs post buffer editing
        frame.validate_buffers()
        frame.update_buffer_attrs()

        return frame

create_data(pad, buf)

Create the fake data.

Parameters:

Name	Type	Description	Default
pad	SourcePad	SourcePad, the source pad generating data.	required
buf	SeriesBuffer	SeriesBuffer, the buffer to create the data for.	required

Returns:

Type	Description
Array	Array, the fake data array.

Source code in src/sgnts/sources/fake_series.py

def create_data(self, pad: SourcePad, buf: SeriesBuffer) -> Array:
    """Create the fake data.

    Args:
        pad:
            SourcePad, the source pad generating data.
        buf:
            SeriesBuffer, the buffer to create the data for.

    Returns:
        Array, the fake data array.
    """
    offset = buf.offset
    ngap = self.ngap
    cnt = self.cnt[pad]
    metadata: dict[str, int] = {}

    assert (
        self.signals is not None
    ), "Signals dictionary must be initialized before generating data"

    signal = self.signals.get(self.rsrcs[pad], {})
    signal_type = signal.get("signal_type", self.signal_type)

    if (ngap == -1 and np.random.rand(1) > 0.5) or (ngap > 0 and cnt % ngap == 0):
        data = None
    elif signal_type == "white":
        data = np.random.randn(*buf.shape)
    elif signal_type in ["sin", "sine"]:
        data = np.sin(
            2
            * np.pi
            * signal.get("fsin", self.fsin)
            * np.tile(
                buf.tarr,
                buf.sample_shape + (1,),
            ),
        )
    elif signal_type == "impulse":
        # return self.create_impulse_data(offset, buf.samples, buf.sample_rate)
        impulse_position = signal.get("impulse_position", self.impulse_position)
        if impulse_position == -1 and self.end is not None:
            impulse_position = np.random.randint(0, int(self.end * buf.sample_rate))
        data = np.zeros(buf.samples)
        current_samples = Offset.tosamples(offset, buf.sample_rate)
        if (
            current_samples <= impulse_position
            and impulse_position < current_samples + buf.samples
        ):
            data[impulse_position - current_samples] = 1
        metadata["impulse_offset"] = Offset.fromsamples(
            impulse_position, buf.sample_rate
        )
    elif signal_type in ["square", "toggle"]:
        # 50%-duty square wave toggling between exactly 0 (off) and
        # `const` (on) at frequency `fsin`. floor(2*f*t) is even on the
        # first half of each 1/f period and odd on the second.
        t = np.tile(buf.tarr, buf.sample_shape + (1,))
        on = np.mod(np.floor(2.0 * signal.get("fsin", self.fsin) * t), 2) == 0
        data = np.where(on, signal.get("const", self.const), 0.0)
    elif signal_type == "const":
        data = np.full(buf.shape, signal.get("const", self.const))
    else:
        msg = f"Unknown signal type '{signal_type}'."
        raise ValueError(msg)

    dtype = signal.get("dtype", self.dtype)
    if data is not None and dtype is not None:
        data = data.astype(np.dtype(dtype), copy=False)

    return data, metadata

new(pad)

New buffers are created on "pad" with an instance specific count and a name derived from the pad name. "EOS" is set if we have surpassed the requested end time.

Parameters:

Name	Type	Description	Default
pad	SourcePad	SourcePad, the source pad to generate TSFrames.	required

Returns:

Type	Description
TSFrame	TSFrame, the TSFrame that carries the buffers with fake data.

Source code in src/sgnts/sources/fake_series.py

def new(self, pad: SourcePad) -> TSFrame:
    """New buffers are created on "pad" with an instance specific count and a name
    derived from the pad name. "EOS" is set if we have surpassed the requested
    end time.

    Args:
        pad:
            SourcePad, the source pad to generate TSFrames.

    Returns:
        TSFrame, the TSFrame that carries the buffers with fake data.
    """
    latest_offset = None
    if self.real_time:
        # A simulated live source only "has" data up to the wall clock:
        # pass it as latest_offset so prepare_frame emits a heartbeat
        # when the frame's span is not yet due (internal() sleeps at
        # most the heartbeat interval, not the full frame). Clamped to
        # the frame offset to allow a start time in the future.
        latest_offset = max(
            Offset.fromsec(gpsnow()),
            self._next_frame_dict[pad].offset,
        )

    metadata = {"name": f"{self.rsrcs[pad]}", "cnt": self.cnt[pad] + 1}

    frame = self.prepare_frame(
        pad, data=None, metadata=metadata, latest_offset=latest_offset
    )
    if frame.offset == frame.end_offset:
        # Zero-length heartbeat: the frame's span is not due yet
        # (real_time) or the source is past its end. Nothing to
        # synthesize and no count consumed.
        return frame
    self.cnt[pad] += 1
    for buf in frame:
        data, _metadata = self.create_data(pad, buf)
        buf.set_data(data)
        metadata.update(_metadata)

    # Update the frame attrs post buffer editing
    frame.validate_buffers()
    frame.update_buffer_attrs()

    return frame

output_prototype(pad)

Declare the numpy spec this pad emits.

FakeSeriesSource is numpy, but ngap makes its first frame an all-gap frame (ngap > 0 gaps at cnt == 0; ngap == -1 gaps randomly), so the spec can't always be inferred from data -- declare it. Backend/device enforcement only checks numpy/cpu, so the dtype here only needs to match what gaps should materialize as.

Source code in src/sgnts/sources/fake_series.py

def output_prototype(self, pad: SourcePad) -> Array:
    """Declare the numpy spec this pad emits.

    FakeSeriesSource is numpy, but ``ngap`` makes its *first* frame an all-gap
    frame (``ngap > 0`` gaps at ``cnt == 0``; ``ngap == -1`` gaps randomly), so
    the spec can't always be inferred from data -- declare it. Backend/device
    enforcement only checks numpy/cpu, so the dtype here only needs to match
    what gaps should materialize as.
    """
    assert self.signals is not None
    signal = self.signals.get(self.rsrcs[pad], {})
    dtype = signal.get("dtype", self.dtype)
    if dtype is not None:
        return np.zeros(0, dtype=np.dtype(dtype))
    signal_type = signal.get("signal_type", self.signal_type)
    if signal_type == "const":
        return np.zeros(0, dtype=np.asarray(signal.get("const", self.const)).dtype)
    return np.zeros(0)