Skip to content

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
Source code in src/sgnts/sources/iter.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
@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.
    """

    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 new(self, pad: SourcePad) -> TSFrame:
        try:
            return next(self.frame_iter)
        except StopIteration:
            # Return heartbeats after iterator is exhausted
            return self.last_frame.heartbeat()