Skip to content

sgnts.sources.segment

SegmentSource dataclass

Bases: 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
Source code in sgnts/sources/segment.py 
11
12
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
@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
    """

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

    def __post_init__(self):
        assert self.segments is not None
        super().__post_init__()
        assert len(self.source_pads) == 1
        # FIXME
        self.segment_slices = TSSlices(
            TSSlice(Offset.fromns(s[0]), Offset.fromns(s[1]))
            for s in self.segments
            if (s[0] >= self.t0 * 1e9 and s[1] <= self.end * 1e9)
        ).simplify()

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

    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
        frame = self.prepare_frame(pad, data=1)

        bufs = []
        for buf in frame:
            nongap_slices = self.segment_slices.search(buf.slice)
            bufs.extend(buf.split(nongap_slices, contiguous=True))

        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 sgnts/sources/segment.py 
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
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
    frame = self.prepare_frame(pad, data=1)

    bufs = []
    for buf in frame:
        nongap_slices = self.segment_slices.search(buf.slice)
        bufs.extend(buf.split(nongap_slices, contiguous=True))

    frame.set_buffers(bufs)

    return frame