Skip to content

sgnts.base.audioadapter

The audioadapter stores buffers of data into a deque.

Audioadapter

The audioadapter stores buffers of data into a deque, and will track the copying and flushing of data from the adapter.

Parameters:

Name Type Description Default
backend type[ArrayBackend]

type[ArrayBackend], the wrapper around array operations

NumpyBackend
Source code in sgnts/base/audioadapter.py
 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
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
class Audioadapter:
    """The audioadapter stores buffers of data into a deque, and will track the copying
    and flushing of data from the adapter.

    Args:
        backend:
            type[ArrayBackend], the wrapper around array operations
    """

    def __init__(self, backend: type[ArrayBackend] = NumpyBackend):
        self.buffers: Deque[SeriesBuffer] = deque()
        self.size: int = 0
        self.gap_size: int = 0
        self.nongap_size: int = 0
        self.sample_rate: int = -1
        self.pre_cat_data: Optional[SeriesBuffer] = None
        self.backend: type[ArrayBackend] = backend

    def __len__(self) -> int:
        return len(self.buffers)

    @property
    def offset(self) -> int:
        """The offset of the first buffer in the audioadapter.

        Returns:
            int, the offset
        """
        if len(self) == 0:
            raise ValueError("Audioadapter not populated")
        return self.buffers[0].offset

    @property
    def end_offset(self) -> int:
        """The end offset of the last buffer in the audioadapter.

        Returns:
            int, the end offset
        """
        if len(self) == 0:
            raise ValueError("Audioadapter not populated")
        return self.offset + Offset.fromsamples(self.size, self.sample_rate)

    @property
    def slice(self) -> tuple[int, int]:
        """The offset slice of the audioadapter.

        Returns:
            tuple[int, int], the offset slice
        """
        return (self.offset, self.end_offset)

    @property
    def is_gap(self) -> bool:
        """True if all buffers are gaps.

        Returns:
            bool, if True, the whole audioadapter is a gap. If False, there are nongap
            buffers in the audioadapter
        """
        return self.nongap_size == 0

    def concatenate_data(
        self, offset_segment: Optional[tuple[int, int]] = None
    ) -> None:
        """Concatenate all the data and gaps info in the buffers, and save as attribute.

        Args:
            offset_segment:
                Optional[tuple[int, int]], only concatenate data within this offset
                segment
        """
        if self.size > 0:
            if offset_segment is not None:
                bufs = self.get_sliced_buffers(offset_segment)
            else:
                bufs = self.buffers
                offset_segment = self.slice

            cat_all = self.backend.cat(
                [b.filleddata(self.backend.zeros) for b in bufs], axis=-1
            )

            self.pre_cat_data = SeriesBuffer(
                offset=offset_segment[0],
                data=cat_all,
                sample_rate=bufs[0].sample_rate,
                shape=bufs[0].shape[:-1]
                + (
                    Offset.tosamples(
                        offset_segment[1] - offset_segment[0], bufs[0].sample_rate
                    ),
                ),
            )

    def push(self, buf: SeriesBuffer) -> None:
        """Push buffer into the deque, update metadata.

        Args:
            buf:
                SeriesBuffer, the buffer to append to the deque of the audioadapter.
        """
        if buf.noffset == 0 and len(self) > 0:
            # if there are no buffers and the very first buffer we receive
            # is a zero length buffer, still push it into the adapter
            return

        # Check if the start time is as expected
        # FIXME should we support discontinuities?
        if self.sample_rate != -1 and buf.offset != self.end_offset:
            raise ValueError(
                f"Got an unexpected buffer offset: {buf.offset=}"
                f" instead of {self.end_offset=} {buf=}"
            )

        if self.sample_rate == -1:
            self.sample_rate = buf.sample_rate
        elif buf.sample_rate != self.sample_rate:
            # buffers in the audioadapter must be the same sample rate
            raise ValueError(
                f"Inconsistent sample rate, buffer sample rate: {buf.sample_rate}"
                f" audioadpater sample rate: {self.sample_rate}"
            )

        # Store gap information
        nsamples = buf.samples
        self.size += nsamples
        if buf.is_gap is True:
            self.gap_size += nsamples
        else:
            self.nongap_size += nsamples

        if len(self) > 0 and self.buffers[-1].duration == 0:
            if buf.duration > 0:
                # Replace heartbeat buffers
                self.buffers[-1] = buf
        else:
            self.buffers.append(buf)

        self.pre_cat_data = None  # reset

    def get_sliced_buffers(
        self, offset_segment: tuple[int, int], pad_start: bool = False
    ) -> Deque[SeriesBuffer]:
        """Return buffers that lie within the offset_segment, slice up buffers if
        neeeded.

        Args:
            offset_segment:
                tuple[int, int], the offset segment to get buffers from
            pad_start:
                bool, default False, if True and if offset segment is earlier than the
                available buffers, will make front-pad the buffers with a gap buffer

        Returns:
            Deque[SeriesBuffer], the sliced buffers within the offset segment
        """
        start = offset_segment[0]
        end = offset_segment[1]

        if end > self.end_offset:
            raise ValueError(
                f"Requested end offset {end} outside of available end offset"
                f" {self.end_offset}"
            )

        if pad_start is False and start < self.offset:
            raise ValueError(
                f"Requested offset {start} outside of available offset {self.offset}"
            )

        bufs = deque(
            b for b in self.buffers if b.offset <= end and b.end_offset >= start
        )

        if pad_start is True and start < bufs[0].offset:
            # pad buffers in front
            buf = bufs[0].pad_buffer(off=start)
            bufs.appendleft(buf)

        # check buffers at each end
        if bufs[0].offset < start:
            bufs[0] = bufs[0].sub_buffer(TSSlice(start, bufs[0].end_offset))
        if bufs[-1].end_offset > end:
            bufs[-1] = bufs[-1].sub_buffer(TSSlice(bufs[-1].offset, end))

        return bufs

    def copy_samples(self, nsamples: int, start_sample: int = 0) -> Array:
        """Copy nsamples from the start_sample of the deque.

        Args:
            nsamples:
                int, the number of samples to copy out of the audioadapter
            start_sample:
                int, start the copying from this sample point

        Returns:
            Array, the array of copied samples
        """
        start_offset = Offset.fromsamples(start_sample, self.sample_rate) + self.offset
        end_offset = Offset.fromsamples(nsamples, self.sample_rate) + start_offset

        return self.copy_samples_by_offset_segment((start_offset, end_offset))

    def copy_samples_by_offset_segment(
        self, offset_segment: tuple[int, int], pad_start: bool = False
    ) -> Array:
        """Copy samples within the offset segment.

        Args:
            offset_segment:
                tuple[int, int], the offset segment
            pad_start:
                bool, default False, pad zeros in front if offset_segment[0] is earlier
                than the available segment

        Returns:
            Array, the array of copied samples
        """
        if self.pre_cat_data is None:
            avail_seg = self.slice
        else:
            avail_seg = (self.pre_cat_data.offset, self.pre_cat_data.end_offset)

        assert offset_segment[1] <= avail_seg[1], (
            f"rate: {self.sample_rate} requested end segment outside of"
            f" available segment, requested: {offset_segment}, available: {avail_seg}"
        )

        if pad_start is False:
            assert offset_segment[0] >= avail_seg[0], (
                "requested start segment outside of available segment,"
                f" requested: {offset_segment}, available: {avail_seg}"
            )

        segment_has_gaps, segment_has_nongaps = self.segment_gaps_info(
            offset_segment, pad_start
        )
        # check gaps before copying
        if self.is_gap or not segment_has_nongaps:
            # no nongaps
            out = None
        else:
            if self.pre_cat_data is None:
                bufs = self.get_sliced_buffers(offset_segment, pad_start=pad_start)
                if len(bufs) == 1:
                    out = bufs[0].data
                else:
                    out = self.backend.cat(
                        [b.filleddata(self.backend.zeros) for b in bufs], axis=-1
                    )
            else:
                # find start sample
                ni = Offset.tosamples(
                    offset_segment[0] - self.pre_cat_data.offset, self.sample_rate
                )
                nsamples = Offset.tosamples(
                    offset_segment[1] - offset_segment[0], self.sample_rate
                )
                # FIXME: this is a thorny issue because of how generous we are
                # with the type of data and the type of Array.  Fixing this will
                # involve being stricter about types and more careful throughout
                # the array_ops module.
                out = self.pre_cat_data.data[..., ni : ni + nsamples]  # type: ignore

        return out

    def flush_samples(self, nsamples: int) -> None:
        """Flush nsamples from the head of the deque.

        Args:
            nsamples:
                int, the number of samples to flush from the head of the audioadapter
        """
        self.flush_samples_by_end_offset(
            self.offset + Offset.fromsamples(nsamples, self.sample_rate)
        )

    def flush_samples_by_end_offset(self, end_offset: int) -> None:
        """Flush nsamples from the head of the deque up to the end of the offset.

        Args:
            end_offset:
                int, the end offset
        """
        avail = self.slice
        if end_offset < avail[0] or end_offset > avail[1]:
            raise ValueError(
                f"offset segment outside of available segment" f" {end_offset} {avail}"
            )

        while self.size > 0:
            b = self.buffers[0]
            if b.end_offset < end_offset:
                # pop out old buffers
                self.buffers.popleft()
                if b.is_gap:
                    self.gap_size -= b.samples
                else:
                    self.nongap_size -= b.samples
                self.size -= b.samples
            elif b.end_offset == end_offset:
                # if b.end_offset == end_offset, have a zero-length buffer in the
                # adapter to record metadata
                self.buffers[0] = b.sub_buffer(
                    slc=TSSlice(end_offset, end_offset), gap=True
                )
                if b.is_gap:
                    self.gap_size -= b.samples
                else:
                    self.nongap_size -= b.samples
                self.size -= b.samples
                break
            else:
                if b.offset < end_offset:
                    # if the end_offset lies within a buffer, split the buffer
                    l, r = b.split(end_offset)
                    self.buffers[0] = r
                    if l.is_gap:
                        self.gap_size -= l.samples
                    else:
                        self.nongap_size -= l.samples
                    self.size -= l.samples
                break

        self.pre_cat_data = None

    def buffers_gaps_info(
        self, offset_segment: tuple[int, int], pad_start: bool = False
    ) -> list[bool]:
        """Return a list of booleans that flag buffers based on whether they are gaps.
        True: is_gap, False: is_nongap

        Args:
            offset_segment:
                tuple[int, int], the offset segment to get gaps info from
            pad_start:
                bool, default False, pad zeros in front if offset_segment[0] is earlier
                than the available segment

        Returns:
            list[bool], a list of booleans that flags whether buffers are gaps
        """
        return [
            b.is_gap
            for b in self.get_sliced_buffers(offset_segment, pad_start=pad_start)
        ]

    def samples_gaps_info(self, offset_segment: tuple[int, int]) -> Array:
        """Return an array of booleans that flag samples based on whether they are gaps.
        True: is_gap, False: is_nongap

        Args:
            offset_segment:
                tuple[int, int], the offset segment to get gaps info from

        Returns:
            Array, an array of booleans that flags whether buffers are gaps
        """
        return self.backend.cat(
            [
                self.backend.full((b.samples,), b.is_gap)
                for b in self.get_sliced_buffers(offset_segment)
            ],
            axis=-1,
        )

    def segment_gaps_info(
        self, offset_segment: tuple[int, int], pad_start: bool = False
    ) -> tuple[bool, bool]:
        """Identify whether there are gaps or nongaps in the requested offset_segment.

        Args:
            offset_segment:
                tuple[int, int], the offset segment to check for gaps and nongaps
            pad_start:
                bool, default False, pad zeros in front if offset_segment[0] is earlier
                than the available segment

        Returns:
            tuple[bool, bool], the tuple representing gaps information in the form
            (has_gaps, has_nongaps)
        """
        if self.is_gap:
            return True, False
        else:
            gaps = self.buffers_gaps_info(offset_segment, pad_start=pad_start)
            return True in gaps, False in gaps

end_offset property

The end offset of the last buffer in the audioadapter.

Returns:

Type Description
int

int, the end offset

is_gap property

True if all buffers are gaps.

Returns:

Type Description
bool

bool, if True, the whole audioadapter is a gap. If False, there are nongap

bool

buffers in the audioadapter

offset property

The offset of the first buffer in the audioadapter.

Returns:

Type Description
int

int, the offset

slice property

The offset slice of the audioadapter.

Returns:

Type Description
tuple[int, int]

tuple[int, int], the offset slice

buffers_gaps_info(offset_segment, pad_start=False)

Return a list of booleans that flag buffers based on whether they are gaps. True: is_gap, False: is_nongap

Parameters:

Name Type Description Default
offset_segment tuple[int, int]

tuple[int, int], the offset segment to get gaps info from

required
pad_start bool

bool, default False, pad zeros in front if offset_segment[0] is earlier than the available segment

False

Returns:

Type Description
list[bool]

list[bool], a list of booleans that flags whether buffers are gaps

Source code in sgnts/base/audioadapter.py
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
def buffers_gaps_info(
    self, offset_segment: tuple[int, int], pad_start: bool = False
) -> list[bool]:
    """Return a list of booleans that flag buffers based on whether they are gaps.
    True: is_gap, False: is_nongap

    Args:
        offset_segment:
            tuple[int, int], the offset segment to get gaps info from
        pad_start:
            bool, default False, pad zeros in front if offset_segment[0] is earlier
            than the available segment

    Returns:
        list[bool], a list of booleans that flags whether buffers are gaps
    """
    return [
        b.is_gap
        for b in self.get_sliced_buffers(offset_segment, pad_start=pad_start)
    ]

concatenate_data(offset_segment=None)

Concatenate all the data and gaps info in the buffers, and save as attribute.

Parameters:

Name Type Description Default
offset_segment Optional[tuple[int, int]]

Optional[tuple[int, int]], only concatenate data within this offset segment

None
Source code in sgnts/base/audioadapter.py
 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
def concatenate_data(
    self, offset_segment: Optional[tuple[int, int]] = None
) -> None:
    """Concatenate all the data and gaps info in the buffers, and save as attribute.

    Args:
        offset_segment:
            Optional[tuple[int, int]], only concatenate data within this offset
            segment
    """
    if self.size > 0:
        if offset_segment is not None:
            bufs = self.get_sliced_buffers(offset_segment)
        else:
            bufs = self.buffers
            offset_segment = self.slice

        cat_all = self.backend.cat(
            [b.filleddata(self.backend.zeros) for b in bufs], axis=-1
        )

        self.pre_cat_data = SeriesBuffer(
            offset=offset_segment[0],
            data=cat_all,
            sample_rate=bufs[0].sample_rate,
            shape=bufs[0].shape[:-1]
            + (
                Offset.tosamples(
                    offset_segment[1] - offset_segment[0], bufs[0].sample_rate
                ),
            ),
        )

copy_samples(nsamples, start_sample=0)

Copy nsamples from the start_sample of the deque.

Parameters:

Name Type Description Default
nsamples int

int, the number of samples to copy out of the audioadapter

required
start_sample int

int, start the copying from this sample point

0

Returns:

Type Description
Array

Array, the array of copied samples

Source code in sgnts/base/audioadapter.py
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def copy_samples(self, nsamples: int, start_sample: int = 0) -> Array:
    """Copy nsamples from the start_sample of the deque.

    Args:
        nsamples:
            int, the number of samples to copy out of the audioadapter
        start_sample:
            int, start the copying from this sample point

    Returns:
        Array, the array of copied samples
    """
    start_offset = Offset.fromsamples(start_sample, self.sample_rate) + self.offset
    end_offset = Offset.fromsamples(nsamples, self.sample_rate) + start_offset

    return self.copy_samples_by_offset_segment((start_offset, end_offset))

copy_samples_by_offset_segment(offset_segment, pad_start=False)

Copy samples within the offset segment.

Parameters:

Name Type Description Default
offset_segment tuple[int, int]

tuple[int, int], the offset segment

required
pad_start bool

bool, default False, pad zeros in front if offset_segment[0] is earlier than the available segment

False

Returns:

Type Description
Array

Array, the array of copied samples

Source code in sgnts/base/audioadapter.py
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
def copy_samples_by_offset_segment(
    self, offset_segment: tuple[int, int], pad_start: bool = False
) -> Array:
    """Copy samples within the offset segment.

    Args:
        offset_segment:
            tuple[int, int], the offset segment
        pad_start:
            bool, default False, pad zeros in front if offset_segment[0] is earlier
            than the available segment

    Returns:
        Array, the array of copied samples
    """
    if self.pre_cat_data is None:
        avail_seg = self.slice
    else:
        avail_seg = (self.pre_cat_data.offset, self.pre_cat_data.end_offset)

    assert offset_segment[1] <= avail_seg[1], (
        f"rate: {self.sample_rate} requested end segment outside of"
        f" available segment, requested: {offset_segment}, available: {avail_seg}"
    )

    if pad_start is False:
        assert offset_segment[0] >= avail_seg[0], (
            "requested start segment outside of available segment,"
            f" requested: {offset_segment}, available: {avail_seg}"
        )

    segment_has_gaps, segment_has_nongaps = self.segment_gaps_info(
        offset_segment, pad_start
    )
    # check gaps before copying
    if self.is_gap or not segment_has_nongaps:
        # no nongaps
        out = None
    else:
        if self.pre_cat_data is None:
            bufs = self.get_sliced_buffers(offset_segment, pad_start=pad_start)
            if len(bufs) == 1:
                out = bufs[0].data
            else:
                out = self.backend.cat(
                    [b.filleddata(self.backend.zeros) for b in bufs], axis=-1
                )
        else:
            # find start sample
            ni = Offset.tosamples(
                offset_segment[0] - self.pre_cat_data.offset, self.sample_rate
            )
            nsamples = Offset.tosamples(
                offset_segment[1] - offset_segment[0], self.sample_rate
            )
            # FIXME: this is a thorny issue because of how generous we are
            # with the type of data and the type of Array.  Fixing this will
            # involve being stricter about types and more careful throughout
            # the array_ops module.
            out = self.pre_cat_data.data[..., ni : ni + nsamples]  # type: ignore

    return out

flush_samples(nsamples)

Flush nsamples from the head of the deque.

Parameters:

Name Type Description Default
nsamples int

int, the number of samples to flush from the head of the audioadapter

required
Source code in sgnts/base/audioadapter.py
282
283
284
285
286
287
288
289
290
291
def flush_samples(self, nsamples: int) -> None:
    """Flush nsamples from the head of the deque.

    Args:
        nsamples:
            int, the number of samples to flush from the head of the audioadapter
    """
    self.flush_samples_by_end_offset(
        self.offset + Offset.fromsamples(nsamples, self.sample_rate)
    )

flush_samples_by_end_offset(end_offset)

Flush nsamples from the head of the deque up to the end of the offset.

Parameters:

Name Type Description Default
end_offset int

int, the end offset

required
Source code in sgnts/base/audioadapter.py
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
def flush_samples_by_end_offset(self, end_offset: int) -> None:
    """Flush nsamples from the head of the deque up to the end of the offset.

    Args:
        end_offset:
            int, the end offset
    """
    avail = self.slice
    if end_offset < avail[0] or end_offset > avail[1]:
        raise ValueError(
            f"offset segment outside of available segment" f" {end_offset} {avail}"
        )

    while self.size > 0:
        b = self.buffers[0]
        if b.end_offset < end_offset:
            # pop out old buffers
            self.buffers.popleft()
            if b.is_gap:
                self.gap_size -= b.samples
            else:
                self.nongap_size -= b.samples
            self.size -= b.samples
        elif b.end_offset == end_offset:
            # if b.end_offset == end_offset, have a zero-length buffer in the
            # adapter to record metadata
            self.buffers[0] = b.sub_buffer(
                slc=TSSlice(end_offset, end_offset), gap=True
            )
            if b.is_gap:
                self.gap_size -= b.samples
            else:
                self.nongap_size -= b.samples
            self.size -= b.samples
            break
        else:
            if b.offset < end_offset:
                # if the end_offset lies within a buffer, split the buffer
                l, r = b.split(end_offset)
                self.buffers[0] = r
                if l.is_gap:
                    self.gap_size -= l.samples
                else:
                    self.nongap_size -= l.samples
                self.size -= l.samples
            break

    self.pre_cat_data = None

get_sliced_buffers(offset_segment, pad_start=False)

Return buffers that lie within the offset_segment, slice up buffers if neeeded.

Parameters:

Name Type Description Default
offset_segment tuple[int, int]

tuple[int, int], the offset segment to get buffers from

required
pad_start bool

bool, default False, if True and if offset segment is earlier than the available buffers, will make front-pad the buffers with a gap buffer

False

Returns:

Type Description
Deque[SeriesBuffer]

Deque[SeriesBuffer], the sliced buffers within the offset segment

Source code in sgnts/base/audioadapter.py
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
190
191
192
193
194
195
196
197
198
199
200
def get_sliced_buffers(
    self, offset_segment: tuple[int, int], pad_start: bool = False
) -> Deque[SeriesBuffer]:
    """Return buffers that lie within the offset_segment, slice up buffers if
    neeeded.

    Args:
        offset_segment:
            tuple[int, int], the offset segment to get buffers from
        pad_start:
            bool, default False, if True and if offset segment is earlier than the
            available buffers, will make front-pad the buffers with a gap buffer

    Returns:
        Deque[SeriesBuffer], the sliced buffers within the offset segment
    """
    start = offset_segment[0]
    end = offset_segment[1]

    if end > self.end_offset:
        raise ValueError(
            f"Requested end offset {end} outside of available end offset"
            f" {self.end_offset}"
        )

    if pad_start is False and start < self.offset:
        raise ValueError(
            f"Requested offset {start} outside of available offset {self.offset}"
        )

    bufs = deque(
        b for b in self.buffers if b.offset <= end and b.end_offset >= start
    )

    if pad_start is True and start < bufs[0].offset:
        # pad buffers in front
        buf = bufs[0].pad_buffer(off=start)
        bufs.appendleft(buf)

    # check buffers at each end
    if bufs[0].offset < start:
        bufs[0] = bufs[0].sub_buffer(TSSlice(start, bufs[0].end_offset))
    if bufs[-1].end_offset > end:
        bufs[-1] = bufs[-1].sub_buffer(TSSlice(bufs[-1].offset, end))

    return bufs

push(buf)

Push buffer into the deque, update metadata.

Parameters:

Name Type Description Default
buf SeriesBuffer

SeriesBuffer, the buffer to append to the deque of the audioadapter.

required
Source code in sgnts/base/audioadapter.py
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
def push(self, buf: SeriesBuffer) -> None:
    """Push buffer into the deque, update metadata.

    Args:
        buf:
            SeriesBuffer, the buffer to append to the deque of the audioadapter.
    """
    if buf.noffset == 0 and len(self) > 0:
        # if there are no buffers and the very first buffer we receive
        # is a zero length buffer, still push it into the adapter
        return

    # Check if the start time is as expected
    # FIXME should we support discontinuities?
    if self.sample_rate != -1 and buf.offset != self.end_offset:
        raise ValueError(
            f"Got an unexpected buffer offset: {buf.offset=}"
            f" instead of {self.end_offset=} {buf=}"
        )

    if self.sample_rate == -1:
        self.sample_rate = buf.sample_rate
    elif buf.sample_rate != self.sample_rate:
        # buffers in the audioadapter must be the same sample rate
        raise ValueError(
            f"Inconsistent sample rate, buffer sample rate: {buf.sample_rate}"
            f" audioadpater sample rate: {self.sample_rate}"
        )

    # Store gap information
    nsamples = buf.samples
    self.size += nsamples
    if buf.is_gap is True:
        self.gap_size += nsamples
    else:
        self.nongap_size += nsamples

    if len(self) > 0 and self.buffers[-1].duration == 0:
        if buf.duration > 0:
            # Replace heartbeat buffers
            self.buffers[-1] = buf
    else:
        self.buffers.append(buf)

    self.pre_cat_data = None  # reset

samples_gaps_info(offset_segment)

Return an array of booleans that flag samples based on whether they are gaps. True: is_gap, False: is_nongap

Parameters:

Name Type Description Default
offset_segment tuple[int, int]

tuple[int, int], the offset segment to get gaps info from

required

Returns:

Type Description
Array

Array, an array of booleans that flags whether buffers are gaps

Source code in sgnts/base/audioadapter.py
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
def samples_gaps_info(self, offset_segment: tuple[int, int]) -> Array:
    """Return an array of booleans that flag samples based on whether they are gaps.
    True: is_gap, False: is_nongap

    Args:
        offset_segment:
            tuple[int, int], the offset segment to get gaps info from

    Returns:
        Array, an array of booleans that flags whether buffers are gaps
    """
    return self.backend.cat(
        [
            self.backend.full((b.samples,), b.is_gap)
            for b in self.get_sliced_buffers(offset_segment)
        ],
        axis=-1,
    )

segment_gaps_info(offset_segment, pad_start=False)

Identify whether there are gaps or nongaps in the requested offset_segment.

Parameters:

Name Type Description Default
offset_segment tuple[int, int]

tuple[int, int], the offset segment to check for gaps and nongaps

required
pad_start bool

bool, default False, pad zeros in front if offset_segment[0] is earlier than the available segment

False

Returns:

Type Description
bool

tuple[bool, bool], the tuple representing gaps information in the form

bool

(has_gaps, has_nongaps)

Source code in sgnts/base/audioadapter.py
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
def segment_gaps_info(
    self, offset_segment: tuple[int, int], pad_start: bool = False
) -> tuple[bool, bool]:
    """Identify whether there are gaps or nongaps in the requested offset_segment.

    Args:
        offset_segment:
            tuple[int, int], the offset segment to check for gaps and nongaps
        pad_start:
            bool, default False, pad zeros in front if offset_segment[0] is earlier
            than the available segment

    Returns:
        tuple[bool, bool], the tuple representing gaps information in the form
        (has_gaps, has_nongaps)
    """
    if self.is_gap:
        return True, False
    else:
        gaps = self.buffers_gaps_info(offset_segment, pad_start=pad_start)
        return True in gaps, False in gaps