Skip to content

sgnts.transforms.and_transform

ANDTransform dataclass

Bases: TSTransform


              flowchart TD
              sgnts.transforms.and_transform.ANDTransform[ANDTransform]
              sgnts.base.base.TSTransform[TSTransform]
              sgnts.base.base.TimeSeriesMixin[TimeSeriesMixin]

                              sgnts.base.base.TSTransform --> sgnts.transforms.and_transform.ANDTransform
                                sgnts.base.base.TimeSeriesMixin --> sgnts.base.base.TSTransform
                



              click sgnts.transforms.and_transform.ANDTransform href "" "sgnts.transforms.and_transform.ANDTransform"
              click sgnts.base.base.TSTransform href "" "sgnts.base.base.TSTransform"
              click sgnts.base.base.TimeSeriesMixin href "" "sgnts.base.base.TimeSeriesMixin"

Perform logical AND operation across multiple input streams based on gap status.

This transform takes multiple input streams with arbitrary sample rates and produces a single output stream at the maximum sample rate among all inputs.

Output behavior: - Where ALL inputs have non-gap data: output is 1 (logical AND) - Where ANY input has a gap: output is a gap buffer (data=None)

This approach is semantically correct as gaps represent "no valid data" rather than zeros. Downstream transforms can decide how to handle gaps (e.g., convert to zeros if needed).

The transform uses TSSlice set logic to determine overlapping regions and efficiently creates output buffers using the split method with contiguous=True.

Parameters:

Name Type Description Default
output_shape tuple[int, ...]

Optional[tuple[int, ...]], shape of output samples (excluding time dimension). If None, defaults to scalar output ()

 tuple()
Notes

Thread safety: Marked thread_safe = True. 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.

``pull`` (inherited ``TimeSeriesMixin.pull``):
per-pad-keyed dict writes; safe across pads. ``new``
(inherited): read-only ``self.outframes`` lookup.
``process``: TSSlice set-logic and buffer splitting on
local data only — no element-level mutation.

Where speedup comes from: the work is mostly Python-bound
(TSSlice ops) so direct compute-parallelism is modest, but
marking thread_safe lets the element run concurrently with
adjacent thread_safe elements without forcing a
serialization point in the wave.

**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 — multi-sink elements will race.
Source code in src/sgnts/transforms/and_transform.py 
 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
@dataclass
class ANDTransform(TSTransform):
    """Perform logical AND operation across multiple input streams based on gap status.

    This transform takes multiple input streams with arbitrary sample rates and
    produces a single output stream at the maximum sample rate among all inputs.

    Output behavior:
    - Where ALL inputs have non-gap data: output is 1 (logical AND)
    - Where ANY input has a gap: output is a gap buffer (data=None)

    This approach is semantically correct as gaps represent "no valid data" rather
    than zeros. Downstream transforms can decide how to handle gaps (e.g., convert
    to zeros if needed).

    The transform uses TSSlice set logic to determine overlapping regions and
    efficiently creates output buffers using the split method with contiguous=True.

    Args:
        output_shape:
            Optional[tuple[int, ...]], shape of output samples (excluding time
            dimension).
            If None, defaults to scalar output ()

    Notes:
        Thread safety:
            Marked ``thread_safe = True``. 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.

            ``pull`` (inherited ``TimeSeriesMixin.pull``):
            per-pad-keyed dict writes; safe across pads. ``new``
            (inherited): read-only ``self.outframes`` lookup.
            ``process``: TSSlice set-logic and buffer splitting on
            local data only — no element-level mutation.

            Where speedup comes from: the work is mostly Python-bound
            (TSSlice ops) so direct compute-parallelism is modest, but
            marking thread_safe lets the element run concurrently with
            adjacent thread_safe elements without forcing a
            serialization point in the wave.

            **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 — multi-sink elements will race.
    """

    thread_safe = True

    backends = frozenset({"numpy"})

    output_shape: tuple[int, ...] = field(default_factory=tuple)

    def configure(self) -> None:
        # Explicitly disable adapter to prevent gap filling
        self.adapter_config.enable = False

    @validator.many_to_one
    def validate(self) -> None:
        pass

    def output_prototype(self, pad):
        # Emits a float64 presence mask regardless of input dtype (numpy-only).
        return self.input_prototype(self.sink_pad_names[0], dtype=float)

    @transform.many_to_one
    def process(
        self, input_frames: dict[SinkPad, TSFrame], output_frame: TSCollectFrame
    ) -> None:
        """Generate output frame with AND logic across all inputs.

        Output contains buffers with 1s where all inputs have data,
        and gap buffers where any input has gaps.
        """
        frames = list(input_frames.values())

        # Use the maximum sample rate among all inputs for output
        max_rate = max(f.sample_rate for f in frames)

        # Collect non-gap slices from each input
        all_nongap_slices = [
            TSSlices([buf.slice for buf in frame if not buf.is_gap]) for frame in frames
        ]

        # Find intersection of all non-gap regions (where ALL inputs have data)
        intersection_slices = TSSlices.intersection_of_multiple(all_nongap_slices)

        # Create initial buffer spanning entire frame with 1s
        frame_slice = TSSlice(frames[0].offset, frames[0].end_offset)
        full_buffer = SeriesBuffer.fromoffsetslice(
            frame_slice,
            sample_rate=max_rate,
            data=None,
            channels=self.output_shape,
        )
        full_buffer.set_data(numpy.ones(full_buffer.shape))  # filled with ones

        # Split buffer at intersection boundaries:
        # - Keeps 1s where we have intersections (all inputs have data)
        # - Creates gap buffers (data=None) between intersections
        # If no intersections exist, create a single gap buffer
        output_buffers = (
            full_buffer.split(intersection_slices, contiguous=True)
            if intersection_slices.slices
            else [full_buffer.new()]  # new() with no data creates a gap buffer
        )

        output_frame.extend(output_buffers)

process(input_frames, output_frame)

Generate output frame with AND logic across all inputs.

Output contains buffers with 1s where all inputs have data, and gap buffers where any input has gaps.

Source code in src/sgnts/transforms/and_transform.py 
 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
@transform.many_to_one
def process(
    self, input_frames: dict[SinkPad, TSFrame], output_frame: TSCollectFrame
) -> None:
    """Generate output frame with AND logic across all inputs.

    Output contains buffers with 1s where all inputs have data,
    and gap buffers where any input has gaps.
    """
    frames = list(input_frames.values())

    # Use the maximum sample rate among all inputs for output
    max_rate = max(f.sample_rate for f in frames)

    # Collect non-gap slices from each input
    all_nongap_slices = [
        TSSlices([buf.slice for buf in frame if not buf.is_gap]) for frame in frames
    ]

    # Find intersection of all non-gap regions (where ALL inputs have data)
    intersection_slices = TSSlices.intersection_of_multiple(all_nongap_slices)

    # Create initial buffer spanning entire frame with 1s
    frame_slice = TSSlice(frames[0].offset, frames[0].end_offset)
    full_buffer = SeriesBuffer.fromoffsetslice(
        frame_slice,
        sample_rate=max_rate,
        data=None,
        channels=self.output_shape,
    )
    full_buffer.set_data(numpy.ones(full_buffer.shape))  # filled with ones

    # Split buffer at intersection boundaries:
    # - Keeps 1s where we have intersections (all inputs have data)
    # - Creates gap buffers (data=None) between intersections
    # If no intersections exist, create a single gap buffer
    output_buffers = (
        full_buffer.split(intersection_slices, contiguous=True)
        if intersection_slices.slices
        else [full_buffer.new()]  # new() with no data creates a gap buffer
    )

    output_frame.extend(output_buffers)