Skip to content

sgnts.transforms.nary

NAry transforms.

Multiply dataclass

Bases: NaryTransform


              flowchart TD
              sgnts.transforms.nary.Multiply[Multiply]
              sgnts.transforms.nary.NaryTransform[NaryTransform]
              sgnts.base.base.TSTransform[TSTransform]
              sgnts.base.base.TimeSeriesMixin[TimeSeriesMixin]

                              sgnts.transforms.nary.NaryTransform --> sgnts.transforms.nary.Multiply
                                sgnts.base.base.TSTransform --> sgnts.transforms.nary.NaryTransform
                                sgnts.base.base.TimeSeriesMixin --> sgnts.base.base.TSTransform
                




              click sgnts.transforms.nary.Multiply href "" "sgnts.transforms.nary.Multiply"
              click sgnts.transforms.nary.NaryTransform href "" "sgnts.transforms.nary.NaryTransform"
              click sgnts.base.base.TSTransform href "" "sgnts.base.base.TSTransform"
              click sgnts.base.base.TimeSeriesMixin href "" "sgnts.base.base.TimeSeriesMixin"

Multiply transform

Source code in src/sgnts/transforms/nary.py 
92
93
94
95
96
97
@dataclass
class Multiply(NaryTransform):
    """Multiply transform"""

    def configure(self) -> None:
        self.op = _multiply

NaryTransform dataclass

Bases: TSTransform


              flowchart TD
              sgnts.transforms.nary.NaryTransform[NaryTransform]
              sgnts.base.base.TSTransform[TSTransform]
              sgnts.base.base.TimeSeriesMixin[TimeSeriesMixin]

                              sgnts.base.base.TSTransform --> sgnts.transforms.nary.NaryTransform
                                sgnts.base.base.TimeSeriesMixin --> sgnts.base.base.TSTransform
                



              click sgnts.transforms.nary.NaryTransform href "" "sgnts.transforms.nary.NaryTransform"
              click sgnts.base.base.TSTransform href "" "sgnts.base.base.TSTransform"
              click sgnts.base.base.TimeSeriesMixin href "" "sgnts.base.base.TimeSeriesMixin"

N-ary transform. Takes N inputs and applies a function to them frame by frame.

Parameters:

Name Type Description Default
op Callable | None

Callable, the operation to apply to the inputs. Must take N arguments, where N is the number of sink pads, and return a single output.

 None
Source code in src/sgnts/transforms/nary.py 
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
@dataclass
class NaryTransform(TSTransform):
    """N-ary transform. Takes N inputs and applies a function to them
    frame by frame.

    Args:
        op:
            Callable, the operation to apply to the inputs. Must take N
            arguments, where N is the number of sink pads, and return a
            single output.
    """

    op: Callable | None = None

    @validator.many_to_one
    def validate(self) -> None:
        assert self.op is not None, "op must be provided"
        self._validate_op()

    def apply(self, *buffers: SeriesBuffer) -> SeriesBuffer:
        """Apply the operator to the given sequence of buffers"""
        # Check if there are any gaps
        if any(buf.is_gap for buf in buffers):
            data = None
        else:
            assert self.op is not None
            data = self.op(*[buf.data for buf in buffers])

        return SeriesBuffer(
            data=data,
            offset=buffers[0].offset,
            sample_rate=buffers[0].sample_rate,
            shape=buffers[0].shape,
        )

    @transform.many_to_one
    def process(
        self, input_frames: dict[SinkPad, TSFrame], output_frame: TSCollectFrame
    ) -> None:
        """Process multiple input frames to single output."""
        input_buffers = [frame.buffers for frame in input_frames.values()]

        # Check all prepared frames have same number of buffers, this
        # is to make sure that zip doesn't silently drop any buffers
        assert all(len(b) == len(input_buffers[0]) for b in input_buffers), (
            "Prepared frames have different number "
            "of buffers, expected same number of "
            "buffers for all sink pads, got:"
            f" {[len(b) for b in input_buffers]}"
        )

        # Apply the operator to zipped groups of buffers
        for buffers in zip(*input_buffers):
            buf = self.apply(*buffers)
            output_frame.append(buf)

    def _validate_op(self):
        """Validate the given operator to make sure it
        has the right number of arguments
        """
        sig = inspect.signature(self.op)

        # Check if the operator has var positional arguments,
        # meaning that it can accept and arbitrary number of arguments,
        # so we don't need to check if the number of pads is compatible
        if not any(
            p.kind == inspect.Parameter.VAR_POSITIONAL for p in sig.parameters.values()
        ):
            assert len(sig.parameters) == len(self.aligned_sink_pads), (
                "Operator must take arguments matching number of aligned pads. "
                f"Got {len(sig.parameters)} arguments, "
                f"expected {len(self.aligned_sink_pads)}"
            )

apply(*buffers)

Apply the operator to the given sequence of buffers

Source code in src/sgnts/transforms/nary.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def apply(self, *buffers: SeriesBuffer) -> SeriesBuffer:
    """Apply the operator to the given sequence of buffers"""
    # Check if there are any gaps
    if any(buf.is_gap for buf in buffers):
        data = None
    else:
        assert self.op is not None
        data = self.op(*[buf.data for buf in buffers])

    return SeriesBuffer(
        data=data,
        offset=buffers[0].offset,
        sample_rate=buffers[0].sample_rate,
        shape=buffers[0].shape,
    )

process(input_frames, output_frame)

Process multiple input frames to single output.

Source code in src/sgnts/transforms/nary.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@transform.many_to_one
def process(
    self, input_frames: dict[SinkPad, TSFrame], output_frame: TSCollectFrame
) -> None:
    """Process multiple input frames to single output."""
    input_buffers = [frame.buffers for frame in input_frames.values()]

    # Check all prepared frames have same number of buffers, this
    # is to make sure that zip doesn't silently drop any buffers
    assert all(len(b) == len(input_buffers[0]) for b in input_buffers), (
        "Prepared frames have different number "
        "of buffers, expected same number of "
        "buffers for all sink pads, got:"
        f" {[len(b) for b in input_buffers]}"
    )

    # Apply the operator to zipped groups of buffers
    for buffers in zip(*input_buffers):
        buf = self.apply(*buffers)
        output_frame.append(buf)

Real dataclass

Bases: NaryTransform


              flowchart TD
              sgnts.transforms.nary.Real[Real]
              sgnts.transforms.nary.NaryTransform[NaryTransform]
              sgnts.base.base.TSTransform[TSTransform]
              sgnts.base.base.TimeSeriesMixin[TimeSeriesMixin]

                              sgnts.transforms.nary.NaryTransform --> sgnts.transforms.nary.Real
                                sgnts.base.base.TSTransform --> sgnts.transforms.nary.NaryTransform
                                sgnts.base.base.TimeSeriesMixin --> sgnts.base.base.TSTransform
                




              click sgnts.transforms.nary.Real href "" "sgnts.transforms.nary.Real"
              click sgnts.transforms.nary.NaryTransform href "" "sgnts.transforms.nary.NaryTransform"
              click sgnts.base.base.TSTransform href "" "sgnts.base.base.TSTransform"
              click sgnts.base.base.TimeSeriesMixin href "" "sgnts.base.base.TimeSeriesMixin"

Extract Real component of single input

Source code in src/sgnts/transforms/nary.py 
108
109
110
111
112
113
@dataclass
class Real(NaryTransform):
    """Extract Real component of single input"""

    def configure(self) -> None:
        self.op = _real