Skip to content

sgnts.transforms.threshold

Threshold dataclass

Bases: TSTransform


              flowchart TD
              sgnts.transforms.threshold.Threshold[Threshold]
              sgnts.base.base.TSTransform[TSTransform]
              sgnts.base.base.TimeSeriesMixin[TimeSeriesMixin]

                              sgnts.base.base.TSTransform --> sgnts.transforms.threshold.Threshold
                                sgnts.base.base.TimeSeriesMixin --> sgnts.base.base.TSTransform
                



              click sgnts.transforms.threshold.Threshold href "" "sgnts.transforms.threshold.Threshold"
              click sgnts.base.base.TSTransform href "" "sgnts.base.base.TSTransform"
              click sgnts.base.base.TimeSeriesMixin href "" "sgnts.base.base.TimeSeriesMixin"

Only allow data above or below a threshold to pass. data will otherwise be marked as gap.

Parameters:

Name Type Description Default
threshold float

float, the absolute value threshold above which to allow data to pass

 float('+inf')
invert bool

bool, If False, only data above a threshold will pass. If True: only data below a threshold will pass

 False
inclusive bool

bool, If True (default) the crossing test is |data| >= threshold; if False it is strict, |data| > threshold.

 True
startwn int

int, the number of samples ahead of the crossing to allow data to pass

 0
stopwn int

int, the number of samples after the crossing to allow data to pass

 0
Notes

Thread safety: Marked thread_safe = True. Pad layout: 1 sink + 1 source (@transform.one_to_one). No same-element pull/new concurrency. internal runs alone.

``pull`` (inherited): per-pad-keyed dict writes. ``new``
(inherited): read-only lookup. ``process``: NumPy
``concatenate``/``flatnonzero``/comparison kernels (all
release the GIL) on local buffers, plus mutation of the
persistent accumulator ``self.nongap_slices``.

Why ``self.nongap_slices`` mutation is safe: it is only
written from ``process``, which is only called from
``internal()``, which is the single ``InternalPad`` of the
element and runs alone. No two threads ever enter
``process`` for the same element concurrently.

**Future editors MUST preserve thread safety**: do NOT
relax the one-to-one decorator without first moving the
``nongap_slices`` accumulation into per-pad-keyed state.
If you ever move accumulation logic into ``pull``, that
would expose it to per-sink-pad concurrency and require
per-pad keying (or external synchronization).
Source code in src/sgnts/transforms/threshold.py 
 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
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
@dataclass
class Threshold(TSTransform):
    """Only allow data above or below a threshold to pass. data will otherwise be marked
    as gap.

    Args:
        threshold:
            float, the absolute value threshold above which to allow data to pass
        invert:
            bool, If False, only data above a threshold will pass. If True: only data
            below a threshold will pass
        inclusive:
            bool, If True (default) the crossing test is ``|data| >= threshold``;
            if False it is strict, ``|data| > threshold``.
        startwn:
            int, the number of samples ahead of the crossing to allow data to pass
        stopwn:
            int, the number of samples after the crossing to allow data to pass

    Notes:
        Thread safety:
            Marked ``thread_safe = True``. Pad layout: 1 sink + 1
            source (``@transform.one_to_one``). No same-element
            ``pull``/``new`` concurrency. ``internal`` runs alone.

            ``pull`` (inherited): per-pad-keyed dict writes. ``new``
            (inherited): read-only lookup. ``process``: NumPy
            ``concatenate``/``flatnonzero``/comparison kernels (all
            release the GIL) on local buffers, plus mutation of the
            persistent accumulator ``self.nongap_slices``.

            Why ``self.nongap_slices`` mutation is safe: it is only
            written from ``process``, which is only called from
            ``internal()``, which is the single ``InternalPad`` of the
            element and runs alone. No two threads ever enter
            ``process`` for the same element concurrently.

            **Future editors MUST preserve thread safety**: do NOT
            relax the one-to-one decorator without first moving the
            ``nongap_slices`` accumulation into per-pad-keyed state.
            If you ever move accumulation logic into ``pull``, that
            would expose it to per-sink-pad concurrency and require
            per-pad keying (or external synchronization).
    """

    thread_safe = True
    backends = ANY_BACKEND

    threshold: float = float("+inf")
    invert: bool = False
    inclusive: bool = True
    startwn: int = 0
    stopwn: int = 0

    def configure(self) -> None:
        self.nongap_slices = TSSlices([])

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

    # Modified from: https://stackoverflow.com/questions/43258896/
    # extract-subarrays-of-numpy-array-whose-values-are-above-a-threshold
    def __split_above_threshold(
        self,
        buffer: SeriesBuffer,
        threshold: float,
        start_window: int = 0,
        stop_window: int = 0,
    ) -> list[TSSlice]:
        """Find subslices in buffer whose data are above threshold, along with
        start_window samples ahead of and stop_window samples after the crossing.

        Args:
            buffer:
                SeriesBuffer, the buffer from which to extract subslices
            threshold:
                float, the crossing threshold
            start_window:
                int, the number of samples ahead of the crossing to allow data to pass
            stop_window:
                int, the number of samples after the crossing to allow data to pass

        Returns:
            list[TSSlice], a list of TSSlices whose data value crossed a threshold,
            along with a window around the crossing
        """
        xp = self._input_specs[self.sink_pads[0].pad_name].namespace
        assert xp is not None
        dev = self._input_specs[self.sink_pads[0].pad_name].device

        signal = xp.asarray(buffer.data, device=dev)
        sample_rate = buffer.sample_rate
        off0 = buffer.offset
        # Pad with one False on each side so a crossing at the array edge
        # still produces a start/stop pair.
        above = (
            xp.abs(signal) >= threshold
            if self.inclusive
            else xp.abs(signal) > threshold
        )
        pad = xp.zeros((1,), dtype=xp.bool, device=dev)
        mask = xp.concat((pad, above, pad))
        idx = xp.nonzero(mask[1:] != mask[:-1])[0]
        return [
            TSSlice(
                off0 + Offset.fromsamples(int(idx[i] - start_window), sample_rate),
                off0 + Offset.fromsamples(int(idx[i + 1] + stop_window), sample_rate),
            )
            for i in range(0, len(idx), 2)
        ]

    @transform.one_to_one
    def process(self, input_frame: TSFrame, output_frame: TSCollectFrame) -> None:
        """Process frame to threshold data based on absolute value."""

        boundary_offsets = TSSlice(
            input_frame[0].offset,
            input_frame[-1].end_offset,
        )
        self.nongap_slices += TSSlices(
            [
                j
                for sub in [
                    self.__split_above_threshold(
                        b,
                        self.threshold,
                        self.startwn,
                        self.stopwn,
                    )
                    for b in input_frame
                    if b
                ]
                for j in sub
            ]
        )
        self.nongap_slices = self.nongap_slices.simplify()

        # restrict to slices that are new enough to matter
        self.nongap_slices = TSSlices(
            [
                s
                for s in self.nongap_slices.slices
                if not s.stop <= boundary_offsets.start
            ]
        )

        aligned_nongap_slices = self.nongap_slices.search(boundary_offsets, align=True)
        if self.invert:
            aligned_nongap_slices = aligned_nongap_slices.invert(boundary_offsets)

        out = sorted(
            [
                b
                for bs in [
                    buf.split(aligned_nongap_slices.search(buf.slice), contiguous=True)
                    for buf in input_frame
                ]
                for b in bs
            ]
        )

        # sanity check that buffers don't overlap
        o0 = out[0]
        for o in out[1:]:
            assert o.offset == o0.end_offset, (
                f"Buffer overlap detected: buffer offset {o.offset} != "
                f"previous buffer end_offset {o0.end_offset}"
            )
            o0 = o

        output_frame.extend(out)

__split_above_threshold(buffer, threshold, start_window=0, stop_window=0)

Find subslices in buffer whose data are above threshold, along with start_window samples ahead of and stop_window samples after the crossing.

Parameters:

Name Type Description Default
buffer SeriesBuffer

SeriesBuffer, the buffer from which to extract subslices

 required
threshold float

float, the crossing threshold

 required
start_window int

int, the number of samples ahead of the crossing to allow data to pass

 0
stop_window int

int, the number of samples after the crossing to allow data to pass

 0

Returns:

Type Description
list[TSSlice]

list[TSSlice], a list of TSSlices whose data value crossed a threshold,
list[TSSlice]

along with a window around the crossing
Source code in src/sgnts/transforms/threshold.py 
 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
def __split_above_threshold(
    self,
    buffer: SeriesBuffer,
    threshold: float,
    start_window: int = 0,
    stop_window: int = 0,
) -> list[TSSlice]:
    """Find subslices in buffer whose data are above threshold, along with
    start_window samples ahead of and stop_window samples after the crossing.

    Args:
        buffer:
            SeriesBuffer, the buffer from which to extract subslices
        threshold:
            float, the crossing threshold
        start_window:
            int, the number of samples ahead of the crossing to allow data to pass
        stop_window:
            int, the number of samples after the crossing to allow data to pass

    Returns:
        list[TSSlice], a list of TSSlices whose data value crossed a threshold,
        along with a window around the crossing
    """
    xp = self._input_specs[self.sink_pads[0].pad_name].namespace
    assert xp is not None
    dev = self._input_specs[self.sink_pads[0].pad_name].device

    signal = xp.asarray(buffer.data, device=dev)
    sample_rate = buffer.sample_rate
    off0 = buffer.offset
    # Pad with one False on each side so a crossing at the array edge
    # still produces a start/stop pair.
    above = (
        xp.abs(signal) >= threshold
        if self.inclusive
        else xp.abs(signal) > threshold
    )
    pad = xp.zeros((1,), dtype=xp.bool, device=dev)
    mask = xp.concat((pad, above, pad))
    idx = xp.nonzero(mask[1:] != mask[:-1])[0]
    return [
        TSSlice(
            off0 + Offset.fromsamples(int(idx[i] - start_window), sample_rate),
            off0 + Offset.fromsamples(int(idx[i + 1] + stop_window), sample_rate),
        )
        for i in range(0, len(idx), 2)
    ]

process(input_frame, output_frame)

Process frame to threshold data based on absolute value.

Source code in src/sgnts/transforms/threshold.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
@transform.one_to_one
def process(self, input_frame: TSFrame, output_frame: TSCollectFrame) -> None:
    """Process frame to threshold data based on absolute value."""

    boundary_offsets = TSSlice(
        input_frame[0].offset,
        input_frame[-1].end_offset,
    )
    self.nongap_slices += TSSlices(
        [
            j
            for sub in [
                self.__split_above_threshold(
                    b,
                    self.threshold,
                    self.startwn,
                    self.stopwn,
                )
                for b in input_frame
                if b
            ]
            for j in sub
        ]
    )
    self.nongap_slices = self.nongap_slices.simplify()

    # restrict to slices that are new enough to matter
    self.nongap_slices = TSSlices(
        [
            s
            for s in self.nongap_slices.slices
            if not s.stop <= boundary_offsets.start
        ]
    )

    aligned_nongap_slices = self.nongap_slices.search(boundary_offsets, align=True)
    if self.invert:
        aligned_nongap_slices = aligned_nongap_slices.invert(boundary_offsets)

    out = sorted(
        [
            b
            for bs in [
                buf.split(aligned_nongap_slices.search(buf.slice), contiguous=True)
                for buf in input_frame
            ]
            for b in bs
        ]
    )

    # sanity check that buffers don't overlap
    o0 = out[0]
    for o in out[1:]:
        assert o.offset == o0.end_offset, (
            f"Buffer overlap detected: buffer offset {o.offset} != "
            f"previous buffer end_offset {o0.end_offset}"
        )
        o0 = o

    output_frame.extend(out)