Skip to content

sgnts.decorators.sink

Decorators for simplifying sink implementations.

single_pad(method)

Decorator for simple single-pad sinks.

Transforms the simpler signature::

def process(self, input_frame: TSFrame) -> None

Into the dict-based signature expected by TSSink::

def process(self, input_frames: dict[SinkPad, TSFrame]) -> None

Usage::

from sgnts.decorators import sink

class MySink(TSSink):
    @sink.single_pad
    def process(self, input_frame: TSFrame) -> None:
        # Process single input frame
        pass
Source code in src/sgnts/decorators/sink.py 
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
def single_pad(
    method: Callable[[T, TSFrame], None],
) -> Callable[[T, dict[SinkPad, TSFrame]], None]:
    """Decorator for simple single-pad sinks.

    Transforms the simpler signature::

        def process(self, input_frame: TSFrame) -> None

    Into the dict-based signature expected by TSSink::

        def process(self, input_frames: dict[SinkPad, TSFrame]) -> None

    Usage::

        from sgnts.decorators import sink

        class MySink(TSSink):
            @sink.single_pad
            def process(self, input_frame: TSFrame) -> None:
                # Process single input frame
                pass
    """

    @wraps(method)
    def wrapper(
        self: T,
        input_frames: dict[SinkPad, TSFrame],
    ) -> None:
        assert len(self.sink_pads) == 1, (
            f"@sink.single_pad requires exactly one sink pad, "
            f"got {len(self.sink_pads)}"
        )
        return method(self, input_frames[self.sink_pads[0]])

    return wrapper