sgnts.sources.iter

Frame iterator source element.

TSIterSource dataclass

Bases: TSSource

              flowchart TD
              sgnts.sources.iter.TSIterSource[TSIterSource]
              sgnts.base.base.TSSource[TSSource]
              sgnts.base.base._TSSource[_TSSource]

                              sgnts.base.base.TSSource --> sgnts.sources.iter.TSIterSource
                                sgnts.base.base._TSSource --> sgnts.base.base.TSSource
                



              click sgnts.sources.iter.TSIterSource href "" "sgnts.sources.iter.TSIterSource"
              click sgnts.base.base.TSSource href "" "sgnts.base.base.TSSource"
              click sgnts.base.base._TSSource href "" "sgnts.base.base._TSSource"
            

A source that iterates through a provided list of TSFrames.

This is useful for testing, allowing you to provide pre-constructed frames (including multi-buffer frames) to a pipeline.

Parameters:

Name	Type	Description	Default
frames	list[TSFrame]	list[TSFrame], the frames to iterate through. After the last frame is sent, an EOS frame will be sent automatically.	required

Notes

Thread safety: Deliberately NOT marked thread_safe. Pad layout: N source pads (configurable). The N source pads' new callbacks would run concurrently in the same wave under Pipeline.run(threaded=...), but new calls next(self.frame_iter) on a single shared iterator. Concurrent next calls would race over iterator state and StopIteration handling, producing undefined interleaving of frames across pads. Single-pad use is safe but multi-pad use is not, so the conservative default is to leave the element on the event-loop thread always.

If parallel iteration is needed, subclass with per-pad
iterators (one ``iter()`` per source pad, indexed by pad
in ``new``) and set ``thread_safe = True`` on the subclass.

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

@dataclass(kw_only=True)
class TSIterSource(TSSource):
    """A source that iterates through a provided list of TSFrames.

    This is useful for testing, allowing you to provide pre-constructed
    frames (including multi-buffer frames) to a pipeline.

    Args:
        frames:
            list[TSFrame], the frames to iterate through. After the last
            frame is sent, an EOS frame will be sent automatically.

    Notes:
        Thread safety:
            **Deliberately NOT marked thread_safe.** Pad layout: N
            source pads (configurable). The N source pads' ``new``
            callbacks would run concurrently in the same wave under
            ``Pipeline.run(threaded=...)``, but ``new`` calls
            ``next(self.frame_iter)`` on a single shared iterator.
            Concurrent ``next`` calls would race over iterator state
            and ``StopIteration`` handling, producing undefined
            interleaving of frames across pads. Single-pad use is
            safe but multi-pad use is not, so the conservative default
            is to leave the element on the event-loop thread always.

            If parallel iteration is needed, subclass with per-pad
            iterators (one ``iter()`` per source pad, indexed by pad
            in ``new``) and set ``thread_safe = True`` on the subclass.
    """

    frames: list[TSFrame]

    def __post_init__(self) -> None:
        super().__post_init__()
        self.frames[-1].EOS = True
        self.last_frame = self.frames[-1]
        self.frame_iter = iter(self.frames)

    def output_prototype(self, pad: SourcePad):
        """Declare the spec from the provided frames.

        Needed when the *first* frame is an all-gap frame (so the spec can't be
        inferred from it) and for the trailing heartbeats emitted after the
        iterator is exhausted -- both must carry the same backend as the real
        frames. A zero-length slice of the first real buffer preserves its
        namespace, device, and dtype exactly. Falls back to the numpy default
        when every provided frame is a gap.
        """
        for frame in self.frames:
            for buf in frame:
                if not buf.is_gap and getattr(buf.data, "dtype", None) is not None:
                    return buf.data[:0]
        return super().output_prototype(pad)

    def new(self, pad: SourcePad) -> TSFrame:
        try:
            return next(self.frame_iter)
        except StopIteration:
            # Return heartbeats after iterator is exhausted
            return self.last_frame.heartbeat()

output_prototype(pad)

Declare the spec from the provided frames.

Needed when the first frame is an all-gap frame (so the spec can't be inferred from it) and for the trailing heartbeats emitted after the iterator is exhausted -- both must carry the same backend as the real frames. A zero-length slice of the first real buffer preserves its namespace, device, and dtype exactly. Falls back to the numpy default when every provided frame is a gap.

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

def output_prototype(self, pad: SourcePad):
    """Declare the spec from the provided frames.

    Needed when the *first* frame is an all-gap frame (so the spec can't be
    inferred from it) and for the trailing heartbeats emitted after the
    iterator is exhausted -- both must carry the same backend as the real
    frames. A zero-length slice of the first real buffer preserves its
    namespace, device, and dtype exactly. Falls back to the numpy default
    when every provided frame is a gap.
    """
    for frame in self.frames:
        for buf in frame:
            if not buf.is_gap and getattr(buf.data, "dtype", None) is not None:
                return buf.data[:0]
    return super().output_prototype(pad)