Skip to content

sgnts.decorators.transform

Decorators for simplifying transform implementations.

many_to_one(method)

Decorator for many-to-one transforms.

Transforms the simpler signature::

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

Into the dict-based signature expected by TSTransform::

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

Usage::

from sgnts.decorators import transform

class MyTransform(TSTransform):
    @transform.many_to_one
    def process(
        self, input_frames: dict[SinkPad, TSFrame], output_frame: TSCollectFrame
    ) -> None:
        # Process multiple inputs to single output
        pass
Source code in src/sgnts/decorators/transform.py 
 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
def many_to_one(
    method: Callable[[T, dict[SinkPad, TSFrame], TSCollectFrame], None],
) -> Callable[[T, dict[SinkPad, TSFrame], dict[SourcePad, TSCollectFrame]], None]:
    """Decorator for many-to-one transforms.

    Transforms the simpler signature::

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

    Into the dict-based signature expected by TSTransform::

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

    Usage::

        from sgnts.decorators import transform

        class MyTransform(TSTransform):
            @transform.many_to_one
            def process(
                self, input_frames: dict[SinkPad, TSFrame], output_frame: TSCollectFrame
            ) -> None:
                # Process multiple inputs to single output
                pass
    """

    @wraps(method)
    def wrapper(
        self: T,
        input_frames: dict[SinkPad, TSFrame],
        output_frames: dict[SourcePad, TSCollectFrame],
    ) -> None:
        assert len(self.source_pads) == 1, (
            f"@transform.many_to_one requires exactly one source pad, "
            f"got {len(self.source_pads)}"
        )
        return method(self, input_frames, output_frames[self.source_pads[0]])

    return wrapper

one_to_one(method)

Decorator for simple one-to-one transforms.

Transforms the simpler signature::

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

Into the dict-based signature expected by TSTransform::

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

Usage::

from sgnts.decorators import transform

class MyTransform(TSTransform):
    @transform.one_to_one
    def process(
        self, input_frame: TSFrame, output_frame: TSCollectFrame
    ) -> None:
        for buf in input_frame:
            output_frame.append(buf)
Source code in src/sgnts/decorators/transform.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
52
53
54
55
56
57
58
def one_to_one(
    method: Callable[[T, TSFrame, TSCollectFrame], None],
) -> Callable[[T, dict[SinkPad, TSFrame], dict[SourcePad, TSCollectFrame]], None]:
    """Decorator for simple one-to-one transforms.

    Transforms the simpler signature::

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

    Into the dict-based signature expected by TSTransform::

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

    Usage::

        from sgnts.decorators import transform

        class MyTransform(TSTransform):
            @transform.one_to_one
            def process(
                self, input_frame: TSFrame, output_frame: TSCollectFrame
            ) -> None:
                for buf in input_frame:
                    output_frame.append(buf)
    """

    @wraps(method)
    def wrapper(
        self: T,
        input_frames: dict[SinkPad, TSFrame],
        output_frames: dict[SourcePad, TSCollectFrame],
    ) -> None:
        assert len(self.sink_pads) == 1 and len(self.source_pads) == 1, (
            f"@transform.one_to_one requires exactly one sink and source pad, "
            f"got {len(self.sink_pads)} sink pads and "
            f"{len(self.source_pads)} source pads"
        )
        return method(
            self, input_frames[self.sink_pads[0]], output_frames[self.source_pads[0]]
        )

    return wrapper