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
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
Source code in sgnts/transforms/threshold.py 
 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
@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
        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
    """

    threshold: float = float("+inf")
    invert: bool = False
    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
        """
        signal = numpy.array(buffer.data)
        sample_rate = buffer.sample_rate
        off0 = buffer.offset
        # NOTE the tuple casting is here because of mypy. Numpy typing seems a
        # bit broken in a few places.
        mask: numpy.ndarray = numpy.concatenate(
            ((False,), tuple(numpy.abs(signal) >= threshold), (False,))
        )
        idx = numpy.flatnonzero(mask[1:] != mask[:-1])
        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 sgnts/transforms/threshold.py 
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
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
    """
    signal = numpy.array(buffer.data)
    sample_rate = buffer.sample_rate
    off0 = buffer.offset
    # NOTE the tuple casting is here because of mypy. Numpy typing seems a
    # bit broken in a few places.
    mask: numpy.ndarray = numpy.concatenate(
        ((False,), tuple(numpy.abs(signal) >= threshold), (False,))
    )
    idx = numpy.flatnonzero(mask[1:] != mask[:-1])
    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 sgnts/transforms/threshold.py 
 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
@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)