Skip to content

sgnts.sinks.collect

TSFrameCollectSink dataclass

Bases: make_ts_element(CollectSink)


              flowchart TD
              sgnts.sinks.collect.TSFrameCollectSink[TSFrameCollectSink]

                              sgnts.base.base.make_ts_element --> sgnts.sinks.collect.TSFrameCollectSink
                


              click sgnts.sinks.collect.TSFrameCollectSink href "" "sgnts.sinks.collect.TSFrameCollectSink"

Sink that collects input SeriesBuffers

sgn.CollectSink with an additional method out_frames that will return a dictionary, keyed by sink pad names, where the values are single TSFrames containing all buffers collected on the sink pads during pipeline operation.

Notes

Thread safety: Marked thread_safe = True (inherited by TSPlotSink, which only adds a post-pipeline plot method that runs after run() returns).

Pad layout: N sink pads (no source pads). The N sink pads'
``pull`` callbacks CAN run concurrently in the same wave;
``internal`` runs alone.

``pull`` (inherited ``CollectSink.pull``): writes
``self.collects[pad_name].append(...)`` — per-pad-keyed
list, distinct keys per pad. Calls ``self.mark_eos(pad)``
on EOS, which writes ``self._at_eos[pad]`` (per-pad-keyed).
Both safe under concurrent calls from different pads.
``process``/``internal``: nothing element-mutating.

Speedup from threading is negligible (no GIL-releasing
compute) but marking thread_safe avoids forcing a
serialization point against neighboring thread_safe
elements. ``out_frames()`` is a post-pipeline read of
``self.collects`` and is not on the threaded path.

**Future editors MUST preserve thread safety**: keep
``pull`` mutations per-pad-keyed.
Source code in src/sgnts/sinks/collect.py 
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
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@dataclass
class TSFrameCollectSink(make_ts_element(CollectSink)):  # type: ignore[misc]
    """Sink that collects input SeriesBuffers

    sgn.CollectSink with an additional method `out_frames` that will
    return a dictionary, keyed by sink pad names, where the values are
    single TSFrames containing all buffers collected on the sink pads
    during pipeline operation.

    Notes:
        Thread safety:
            Marked ``thread_safe = True`` (inherited by
            ``TSPlotSink``, which only adds a post-pipeline ``plot``
            method that runs after ``run()`` returns).

            Pad layout: N sink pads (no source pads). The N sink pads'
            ``pull`` callbacks CAN run concurrently in the same wave;
            ``internal`` runs alone.

            ``pull`` (inherited ``CollectSink.pull``): writes
            ``self.collects[pad_name].append(...)`` — per-pad-keyed
            list, distinct keys per pad. Calls ``self.mark_eos(pad)``
            on EOS, which writes ``self._at_eos[pad]`` (per-pad-keyed).
            Both safe under concurrent calls from different pads.
            ``process``/``internal``: nothing element-mutating.

            Speedup from threading is negligible (no GIL-releasing
            compute) but marking thread_safe avoids forcing a
            serialization point against neighboring thread_safe
            elements. ``out_frames()`` is a post-pipeline read of
            ``self.collects`` and is not on the threaded path.

            **Future editors MUST preserve thread safety**: keep
            ``pull`` mutations per-pad-keyed.

    """

    thread_safe = True

    def __post_init__(self):
        self.extract_data = False
        self.skip_empty = False
        super().__post_init__()

    def out_frames(self) -> dict[str, TSFrame]:
        """The collected frames."""
        out = {}
        for pad_name, frames in self.collects.items():
            buffers = []
            for frame in frames:
                # Filter out heartbeat buffers (zero-duration buffers)
                buffers.extend([buf for buf in frame.buffers if buf.duration > 0])
            out[pad_name] = TSFrame(buffers=buffers)
        return out

out_frames()

The collected frames.

Source code in src/sgnts/sinks/collect.py 
55
56
57
58
59
60
61
62
63
64
def out_frames(self) -> dict[str, TSFrame]:
    """The collected frames."""
    out = {}
    for pad_name, frames in self.collects.items():
        buffers = []
        for frame in frames:
            # Filter out heartbeat buffers (zero-duration buffers)
            buffers.extend([buf for buf in frame.buffers if buf.duration > 0])
        out[pad_name] = TSFrame(buffers=buffers)
    return out