Skip to content

sgnts.transforms.gate

Gate dataclass

Bases: TSTransform


              flowchart TD
              sgnts.transforms.gate.Gate[Gate]
              sgnts.base.base.TSTransform[TSTransform]
              sgnts.base.base.TimeSeriesMixin[TimeSeriesMixin]

                              sgnts.base.base.TSTransform --> sgnts.transforms.gate.Gate
                                sgnts.base.base.TimeSeriesMixin --> sgnts.base.base.TSTransform
                



              click sgnts.transforms.gate.Gate href "" "sgnts.transforms.gate.Gate"
              click sgnts.base.base.TSTransform href "" "sgnts.base.base.TSTransform"
              click sgnts.base.base.TimeSeriesMixin href "" "sgnts.base.base.TimeSeriesMixin"

Uses one sink pad's buffers to control the state of anothers. The control buffer state is defined by either being gap or not. The actual content of the data is ignored otherwise.

Parameters:

Name Type Description Default
control str

str, the name of the pad to use as a control signal

 required
Notes

Thread safety: Marked thread_safe = True. Pad layout: 2 sink pads (data + control) + 1 source pad (@validator.num_pads(sink_pads=2, source_pads=1)). Both 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. ``new``
(inherited): read-only ``self.outframes`` lookup.
``process``: buffer split + TSSlice search on local data;
reads ``self.controlpad``/``self.sinkpad``/
``self.source_pad`` (set in ``configure()``, read-only).

Direct speedup from threading is modest (Python-bound work),
but marking thread_safe lets the element run alongside
other thread_safe elements without a serialization point.

**Future editors MUST preserve thread safety**: keep
``process`` purely functional on its inputs. Do NOT add
element-level state mutated from ``pull`` outside of
per-pad-keyed containers.
Source code in src/sgnts/transforms/gate.py 
10
11
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
@dataclass(kw_only=True)
class Gate(TSTransform):
    """Uses one sink pad's buffers to control the state of anothers. The control buffer
    state is defined by either being gap or not. The actual content of the data is
    ignored otherwise.

    Args:
        control:
            str, the name of the pad to use as a control signal

    Notes:
        Thread safety:
            Marked ``thread_safe = True``. Pad layout: 2 sink pads
            (data + control) + 1 source pad
            (``@validator.num_pads(sink_pads=2, source_pads=1)``).
            Both 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. ``new``
            (inherited): read-only ``self.outframes`` lookup.
            ``process``: buffer split + TSSlice search on local data;
            reads ``self.controlpad``/``self.sinkpad``/
            ``self.source_pad`` (set in ``configure()``, read-only).

            Direct speedup from threading is modest (Python-bound work),
            but marking thread_safe lets the element run alongside
            other thread_safe elements without a serialization point.

            **Future editors MUST preserve thread safety**: keep
            ``process`` purely functional on its inputs. Do NOT add
            element-level state mutated from ``pull`` outside of
            per-pad-keyed containers.
    """

    thread_safe = True

    # Gates on gap state and slices buffers; does not touch data values.
    backends = ANY_BACKEND

    control: str

    def configure(self) -> None:
        self.controlpad = self.snks[self.control]
        data_pad_name = list(set(self.sink_pad_names) - set([self.control]))[0]
        self.sinkpad = self.snks[data_pad_name]
        self.source_pad = self.source_pads[0]

    @validator.num_pads(sink_pads=2, source_pads=1)
    def validate(self) -> None:
        assert self.control and self.control in self.sink_pad_names, (
            f"Control pad '{self.control}' must be specified and exist "
            f"in sink_pad_names: {self.sink_pad_names}"
        )

    def output_prototype(self, pad):
        # Make sure prototype is made from data pad, not control pad
        return self.input_prototype(self.sinkpad.pad_name)

    @transform.many_to_one
    def process(
        self, input_frames: dict[SinkPad, TSFrame], output_frame: TSCollectFrame
    ) -> None:
        """Gate input based on control pad."""
        nongap_slices = TSSlices([b.slice for b in input_frames[self.controlpad] if b])
        bufs = sorted(
            [
                b
                for bs in [
                    buf.split(nongap_slices.search(buf.slice), contiguous=True)
                    for buf in input_frames[self.sinkpad]
                ]
                for b in bs
            ]
        )
        output_frame.extend(bufs)

process(input_frames, output_frame)

Gate input based on control pad.

Source code in src/sgnts/transforms/gate.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@transform.many_to_one
def process(
    self, input_frames: dict[SinkPad, TSFrame], output_frame: TSCollectFrame
) -> None:
    """Gate input based on control pad."""
    nongap_slices = TSSlices([b.slice for b in input_frames[self.controlpad] if b])
    bufs = sorted(
        [
            b
            for bs in [
                buf.split(nongap_slices.search(buf.slice), contiguous=True)
                for buf in input_frames[self.sinkpad]
            ]
            for b in bs
        ]
    )
    output_frame.extend(bufs)