Skip to content

sgnts.base

AdapterConfig dataclass

Config to hold parameters used for the audioadapter in _TSTransSink.

Parameters:

Name Type Description Default
overlap tuple[int, int]

tuple[int, int], the overlap before and after the data segment to process, in offsets

 (0, 0)
stride int

int, the stride to produce, in offsets

 0
pad_zeros_startup bool

bool, when overlap is provided, whether to pad zeros in front of the first buffer, or wait until there is enough data.

 False
skip_gaps bool

bool, produce a whole gap buffer if there are any gaps in the copied data segment

 False
backend type[ArrayBackend]

type[ArrayBackend], the ArrayBackend wrapper

 NumpyBackend
Source code in sgnts/base/audioadapter.py 
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
@dataclass
class AdapterConfig:
    """Config to hold parameters used for the audioadapter in _TSTransSink.

    Args:
        overlap:
            tuple[int, int], the overlap before and after the data segment to process,
            in offsets
        stride:
            int, the stride to produce, in offsets
        pad_zeros_startup:
            bool, when overlap is provided, whether to pad zeros in front of the
            first buffer, or wait until there is enough data.
        skip_gaps:
            bool, produce a whole gap buffer if there are any gaps in the copied data
            segment
        backend:
            type[ArrayBackend], the ArrayBackend wrapper
    """

    overlap: tuple[int, int] = (0, 0)
    stride: int = 0
    pad_zeros_startup: bool = False
    skip_gaps: bool = False
    backend: type[ArrayBackend] = NumpyBackend

    def valid_buffer(self, buf, data: Optional[Union[int, Array]] = 0):
        """
        Return a new buffer corresponding to the non overlapping part of a
        buffer "buf" as defined by this classes overlap properties As a special case,
        if the buffer is shape zero (a heartbeat buffer) a new heartbeat buffer is
        returned with the offsets shifted by overlap[0].
        Otherwise, in order for the buffer to be valid it must be what is expected
        based on the adapter's overlap and stride etc.
        """

        if buf.shape == (0,):
            new_slice = TSSlice(
                buf.slice[0] + self.overlap[0], buf.slice[0] + self.overlap[0]
            )
            return buf.new(new_slice, data=None)
        else:
            expected_shape = (
                Offset.tosamples(self.overlap[0], buf.sample_rate)
                + Offset.tosamples(self.overlap[1], buf.sample_rate)
                + Offset.sample_stride(buf.sample_rate),
            )
            assert buf.shape == expected_shape
            new_slice = TSSlice(
                buf.slice[0] + self.overlap[0], buf.slice[1] - self.overlap[1]
            )
            return buf.new(new_slice, data)

valid_buffer(buf, data=0)

Return a new buffer corresponding to the non overlapping part of a buffer "buf" as defined by this classes overlap properties As a special case, if the buffer is shape zero (a heartbeat buffer) a new heartbeat buffer is returned with the offsets shifted by overlap[0]. Otherwise, in order for the buffer to be valid it must be what is expected based on the adapter's overlap and stride etc.

Source code in sgnts/base/audioadapter.py 
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
def valid_buffer(self, buf, data: Optional[Union[int, Array]] = 0):
    """
    Return a new buffer corresponding to the non overlapping part of a
    buffer "buf" as defined by this classes overlap properties As a special case,
    if the buffer is shape zero (a heartbeat buffer) a new heartbeat buffer is
    returned with the offsets shifted by overlap[0].
    Otherwise, in order for the buffer to be valid it must be what is expected
    based on the adapter's overlap and stride etc.
    """

    if buf.shape == (0,):
        new_slice = TSSlice(
            buf.slice[0] + self.overlap[0], buf.slice[0] + self.overlap[0]
        )
        return buf.new(new_slice, data=None)
    else:
        expected_shape = (
            Offset.tosamples(self.overlap[0], buf.sample_rate)
            + Offset.tosamples(self.overlap[1], buf.sample_rate)
            + Offset.sample_stride(buf.sample_rate),
        )
        assert buf.shape == expected_shape
        new_slice = TSSlice(
            buf.slice[0] + self.overlap[0], buf.slice[1] - self.overlap[1]
        )
        return buf.new(new_slice, data)

TSTransform dataclass

Bases: TimeSeriesMixin[TSFrame], TransformElement[TSFrame]

A time-series transform element.

Source code in sgnts/base/base.py 
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
@dataclass
class TSTransform(TimeSeriesMixin[TSFrame], TransformElement[TSFrame]):
    """A time-series transform element."""

    def new(self, pad: SourcePad) -> TSFrame:
        """The transform function must be provided by the subclass.

        It should take the source pad as an argument and return a new
        TSFrame.

        Args:
            pad:
                SourcePad, The source pad that is producing the transformed frame

        Returns:
            TSFrame, The transformed frame

        """
        raise NotImplementedError

new(pad)

The transform function must be provided by the subclass.

It should take the source pad as an argument and return a new TSFrame.

Parameters:

Name Type Description Default
pad SourcePad

SourcePad, The source pad that is producing the transformed frame

 required

Returns:

Type Description
TSFrame

TSFrame, The transformed frame
Source code in sgnts/base/base.py 
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
def new(self, pad: SourcePad) -> TSFrame:
    """The transform function must be provided by the subclass.

    It should take the source pad as an argument and return a new
    TSFrame.

    Args:
        pad:
            SourcePad, The source pad that is producing the transformed frame

    Returns:
        TSFrame, The transformed frame

    """
    raise NotImplementedError

TSSink dataclass

Bases: TimeSeriesMixin[TSFrame], SinkElement[TSFrame]

A time-series sink element.

Source code in sgnts/base/base.py 
425
426
427
428
429
@dataclass
class TSSink(TimeSeriesMixin[TSFrame], SinkElement[TSFrame]):
    """A time-series sink element."""

    pass

TSSource dataclass

Bases: _TSSource

A time-series source that generates data in fixed-size buffers where the user can specify the start time and end time. If you want a data driven source consider using TSResourceSource.

Parameters:

Name Type Description Default
t0 float | None

float, start time of first buffer, in seconds

 None
end float | None

float, end time of the last buffer, in seconds

 None
duration float | None

float, alternative to end option, specify the duration of time to be covered in seconds. Cannot be given if end is given.

 None
Source code in sgnts/base/base.py 
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
@dataclass
class TSSource(_TSSource):
    """A time-series source that generates data in fixed-size buffers where the
       user can specify the start time and end time. If you want a data driven
       source consider using TSResourceSource.

    Args:
        t0:
            float, start time of first buffer, in seconds
        end:
            float, end time of the last buffer, in seconds
        duration:
            float, alternative to end option, specify the duration of
            time to be covered in seconds. Cannot be given if end is given.
    """

    t0: float | None = None
    end: float | None = None
    duration: float | None = None

    def __post_init__(self):
        super().__post_init__()

        if self.t0 is None:
            raise ValueError("You must specifiy a t0")

        if self.end is not None and self.duration is not None:
            raise ValueError("may specify either end or duration, not both")

        if self.duration is not None:
            self.end = self.t0 + self.duration

        if self.end is not None:
            assert self.end > self.t0, "end is before t0"

    @property
    def end_offset(self):
        if self.end is None:
            return float("inf")
        return Offset.fromsec(self.end - Offset.offset_ref_t0 / Time.SECONDS)

    @property
    def start_offset(self):
        return Offset.fromsec(self.t0 - Offset.offset_ref_t0 / Time.SECONDS)

    def set_pad_buffer_params(
        self,
        pad: SourcePad,
        sample_shape: tuple[int, ...],
        rate: int,
    ) -> None:
        """Set variables on the pad that are needed to construct SeriesBuffers.

        These should remain constant throughout the duration of the
        pipeline so this method may only be called once.

        Args:
            pad:
                SourcePad, the pad to setup buffers on
            sample_shape:
                tuple[int, ...], the shape of a single sample of the
                data, or put another way, the shape of the data except
                for the last (time) dimension,
                i.e. sample_shape=data.shape[:-1]
            rate:
                int, the sample rate of the data the pad will produce

        """
        # Make sure this has only been called once per pad
        assert (
            pad not in self._new_buffer_dict
        ), f"Pad {pad.name} already exists in _new_buffer_dict - duplicate pad entry"

        self._new_buffer_dict[pad] = {
            "sample_rate": rate,
            "shape": sample_shape + (self.num_samples(rate),),
        }
        self._next_frame_dict[pad] = TSFrame.from_buffer_kwargs(
            offset=self.start_offset, data=None, **self._new_buffer_dict[pad]
        )

set_pad_buffer_params(pad, sample_shape, rate)

Set variables on the pad that are needed to construct SeriesBuffers.

These should remain constant throughout the duration of the pipeline so this method may only be called once.

Parameters:

Name Type Description Default
pad SourcePad

SourcePad, the pad to setup buffers on

 required
sample_shape tuple[int, ...]

tuple[int, ...], the shape of a single sample of the data, or put another way, the shape of the data except for the last (time) dimension, i.e. sample_shape=data.shape[:-1]

 required
rate int

int, the sample rate of the data the pad will produce

 required
Source code in sgnts/base/base.py 
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
def set_pad_buffer_params(
    self,
    pad: SourcePad,
    sample_shape: tuple[int, ...],
    rate: int,
) -> None:
    """Set variables on the pad that are needed to construct SeriesBuffers.

    These should remain constant throughout the duration of the
    pipeline so this method may only be called once.

    Args:
        pad:
            SourcePad, the pad to setup buffers on
        sample_shape:
            tuple[int, ...], the shape of a single sample of the
            data, or put another way, the shape of the data except
            for the last (time) dimension,
            i.e. sample_shape=data.shape[:-1]
        rate:
            int, the sample rate of the data the pad will produce

    """
    # Make sure this has only been called once per pad
    assert (
        pad not in self._new_buffer_dict
    ), f"Pad {pad.name} already exists in _new_buffer_dict - duplicate pad entry"

    self._new_buffer_dict[pad] = {
        "sample_rate": rate,
        "shape": sample_shape + (self.num_samples(rate),),
    }
    self._next_frame_dict[pad] = TSFrame.from_buffer_kwargs(
        offset=self.start_offset, data=None, **self._new_buffer_dict[pad]
    )