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
Source code in 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
@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
    """

    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}"
        )

    @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 sgnts/transforms/gate.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@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)