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``,
``self.segment_slices`` and ``self._proto`` (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

@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``,
            ``self.segment_slices`` and ``self._proto`` (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)})"
            )

        # One spec for everything this pad emits. A pad's spec (namespace,
        # device, and dtype) is fixed by its first frame, so derive a
        # zero-length prototype from the values themselves: array values keep
        # their own namespace and device, scalars materialize as numpy (the
        # default backend), and zero-length addition promotes dtypes across
        # segments -- integer or complex values emitted against a float64
        # first-gap spec would otherwise be a mid-stream spec change and
        # hard-fail in the pipeline.
        if self.values:
            examples = [
                v if array_namespace(v) is not None else numpy.asarray(v)
                for v in self.values
            ]
            xp = array_namespace(*examples)
            assert xp is not None, "values must share one array namespace"
            proto = xp.reshape(examples[0], (-1,))[:0]
            for example in examples[1:]:
                proto = proto + xp.reshape(example, (-1,))[:0]
            self._proto = proto
        else:
            # The default path emits numpy.ones.
            self._proto = numpy.ones(0)

        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 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 it. The prototype is derived from the
        provided ``values`` (numpy float64 for the default ones): array values
        contribute their namespace and device, scalars default to numpy, and
        dtypes promote across segments. Spec enforcement compares dtype along
        with backend and device, so the declaration must match every buffer
        this source later emits.
        """
        return self._proto

    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:
                                value = self.values[orig_idx]
                                if array_namespace(value) is None:
                                    # Scalar shorthand: materialize at the
                                    # declared spec (namespace/device/dtype of
                                    # the prototype) rather than set_data's
                                    # deprecated scalar path, whose dtype
                                    # varies per value (ones for 1, int64
                                    # full for other ints) and would change
                                    # the pad's spec mid-stream.
                                    value = (
                                        new_zeros(self._proto, split_buf.shape) + value
                                    )
                                split_buf.set_data(value)
                            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

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:
                            value = self.values[orig_idx]
                            if array_namespace(value) is None:
                                # Scalar shorthand: materialize at the
                                # declared spec (namespace/device/dtype of
                                # the prototype) rather than set_data's
                                # deprecated scalar path, whose dtype
                                # varies per value (ones for 1, int64
                                # full for other ints) and would change
                                # the pad's spec mid-stream.
                                value = (
                                    new_zeros(self._proto, split_buf.shape) + value
                                )
                            split_buf.set_data(value)
                        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 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 it. The prototype is derived from the provided values (numpy float64 for the default ones): array values contribute their namespace and device, scalars default to numpy, and dtypes promote across segments. Spec enforcement compares dtype along with backend and device, so the declaration must match every buffer this source later emits.

Source code in src/sgnts/sources/segment.py

def output_prototype(self, pad: SourcePad) -> Array:
    """Declare the 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 it. The prototype is derived from the
    provided ``values`` (numpy float64 for the default ones): array values
    contribute their namespace and device, scalars default to numpy, and
    dtypes promote across segments. Spec enforcement compares dtype along
    with backend and device, so the declaration must match every buffer
    this source later emits.
    """
    return self._proto