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
align_to Optional[int]

int or None, alignment boundary in offsets When set, output offsets will be aligned to multiples of this value. For example: - Offset.fromsec(1) aligns to integer seconds - Offset.fromsamples(1024, rate) aligns to 1024-sample boundaries Default: None (no alignment)

 None
align_buffers bool

bool, when True, aligns buffer slices to the minimum sampling rate across all pads. This expands gaps and shrinks data slices to ensure all buffers align to integer sample boundaries at the lowest rate. Default: False

 False
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
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@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
        align_to:
            int or None, alignment boundary in offsets
            When set, output offsets will be aligned to multiples of this value.
            For example:
            - Offset.fromsec(1) aligns to integer seconds
            - Offset.fromsamples(1024, rate) aligns to 1024-sample boundaries
            Default: None (no alignment)
        align_buffers:
            bool, when True, aligns buffer slices to the minimum sampling rate
            across all pads. This expands gaps and shrinks data slices to ensure
            all buffers align to integer sample boundaries at the lowest rate.
            Default: False
    """

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

    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 
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
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]


              flowchart TD
              sgnts.base.TSTransform[TSTransform]
              sgnts.base.base.TimeSeriesMixin[TimeSeriesMixin]

                              sgnts.base.base.TimeSeriesMixin --> sgnts.base.TSTransform
                


              click sgnts.base.TSTransform href "" "sgnts.base.TSTransform"
              click sgnts.base.base.TimeSeriesMixin href "" "sgnts.base.base.TimeSeriesMixin"

A time-series transform element.

Source code in sgnts/base/base.py 
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
@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 
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
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]


              flowchart TD
              sgnts.base.TSSink[TSSink]
              sgnts.base.base.TimeSeriesMixin[TimeSeriesMixin]

                              sgnts.base.base.TimeSeriesMixin --> sgnts.base.TSSink
                


              click sgnts.base.TSSink href "" "sgnts.base.TSSink"
              click sgnts.base.base.TimeSeriesMixin href "" "sgnts.base.base.TimeSeriesMixin"

A time-series sink element.

Source code in sgnts/base/base.py 
617
618
619
620
621
@dataclass
class TSSink(TimeSeriesMixin[TSFrame], SinkElement[TSFrame]):
    """A time-series sink element."""

    pass

TSSource dataclass

Bases: _TSSource


              flowchart TD
              sgnts.base.TSSource[TSSource]
              sgnts.base.base._TSSource[_TSSource]

                              sgnts.base.base._TSSource --> sgnts.base.TSSource
                


              click sgnts.base.TSSource href "" "sgnts.base.TSSource"
              click sgnts.base.base._TSSource href "" "sgnts.base.base._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 
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
@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 
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
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]
    )