sgnts.base.buffer

Event dataclass

Bases: TimeLike

              flowchart TD
              sgnts.base.buffer.Event[Event]
              sgnts.base.buffer.TimeLike[TimeLike]

                              sgnts.base.buffer.TimeLike --> sgnts.base.buffer.Event
                


              click sgnts.base.buffer.Event href "" "sgnts.base.buffer.Event"
              click sgnts.base.buffer.TimeLike href "" "sgnts.base.buffer.TimeLike"
            

Event with metadata.

Parameters:

Name	Type	Description	Default
offset	int	int, the offset of the buffer. See Offset class for definitions.	required
data	Any	Any, data of the event	None

Source code in sgnts/base/buffer.py

@dataclass
class Event(TimeLike):
    """Event with metadata.

    Args:
        offset:
            int, the offset of the buffer. See Offset class for definitions.
        data:
            Any, data of the event
    """

    offset: int
    data: Any = None

    @classmethod
    def from_time(cls, time: int, data: Any = None) -> Event:
        """Create an Event from a reference time (in nanoseconds)."""
        offset = Offset.fromns(time) - Offset.offset_ref_t0
        return cls(offset=offset, data=data)

from_time(time, data=None) classmethod

Create an Event from a reference time (in nanoseconds).

Source code in sgnts/base/buffer.py

@classmethod
def from_time(cls, time: int, data: Any = None) -> Event:
    """Create an Event from a reference time (in nanoseconds)."""
    offset = Offset.fromns(time) - Offset.offset_ref_t0
    return cls(offset=offset, data=data)

EventBuffer dataclass

Bases: TimeSpanLike

              flowchart TD
              sgnts.base.buffer.EventBuffer[EventBuffer]
              sgnts.base.buffer.TimeSpanLike[TimeSpanLike]
              sgnts.base.buffer.TimeLike[TimeLike]

                              sgnts.base.buffer.TimeSpanLike --> sgnts.base.buffer.EventBuffer
                                sgnts.base.buffer.TimeLike --> sgnts.base.buffer.TimeSpanLike
                



              click sgnts.base.buffer.EventBuffer href "" "sgnts.base.buffer.EventBuffer"
              click sgnts.base.buffer.TimeSpanLike href "" "sgnts.base.buffer.TimeSpanLike"
              click sgnts.base.buffer.TimeLike href "" "sgnts.base.buffer.TimeLike"
            

Event buffer with associated metadata.

Parameters:

Name	Type	Description	Default
offset	int	int, the offset of the buffer. See Offset class for definitions.	required
noffset	int	int, the number of offsets the buffer spans, or the duration.	required
data	Sequence[Any]	Sequence[Any], event data covering the span in question.	list()

Source code in sgnts/base/buffer.py

@dataclass(eq=False)
class EventBuffer(TimeSpanLike):
    """Event buffer with associated metadata.

    Args:
        offset:
            int, the offset of the buffer. See Offset class for definitions.
        noffset:
            int, the number of offsets the buffer spans, or the duration.
        data:
            Sequence[Any], event data covering the span in question.
    """

    offset: int
    noffset: int
    data: Sequence[Any] = field(default_factory=list)

    def __post_init__(self):
        if not isinstance(self.offset, int) or not isinstance(self.noffset, int):
            msg = "offset and noffset must be integers"
            raise ValueError(msg)

    @classmethod
    def from_span(
        cls, start: int, end: int, data: Sequence[Any] | None = None
    ) -> EventBuffer:
        """Create an EventBuffer from start/end times (in nanoseconds)."""
        if not isinstance(start, int) or not isinstance(end, int) or not (start <= end):
            raise ValueError(
                "start and end must be integers and start must be <= end,"
                f"got {start} and {end}"
            )
        offset = Offset.fromns(start)
        noffset = Offset.fromns(end) - offset
        if data is None:
            data = []
        return cls(offset=offset, noffset=noffset, data=data)

    def __iter__(self):
        return iter(self.data)

    def __getitem__(self, idx: int) -> Any:
        return self.data[idx]

    @property
    def events(self) -> Sequence[Any]:
        """The event data."""
        return self.data

    def __repr__(self):
        with numpy.printoptions(threshold=3, edgeitems=1):
            return "EventBuffer(offset=%d, end_offset=%d, data=%s)" % (
                self.offset,
                self.end_offset,
                self.data,
            )

    def __bool__(self):
        return bool(self.data)

    @property
    def is_gap(self):
        return not self.data

    def __contains__(self, item):
        # FIXME, is this what we want?
        if isinstance(item, int):
            # The end offset is not actually in the buffer hence the second "<" vs "<="
            return self.offset <= item < self.end_offset
        elif isinstance(item, EventBuffer):
            return (self.offset <= item.offset) and (item.end_offset <= self.end_offset)
        else:
            return False

events property

The event data.

from_span(start, end, data=None) classmethod

Create an EventBuffer from start/end times (in nanoseconds).

Source code in sgnts/base/buffer.py

@classmethod
def from_span(
    cls, start: int, end: int, data: Sequence[Any] | None = None
) -> EventBuffer:
    """Create an EventBuffer from start/end times (in nanoseconds)."""
    if not isinstance(start, int) or not isinstance(end, int) or not (start <= end):
        raise ValueError(
            "start and end must be integers and start must be <= end,"
            f"got {start} and {end}"
        )
    offset = Offset.fromns(start)
    noffset = Offset.fromns(end) - offset
    if data is None:
        data = []
    return cls(offset=offset, noffset=noffset, data=data)

EventFrame dataclass

Bases: Frame, TimeSpanLike

              flowchart TD
              sgnts.base.buffer.EventFrame[EventFrame]
              sgnts.base.buffer.TimeSpanLike[TimeSpanLike]
              sgnts.base.buffer.TimeLike[TimeLike]

                              sgnts.base.buffer.TimeSpanLike --> sgnts.base.buffer.EventFrame
                                sgnts.base.buffer.TimeLike --> sgnts.base.buffer.TimeSpanLike
                



              click sgnts.base.buffer.EventFrame href "" "sgnts.base.buffer.EventFrame"
              click sgnts.base.buffer.TimeSpanLike href "" "sgnts.base.buffer.TimeSpanLike"
              click sgnts.base.buffer.TimeLike href "" "sgnts.base.buffer.TimeLike"
            

An sgn Frame object that holds a dictionary of events.

Parameters:

Name	Type	Description	Default
events		dict, Dictionary of EventBuffers	required

Source code in sgnts/base/buffer.py

@dataclass(eq=False)
class EventFrame(Frame, TimeSpanLike):
    """An sgn Frame object that holds a dictionary of events.

    Args:
        events:
            dict, Dictionary of EventBuffers
    """

    data: list[EventBuffer] = field(default_factory=list)

    def __post_init__(self):
        super().__post_init__()
        assert len(self.data) > 0
        if (
            not isinstance(self.start, int)
            or not isinstance(self.end, int)
            or not (self.start <= self.end)
        ):
            raise ValueError(
                "start and end must be integers and start must be <= end,"
                f"got {self.start} and {self.end}"
            )

    def __iter__(self):
        return iter(self.data)

    def __getitem__(self, idx: int) -> EventBuffer:
        return self.data[idx]

    def __contains__(self, other):
        return other.slice in self.slice

    @property
    def events(self):
        """The list of Events."""
        return [event for buffer in self.data for event in buffer.data]

    def __repr__(self):
        out = (
            f"EventFrame(EOS={self.EOS}, is_gap={self.is_gap}, "
            f"metadata={self.metadata}, buffers={{\n"
        )
        for buf in self.data:
            out += f"    {buf},\n"
        out += "}})"
        return out

    @property
    def offset(self) -> int:
        """The offset of the EventFrame, which is the offset of the first buffer.

        Returns:
            int, the offset of the EventFrame
        """
        return self.data[0].offset

    @offset.setter
    def offset(self, other: int) -> None:
        msg = "cannot set offset on an EventFrame"
        raise AttributeError(msg)

    @property
    def noffset(self) -> int:
        """The end offset of the EventFrame, which is the end offset of the last buffer.

        Returns:
            int, the end offset of the EventFrame
        """
        return self.data[-1].end_offset - self.offset

    @noffset.setter
    def noffset(self, other: int) -> None:
        msg = "cannot set noffset on an EventFrame"
        raise AttributeError(msg)

events property

The list of Events.

noffset property writable

The end offset of the EventFrame, which is the end offset of the last buffer.

Returns:

Type	Description
int	int, the end offset of the EventFrame

offset property writable

The offset of the EventFrame, which is the offset of the first buffer.

Returns:

Type	Description
int	int, the offset of the EventFrame

SeriesBuffer dataclass

Bases: TimeSpanLike

              flowchart TD
              sgnts.base.buffer.SeriesBuffer[SeriesBuffer]
              sgnts.base.buffer.TimeSpanLike[TimeSpanLike]
              sgnts.base.buffer.TimeLike[TimeLike]

                              sgnts.base.buffer.TimeSpanLike --> sgnts.base.buffer.SeriesBuffer
                                sgnts.base.buffer.TimeLike --> sgnts.base.buffer.TimeSpanLike
                



              click sgnts.base.buffer.SeriesBuffer href "" "sgnts.base.buffer.SeriesBuffer"
              click sgnts.base.buffer.TimeSpanLike href "" "sgnts.base.buffer.TimeSpanLike"
              click sgnts.base.buffer.TimeLike href "" "sgnts.base.buffer.TimeLike"
            

Timeseries buffer with associated metadata.

Parameters:

Name	Type	Description	Default
offset	int	int, the offset of the buffer. See Offset class for definitions.	required
sample_rate	int	int, the sample rate belonging to the set of Offset.ALLOWED_RATES	required
data	Optional[Union[int, Array]]	Optional[Union[int, Array]], the timeseries data or None.	None
shape	tuple	tuple, the shape of the data regardless of gaps. Required if data is None or int, and represents the shape of the absent data.	(-1,)
backend	type[ArrayBackend]	type[ArrayBackend], default NumpyBackend, the wrapper around array operations	NumpyBackend

Source code in sgnts/base/buffer.py

@dataclass(eq=False)
class SeriesBuffer(TimeSpanLike):
    """Timeseries buffer with associated metadata.

    Args:
        offset:
            int, the offset of the buffer. See Offset class for definitions.
        sample_rate:
            int, the sample rate belonging to the set of Offset.ALLOWED_RATES
        data:
            Optional[Union[int, Array]], the timeseries data or None.
        shape:
            tuple, the shape of the data regardless of gaps. Required if data is None
            or int, and represents the shape of the absent data.
        backend:
            type[ArrayBackend], default NumpyBackend, the wrapper around array
            operations
    """

    offset: int
    sample_rate: int
    data: Optional[Union[int, Array]] = None
    shape: tuple = (-1,)
    backend: type[ArrayBackend] = NumpyBackend

    def __post_init__(self):
        assert isinstance(self.offset, int)
        if self.sample_rate not in Offset.ALLOWED_RATES:
            raise ValueError(
                "%s not in allowed rates %s" % (self.sample_rate, Offset.ALLOWED_RATES)
            )
        if self.data is None:
            if self.shape == (-1,):
                raise ValueError("if data is None self.shape must be given")
        elif isinstance(self.data, int) and self.data == 1:
            if self.shape == (-1,):
                raise ValueError("if data is 1 self.shape must be given")
            self.data = self.backend.ones(self.shape)
        elif isinstance(self.data, int) and self.data == 0:
            if self.shape == (-1,):
                raise ValueError("if data is 0 self.shape must be given")
            self.data = self.backend.zeros(self.shape)
        elif self.shape == (-1,):
            self.shape = self.data.shape
        else:
            if self.shape != self.data.shape:
                raise ValueError(
                    "Array size mismatch: self.shape and self.data.shape "
                    "must agree,"
                    f"got {self.shape} and {self.data.shape} "
                    f"with data {self.data}"
                )

        assert isinstance(self.shape, tuple)
        assert len(self.shape) > 0, f"Buffer shape cannot be empty, got {self.shape}"

        for t in self.shape:
            assert isinstance(t, int)

        # set the data specification
        self.spec = SeriesDataSpec(
            sample_rate=self.sample_rate, data_type=self.backend.DTYPE
        )

    @staticmethod
    def fromoffsetslice(
        offslice: TSSlice,
        sample_rate: int,
        data: Optional[Union[int, Array]] = None,
        channels: tuple[int, ...] = (),
    ) -> "SeriesBuffer":
        """Create a SeriesBuffer from a requested offset slice.

        Args:
            offslice:
                TSSlice, the offset slices the buffer spans
            sample_rate:
                int, the sample rate of the buffer
            data:
                Optional[Union[int, Array]], the data in the buffer
            channels:
                tuple[int, ...], the number of channels except the last dimension of the
                shape of the data, i.e., channels = data.shape[:-1]

        Returns:
            SeriesBuffer, the buffer that spans the requested offset slice
        """
        shape = channels + (
            Offset.tosamples(offslice.stop - offslice.start, sample_rate),
        )
        return SeriesBuffer(
            offset=offslice.start, sample_rate=sample_rate, data=data, shape=shape
        )

    def new(
        self,
        offslice: Optional[TSSlice] = None,
        data: Optional[Union[int, Array]] = None,
    ):
        """
        Return a new buffer from an existing one and optionally change the offsets.
        """
        return SeriesBuffer.fromoffsetslice(
            self.slice if offslice is None else offslice,
            self.sample_rate,
            data,
            self.shape[:-1],
        )

    def __repr__(self):
        with numpy.printoptions(threshold=3, edgeitems=1):
            return (
                "SeriesBuffer(offset=%d, offset_end=%d, shape=%s, sample_rate=%d,"
                " duration=%d, data=%s)"
                % (
                    self.offset,
                    self.end_offset,
                    self.shape,
                    self.sample_rate,
                    self.duration,
                    self.data,
                )
            )

    @property
    def properties(self):
        return {
            "offset": self.offset,
            "end_offset": self.end_offset,
            "t0": self.t0,
            "end": self.end,
            "shape": self.shape,
            "sample_shape": self.sample_shape,
            "sample_rate": self.sample_rate,
        }

    def __bool__(self):
        return self.data is not None

    def __len__(self):
        return 0 if self.data is None else len(self.data)

    def set_data(self, data: Optional[Array] = None) -> None:
        """Set the data attribute to the newly provided data.

        Args:
            data:
                Optional[Array], the new data to set to
        """
        if isinstance(data, int) and data == 1:
            self.data = self.backend.ones(self.shape)
        elif isinstance(data, int) and data == 0:
            self.data = self.backend.zeros(self.shape)
        elif isinstance(data, (int, float, complex)):
            # Handle any numeric value by creating an array filled with that value
            self.data = self.backend.full(self.shape, data)
        elif data is not None and self.shape != data.shape:
            raise ValueError("Data are incompatible shapes")
        else:
            # it really isn't clear to me if this should be by reference or copy...
            self.data = data

    @property
    def tarr(self) -> Array:
        """An array of time stamps for each sample of the data in the buffer, in
        seconds.

        Returns:
            Array, the time array
        """
        return (
            self.backend.arange(self.samples) / self.sample_rate
            + self.t0 / Time.SECONDS
        )

    def __eq__(self, value: Union[SeriesBuffer, Any]) -> bool:
        # FIXME this is a bit convoluted.  In order for some of these tests to
        # be triggered strange manipulation of objects would have to occur.
        # Consider making the SeriesBuffer properties read only where possible.
        is_series_buffer = isinstance(value, SeriesBuffer)
        if not is_series_buffer:
            return False
        if not (value.shape == self.shape):
            return False
        # FIXME is this the right check? Or do we want to check dtype? Under
        # what circumstances will this check fail?
        if type(self.data) is not type(value.data):
            return False
        if isinstance(self.data, NumpyArray) and isinstance(value.data, NumpyArray):
            share_data = NumpyBackend.all(self.data == value.data)
        elif isinstance(self.data, TorchArray) and isinstance(value.data, TorchArray):
            share_data = TorchBackend.all(self.data == value.data)
        elif self.data is None and value.data is None:
            share_data = True
        else:
            # Will need to expand this conditional if/when other data types are added
            raise ValueError("invalid data object")
        share_offset = value.offset == self.offset
        share_sample_rate = value.sample_rate == self.sample_rate
        return share_data and share_offset and share_sample_rate

    @property
    def noffset(self) -> int:
        """The number of offsets the buffer spans, which is the buffer's duration in
        terms of offsets.

        Returns:
            int, the offset duration
        """
        return Offset.fromsamples(self.samples, self.sample_rate)

    @noffset.setter
    def noffset(self, other: int) -> None:
        msg = "cannot set noffset on a SeriesBuffer"
        raise AttributeError(msg)

    @property
    def samples(self) -> int:
        """The number of samples the buffer carries.

        Return:
            int, the number of samples
        """
        assert len(self.shape) > 0, f"Buffer shape cannot be empty, got {self.shape}"
        return self.shape[-1]

    @property
    def sample_shape(self) -> tuple:
        """return the sample shape"""
        return self.shape[:-1]

    @property
    def is_gap(self) -> bool:
        """Whether the buffer is a gap. This is determined by whether the data is None.

        Returns:
            bool, whether the buffer is a gap
        """
        return self.data is None

    def filleddata(self, zeros_func=None) -> Array:
        """Fill the data with zeros if buffer is a gap, otherwise return the data.

        Args:
            zeros_func:
                the function to produce a zeros array

        Returns:
            Array, the filled data
        """
        if zeros_func is None:
            zeros_func = self.backend.zeros

        if self.data is not None:
            return self.data
        else:
            return zeros_func(self.shape)

    def __contains__(self, item):
        # FIXME, is this what we want?
        if isinstance(item, int):
            # The end offset is not actually in the buffer hence the second "<" vs "<="
            return self.offset <= item < self.end_offset
        elif isinstance(item, SeriesBuffer):
            return (self.offset <= item.offset) and (item.end_offset <= self.end_offset)
        else:
            return False

    def _insert(self, data: Array, offset) -> None:
        """TODO workshop the name
        Adds data from a whose slice is
        fully contained within self's into self.
        Does not do safety checks."""
        insertion_index = Offset.tosamples(
            offset - self.offset, sample_rate=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.
        self.data[
            ..., insertion_index : insertion_index + data.shape[-1]
        ] += data  # type: ignore

    @property
    def _backend_from_data(self):
        if isinstance(self.data, NumpyArray):
            return NumpyBackend
        elif isinstance(self.data, TorchArray):
            if (
                self.data.device != TorchBackend.DEVICE
                or self.data.dtype != TorchBackend.DTYPE
            ):
                raise ValueError("TorchArray and data backends are incompatable")
            return TorchBackend
        else:
            return None

    def __add__(self, item: "SeriesBuffer") -> "SeriesBuffer":
        """Add two `SeriesBuffer`s, padding as necessary.

        Args:
            item:
                SeriesBuffer, The other component of the addition. Must be a
                SeriesBuffer, must have the same sample rate as self, and its data must
                be the same type (e.g. numpy array or pytorch Tensor)

        Returns:
            SeriesBuffer, The SeriesBuffer resulting from the addition
        """
        # Choose the correct backend
        # Handle polymorphism more smoothly in the future?
        # It's python so maybe this is the best option available
        if not isinstance(item, SeriesBuffer):
            raise TypeError("Both arguments must be of the SeriesBuffer type")
        # A bit convoluted, cases are:
        # - if both None then output gap
        # - if one None fill the gap and add with other's backend
        # - if neither None but disagree raise an error
        backend = self._backend_from_data
        if (
            (backend != item._backend_from_data)
            and (item._backend_from_data is not None)
            and (backend is not None)
        ):
            raise TypeError("Incompatible data types")
        if backend is None and item._backend_from_data is not None:
            backend = item._backend_from_data
        if self.shape[:-1] != item.shape[:-1]:
            raise ValueError("All dimensions except the padding dimension must match")
        if self.sample_rate != item.sample_rate:
            raise ValueError("Sample rates must match")
        new_buffer = self.fromoffsetslice(
            self.slice | item.slice,
            sample_rate=self.sample_rate,
            data=None,
            channels=self.shape[:-1],
        )
        if backend is None:
            return new_buffer

        new_buffer.data = new_buffer.filleddata(backend.zeros)
        self_filled_data = self.filleddata(backend.zeros)
        item_filled_data = item.filleddata(backend.zeros)

        new_buffer._insert(self_filled_data, self.offset)
        new_buffer._insert(item_filled_data, item.offset)

        return new_buffer

    def pad_buffer(
        self, off: int, data: Optional[Union[int, Array]] = None
    ) -> "SeriesBuffer":
        """Generate a buffer to pad before this buffer.

        Args:
            off:
                int, the offset to start the padding. Must be earlier than this buffer.
            data:
                Optional[Union[int, Array]], the data of the pad buffer

        Returns:
            SeriesBuffer, the pad buffer
        """
        assert (
            off < self.offset
        ), f"Requested offset {off} must be before buffer offset {self.offset}"
        return SeriesBuffer(
            offset=off,
            sample_rate=self.sample_rate,
            data=data,
            shape=self.shape[:-1]
            + (Offset.tosamples(self.offset - off, self.sample_rate),),
        )

    def sub_buffer(self, slc: TSSlice, gap: bool = False) -> "SeriesBuffer":
        """Generate a sub buffer whose offset slice is within this buffer.

        Args:
            slc:
                TSSlice, the offset slice of the sub buffer
            gap:
                bool, if True, set the sub buffer to a gap

        Returns:
            SeriesBuffer, the sub buffer
        """
        assert (
            slc in self.slice
        ), f"Requested slice {slc} not contained in buffer slice {self.slice}"
        startsamples, stopsamples = Offset.tosamples(
            slc.start - self.offset, self.sample_rate
        ), Offset.tosamples(slc.stop - self.offset, self.sample_rate)
        if not gap and self.data is not None and not isinstance(self.data, int):
            data = self.data[..., startsamples:stopsamples]
        else:
            data = None

        return SeriesBuffer(
            offset=slc.start,
            sample_rate=self.sample_rate,
            data=data,
            shape=self.shape[:-1] + (stopsamples - startsamples,),
        )

    def split(
        self, boundaries: Union[int, TSSlices], contiguous: bool = False
    ) -> list["SeriesBuffer"]:
        """Split the buffer according to the requested offset boundaries.

        Args:
            boundaries:
                Union[int, TSSlices], the offset boundaries to split the buffer into.
            contiguous:
                bool, if True, will generate gap buffers when there are discontinuities

        Returns:
            list[SeriesBuffer], a list of SeriesBuffers split up according to the
            offset boundaries
        """
        out = []
        if isinstance(boundaries, int):
            boundaries = TSSlices(self.slice.split(boundaries))
        if not isinstance(boundaries, TSSlices):
            raise NotImplementedError
        for slc in boundaries.slices:
            assert (
                slc in self.slice
            ), f"Slice {slc} must be within buffer bounds {self.slice}"
            out.append(self.sub_buffer(slc))
        if contiguous:
            gap_boundaries = boundaries.invert(self.slice)
            for slc in gap_boundaries.slices:
                out.append(self.sub_buffer(slc, gap=True))
        return sorted(out)

is_gap property

Whether the buffer is a gap. This is determined by whether the data is None.

Returns:

Type	Description
bool	bool, whether the buffer is a gap

noffset property writable

The number of offsets the buffer spans, which is the buffer's duration in terms of offsets.

Returns:

Type	Description
int	int, the offset duration

sample_shape property

return the sample shape

samples property

The number of samples the buffer carries.

Return

int, the number of samples

tarr property

An array of time stamps for each sample of the data in the buffer, in seconds.

Returns:

Type	Description
Array	Array, the time array

__add__(item)

Add two SeriesBuffers, padding as necessary.

Parameters:

Name	Type	Description	Default
item	'SeriesBuffer'	SeriesBuffer, The other component of the addition. Must be a SeriesBuffer, must have the same sample rate as self, and its data must be the same type (e.g. numpy array or pytorch Tensor)	required

Returns:

Type	Description
'SeriesBuffer'	SeriesBuffer, The SeriesBuffer resulting from the addition

Source code in sgnts/base/buffer.py

def __add__(self, item: "SeriesBuffer") -> "SeriesBuffer":
    """Add two `SeriesBuffer`s, padding as necessary.

    Args:
        item:
            SeriesBuffer, The other component of the addition. Must be a
            SeriesBuffer, must have the same sample rate as self, and its data must
            be the same type (e.g. numpy array or pytorch Tensor)

    Returns:
        SeriesBuffer, The SeriesBuffer resulting from the addition
    """
    # Choose the correct backend
    # Handle polymorphism more smoothly in the future?
    # It's python so maybe this is the best option available
    if not isinstance(item, SeriesBuffer):
        raise TypeError("Both arguments must be of the SeriesBuffer type")
    # A bit convoluted, cases are:
    # - if both None then output gap
    # - if one None fill the gap and add with other's backend
    # - if neither None but disagree raise an error
    backend = self._backend_from_data
    if (
        (backend != item._backend_from_data)
        and (item._backend_from_data is not None)
        and (backend is not None)
    ):
        raise TypeError("Incompatible data types")
    if backend is None and item._backend_from_data is not None:
        backend = item._backend_from_data
    if self.shape[:-1] != item.shape[:-1]:
        raise ValueError("All dimensions except the padding dimension must match")
    if self.sample_rate != item.sample_rate:
        raise ValueError("Sample rates must match")
    new_buffer = self.fromoffsetslice(
        self.slice | item.slice,
        sample_rate=self.sample_rate,
        data=None,
        channels=self.shape[:-1],
    )
    if backend is None:
        return new_buffer

    new_buffer.data = new_buffer.filleddata(backend.zeros)
    self_filled_data = self.filleddata(backend.zeros)
    item_filled_data = item.filleddata(backend.zeros)

    new_buffer._insert(self_filled_data, self.offset)
    new_buffer._insert(item_filled_data, item.offset)

    return new_buffer

filleddata(zeros_func=None)

Fill the data with zeros if buffer is a gap, otherwise return the data.

Parameters:

Name	Type	Description	Default
zeros_func		the function to produce a zeros array	None

Returns:

Type	Description
Array	Array, the filled data

Source code in sgnts/base/buffer.py

def filleddata(self, zeros_func=None) -> Array:
    """Fill the data with zeros if buffer is a gap, otherwise return the data.

    Args:
        zeros_func:
            the function to produce a zeros array

    Returns:
        Array, the filled data
    """
    if zeros_func is None:
        zeros_func = self.backend.zeros

    if self.data is not None:
        return self.data
    else:
        return zeros_func(self.shape)

fromoffsetslice(offslice, sample_rate, data=None, channels=()) staticmethod

Create a SeriesBuffer from a requested offset slice.

Parameters:

Name	Type	Description	Default
offslice	TSSlice	TSSlice, the offset slices the buffer spans	required
sample_rate	int	int, the sample rate of the buffer	required
data	Optional[Union[int, Array]]	Optional[Union[int, Array]], the data in the buffer	None
channels	tuple[int, ...]	tuple[int, ...], the number of channels except the last dimension of the shape of the data, i.e., channels = data.shape[:-1]	()

Returns:

Type	Description
'SeriesBuffer'	SeriesBuffer, the buffer that spans the requested offset slice

Source code in sgnts/base/buffer.py

@staticmethod
def fromoffsetslice(
    offslice: TSSlice,
    sample_rate: int,
    data: Optional[Union[int, Array]] = None,
    channels: tuple[int, ...] = (),
) -> "SeriesBuffer":
    """Create a SeriesBuffer from a requested offset slice.

    Args:
        offslice:
            TSSlice, the offset slices the buffer spans
        sample_rate:
            int, the sample rate of the buffer
        data:
            Optional[Union[int, Array]], the data in the buffer
        channels:
            tuple[int, ...], the number of channels except the last dimension of the
            shape of the data, i.e., channels = data.shape[:-1]

    Returns:
        SeriesBuffer, the buffer that spans the requested offset slice
    """
    shape = channels + (
        Offset.tosamples(offslice.stop - offslice.start, sample_rate),
    )
    return SeriesBuffer(
        offset=offslice.start, sample_rate=sample_rate, data=data, shape=shape
    )

new(offslice=None, data=None)

Return a new buffer from an existing one and optionally change the offsets.

Source code in sgnts/base/buffer.py

def new(
    self,
    offslice: Optional[TSSlice] = None,
    data: Optional[Union[int, Array]] = None,
):
    """
    Return a new buffer from an existing one and optionally change the offsets.
    """
    return SeriesBuffer.fromoffsetslice(
        self.slice if offslice is None else offslice,
        self.sample_rate,
        data,
        self.shape[:-1],
    )

pad_buffer(off, data=None)

Generate a buffer to pad before this buffer.

Parameters:

Name	Type	Description	Default
off	int	int, the offset to start the padding. Must be earlier than this buffer.	required
data	Optional[Union[int, Array]]	Optional[Union[int, Array]], the data of the pad buffer	None

Returns:

Type	Description
'SeriesBuffer'	SeriesBuffer, the pad buffer

Source code in sgnts/base/buffer.py

def pad_buffer(
    self, off: int, data: Optional[Union[int, Array]] = None
) -> "SeriesBuffer":
    """Generate a buffer to pad before this buffer.

    Args:
        off:
            int, the offset to start the padding. Must be earlier than this buffer.
        data:
            Optional[Union[int, Array]], the data of the pad buffer

    Returns:
        SeriesBuffer, the pad buffer
    """
    assert (
        off < self.offset
    ), f"Requested offset {off} must be before buffer offset {self.offset}"
    return SeriesBuffer(
        offset=off,
        sample_rate=self.sample_rate,
        data=data,
        shape=self.shape[:-1]
        + (Offset.tosamples(self.offset - off, self.sample_rate),),
    )

set_data(data=None)

Set the data attribute to the newly provided data.

Parameters:

Name	Type	Description	Default
data	Optional[Array]	Optional[Array], the new data to set to	None

Source code in sgnts/base/buffer.py

def set_data(self, data: Optional[Array] = None) -> None:
    """Set the data attribute to the newly provided data.

    Args:
        data:
            Optional[Array], the new data to set to
    """
    if isinstance(data, int) and data == 1:
        self.data = self.backend.ones(self.shape)
    elif isinstance(data, int) and data == 0:
        self.data = self.backend.zeros(self.shape)
    elif isinstance(data, (int, float, complex)):
        # Handle any numeric value by creating an array filled with that value
        self.data = self.backend.full(self.shape, data)
    elif data is not None and self.shape != data.shape:
        raise ValueError("Data are incompatible shapes")
    else:
        # it really isn't clear to me if this should be by reference or copy...
        self.data = data

split(boundaries, contiguous=False)

Split the buffer according to the requested offset boundaries.

Parameters:

Name	Type	Description	Default
boundaries	Union[int, TSSlices]	Union[int, TSSlices], the offset boundaries to split the buffer into.	required
contiguous	bool	bool, if True, will generate gap buffers when there are discontinuities	False

Returns:

Type	Description
list['SeriesBuffer']	list[SeriesBuffer], a list of SeriesBuffers split up according to the
list['SeriesBuffer']	offset boundaries

Source code in sgnts/base/buffer.py

def split(
    self, boundaries: Union[int, TSSlices], contiguous: bool = False
) -> list["SeriesBuffer"]:
    """Split the buffer according to the requested offset boundaries.

    Args:
        boundaries:
            Union[int, TSSlices], the offset boundaries to split the buffer into.
        contiguous:
            bool, if True, will generate gap buffers when there are discontinuities

    Returns:
        list[SeriesBuffer], a list of SeriesBuffers split up according to the
        offset boundaries
    """
    out = []
    if isinstance(boundaries, int):
        boundaries = TSSlices(self.slice.split(boundaries))
    if not isinstance(boundaries, TSSlices):
        raise NotImplementedError
    for slc in boundaries.slices:
        assert (
            slc in self.slice
        ), f"Slice {slc} must be within buffer bounds {self.slice}"
        out.append(self.sub_buffer(slc))
    if contiguous:
        gap_boundaries = boundaries.invert(self.slice)
        for slc in gap_boundaries.slices:
            out.append(self.sub_buffer(slc, gap=True))
    return sorted(out)

sub_buffer(slc, gap=False)

Generate a sub buffer whose offset slice is within this buffer.

Parameters:

Name	Type	Description	Default
slc	TSSlice	TSSlice, the offset slice of the sub buffer	required
gap	bool	bool, if True, set the sub buffer to a gap	False

Returns:

Type	Description
'SeriesBuffer'	SeriesBuffer, the sub buffer

Source code in sgnts/base/buffer.py

def sub_buffer(self, slc: TSSlice, gap: bool = False) -> "SeriesBuffer":
    """Generate a sub buffer whose offset slice is within this buffer.

    Args:
        slc:
            TSSlice, the offset slice of the sub buffer
        gap:
            bool, if True, set the sub buffer to a gap

    Returns:
        SeriesBuffer, the sub buffer
    """
    assert (
        slc in self.slice
    ), f"Requested slice {slc} not contained in buffer slice {self.slice}"
    startsamples, stopsamples = Offset.tosamples(
        slc.start - self.offset, self.sample_rate
    ), Offset.tosamples(slc.stop - self.offset, self.sample_rate)
    if not gap and self.data is not None and not isinstance(self.data, int):
        data = self.data[..., startsamples:stopsamples]
    else:
        data = None

    return SeriesBuffer(
        offset=slc.start,
        sample_rate=self.sample_rate,
        data=data,
        shape=self.shape[:-1] + (stopsamples - startsamples,),
    )

SeriesDataSpec dataclass

Bases: DataSpec

              flowchart TD
              sgnts.base.buffer.SeriesDataSpec[SeriesDataSpec]

              

              click sgnts.base.buffer.SeriesDataSpec href "" "sgnts.base.buffer.SeriesDataSpec"
            

Data specification for timeseries.

Parameters:

Name	Type	Description	Default
sample_rate	int	int, the sample rate associated with the data.	required
data_type	Any	Any, the data type associated with the data.	required

Source code in sgnts/base/buffer.py

@dataclass(frozen=True)
class SeriesDataSpec(DataSpec):
    """Data specification for timeseries.

    Args:
        sample_rate:
            int, the sample rate associated with the data.
        data_type:
            Any, the data type associated with the data.
    """

    sample_rate: int
    data_type: Any

TSFrame dataclass

Bases: Frame, TimeSpanLike

              flowchart TD
              sgnts.base.buffer.TSFrame[TSFrame]
              sgnts.base.buffer.TimeSpanLike[TimeSpanLike]
              sgnts.base.buffer.TimeLike[TimeLike]

                              sgnts.base.buffer.TimeSpanLike --> sgnts.base.buffer.TSFrame
                                sgnts.base.buffer.TimeLike --> sgnts.base.buffer.TimeSpanLike
                



              click sgnts.base.buffer.TSFrame href "" "sgnts.base.buffer.TSFrame"
              click sgnts.base.buffer.TimeSpanLike href "" "sgnts.base.buffer.TimeSpanLike"
              click sgnts.base.buffer.TimeLike href "" "sgnts.base.buffer.TimeLike"
            

An sgn Frame object that holds a list of buffers

Parameters:

Name	Type	Description	Default
buffers	list[SeriesBuffer]	list[SeriesBuffer], An iterable of SeriesBuffers	list()

Source code in sgnts/base/buffer.py

@dataclass(eq=False)
class TSFrame(Frame, TimeSpanLike):
    """An sgn Frame object that holds a list of buffers

    Args:
        buffers:
            list[SeriesBuffer], An iterable of SeriesBuffers
    """

    buffers: list[SeriesBuffer] = field(default_factory=list)

    def __post_init__(self):
        super().__post_init__()
        assert len(self.buffers) > 0, "Cannot create TSFrame with empty buffers list"
        self.validate_buffers()
        self.update_buffer_attrs()
        self.spec = self.buffers[0].spec

    def __getitem__(self, item):
        return self.buffers[item]

    def __iter__(self):
        return iter(self.buffers)

    def __repr__(self):
        out = (
            f"TSFrame(EOS={self.EOS}, is_gap={self.is_gap}, "
            f"metadata={self.metadata}, buffers=[\n"
        )
        for buf in self:
            out += f"    {buf},\n"
        out += "])"
        return out

    def __len__(self):
        return len(self.buffers)

    def validate_buffers(self) -> None:
        """Sanity check that the buffers don't overlap nor have discontinuities.

        Args:
            bufs:
                list[SeriesBuffer], the buffers to perform the sanity check on
        """
        # FIXME: is there a smart way using TSSlics?

        if len(self.buffers) > 1:
            slices = [buf.slice for buf in self.buffers]
            off0 = slices[0].stop
            for sl in slices[1:]:
                assert off0 == sl.start, (
                    f"Buffer offset {off0} must match slice start {sl.start} "
                    f"for contiguous buffers"
                )
                off0 = sl.stop

        # Check all backends are the same
        backends = {buf.backend for buf in self.buffers}
        assert (
            len(backends) == 1
        ), f"All buffers must have the same backend, got {backends}"

        # check that data specifications are all the same
        data_specs = {buf.spec for buf in self.buffers}
        assert (
            len(data_specs) == 1
        ), f"All buffers must have the same data specifications, got {data_specs}"

    def update_buffer_attrs(self):
        """Helper method for updating buffer dependent attributes.

        This is useful since buffers are mutable, and there are cases where we modify
        the buffer contents after the TSFrame has been created, e.g., when preparing a
        return frame in a "new" method.
        """
        self.is_gap = all([b.is_gap for b in self.buffers])

    def set_buffers(self, bufs: list[SeriesBuffer]) -> None:
        """Set the buffers attribute to the bufs provided.

        Args:
            bufs:
                list[SeriesBuffers], the list of buffers to set to
        """
        self.buffers = bufs
        self.validate_buffers()
        self.update_buffer_attrs()

    @property
    def offset(self) -> int:
        """The offset of the TSFrame, which is the offset of the first buffer.

        Returns:
            int, the offset of the TSFrame
        """
        return self.buffers[0].offset

    @offset.setter
    def offset(self, other: int) -> None:
        msg = "cannot set offset on a TSFrame"
        raise AttributeError(msg)

    @property
    def noffset(self) -> int:
        """The end offset of the TSFrame, which is the end offset of the last buffer.

        Returns:
            int, the end offset of the TSFrame
        """
        return self.buffers[-1].end_offset - self.offset

    @noffset.setter
    def noffset(self, other: int) -> None:
        msg = "cannot set noffset on a TSFrame"
        raise AttributeError(msg)

    @property
    def shape(self) -> tuple[int, ...]:
        """The shape of the TSFrame.

        Returns:
            tuple[int, ...], the shape of the TSFrame
        """
        return self.buffers[0].shape[:-1] + (sum(b.samples for b in self.buffers),)

    @property
    def sample_shape(self) -> tuple:
        """return the sample shape"""
        return self.buffers[0].sample_shape

    @property
    def sample_rate(self) -> int:
        """The sample rate of the TSFrame.

        Returns:
            int, the sample rate
        """
        return self.buffers[0].sample_rate

    @classmethod
    def from_buffer_kwargs(cls, **kwargs):
        """A short hand for the following:

        >>> buf = SeriesBuffer(**kwargs)
        >>> frame = TSFrame(buffers=[buf])
        """
        return cls(buffers=[SeriesBuffer(**kwargs)])

    @property
    def backend(self) -> type[ArrayBackend]:
        """The backend of the buffers.

        Returns:
            type[ArrayBackend], the backend of the buffers
        """
        return self.buffers[0].backend

    def heartbeat(self, EOS=False):
        frame = TSFrame.from_buffer_kwargs(
            offset=self.offset,
            sample_rate=self.sample_rate,
            shape=self.sample_shape + (0,),
            data=None,
        )
        frame.EOS = EOS
        return frame

    def __next__(self):
        """
        return a new empty frame that is like the current one but advanced to
        the next offset, e.g.,

        >>> frame = TSFrame.from_buffer_kwargs(offset=0,
                        sample_rate=2048, shape=(2048,))
        >>> print (frame)

                SeriesBuffer(offset=0, offset_end=16384, shape=(2048,),
                             sample_rate=2048, duration=1000000000, data=None)
        >>> print (next(frame))
        """
        return self.from_buffer_kwargs(
            offset=self.end_offset, sample_rate=self.sample_rate, shape=self.shape
        )

    def __contains__(self, other):
        return other.slice in self.slice

    def intersect(self, other):
        """
        Intersect self with another frame and return up to three
        frames, the frame before, the intersecting frame and the frame after.  For
        example, given two frames A and B:

        A:
                SeriesBuffer(offset=0, offset_end=4096, shape=(32,),
                             sample_rate=128, duration=250000000, data=None)
                SeriesBuffer(offset=4096, offset_end=20480, shape=(128,),
                             sample_rate=128, duration=1000000000, data=None)
        B:
                SeriesBuffer(offset=2048, offset_end=10240, shape=(64,),
                             sample_rate=128, duration=500000000, data=None)
                SeriesBuffer(offset=10240, offset_end=174080, shape=(1280,),
                             sample_rate=128, duration=10000000000, data=None)

        B.intersect(A):

                before Frame:
                SeriesBuffer(offset=0, offset_end=2048, shape=(16,),
                             sample_rate=128, duration=125000000, data=None)

                intersecting Frame:
                SeriesBuffer(offset=2048, offset_end=4096, shape=(16,),
                             sample_rate=128, duration=125000000, data=None)
                SeriesBuffer(offset=4096, offset_end=20480, shape=(128,),
                             sample_rate=128, duration=1000000000, data=None)

                after Frame: None

        A.intersect(B):

                before Frame: None

                intersecting Frame:
                SeriesBuffer(offset=2048, offset_end=10240, shape=(64,),
                             sample_rate=128, duration=500000000, data=None)
                SeriesBuffer(offset=10240, offset_end=20480, shape=(80,),
                             sample_rate=128, duration=625000000, data=None)

                after Frame:
                SeriesBuffer(offset=20480, offset_end=174080, shape=(1200,),
                             sample_rate=128, duration=9375000000, data=None)
        """
        bbuf = []
        inbuf = []
        abuf = []
        for buf in other.buffers:
            if buf.end_offset <= self.offset:
                bbuf.append(buf)
            elif buf.offset >= self.end_offset:
                abuf.append(buf)
            elif buf in self:
                inbuf.append(buf)
            else:
                outside_slices = TSSlices(self.slice - buf.slice).search(buf.slice)
                outside_bufs = buf.split(outside_slices)
                for obuf in outside_bufs:
                    assert (obuf.end_offset <= self.offset) or (
                        obuf.offset >= self.end_offset
                    ), (
                        f"Buffer overlap detected - output buffer "
                        f"[{obuf.offset}, {obuf.end_offset}] must not overlap "
                        f"with frame range [{self.offset}, {self.end_offset}]"
                    )
                    if obuf.end_offset <= self.offset:
                        bbuf.append(obuf)
                    else:
                        abuf.append(obuf)
                inbuf.extend(buf.split(TSSlices([self.slice & buf.slice])))
        return (
            None if not bbuf else TSFrame(buffers=bbuf),
            None if not inbuf else TSFrame(buffers=inbuf),
            None if not abuf else TSFrame(buffers=abuf),
        )

    def filleddata(self) -> "TSFrame":
        """Combine the buffers of the frame into a single buffer,
        analogous to itertools.chain.

        Returns:
            TSFrame, the frame with a single buffer
        """
        arrays = [buf.filleddata() for buf in self.buffers]
        data = self.backend.cat(arrays, axis=-1)
        return TSFrame.from_buffer_kwargs(
            offset=self.offset,
            sample_rate=self.sample_rate,
            shape=self.shape,
            data=data,
        )

backend property

The backend of the buffers.

Returns:

Type	Description
type[ArrayBackend]	type[ArrayBackend], the backend of the buffers

noffset property writable

The end offset of the TSFrame, which is the end offset of the last buffer.

Returns:

Type	Description
int	int, the end offset of the TSFrame

offset property writable

The offset of the TSFrame, which is the offset of the first buffer.

Returns:

Type	Description
int	int, the offset of the TSFrame

sample_rate property

The sample rate of the TSFrame.

Returns:

Type	Description
int	int, the sample rate

sample_shape property

return the sample shape

shape property

The shape of the TSFrame.

Returns:

Type	Description
tuple[int, ...]	tuple[int, ...], the shape of the TSFrame

__next__()

return a new empty frame that is like the current one but advanced to the next offset, e.g.,

frame = TSFrame.from_buffer_kwargs(offset=0, sample_rate=2048, shape=(2048,)) print (frame)

    SeriesBuffer(offset=0, offset_end=16384, shape=(2048,),
                 sample_rate=2048, duration=1000000000, data=None)

print (next(frame))

Source code in sgnts/base/buffer.py

def __next__(self):
    """
    return a new empty frame that is like the current one but advanced to
    the next offset, e.g.,

    >>> frame = TSFrame.from_buffer_kwargs(offset=0,
                    sample_rate=2048, shape=(2048,))
    >>> print (frame)

            SeriesBuffer(offset=0, offset_end=16384, shape=(2048,),
                         sample_rate=2048, duration=1000000000, data=None)
    >>> print (next(frame))
    """
    return self.from_buffer_kwargs(
        offset=self.end_offset, sample_rate=self.sample_rate, shape=self.shape
    )

filleddata()

Combine the buffers of the frame into a single buffer, analogous to itertools.chain.

Returns:

Type	Description
'TSFrame'	TSFrame, the frame with a single buffer

Source code in sgnts/base/buffer.py

def filleddata(self) -> "TSFrame":
    """Combine the buffers of the frame into a single buffer,
    analogous to itertools.chain.

    Returns:
        TSFrame, the frame with a single buffer
    """
    arrays = [buf.filleddata() for buf in self.buffers]
    data = self.backend.cat(arrays, axis=-1)
    return TSFrame.from_buffer_kwargs(
        offset=self.offset,
        sample_rate=self.sample_rate,
        shape=self.shape,
        data=data,
    )

from_buffer_kwargs(**kwargs) classmethod

A short hand for the following:

buf = SeriesBuffer(**kwargs) frame = TSFrame(buffers=[buf])

Source code in sgnts/base/buffer.py

@classmethod
def from_buffer_kwargs(cls, **kwargs):
    """A short hand for the following:

    >>> buf = SeriesBuffer(**kwargs)
    >>> frame = TSFrame(buffers=[buf])
    """
    return cls(buffers=[SeriesBuffer(**kwargs)])

intersect(other)

Intersect self with another frame and return up to three frames, the frame before, the intersecting frame and the frame after. For example, given two frames A and B:

A

SeriesBuffer(offset=0, offset_end=4096, shape=(32,), sample_rate=128, duration=250000000, data=None) SeriesBuffer(offset=4096, offset_end=20480, shape=(128,), sample_rate=128, duration=1000000000, data=None)

B: SeriesBuffer(offset=2048, offset_end=10240, shape=(64,), sample_rate=128, duration=500000000, data=None) SeriesBuffer(offset=10240, offset_end=174080, shape=(1280,), sample_rate=128, duration=10000000000, data=None)

B.intersect(A):

    before Frame:
    SeriesBuffer(offset=0, offset_end=2048, shape=(16,),
                 sample_rate=128, duration=125000000, data=None)

    intersecting Frame:
    SeriesBuffer(offset=2048, offset_end=4096, shape=(16,),
                 sample_rate=128, duration=125000000, data=None)
    SeriesBuffer(offset=4096, offset_end=20480, shape=(128,),
                 sample_rate=128, duration=1000000000, data=None)

    after Frame: None

A.intersect(B):

    before Frame: None

    intersecting Frame:
    SeriesBuffer(offset=2048, offset_end=10240, shape=(64,),
                 sample_rate=128, duration=500000000, data=None)
    SeriesBuffer(offset=10240, offset_end=20480, shape=(80,),
                 sample_rate=128, duration=625000000, data=None)

    after Frame:
    SeriesBuffer(offset=20480, offset_end=174080, shape=(1200,),
                 sample_rate=128, duration=9375000000, data=None)

Source code in sgnts/base/buffer.py

def intersect(self, other):
    """
    Intersect self with another frame and return up to three
    frames, the frame before, the intersecting frame and the frame after.  For
    example, given two frames A and B:

    A:
            SeriesBuffer(offset=0, offset_end=4096, shape=(32,),
                         sample_rate=128, duration=250000000, data=None)
            SeriesBuffer(offset=4096, offset_end=20480, shape=(128,),
                         sample_rate=128, duration=1000000000, data=None)
    B:
            SeriesBuffer(offset=2048, offset_end=10240, shape=(64,),
                         sample_rate=128, duration=500000000, data=None)
            SeriesBuffer(offset=10240, offset_end=174080, shape=(1280,),
                         sample_rate=128, duration=10000000000, data=None)

    B.intersect(A):

            before Frame:
            SeriesBuffer(offset=0, offset_end=2048, shape=(16,),
                         sample_rate=128, duration=125000000, data=None)

            intersecting Frame:
            SeriesBuffer(offset=2048, offset_end=4096, shape=(16,),
                         sample_rate=128, duration=125000000, data=None)
            SeriesBuffer(offset=4096, offset_end=20480, shape=(128,),
                         sample_rate=128, duration=1000000000, data=None)

            after Frame: None

    A.intersect(B):

            before Frame: None

            intersecting Frame:
            SeriesBuffer(offset=2048, offset_end=10240, shape=(64,),
                         sample_rate=128, duration=500000000, data=None)
            SeriesBuffer(offset=10240, offset_end=20480, shape=(80,),
                         sample_rate=128, duration=625000000, data=None)

            after Frame:
            SeriesBuffer(offset=20480, offset_end=174080, shape=(1200,),
                         sample_rate=128, duration=9375000000, data=None)
    """
    bbuf = []
    inbuf = []
    abuf = []
    for buf in other.buffers:
        if buf.end_offset <= self.offset:
            bbuf.append(buf)
        elif buf.offset >= self.end_offset:
            abuf.append(buf)
        elif buf in self:
            inbuf.append(buf)
        else:
            outside_slices = TSSlices(self.slice - buf.slice).search(buf.slice)
            outside_bufs = buf.split(outside_slices)
            for obuf in outside_bufs:
                assert (obuf.end_offset <= self.offset) or (
                    obuf.offset >= self.end_offset
                ), (
                    f"Buffer overlap detected - output buffer "
                    f"[{obuf.offset}, {obuf.end_offset}] must not overlap "
                    f"with frame range [{self.offset}, {self.end_offset}]"
                )
                if obuf.end_offset <= self.offset:
                    bbuf.append(obuf)
                else:
                    abuf.append(obuf)
            inbuf.extend(buf.split(TSSlices([self.slice & buf.slice])))
    return (
        None if not bbuf else TSFrame(buffers=bbuf),
        None if not inbuf else TSFrame(buffers=inbuf),
        None if not abuf else TSFrame(buffers=abuf),
    )

set_buffers(bufs)

Set the buffers attribute to the bufs provided.

Parameters:

Name	Type	Description	Default
bufs	list[SeriesBuffer]	list[SeriesBuffers], the list of buffers to set to	required

Source code in sgnts/base/buffer.py

def set_buffers(self, bufs: list[SeriesBuffer]) -> None:
    """Set the buffers attribute to the bufs provided.

    Args:
        bufs:
            list[SeriesBuffers], the list of buffers to set to
    """
    self.buffers = bufs
    self.validate_buffers()
    self.update_buffer_attrs()

update_buffer_attrs()

Helper method for updating buffer dependent attributes.

This is useful since buffers are mutable, and there are cases where we modify the buffer contents after the TSFrame has been created, e.g., when preparing a return frame in a "new" method.

Source code in sgnts/base/buffer.py

def update_buffer_attrs(self):
    """Helper method for updating buffer dependent attributes.

    This is useful since buffers are mutable, and there are cases where we modify
    the buffer contents after the TSFrame has been created, e.g., when preparing a
    return frame in a "new" method.
    """
    self.is_gap = all([b.is_gap for b in self.buffers])

validate_buffers()

Sanity check that the buffers don't overlap nor have discontinuities.

Parameters:

Name	Type	Description	Default
bufs		list[SeriesBuffer], the buffers to perform the sanity check on	required

Source code in sgnts/base/buffer.py

def validate_buffers(self) -> None:
    """Sanity check that the buffers don't overlap nor have discontinuities.

    Args:
        bufs:
            list[SeriesBuffer], the buffers to perform the sanity check on
    """
    # FIXME: is there a smart way using TSSlics?

    if len(self.buffers) > 1:
        slices = [buf.slice for buf in self.buffers]
        off0 = slices[0].stop
        for sl in slices[1:]:
            assert off0 == sl.start, (
                f"Buffer offset {off0} must match slice start {sl.start} "
                f"for contiguous buffers"
            )
            off0 = sl.stop

    # Check all backends are the same
    backends = {buf.backend for buf in self.buffers}
    assert (
        len(backends) == 1
    ), f"All buffers must have the same backend, got {backends}"

    # check that data specifications are all the same
    data_specs = {buf.spec for buf in self.buffers}
    assert (
        len(data_specs) == 1
    ), f"All buffers must have the same data specifications, got {data_specs}"

TimeLike

Bases: Protocol

              flowchart TD
              sgnts.base.buffer.TimeLike[TimeLike]

              

              click sgnts.base.buffer.TimeLike href "" "sgnts.base.buffer.TimeLike"
            

Source code in sgnts/base/buffer.py

@runtime_checkable
class TimeLike(Protocol):
    offset: int

    @property
    def time(self) -> int:
        """The reference time, in integer nanoseconds.

        Returns:
            int, buffer time
        """
        return Offset.offset_ref_t0 + Offset.tons(self.offset)

    @property
    def t0(self) -> int:
        """The start (reference) time, in integer nanoseconds.

        Returns:
            int, buffer start time
        """
        return self.time

    @property
    def start(self) -> int:
        """The start (reference) time, in integer nanoseconds.

        Returns:
            int, buffer start time
        """
        return self.time

start property

The start (reference) time, in integer nanoseconds.

Returns:

Type	Description
int	int, buffer start time

t0 property

The start (reference) time, in integer nanoseconds.

Returns:

Type	Description
int	int, buffer start time

time property

The reference time, in integer nanoseconds.

Returns:

Type	Description
int	int, buffer time

TimeSpanLike

Bases: TimeLike, Protocol

              flowchart TD
              sgnts.base.buffer.TimeSpanLike[TimeSpanLike]
              sgnts.base.buffer.TimeLike[TimeLike]

                              sgnts.base.buffer.TimeLike --> sgnts.base.buffer.TimeSpanLike
                


              click sgnts.base.buffer.TimeSpanLike href "" "sgnts.base.buffer.TimeSpanLike"
              click sgnts.base.buffer.TimeLike href "" "sgnts.base.buffer.TimeLike"
            

Source code in sgnts/base/buffer.py

@total_ordering
@runtime_checkable
class TimeSpanLike(TimeLike, Protocol):
    noffset: int

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, TimeSpanLike):
            return NotImplemented
        return self.end_offset == other.end_offset

    def __lt__(self, other: TimeSpanLike) -> bool:
        return self.end_offset < other.end_offset

    @property
    def end_offset(self) -> int:
        """The end offset.

        Returns:
            int, end offset
        """
        return self.offset + self.noffset

    @property
    def slice(self) -> TSSlice:
        """The offset slice that the item spans.

        Returns:
            TSSlices, the offset slice
        """
        return TSSlice(self.offset, self.end_offset)

    @property
    def duration(self) -> int:
        """The duration of the buffer, in integer nanoseconds.

        Returns:
            int, the buffer duration
        """
        return Offset.tons(self.noffset)

    @property
    def end(self) -> int:
        """The end time of the buffer, in integer nanoseconds.

        Returns:
            int, buffer end time
        """
        return self.t0 + self.duration

duration property

The duration of the buffer, in integer nanoseconds.

Returns:

Type	Description
int	int, the buffer duration

end property

The end time of the buffer, in integer nanoseconds.

Returns:

Type	Description
int	int, buffer end time

end_offset property

The end offset.

Returns:

Type	Description
int	int, end offset

slice property

The offset slice that the item spans.

Returns:

Type	Description
TSSlice	TSSlices, the offset slice