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 
140
141
142
143
144
145
@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
Notes

Thread safety: Marked thread_safe = True (inherited by Multiply, Real, and other subclasses). With Pipeline.run(threaded=N) the pad callbacks for this element are dispatched onto worker threads.

Pad layout: N sink pads + 1 source pad
(``@validator.many_to_one``). The N sink pads' ``pull``
callbacks CAN run concurrently in the same wave.
``internal`` runs alone.

Where the (potential) GIL-releasing work lives: ``internal()``
→ ``process()`` → ``self.op(*data)``. Speedup depends on
``op``: typical NumPy/Torch ops release the GIL and scale;
pure-Python ops will not.

Per-pad concurrency analysis:

- ``pull`` (inherited ``TimeSeriesMixin.pull``):
  per-pad-keyed dict writes; safe across pads.
- ``new`` (inherited): read-only lookup.
- ``process``: zips per-pad input buffers, calls
  ``self.op`` per buffer-tuple, appends to local output.
  No element-level mutation.

**Future editors MUST preserve thread safety**: keep
``apply``/``process`` purely functional on their inputs.
Subclasses that set ``self.op`` from ``configure()`` are
fine; if you set ``self.op`` from ``pull``/``new``/
``process`` you'd race across same-element pulls. ``op``
itself must be reentrant — that's the user's
responsibility.
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
 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
@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.

    Notes:
        Thread safety:
            Marked ``thread_safe = True`` (inherited by ``Multiply``,
            ``Real``, and other subclasses). With
            ``Pipeline.run(threaded=N)`` the pad callbacks for this
            element are dispatched onto worker threads.

            Pad layout: N sink pads + 1 source pad
            (``@validator.many_to_one``). The N sink pads' ``pull``
            callbacks CAN run concurrently in the same wave.
            ``internal`` runs alone.

            Where the (potential) GIL-releasing work lives: ``internal()``
            → ``process()`` → ``self.op(*data)``. Speedup depends on
            ``op``: typical NumPy/Torch ops release the GIL and scale;
            pure-Python ops will not.

            Per-pad concurrency analysis:

            - ``pull`` (inherited ``TimeSeriesMixin.pull``):
              per-pad-keyed dict writes; safe across pads.
            - ``new`` (inherited): read-only lookup.
            - ``process``: zips per-pad input buffers, calls
              ``self.op`` per buffer-tuple, appends to local output.
              No element-level mutation.

            **Future editors MUST preserve thread safety**: keep
            ``apply``/``process`` purely functional on their inputs.
            Subclasses that set ``self.op`` from ``configure()`` are
            fine; if you set ``self.op`` from ``pull``/``new``/
            ``process`` you'd race across same-element pulls. ``op``
            itself must be reentrant — that's the user's
            responsibility.
    """

    thread_safe = True

    # Default op (_real) is numpy; subclasses with universal ops may widen this.
    backends = frozenset({"numpy"})

    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 output_prototype(self, pad):
        # Run the op on zero-length input prototypes: the output dtype is exactly
        # what op produces (Real's np.real -> float, Multiply -> result_type), while
        # namespace/device ride through from the inputs. op is element-wise, so a
        # length-0 input is safe -- the one general way to declare an op's output
        # without restating its dtype.
        assert self.op is not None
        return self.op(*[self.input_prototype(p) for p in self.sink_pad_names])

    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 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
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 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
@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 
156
157
158
159
160
161
@dataclass
class Real(NaryTransform):
    """Extract Real component of single input"""

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