Skip to content

sgnts.sinks.appsink

TSAppSink dataclass

Bases: TSSink


              flowchart TD
              sgnts.sinks.appsink.TSAppSink[TSAppSink]
              sgnts.base.base.TSSink[TSSink]
              sgnts.base.base.TimeSeriesMixin[TimeSeriesMixin]

                              sgnts.base.base.TSSink --> sgnts.sinks.appsink.TSAppSink
                                sgnts.base.base.TimeSeriesMixin --> sgnts.base.base.TSSink
                



              click sgnts.sinks.appsink.TSAppSink href "" "sgnts.sinks.appsink.TSAppSink"
              click sgnts.base.base.TSSink href "" "sgnts.base.base.TSSink"
              click sgnts.base.base.TimeSeriesMixin href "" "sgnts.base.base.TimeSeriesMixin"

Application sink that executes user-defined callbacks on each buffer.

Similar to GStreamer's appsink, this element allows application code to receive and process buffers via callbacks.

Callbacks are registered per-pad and are invoked for each SeriesBuffer that arrives on that pad.

Example

def my_callback(pad: SinkPad, buf: SeriesBuffer) -> None: print(f"Received buffer on {pad.name}: {buf.slice}")

Via constructor

sink = TSAppSink( name="app_sink", sink_pad_names=("H1", "L1"), callbacks={"H1": my_callback} )

Via method

sink.set_callback("L1", another_callback)

Parameters:

Name Type Description Default
callbacks dict[str, BufferCallback]

Optional dict mapping pad names to callback functions.

 dict()
emit_gaps bool

If True, callbacks are also invoked for gap buffers. Default is True (gap buffers are included).

 True
Notes

Thread safety: Marked thread_safe = True. 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 ``TimeSeriesMixin.pull``):
per-pad-keyed dict writes; safe across pads. ``process``
(called from ``internal``, runs alone): reads
``self._callbacks`` (populated in ``configure()``,
read-only) and invokes the per-pad callback for each
buffer. Calls ``self.mark_eos(pad)`` which writes
``self._at_eos[pad]`` (per-pad-keyed).

Speedup from threading depends on what the user callback
does: NumPy/SciPy/I/O callbacks release the GIL and
scale; pure-Python callbacks won't.

**Future editors MUST preserve thread safety** of the sink
itself: keep callback registration in ``configure()``
(read-only at runtime) and don't add element-level state
mutated from ``pull`` outside per-pad-keyed containers.
User-supplied callbacks are the user's responsibility —
they must be reentrant if they share state across pads.
Source code in src/sgnts/sinks/appsink.py 
 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
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
@dataclass
class TSAppSink(TSSink):
    """
    Application sink that executes user-defined callbacks on each buffer.

    Similar to GStreamer's appsink, this element allows application code
    to receive and process buffers via callbacks.

    Callbacks are registered per-pad and are invoked for each SeriesBuffer
    that arrives on that pad.

    Example:
        def my_callback(pad: SinkPad, buf: SeriesBuffer) -> None:
            print(f"Received buffer on {pad.name}: {buf.slice}")

        # Via constructor
        sink = TSAppSink(
            name="app_sink",
            sink_pad_names=("H1", "L1"),
            callbacks={"H1": my_callback}
        )

        # Via method
        sink.set_callback("L1", another_callback)

    Args:
        callbacks: Optional dict mapping pad names to callback functions.
        emit_gaps: If True, callbacks are also invoked for gap buffers.
                   Default is True (gap buffers are included).

    Notes:
        Thread safety:
            Marked ``thread_safe = True``. 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 ``TimeSeriesMixin.pull``):
            per-pad-keyed dict writes; safe across pads. ``process``
            (called from ``internal``, runs alone): reads
            ``self._callbacks`` (populated in ``configure()``,
            read-only) and invokes the per-pad callback for each
            buffer. Calls ``self.mark_eos(pad)`` which writes
            ``self._at_eos[pad]`` (per-pad-keyed).

            Speedup from threading depends on what the user callback
            does: NumPy/SciPy/I/O callbacks release the GIL and
            scale; pure-Python callbacks won't.

            **Future editors MUST preserve thread safety** of the sink
            itself: keep callback registration in ``configure()``
            (read-only at runtime) and don't add element-level state
            mutated from ``pull`` outside per-pad-keyed containers.
            User-supplied callbacks are the user's responsibility —
            they must be reentrant if they share state across pads.
    """

    thread_safe = True

    callbacks: dict[str, BufferCallback] = field(default_factory=dict)
    emit_gaps: bool = True

    # Internal storage for registered callbacks
    _callbacks: dict[SinkPad, BufferCallback] = field(
        default_factory=dict, init=False, repr=False
    )

    def configure(self) -> None:
        """Register callbacks from constructor dict."""
        super().configure()

        # Convert name-based callbacks to pad-based
        for pad_name, callback in self.callbacks.items():
            if pad_name not in self.snks:
                raise ValueError(
                    f"Unknown pad name '{pad_name}'. "
                    f"Available pads: {list(self.snks.keys())}"
                )
            self._callbacks[self.snks[pad_name]] = callback

    def set_callback(self, pad_name: str, callback: BufferCallback) -> None:
        """
        Register a callback for a specific pad.

        Args:
            pad_name: Name of the sink pad.
            callback: Function to call for each buffer on this pad.
                      Signature: (pad: SinkPad, buf: SeriesBuffer) -> None

        Raises:
            ValueError: If pad_name is not a valid sink pad.
        """
        if pad_name not in self.snks:
            raise ValueError(
                f"Unknown pad name '{pad_name}'. "
                f"Available pads: {list(self.snks.keys())}"
            )
        self._callbacks[self.snks[pad_name]] = callback

    def remove_callback(self, pad_name: str) -> None:
        """
        Remove the callback for a specific pad.

        Args:
            pad_name: Name of the sink pad.
        """
        if pad_name in self.snks:
            pad = self.snks[pad_name]
            self._callbacks.pop(pad, None)

    def get_callback(self, pad_name: str) -> BufferCallback | None:
        """
        Get the callback registered for a pad.

        Args:
            pad_name: Name of the sink pad.

        Returns:
            The callback function or None if not registered.
        """
        if pad_name not in self.snks:
            return None
        return self._callbacks.get(self.snks[pad_name])

    def process(self, input_frames: dict[SinkPad, TSFrame]) -> None:
        """Process incoming frames by invoking callbacks for each buffer."""
        for pad, frame in input_frames.items():
            callback = self._callbacks.get(pad)

            if callback is None:
                # No callback registered for this pad - skip
                continue

            for buf in frame:
                # Skip gaps unless emit_gaps is True
                if buf.is_gap and not self.emit_gaps:
                    continue

                callback(pad, buf)

            # Handle EOS
            if frame.EOS:
                self.mark_eos(pad)

configure()

Register callbacks from constructor dict.

Source code in src/sgnts/sinks/appsink.py 
78
79
80
81
82
83
84
85
86
87
88
89
def configure(self) -> None:
    """Register callbacks from constructor dict."""
    super().configure()

    # Convert name-based callbacks to pad-based
    for pad_name, callback in self.callbacks.items():
        if pad_name not in self.snks:
            raise ValueError(
                f"Unknown pad name '{pad_name}'. "
                f"Available pads: {list(self.snks.keys())}"
            )
        self._callbacks[self.snks[pad_name]] = callback

get_callback(pad_name)

Get the callback registered for a pad.

Parameters:

Name Type Description Default
pad_name str

Name of the sink pad.

 required

Returns:

Type Description
BufferCallback | None

The callback function or None if not registered.
Source code in src/sgnts/sinks/appsink.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
def get_callback(self, pad_name: str) -> BufferCallback | None:
    """
    Get the callback registered for a pad.

    Args:
        pad_name: Name of the sink pad.

    Returns:
        The callback function or None if not registered.
    """
    if pad_name not in self.snks:
        return None
    return self._callbacks.get(self.snks[pad_name])

process(input_frames)

Process incoming frames by invoking callbacks for each buffer.

Source code in src/sgnts/sinks/appsink.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def process(self, input_frames: dict[SinkPad, TSFrame]) -> None:
    """Process incoming frames by invoking callbacks for each buffer."""
    for pad, frame in input_frames.items():
        callback = self._callbacks.get(pad)

        if callback is None:
            # No callback registered for this pad - skip
            continue

        for buf in frame:
            # Skip gaps unless emit_gaps is True
            if buf.is_gap and not self.emit_gaps:
                continue

            callback(pad, buf)

        # Handle EOS
        if frame.EOS:
            self.mark_eos(pad)

remove_callback(pad_name)

Remove the callback for a specific pad.

Parameters:

Name Type Description Default
pad_name str

Name of the sink pad.

 required
Source code in src/sgnts/sinks/appsink.py 
110
111
112
113
114
115
116
117
118
119
def remove_callback(self, pad_name: str) -> None:
    """
    Remove the callback for a specific pad.

    Args:
        pad_name: Name of the sink pad.
    """
    if pad_name in self.snks:
        pad = self.snks[pad_name]
        self._callbacks.pop(pad, None)

set_callback(pad_name, callback)

Register a callback for a specific pad.

Parameters:

Name Type Description Default
pad_name str

Name of the sink pad.

 required
callback BufferCallback

Function to call for each buffer on this pad. Signature: (pad: SinkPad, buf: SeriesBuffer) -> None

 required

Raises:

Type Description
ValueError

If pad_name is not a valid sink pad.
Source code in src/sgnts/sinks/appsink.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def set_callback(self, pad_name: str, callback: BufferCallback) -> None:
    """
    Register a callback for a specific pad.

    Args:
        pad_name: Name of the sink pad.
        callback: Function to call for each buffer on this pad.
                  Signature: (pad: SinkPad, buf: SeriesBuffer) -> None

    Raises:
        ValueError: If pad_name is not a valid sink pad.
    """
    if pad_name not in self.snks:
        raise ValueError(
            f"Unknown pad name '{pad_name}'. "
            f"Available pads: {list(self.snks.keys())}"
        )
    self._callbacks[self.snks[pad_name]] = callback