Skip to content

sgnts.sources.segment

SegmentSource dataclass

Bases: TSSource


              flowchart TD
              sgnts.sources.segment.SegmentSource[SegmentSource]
              sgnts.base.base.TSSource[TSSource]
              sgnts.base.base._TSSource[_TSSource]

                              sgnts.base.base.TSSource --> sgnts.sources.segment.SegmentSource
                                sgnts.base.base._TSSource --> sgnts.base.base.TSSource
                



              click sgnts.sources.segment.SegmentSource href "" "sgnts.sources.segment.SegmentSource"
              click sgnts.base.base.TSSource href "" "sgnts.base.base.TSSource"
              click sgnts.base.base._TSSource href "" "sgnts.base.base._TSSource"

Produce non-gap buffers for segments, and gap buffers otherwise.

Parameters:

Name Type Description Default
rate int

int, the sample rate of the data

 2048
segments Optional[tuple[tuple[int, int], ...]]

tuple[tuple[int, int], ...], a tuple of segment tuples corresponding to time in ns

 None
values Optional[tuple[Union[int, Array], ...]]

Optional[tuple[Union[int, Array], ...]], optional tuple of values to set for each segment's non-gap buffers. Must be same length as segments. If None, defaults to 1 for all non-gap buffers.

 None
Notes

Thread safety: Marked thread_safe = True. Pad layout: exactly 1 source pad (asserted in __post_init__). No same-element new concurrency. internal runs alone.

``new``: reads ``self.segment_data`` and
``self.segment_slices`` (computed in ``__post_init__``,
read-only thereafter); constructs and returns a local
frame. No element-level mutation.

Speedup from threading is modest (TSSlices set logic is
Python-bound), but marking thread_safe lets the source
run alongside other thread_safe elements without a
serialization point.

**Future editors MUST preserve thread safety**: do not
relax the single-source-pad invariant without first
ensuring all per-pad state is per-pad-keyed.
Source code in src/sgnts/sources/segment.py 
 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
 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
@dataclass
class SegmentSource(TSSource):
    """Produce non-gap buffers for segments, and gap buffers otherwise.

    Args:
        rate:
            int, the sample rate of the data
        segments:
            tuple[tuple[int, int], ...], a tuple of segment tuples corresponding to
            time in ns
        values:
            Optional[tuple[Union[int, Array], ...]], optional tuple of values to set
            for each segment's non-gap buffers. Must be same length as segments.
            If None, defaults to 1 for all non-gap buffers.

    Notes:
        Thread safety:
            Marked ``thread_safe = True``. Pad layout: exactly 1
            source pad (asserted in ``__post_init__``). No same-element
            ``new`` concurrency. ``internal`` runs alone.

            ``new``: reads ``self.segment_data`` and
            ``self.segment_slices`` (computed in ``__post_init__``,
            read-only thereafter); constructs and returns a local
            frame. No element-level mutation.

            Speedup from threading is modest (TSSlices set logic is
            Python-bound), but marking thread_safe lets the source
            run alongside other thread_safe elements without a
            serialization point.

            **Future editors MUST preserve thread safety**: do not
            relax the single-source-pad invariant without first
            ensuring all per-pad state is per-pad-keyed.
    """

    thread_safe = True

    rate: int = 2048
    segments: Optional[tuple[tuple[int, int], ...]] = None
    values: Optional[tuple[Union[int, Array], ...]] = None

    def __post_init__(self):
        assert (
            self.segments is not None
        ), "Segments must be provided during initialization"

        # Assert that segments are non-overlapping (but adjacent is OK)
        # Check by ensuring no two segments have overlapping interiors
        for i in range(len(self.segments)):
            for j in range(i + 1, len(self.segments)):
                seg1_start, seg1_end = self.segments[i]
                seg2_start, seg2_end = self.segments[j]
                # Check if segments overlap (not just touch at boundary)
                if seg1_start < seg2_end and seg2_start < seg1_end:
                    # They overlap if seg1 starts before seg2 ends AND seg2
                    # starts before seg1 ends
                    # But we need to exclude the case where they just touch at
                    # a boundary
                    if not (seg1_end == seg2_start or seg2_end == seg1_start):
                        raise AssertionError(
                            f"Input segments must be non-overlapping. "
                            f"Segments {i} ({seg1_start}, {seg1_end}) and "
                            f"{j} ({seg2_start}, {seg2_end}) overlap."
                        )

        # Validate values if provided
        if self.values is not None:
            assert len(self.values) == len(self.segments), (
                f"Length of values ({len(self.values)}) must match "
                f"length of segments ({len(self.segments)})"
            )

        super().__post_init__()
        assert (
            len(self.source_pads) == 1
        ), f"SegmentSource requires exactly one source pad, got {len(self.source_pads)}"

        # Filter segments that overlap with the source time range and track
        # their indices
        self.segment_data = []  # List of (slice, original_index) tuples
        start_ns = self.start * 1e9
        end_ns = self.end * 1e9

        for i, s in enumerate(self.segments):
            # Include segments that have any overlap with the time range
            if s[0] < end_ns and s[1] > start_ns:
                # Clip segment to the source time range
                seg_start = max(s[0], start_ns)
                seg_end = min(s[1], end_ns)
                slice_obj = TSSlice(
                    Offset.fromns(seg_start, sample_rate=self.rate),
                    Offset.fromns(seg_end, sample_rate=self.rate),
                )
                self.segment_data.append((slice_obj, i))

        # Create TSSlices from just the slices
        self.segment_slices = TSSlices([sd[0] for sd in self.segment_data])

        for pad in self.source_pads:
            self.set_pad_buffer_params(pad=pad, sample_shape=(), rate=self.rate)

    def output_prototype(self, pad: SourcePad) -> Array:
        """Declare the numpy spec this pad emits.

        A SegmentSource's first frame is typically an all-gap frame (the start
        time usually precedes the first segment), so the spec can't be inferred
        from data -- declare numpy. Its segment ``values`` may be int / float /
        complex, but all are numpy/cpu, which is what backend+device enforcement
        checks; the declared dtype only governs how gaps materialize.
        """
        return numpy.zeros(0)

    def new(self, pad: SourcePad) -> TSFrame:
        """New TSFrames are created on "pad" with stride matching the stride specified
        in Offset.SAMPLE_STRIDE_AT_MAX_RATE. EOS is set if we have reach the requested
        "end" time. Non-gap buffers will be produced when they are within the segments
        provided, and gap buffers will be produced otherwise.

        Args:
            pad:
                SourcePad, the pad for which to produce a new TSFrame

        Returns:
            TSFrame, the TSFrame with non-gap buffers within segments and gap buffers
            outside segments.
        """
        # FIXME: Find a better way to set EOS
        # Create frame with default data=None (gap buffers)
        frame = self.prepare_frame(pad, data=None)

        bufs = []
        for buf in frame:
            # Find which segments overlap with this buffer
            nongap_slices = self.segment_slices.search(buf.slice)

            if nongap_slices and nongap_slices.slices:
                # Split the buffer based on gap/non-gap regions
                split_bufs = buf.split(nongap_slices, contiguous=True)

                # For each split buffer, determine if it's gap or non-gap
                for split_buf in split_bufs:
                    # Check if this buffer overlaps with any segment
                    for slice_obj, orig_idx in self.segment_data:
                        overlap = split_buf.slice & slice_obj
                        # Only consider finite overlaps (not just boundary touches)
                        if overlap and overlap.isfinite():  # Has finite overlap
                            # Set the appropriate value for this non-gap buffer
                            if self.values is not None:
                                split_buf.set_data(self.values[orig_idx])
                            else:
                                # Default to ones
                                split_buf.set_data(numpy.ones(split_buf.shape))
                            break

                    # Gap buffers keep data=None (already set)
                    bufs.append(split_buf)
            else:
                # No overlap with any segment, keep as gap buffer
                bufs.append(buf)

        frame.set_buffers(bufs)

        return frame

new(pad)

New TSFrames are created on "pad" with stride matching the stride specified in Offset.SAMPLE_STRIDE_AT_MAX_RATE. EOS is set if we have reach the requested "end" time. Non-gap buffers will be produced when they are within the segments provided, and gap buffers will be produced otherwise.

Parameters:

Name Type Description Default
pad SourcePad

SourcePad, the pad for which to produce a new TSFrame

 required

Returns:

Type Description
TSFrame

TSFrame, the TSFrame with non-gap buffers within segments and gap buffers
TSFrame

outside segments.
Source code in src/sgnts/sources/segment.py 
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
def new(self, pad: SourcePad) -> TSFrame:
    """New TSFrames are created on "pad" with stride matching the stride specified
    in Offset.SAMPLE_STRIDE_AT_MAX_RATE. EOS is set if we have reach the requested
    "end" time. Non-gap buffers will be produced when they are within the segments
    provided, and gap buffers will be produced otherwise.

    Args:
        pad:
            SourcePad, the pad for which to produce a new TSFrame

    Returns:
        TSFrame, the TSFrame with non-gap buffers within segments and gap buffers
        outside segments.
    """
    # FIXME: Find a better way to set EOS
    # Create frame with default data=None (gap buffers)
    frame = self.prepare_frame(pad, data=None)

    bufs = []
    for buf in frame:
        # Find which segments overlap with this buffer
        nongap_slices = self.segment_slices.search(buf.slice)

        if nongap_slices and nongap_slices.slices:
            # Split the buffer based on gap/non-gap regions
            split_bufs = buf.split(nongap_slices, contiguous=True)

            # For each split buffer, determine if it's gap or non-gap
            for split_buf in split_bufs:
                # Check if this buffer overlaps with any segment
                for slice_obj, orig_idx in self.segment_data:
                    overlap = split_buf.slice & slice_obj
                    # Only consider finite overlaps (not just boundary touches)
                    if overlap and overlap.isfinite():  # Has finite overlap
                        # Set the appropriate value for this non-gap buffer
                        if self.values is not None:
                            split_buf.set_data(self.values[orig_idx])
                        else:
                            # Default to ones
                            split_buf.set_data(numpy.ones(split_buf.shape))
                        break

                # Gap buffers keep data=None (already set)
                bufs.append(split_buf)
        else:
            # No overlap with any segment, keep as gap buffer
            bufs.append(buf)

    frame.set_buffers(bufs)

    return frame

output_prototype(pad)

Declare the numpy spec this pad emits.

A SegmentSource's first frame is typically an all-gap frame (the start time usually precedes the first segment), so the spec can't be inferred from data -- declare numpy. Its segment values may be int / float / complex, but all are numpy/cpu, which is what backend+device enforcement checks; the declared dtype only governs how gaps materialize.

Source code in src/sgnts/sources/segment.py 
115
116
117
118
119
120
121
122
123
124
def output_prototype(self, pad: SourcePad) -> Array:
    """Declare the numpy spec this pad emits.

    A SegmentSource's first frame is typically an all-gap frame (the start
    time usually precedes the first segment), so the spec can't be inferred
    from data -- declare numpy. Its segment ``values`` may be int / float /
    complex, but all are numpy/cpu, which is what backend+device enforcement
    checks; the declared dtype only governs how gaps materialize.
    """
    return numpy.zeros(0)