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 src/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: float, data: Any = None) -> Event:
        """Create an Event from a reference time (in seconds)."""
        offset = Offset.fromsec(time) - Offset.offset_ref_start
        return cls(offset=offset, data=data)

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

from_time(time, data=None) classmethod

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

Source code in src/sgnts/base/buffer.py

@classmethod
def from_time(cls, time: float, data: Any = None) -> Event:
    """Create an Event from a reference time (in seconds)."""
    offset = Offset.fromsec(time) - Offset.offset_ref_start
    return cls(offset=offset, data=data)

from_time_ns(time, data=None) classmethod

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

Source code in src/sgnts/base/buffer.py

@classmethod
def from_time_ns(cls, time: int, data: Any = None) -> Event:
    """Create an Event from a reference time (in nanoseconds)."""
    offset = Offset.fromns(time) - Offset.offset_ref_start
    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 src/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: float, end: float, data: Sequence[Any] | None = None
    ) -> EventBuffer:
        """Create an EventBuffer from start/end times (in seconds)."""
        if (
            not isinstance(start, (int, float))
            or not isinstance(end, (int, float))
            or not (start <= end)
        ):
            raise ValueError(
                "start and end must be numeric and start must be <= end, "
                f"got {start} and {end}"
            )
        offset = Offset.fromsec(start)
        noffset = Offset.fromsec(end) - offset
        if data is None:
            data = []
        return cls(offset=offset, noffset=noffset, data=data)

    @classmethod
    def from_span_ns(
        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 self.data is None or len(self.data) == 0

    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 seconds).

Source code in src/sgnts/base/buffer.py

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

from_span_ns(start, end, data=None) classmethod

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

Source code in src/sgnts/base/buffer.py

@classmethod
def from_span_ns(
    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: TimeSpanFrame

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

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




              click sgnts.base.buffer.EventFrame href "" "sgnts.base.buffer.EventFrame"
              click sgnts.base.buffer.TimeSpanFrame href "" "sgnts.base.buffer.TimeSpanFrame"
              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 EventBuffers.

EventFrame can be created with data (offset/noffset computed from buffers) or empty with explicit offset/noffset for incremental population.

Parameters:

Name	Type	Description	Default
data	list[EventBuffer]	list[EventBuffer], EventBuffers to hold	list()
offset	int	int, explicit offset when creating empty frame	0
noffset	int	int, explicit noffset (duration) when creating empty frame	0

Source code in src/sgnts/base/buffer.py

@dataclass(eq=False)
class EventFrame(TimeSpanFrame):
    """An sgn Frame object that holds a list of EventBuffers.

    EventFrame can be created with data (offset/noffset computed from buffers)
    or empty with explicit offset/noffset for incremental population.

    Args:
        data: list[EventBuffer], EventBuffers to hold
        offset: int, explicit offset when creating empty frame
        noffset: int, explicit noffset (duration) when creating empty frame
    """

    data: list[EventBuffer] = field(default_factory=list)
    offset: int = 0
    noffset: int = 0

    def __post_init__(self):
        super().__post_init__()
        # If data exists, compute offset/noffset from data
        if self.data:
            # Ensure user didn't try to manually set offset/noffset
            if self.offset != 0 or self.noffset != 0:
                raise ValueError(
                    "Cannot specify offset/noffset when providing data - "
                    "they are computed from data"
                )
            # Compute from data
            self.offset = self.data[0].offset
            self.noffset = self.data[-1].end_offset - self.offset

            # Validate computed values
            if (
                not isinstance(self.start, (int, float))
                or not isinstance(self.end, (int, float))
                or not (self.start <= self.end)
            ):
                raise ValueError(
                    "start and end must be numeric 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

    def append(self, item: EventBuffer) -> None:
        """Append EventBuffer with validation.

        Validates that buffer falls within frame bounds (offset to offset+noffset)
        and is contiguous with previous buffers.

        Args:
            item: EventBuffer to append

        Raises:
            AssertionError: If validation fails.
        """
        frame_end_offset = self.offset + self.noffset

        # Check buffer falls within bounds
        assert (
            self.offset <= item.offset
        ), f"Buffer offset {item.offset:_} starts before frame offset {self.offset:_}"
        assert item.end_offset <= frame_end_offset, (
            f"Buffer end_offset {item.end_offset:_} extends beyond frame "
            f"end_offset {frame_end_offset:_}"
        )

        # Check contiguity with previous buffer
        if self.data:
            assert item.offset == self.data[-1].end_offset, (
                f"Buffer offset {item.offset:_} is not contiguous with "
                f"previous buffer end {self.data[-1].end_offset:_}"
            )

        self.data.append(item)

    def validate_span(self) -> None:
        """Validate that data fully spans the offset/noffset range.

        Checks that:
        - First buffer starts at frame offset
        - Last buffer ends at frame offset+noffset (the frame's end_offset)
        - All buffers are contiguous

        Raises:
            AssertionError: If validation fails.
        """
        if self.data:
            frame_end_offset = self.offset + self.noffset

            assert self.data[0].offset == self.offset, (
                f"First buffer offset {self.data[0].offset:_} != "
                f"frame offset {self.offset:_}"
            )
            assert self.data[-1].end_offset == frame_end_offset, (
                f"Last buffer end_offset {self.data[-1].end_offset:_} != "
                f"frame end_offset {frame_end_offset:_}"
            )
            # Check all buffers are contiguous
            for i in range(1, len(self.data)):
                assert self.data[i].offset == self.data[i - 1].end_offset, (
                    f"Gap between buffer {i-1} (end={self.data[i-1].end_offset:_}) "
                    f"and buffer {i} (start={self.data[i].offset:_})"
                )

events property

The list of Events.

append(item)

Append EventBuffer with validation.

Validates that buffer falls within frame bounds (offset to offset+noffset) and is contiguous with previous buffers.

Parameters:

Name	Type	Description	Default
item	EventBuffer	EventBuffer to append	required

Raises:

Type	Description
AssertionError	If validation fails.

Source code in src/sgnts/base/buffer.py

def append(self, item: EventBuffer) -> None:
    """Append EventBuffer with validation.

    Validates that buffer falls within frame bounds (offset to offset+noffset)
    and is contiguous with previous buffers.

    Args:
        item: EventBuffer to append

    Raises:
        AssertionError: If validation fails.
    """
    frame_end_offset = self.offset + self.noffset

    # Check buffer falls within bounds
    assert (
        self.offset <= item.offset
    ), f"Buffer offset {item.offset:_} starts before frame offset {self.offset:_}"
    assert item.end_offset <= frame_end_offset, (
        f"Buffer end_offset {item.end_offset:_} extends beyond frame "
        f"end_offset {frame_end_offset:_}"
    )

    # Check contiguity with previous buffer
    if self.data:
        assert item.offset == self.data[-1].end_offset, (
            f"Buffer offset {item.offset:_} is not contiguous with "
            f"previous buffer end {self.data[-1].end_offset:_}"
        )

    self.data.append(item)

validate_span()

Validate that data fully spans the offset/noffset range.

Checks that: - First buffer starts at frame offset - Last buffer ends at frame offset+noffset (the frame's end_offset) - All buffers are contiguous

Raises:

Type	Description
AssertionError	If validation fails.

Source code in src/sgnts/base/buffer.py

def validate_span(self) -> None:
    """Validate that data fully spans the offset/noffset range.

    Checks that:
    - First buffer starts at frame offset
    - Last buffer ends at frame offset+noffset (the frame's end_offset)
    - All buffers are contiguous

    Raises:
        AssertionError: If validation fails.
    """
    if self.data:
        frame_end_offset = self.offset + self.noffset

        assert self.data[0].offset == self.offset, (
            f"First buffer offset {self.data[0].offset:_} != "
            f"frame offset {self.offset:_}"
        )
        assert self.data[-1].end_offset == frame_end_offset, (
            f"Last buffer end_offset {self.data[-1].end_offset:_} != "
            f"frame end_offset {frame_end_offset:_}"
        )
        # Check all buffers are contiguous
        for i in range(1, len(self.data)):
            assert self.data[i].offset == self.data[i - 1].end_offset, (
                f"Gap between buffer {i-1} (end={self.data[i-1].end_offset:_}) "
                f"and buffer {i} (start={self.data[i].offset:_})"
            )

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,)

Notes

The backend (numpy / torch / ...) is a property of data, not a configured field. Pass a real array and everything is inferred from it. The data=0/data=1 shorthand and gap fills materialize numpy arrays; for another backend, pass an array of that backend directly.

Source code in src/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.

    Notes:
        The backend (numpy / torch / ...) is a property of ``data``, not a
        configured field. Pass a real array and everything is inferred from it.
        The ``data=0``/``data=1`` shorthand and gap fills materialize **numpy**
        arrays; for another backend, pass an array of that backend directly.
    """

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

    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")
            _warn_scalar_data()
            self.data = numpy.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")
            _warn_scalar_data()
            self.data = numpy.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)

        # The backend identity is read from the array itself. A bare gap (or a
        # non-array placeholder) gets an incomplete spec; its frame supplies a
        # complete one for materialization.
        if backend_name(self.data) is not None:
            self.spec = SeriesDataSpec.from_data(self.data, self.sample_rate)
        else:
            self.spec = SeriesDataSpec(
                sample_rate=self.sample_rate,
                data_type=getattr(self.data, "dtype", None),
            )

    def __and__(self, other):
        sl = self.slice & other.slice
        if sl:
            return self.sub_buffer(sl)
        else:
            return None

    def isfinite(self):
        return self.slice.isfinite()

    def copy(
        self,
        offset: int | None = None,
        sample_rate: int | None = None,
        data: int | Array | None = None,
        is_gap: bool | None = None,
        shape: tuple | None = None,
    ) -> SeriesBuffer:
        """Returns a copy of the TSFrame with requested modifications.

        Any attributes not being changed will inherit from the original
        TSFrame.

        Args:
            offset:
                int, optional, the offset of the buffer. See Offset class for
                definitions.
            sample_rate:
                int, optional, the sample rate belonging to the set of
                Offset.ALLOWED_RATES
            data:
                int | Array, optional, the timeseries data.
            is_gap:
                bool, optional, set the buffer as a gap (or non-gap).
            shape:
                tuple, optional, the shape of the data regardless of gaps.
                Required if data is None or int, and represents the shape of
                the absent data.
        """
        offset = self.offset if offset is None else offset
        sample_rate = self.sample_rate if sample_rate is None else sample_rate
        shape = self.shape if shape is None else shape

        # using data=None as a case to decide whether to modify the buffer's
        # data with user-specified data needs extra care due to data=None also
        # being used to define the presence of a gap, so instead we enumerate
        # the possible cases based on whether is_gap is set and what the value
        # is if it is set.
        # NOTE: this can be simplified but is written as such to be explicit
        if is_gap is None:  # inherit buffer's gap status
            buf_data = self.data if data is None else data
        elif is_gap:  # change to gap
            buf_data = None
        else:  # change to non-gap
            buf_data = data

        return SeriesBuffer(
            offset=offset,
            sample_rate=sample_rate,
            data=buf_data,
            shape=shape,
        )

    @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
        """
        assert (
            offslice.units == TimeUnits.OFFSETS
        ), f"offset slice must be in offsets, got {offslice.units}"
        shape = channels + (
            Offset.tosamples(int(offslice.stop - offslice.start), sample_rate),
        )
        return SeriesBuffer(
            offset=int(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={}, "
                "sample_rate={:d}, duration_ns={:_d}, data={})".format(
                    self.offset,
                    self.end_offset,
                    self.shape,
                    self.sample_rate,
                    self.duration_ns,
                    self.data,
                )
            )

    @property
    def properties(self):
        return {
            "offset": self.offset,
            "end_offset": self.end_offset,
            "start": self.start,
            "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:
            _warn_scalar_data()
            self.data = numpy.ones(self.shape)
        elif isinstance(data, int) and data == 0:
            _warn_scalar_data()
            self.data = numpy.zeros(self.shape)
        elif isinstance(data, (int, float, complex)):
            # Handle any numeric value by creating an array filled with that value
            _warn_scalar_data()
            self.data = numpy.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 numpy.arange(self.samples) / self.sample_rate + self.start

    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
        # Compare elements in the data's own namespace (the array is the backend).
        xp = array_namespace(self.data)
        if xp is not None:
            share_data = xp.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 spanned by this buffer.

        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, reference: Optional[Array] = None) -> Array:
        """The buffer's data, with a gap materialized as zeros.

        A bare buffer has no backend of its own, so a gap's zeros are created to
        match ``reference`` ("the array is the backend"). This is the canonical
        way to fill a single buffer's gap.

        .. deprecated::
            Calling with **no** ``reference`` cannot know the backend a gap's
            zeros should be in and falls back to **numpy**. Pass a ``reference``
            array, or use :meth:`TSFrame.filleddata`, which materializes gaps in
            the frame's declared backend.

        Args:
            reference:
                Array, a reference whose namespace / dtype / device the gap zeros
                match. Required to fill a gap in a non-numpy backend.

        Returns:
            Array, the data (non-gap), or zeros matching ``reference`` (gap).
        """
        if not self.is_gap:
            return self.data
        if reference is not None:
            return new_zeros(reference, self.shape)
        warnings.warn(
            "SeriesBuffer.filleddata() with no reference falls back to numpy for "
            "a gap. Pass a reference array, or use TSFrame.filleddata().",
            DeprecationWarning,
            stacklevel=2,
        )
        return numpy.zeros(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

    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 backend (e.g. numpy array or pytorch Tensor)

        Returns:
            SeriesBuffer, The SeriesBuffer resulting from the addition
        """
        if not isinstance(item, SeriesBuffer):
            raise TypeError("Both arguments must be of the SeriesBuffer type")
        # Cases: both gaps -> gap; one gap -> fill it in the other's backend;
        # both present but different backends -> error.
        self_name = backend_name(self.data)
        item_name = backend_name(item.data)
        if self_name is not None and item_name is not None and self_name != item_name:
            raise TypeError("Incompatible data types")
        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],
        )
        # A reference array (whichever operand has data) fixes the backend, dtype
        # and device of the zeros we fill with.
        ref = self.data if self.data is not None else item.data
        if ref is None:  # both gaps -> output gap
            return new_buffer

        new_buffer.data = new_buffer.filleddata(ref)
        self_filled_data = self.filleddata(ref)
        item_filled_data = item.filleddata(ref)

        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(
            int(slc.start - self.offset), self.sample_rate
        ), Offset.tosamples(int(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=int(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)

    def plot(
        self,
        ax=None,
        label: Optional[str] = None,
        channel: Optional[Union[int, tuple]] = None,
        gap_color: str = "red",
        gap_alpha: float = 0.3,
        show_gaps: bool = True,
        time_unit: Literal["s", "ms", "ns", "gps"] = "gps",
        **kwargs,
    ):
        """Plot the buffer's time-series data.

        Requires matplotlib. Install with: pip install sgn-ts[plot]

        Args:
            ax:
                matplotlib Axes, optional. If None, creates a new figure and axes.
            label:
                str, optional. Legend label for this buffer's data line.
            channel:
                int or tuple[int, ...], optional. For multi-dimensional data,
                specifies which channel(s) to plot. If None and data is
                multi-dimensional, plots all channels.
            gap_color:
                str, color for gap region shading. Default 'red'.
            gap_alpha:
                float, alpha transparency for gap region shading. Default 0.3.
            show_gaps:
                bool, whether to show gap indicators. Default True.
            time_unit:
                str, time unit for x-axis: 's' (seconds since start), 'ms',
                'ns', or 'gps' (absolute GPS time). Default 'gps'.
            **kwargs:
                Additional keyword arguments passed to ax.plot().

        Returns:
            tuple: (fig, ax) matplotlib figure and axes objects
        """
        from sgnts.plotting import plot_buffer

        return plot_buffer(
            self,
            ax=ax,
            label=label,
            channel=channel,
            gap_color=gap_color,
            gap_alpha=gap_alpha,
            show_gaps=show_gaps,
            time_unit=time_unit,
            **kwargs,
        )

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 spanned by this buffer.

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 backend (e.g. numpy array or pytorch Tensor)	required

Returns:

Type	Description
SeriesBuffer	SeriesBuffer, The SeriesBuffer resulting from the addition

Source code in src/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 backend (e.g. numpy array or pytorch Tensor)

    Returns:
        SeriesBuffer, The SeriesBuffer resulting from the addition
    """
    if not isinstance(item, SeriesBuffer):
        raise TypeError("Both arguments must be of the SeriesBuffer type")
    # Cases: both gaps -> gap; one gap -> fill it in the other's backend;
    # both present but different backends -> error.
    self_name = backend_name(self.data)
    item_name = backend_name(item.data)
    if self_name is not None and item_name is not None and self_name != item_name:
        raise TypeError("Incompatible data types")
    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],
    )
    # A reference array (whichever operand has data) fixes the backend, dtype
    # and device of the zeros we fill with.
    ref = self.data if self.data is not None else item.data
    if ref is None:  # both gaps -> output gap
        return new_buffer

    new_buffer.data = new_buffer.filleddata(ref)
    self_filled_data = self.filleddata(ref)
    item_filled_data = item.filleddata(ref)

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

    return new_buffer

copy(offset=None, sample_rate=None, data=None, is_gap=None, shape=None)

Returns a copy of the TSFrame with requested modifications.

Any attributes not being changed will inherit from the original TSFrame.

Parameters:

Name	Type	Description	Default
offset	int | None	int, optional, the offset of the buffer. See Offset class for definitions.	None
sample_rate	int | None	int, optional, the sample rate belonging to the set of Offset.ALLOWED_RATES	None
data	int | Array | None	int | Array, optional, the timeseries data.	None
is_gap	bool | None	bool, optional, set the buffer as a gap (or non-gap).	None
shape	tuple | None	tuple, optional, the shape of the data regardless of gaps. Required if data is None or int, and represents the shape of the absent data.	None

Source code in src/sgnts/base/buffer.py

def copy(
    self,
    offset: int | None = None,
    sample_rate: int | None = None,
    data: int | Array | None = None,
    is_gap: bool | None = None,
    shape: tuple | None = None,
) -> SeriesBuffer:
    """Returns a copy of the TSFrame with requested modifications.

    Any attributes not being changed will inherit from the original
    TSFrame.

    Args:
        offset:
            int, optional, the offset of the buffer. See Offset class for
            definitions.
        sample_rate:
            int, optional, the sample rate belonging to the set of
            Offset.ALLOWED_RATES
        data:
            int | Array, optional, the timeseries data.
        is_gap:
            bool, optional, set the buffer as a gap (or non-gap).
        shape:
            tuple, optional, the shape of the data regardless of gaps.
            Required if data is None or int, and represents the shape of
            the absent data.
    """
    offset = self.offset if offset is None else offset
    sample_rate = self.sample_rate if sample_rate is None else sample_rate
    shape = self.shape if shape is None else shape

    # using data=None as a case to decide whether to modify the buffer's
    # data with user-specified data needs extra care due to data=None also
    # being used to define the presence of a gap, so instead we enumerate
    # the possible cases based on whether is_gap is set and what the value
    # is if it is set.
    # NOTE: this can be simplified but is written as such to be explicit
    if is_gap is None:  # inherit buffer's gap status
        buf_data = self.data if data is None else data
    elif is_gap:  # change to gap
        buf_data = None
    else:  # change to non-gap
        buf_data = data

    return SeriesBuffer(
        offset=offset,
        sample_rate=sample_rate,
        data=buf_data,
        shape=shape,
    )

filleddata(reference=None)

The buffer's data, with a gap materialized as zeros.

A bare buffer has no backend of its own, so a gap's zeros are created to match reference ("the array is the backend"). This is the canonical way to fill a single buffer's gap.

.. deprecated:: Calling with no reference cannot know the backend a gap's zeros should be in and falls back to numpy. Pass a reference array, or use :meth:TSFrame.filleddata, which materializes gaps in the frame's declared backend.

Parameters:

Name	Type	Description	Default
reference	Optional[Array]	Array, a reference whose namespace / dtype / device the gap zeros match. Required to fill a gap in a non-numpy backend.	None

Returns:

Type	Description
Array	Array, the data (non-gap), or zeros matching reference (gap).

Source code in src/sgnts/base/buffer.py

def filleddata(self, reference: Optional[Array] = None) -> Array:
    """The buffer's data, with a gap materialized as zeros.

    A bare buffer has no backend of its own, so a gap's zeros are created to
    match ``reference`` ("the array is the backend"). This is the canonical
    way to fill a single buffer's gap.

    .. deprecated::
        Calling with **no** ``reference`` cannot know the backend a gap's
        zeros should be in and falls back to **numpy**. Pass a ``reference``
        array, or use :meth:`TSFrame.filleddata`, which materializes gaps in
        the frame's declared backend.

    Args:
        reference:
            Array, a reference whose namespace / dtype / device the gap zeros
            match. Required to fill a gap in a non-numpy backend.

    Returns:
        Array, the data (non-gap), or zeros matching ``reference`` (gap).
    """
    if not self.is_gap:
        return self.data
    if reference is not None:
        return new_zeros(reference, self.shape)
    warnings.warn(
        "SeriesBuffer.filleddata() with no reference falls back to numpy for "
        "a gap. Pass a reference array, or use TSFrame.filleddata().",
        DeprecationWarning,
        stacklevel=2,
    )
    return numpy.zeros(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 src/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
    """
    assert (
        offslice.units == TimeUnits.OFFSETS
    ), f"offset slice must be in offsets, got {offslice.units}"
    shape = channels + (
        Offset.tosamples(int(offslice.stop - offslice.start), sample_rate),
    )
    return SeriesBuffer(
        offset=int(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 src/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 src/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),),
    )

plot(ax=None, label=None, channel=None, gap_color='red', gap_alpha=0.3, show_gaps=True, time_unit='gps', **kwargs)

Plot the buffer's time-series data.

Requires matplotlib. Install with: pip install sgn-ts[plot]

Parameters:

Name	Type	Description	Default
ax		matplotlib Axes, optional. If None, creates a new figure and axes.	None
label	Optional[str]	str, optional. Legend label for this buffer's data line.	None
channel	Optional[Union[int, tuple]]	int or tuple[int, ...], optional. For multi-dimensional data, specifies which channel(s) to plot. If None and data is multi-dimensional, plots all channels.	None
gap_color	str	str, color for gap region shading. Default 'red'.	'red'
gap_alpha	float	float, alpha transparency for gap region shading. Default 0.3.	0.3
show_gaps	bool	bool, whether to show gap indicators. Default True.	True
time_unit	Literal['s', 'ms', 'ns', 'gps']	str, time unit for x-axis: 's' (seconds since start), 'ms', 'ns', or 'gps' (absolute GPS time). Default 'gps'.	'gps'
**kwargs		Additional keyword arguments passed to ax.plot().	{}

Returns:

Name	Type	Description
tuple		(fig, ax) matplotlib figure and axes objects

Source code in src/sgnts/base/buffer.py

def plot(
    self,
    ax=None,
    label: Optional[str] = None,
    channel: Optional[Union[int, tuple]] = None,
    gap_color: str = "red",
    gap_alpha: float = 0.3,
    show_gaps: bool = True,
    time_unit: Literal["s", "ms", "ns", "gps"] = "gps",
    **kwargs,
):
    """Plot the buffer's time-series data.

    Requires matplotlib. Install with: pip install sgn-ts[plot]

    Args:
        ax:
            matplotlib Axes, optional. If None, creates a new figure and axes.
        label:
            str, optional. Legend label for this buffer's data line.
        channel:
            int or tuple[int, ...], optional. For multi-dimensional data,
            specifies which channel(s) to plot. If None and data is
            multi-dimensional, plots all channels.
        gap_color:
            str, color for gap region shading. Default 'red'.
        gap_alpha:
            float, alpha transparency for gap region shading. Default 0.3.
        show_gaps:
            bool, whether to show gap indicators. Default True.
        time_unit:
            str, time unit for x-axis: 's' (seconds since start), 'ms',
            'ns', or 'gps' (absolute GPS time). Default 'gps'.
        **kwargs:
            Additional keyword arguments passed to ax.plot().

    Returns:
        tuple: (fig, ax) matplotlib figure and axes objects
    """
    from sgnts.plotting import plot_buffer

    return plot_buffer(
        self,
        ax=ax,
        label=label,
        channel=channel,
        gap_color=gap_color,
        gap_alpha=gap_alpha,
        show_gaps=show_gaps,
        time_unit=time_unit,
        **kwargs,
    )

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 src/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:
        _warn_scalar_data()
        self.data = numpy.ones(self.shape)
    elif isinstance(data, int) and data == 0:
        _warn_scalar_data()
        self.data = numpy.zeros(self.shape)
    elif isinstance(data, (int, float, complex)):
        # Handle any numeric value by creating an array filled with that value
        _warn_scalar_data()
        self.data = numpy.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 src/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 src/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(
        int(slc.start - self.offset), self.sample_rate
    ), Offset.tosamples(int(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=int(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"
            

Backend identity and sample rate for a timeseries.

Embodies "the array is the backend": for real data the namespace, dtype, and device are read from the array. A bare gap (no array) has an incomplete spec; the frame it belongs to carries a complete spec — derived from its data, inherited from upstream, or the numpy default — so that even an all-gap frame knows how to materialize zeros.

Parameters:

Name	Type	Description	Default
sample_rate	int	int, the sample rate associated with the data.	required
data_type	Any	Any, the array dtype (None when unknown, e.g. a bare gap).	None
namespace	Any	Any, the Array API namespace (xp); None when unknown.	None
device	Any	Any, the device the data lives on; None when unknown.	None

Source code in src/sgnts/base/buffer.py

@dataclass(frozen=True)
class SeriesDataSpec(DataSpec):
    """Backend identity and sample rate for a timeseries.

    Embodies "the array is the backend": for real data the namespace, dtype, and
    device are read from the array. A bare gap (no array) has an *incomplete*
    spec; the **frame** it belongs to carries a complete spec — derived from its
    data, inherited from upstream, or the numpy default — so that even an all-gap
    frame knows how to materialize zeros.

    Args:
        sample_rate:
            int, the sample rate associated with the data.
        data_type:
            Any, the array dtype (None when unknown, e.g. a bare gap).
        namespace:
            Any, the Array API namespace (``xp``); None when unknown.
        device:
            Any, the device the data lives on; None when unknown.
    """

    # dtype is a first-class part of a stream's identity, alongside namespace and
    # device: a link carries one dtype for its life. A transform that *changes*
    # dtype (e.g. float -> complex) declares its output by example via
    # ``output_prototype`` so even its all-gap frames materialize in the right dtype
    # -- see e.g. Amplify / BitVector. The default inherits the input, correct for
    # the dtype-preserving majority.
    sample_rate: int
    data_type: Any = None
    namespace: Any = None
    device: Any = None

    @classmethod
    def from_data(cls, data: Any, sample_rate: int) -> "SeriesDataSpec":
        """A complete spec derived from a real array."""
        return cls(
            sample_rate=sample_rate,
            data_type=data.dtype,
            namespace=array_namespace(data),
            device=device(data),
        )

    @classmethod
    def numpy_default(cls, sample_rate: int) -> "SeriesDataSpec":
        """The from-nothing default: numpy / float64 / cpu."""
        return cls.from_data(numpy.zeros(0), sample_rate)

    @classmethod
    def from_first_data(cls, buffers) -> "SeriesDataSpec | None":
        """Spec from the first non-gap buffer's array, or None if all are gaps."""
        for b in buffers:
            if not b.is_gap and backend_name(b.data) is not None:
                return cls.from_data(b.data, b.sample_rate)
        return None

    @property
    def is_complete(self) -> bool:
        """Whether this spec can materialize arrays (has a namespace)."""
        return self.namespace is not None

    def zeros(self, shape: tuple) -> Any:
        """Materialize a zeros array of ``shape`` in this spec's backend."""
        return self.namespace.zeros(shape, dtype=self.data_type, device=self.device)

is_complete property

Whether this spec can materialize arrays (has a namespace).

from_data(data, sample_rate) classmethod

A complete spec derived from a real array.

Source code in src/sgnts/base/buffer.py

@classmethod
def from_data(cls, data: Any, sample_rate: int) -> "SeriesDataSpec":
    """A complete spec derived from a real array."""
    return cls(
        sample_rate=sample_rate,
        data_type=data.dtype,
        namespace=array_namespace(data),
        device=device(data),
    )

from_first_data(buffers) classmethod

Spec from the first non-gap buffer's array, or None if all are gaps.

Source code in src/sgnts/base/buffer.py

@classmethod
def from_first_data(cls, buffers) -> "SeriesDataSpec | None":
    """Spec from the first non-gap buffer's array, or None if all are gaps."""
    for b in buffers:
        if not b.is_gap and backend_name(b.data) is not None:
            return cls.from_data(b.data, b.sample_rate)
    return None

numpy_default(sample_rate) classmethod

The from-nothing default: numpy / float64 / cpu.

Source code in src/sgnts/base/buffer.py

@classmethod
def numpy_default(cls, sample_rate: int) -> "SeriesDataSpec":
    """The from-nothing default: numpy / float64 / cpu."""
    return cls.from_data(numpy.zeros(0), sample_rate)

zeros(shape)

Materialize a zeros array of shape in this spec's backend.

Source code in src/sgnts/base/buffer.py

def zeros(self, shape: tuple) -> Any:
    """Materialize a zeros array of ``shape`` in this spec's backend."""
    return self.namespace.zeros(shape, dtype=self.data_type, device=self.device)

TSCollectFrame dataclass

Bases: TimeSpanFrame

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

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




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

A collector for incrementally building a TSFrame with validation.

TSCollectFrame provides atomic all-or-nothing buffer collection: - Buffers are collected in a temporary list - Validation occurs on close() - Only commits to parent TSFrame if all validations pass - Can be used as a context manager for automatic cleanup

Parameters:

Name	Type	Description	Default
parent_frame	TSFrame	TSFrame, the frame to populate	required

Usage

Context manager (automatic close)

frame = TSFrame(offset=0, noffset=1000) with frame.fill() as collector: collector.append(buf1) collector.append(buf2)

frame now has buffers

Manual (explicit control)

frame = TSFrame(offset=0, noffset=1000) collector = frame.fill() collector.append(buf1) collector.close()

Source code in src/sgnts/base/buffer.py

@dataclass(eq=False, kw_only=True)
class TSCollectFrame(TimeSpanFrame):
    """A collector for incrementally building a TSFrame with validation.

    TSCollectFrame provides atomic all-or-nothing buffer collection:
    - Buffers are collected in a temporary list
    - Validation occurs on close()
    - Only commits to parent TSFrame if all validations pass
    - Can be used as a context manager for automatic cleanup

    Args:
        parent_frame: TSFrame, the frame to populate

    Usage:
        # Context manager (automatic close)
        frame = TSFrame(offset=0, noffset=1000)
        with frame.fill() as collector:
            collector.append(buf1)
            collector.append(buf2)
        # frame now has buffers

        # Manual (explicit control)
        frame = TSFrame(offset=0, noffset=1000)
        collector = frame.fill()
        collector.append(buf1)
        collector.close()
    """

    parent_frame: TSFrame
    _buffers: list[SeriesBuffer] = field(default_factory=list, init=False, repr=False)
    _closed: bool = field(default=False, init=False, repr=False)

    def __post_init__(self):
        super().__post_init__()
        # Inherit offset/noffset from parent
        self.offset = self.parent_frame.offset
        self.noffset = self.parent_frame.noffset
        self.EOS = self.parent_frame.EOS
        self.metadata = self.parent_frame.metadata

    def __iter__(self):
        """Iterate over collected buffers."""
        return iter(self._buffers)

    def __len__(self) -> int:
        """Return number of collected buffers."""
        return len(self._buffers)

    def __enter__(self) -> TSCollectFrame:
        """Enter context manager."""
        if len(self.parent_frame.buffers) > 0:
            raise ValueError(
                "Cannot use fill() on a TSFrame that already has buffers. "
                "TSCollectFrame can only populate empty frames."
            )
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Exit context manager - close if no exception occurred."""
        if exc_type is None:
            self.close()
        return False

    def append(self, item: SeriesBuffer) -> None:
        """Append SeriesBuffer to temporary collection.

        Validates that buffer falls within frame bounds and is contiguous
        with previous buffers. Does not commit to parent frame until close().

        Args:
            item: SeriesBuffer to append
        """
        if self._closed:
            raise ValueError("Cannot append to closed TSCollectFrame")

        frame_end_offset = self.offset + self.noffset

        # Check buffer falls within bounds
        assert (
            self.offset <= item.offset
        ), f"Buffer offset {item.offset:_} starts before frame offset {self.offset:_}"
        assert item.end_offset <= frame_end_offset, (
            f"Buffer end_offset {item.end_offset:_} extends beyond frame "
            f"end_offset {frame_end_offset:_}"
        )

        # Check contiguity with previous buffer
        if self._buffers:
            assert item.offset == self._buffers[-1].end_offset, (
                f"Buffer offset {item.offset:_} is not contiguous with "
                f"previous buffer end {self._buffers[-1].end_offset:_}"
            )
        else:
            # First buffer must start at frame offset
            assert item.offset == self.offset, (
                f"First buffer offset {item.offset:_} must match "
                f"frame offset {self.offset:_}"
            )

        self._buffers.append(item)

    def extend(self, items: Iterable[SeriesBuffer]) -> None:
        """Extend with multiple SeriesBuffers, validating each.

        Args:
            items: Iterable of SeriesBuffers to append
        """
        for item in items:
            self.append(item)

    def __iadd__(self, item: SeriesBuffer) -> TSCollectFrame:
        """Support += operator for appending."""
        self.append(item)
        return self

    def validate_span(self) -> None:
        """Validate that buffers fully span the offset/noffset range.

        Checks that:
        - First buffer starts at frame offset
        - Last buffer ends at frame offset+noffset (the frame's end_offset)
        """
        if not self._buffers:
            raise ValueError("Cannot validate empty TSCollectFrame - no buffers added")

        frame_end_offset = self.offset + self.noffset

        assert self._buffers[0].offset == self.offset, (
            f"First buffer offset {self._buffers[0].offset:_} != "
            f"frame offset {self.offset:_}"
        )
        assert self._buffers[-1].end_offset == frame_end_offset, (
            f"Last buffer end_offset {self._buffers[-1].end_offset:_} != "
            f"frame end_offset {frame_end_offset:_}"
        )

    def close(self) -> None:
        """Validate and commit buffers to parent TSFrame.

        This validates that buffers span the frame's offset/noffset range,
        then atomically commits them to the parent frame using set_buffers(),
        which performs additional validation (contiguity, consistent specs, etc.).

        After close(), this TSCollectFrame cannot be used again.
        """
        if self._closed:
            raise ValueError("TSCollectFrame already closed")

        # Validate that buffers span the frame's range
        self.validate_span()

        # Atomically commit to parent frame
        # set_buffers() handles contiguity, backend, and spec validation
        self.parent_frame.set_buffers(self._buffers)

        # Mark as closed
        self._closed = True

__enter__()

Enter context manager.

Source code in src/sgnts/base/buffer.py

def __enter__(self) -> TSCollectFrame:
    """Enter context manager."""
    if len(self.parent_frame.buffers) > 0:
        raise ValueError(
            "Cannot use fill() on a TSFrame that already has buffers. "
            "TSCollectFrame can only populate empty frames."
        )
    return self

__exit__(exc_type, exc_val, exc_tb)

Exit context manager - close if no exception occurred.

Source code in src/sgnts/base/buffer.py

def __exit__(self, exc_type, exc_val, exc_tb):
    """Exit context manager - close if no exception occurred."""
    if exc_type is None:
        self.close()
    return False

__iadd__(item)

Support += operator for appending.

Source code in src/sgnts/base/buffer.py

def __iadd__(self, item: SeriesBuffer) -> TSCollectFrame:
    """Support += operator for appending."""
    self.append(item)
    return self

__iter__()

Iterate over collected buffers.

Source code in src/sgnts/base/buffer.py

def __iter__(self):
    """Iterate over collected buffers."""
    return iter(self._buffers)

__len__()

Return number of collected buffers.

Source code in src/sgnts/base/buffer.py

def __len__(self) -> int:
    """Return number of collected buffers."""
    return len(self._buffers)

append(item)

Append SeriesBuffer to temporary collection.

Validates that buffer falls within frame bounds and is contiguous with previous buffers. Does not commit to parent frame until close().

Parameters:

Name	Type	Description	Default
item	SeriesBuffer	SeriesBuffer to append	required

Source code in src/sgnts/base/buffer.py

def append(self, item: SeriesBuffer) -> None:
    """Append SeriesBuffer to temporary collection.

    Validates that buffer falls within frame bounds and is contiguous
    with previous buffers. Does not commit to parent frame until close().

    Args:
        item: SeriesBuffer to append
    """
    if self._closed:
        raise ValueError("Cannot append to closed TSCollectFrame")

    frame_end_offset = self.offset + self.noffset

    # Check buffer falls within bounds
    assert (
        self.offset <= item.offset
    ), f"Buffer offset {item.offset:_} starts before frame offset {self.offset:_}"
    assert item.end_offset <= frame_end_offset, (
        f"Buffer end_offset {item.end_offset:_} extends beyond frame "
        f"end_offset {frame_end_offset:_}"
    )

    # Check contiguity with previous buffer
    if self._buffers:
        assert item.offset == self._buffers[-1].end_offset, (
            f"Buffer offset {item.offset:_} is not contiguous with "
            f"previous buffer end {self._buffers[-1].end_offset:_}"
        )
    else:
        # First buffer must start at frame offset
        assert item.offset == self.offset, (
            f"First buffer offset {item.offset:_} must match "
            f"frame offset {self.offset:_}"
        )

    self._buffers.append(item)

close()

Validate and commit buffers to parent TSFrame.

This validates that buffers span the frame's offset/noffset range, then atomically commits them to the parent frame using set_buffers(), which performs additional validation (contiguity, consistent specs, etc.).

After close(), this TSCollectFrame cannot be used again.

Source code in src/sgnts/base/buffer.py

def close(self) -> None:
    """Validate and commit buffers to parent TSFrame.

    This validates that buffers span the frame's offset/noffset range,
    then atomically commits them to the parent frame using set_buffers(),
    which performs additional validation (contiguity, consistent specs, etc.).

    After close(), this TSCollectFrame cannot be used again.
    """
    if self._closed:
        raise ValueError("TSCollectFrame already closed")

    # Validate that buffers span the frame's range
    self.validate_span()

    # Atomically commit to parent frame
    # set_buffers() handles contiguity, backend, and spec validation
    self.parent_frame.set_buffers(self._buffers)

    # Mark as closed
    self._closed = True

extend(items)

Extend with multiple SeriesBuffers, validating each.

Parameters:

Name	Type	Description	Default
items	Iterable[SeriesBuffer]	Iterable of SeriesBuffers to append	required

Source code in src/sgnts/base/buffer.py

def extend(self, items: Iterable[SeriesBuffer]) -> None:
    """Extend with multiple SeriesBuffers, validating each.

    Args:
        items: Iterable of SeriesBuffers to append
    """
    for item in items:
        self.append(item)

validate_span()

Validate that buffers fully span the offset/noffset range.

Checks that: - First buffer starts at frame offset - Last buffer ends at frame offset+noffset (the frame's end_offset)

Source code in src/sgnts/base/buffer.py

def validate_span(self) -> None:
    """Validate that buffers fully span the offset/noffset range.

    Checks that:
    - First buffer starts at frame offset
    - Last buffer ends at frame offset+noffset (the frame's end_offset)
    """
    if not self._buffers:
        raise ValueError("Cannot validate empty TSCollectFrame - no buffers added")

    frame_end_offset = self.offset + self.noffset

    assert self._buffers[0].offset == self.offset, (
        f"First buffer offset {self._buffers[0].offset:_} != "
        f"frame offset {self.offset:_}"
    )
    assert self._buffers[-1].end_offset == frame_end_offset, (
        f"Last buffer end_offset {self._buffers[-1].end_offset:_} != "
        f"frame end_offset {frame_end_offset:_}"
    )

TSFrame dataclass

Bases: TimeSpanFrame

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

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




              click sgnts.base.buffer.TSFrame href "" "sgnts.base.buffer.TSFrame"
              click sgnts.base.buffer.TimeSpanFrame href "" "sgnts.base.buffer.TimeSpanFrame"
              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

TSFrame can be created with data (offset/noffset computed from buffers) or empty with explicit offset/noffset for incremental population.

Parameters:

Name	Type	Description	Default
buffers	list[SeriesBuffer]	list[SeriesBuffer], SeriesBuffers to hold	list()
offset	int	int, explicit offset when creating empty frame	0
noffset	int	int, explicit noffset (duration) when creating empty frame	0

Source code in src/sgnts/base/buffer.py

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

    TSFrame can be created with data (offset/noffset computed from buffers)
    or empty with explicit offset/noffset for incremental population.

    Args:
        buffers: list[SeriesBuffer], SeriesBuffers to hold
        offset: int, explicit offset when creating empty frame
        noffset: int, explicit noffset (duration) when creating empty frame
    """

    buffers: list[SeriesBuffer] = field(default_factory=list)
    offset: int = 0
    noffset: int = 0

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

        # Optional inherited/stamped backend identity, used only to materialize
        # an all-gap frame (set by sources, a Converter, or upstream inheritance).
        self._spec_hint: SeriesDataSpec | None = None

        # If buffers exist, compute offset/noffset from buffers
        if self.buffers:
            # Ensure user didn't try to manually set offset/noffset
            if self.offset != 0 or self.noffset != 0:
                raise ValueError(
                    "Cannot specify offset/noffset when providing buffers - "
                    "they are computed from buffers"
                )

            # Compute from buffers
            self.offset = self.buffers[0].offset
            self.noffset = self.buffers[-1].end_offset - self.offset

            # Validate and update buffer-dependent attributes
            self.validate_buffers()
            self.update_buffer_attrs()
            self.spec = self._compute_spec()
        else:
            # Empty frame - offset/noffset are used as-is
            # Set default attributes for empty frame
            self.is_gap = False
            self.spec = None

    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 fill(self) -> TSCollectFrame:
        """Create a TSCollectFrame for atomically populating this frame.

        Returns a TSCollectFrame that can be used to incrementally add buffers
        with validation. The buffers are only committed to this frame when
        close() is called (or automatically via context manager).

        Returns:
            TSCollectFrame: A collector for building this frame

        Usage:
            # Context manager (recommended - automatic close)
            frame = TSFrame(offset=0, noffset=1000)
            with frame.fill() as collector:
                collector.append(buf1)
                collector.append(buf2)
            # frame now has buffers

            # Manual (explicit control)
            frame = TSFrame(offset=0, noffset=1000)
            collector = frame.fill()
            collector.append(buf1)
            collector.close()  # commits to frame
        """
        return TSCollectFrame(parent_frame=self)

    def validate_span(self) -> None:
        """Validate that buffers fully span the offset/noffset range.

        Checks that:
        - First buffer starts at frame offset
        - Last buffer ends at frame offset+noffset (the frame's end_offset)
        - All buffers are contiguous
        """
        if self.buffers:
            frame_end_offset = self.offset + self.noffset

            assert self.buffers[0].offset == self.offset, (
                f"First buffer offset {self.buffers[0].offset:_} != "
                f"frame offset {self.offset:_}"
            )
            assert self.buffers[-1].end_offset == frame_end_offset, (
                f"Last buffer end_offset {self.buffers[-1].end_offset:_} != "
                f"frame end_offset {frame_end_offset:_}"
            )
            # validate_buffers checks contiguity
            self.validate_buffers()

    def validate_buffers(self) -> None:
        """Sanity check that the buffers don't overlap nor have discontinuities."""
        # 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

        # All non-gap buffers must share one backend (the array is the backend);
        # gaps carry no array and are exempt. (We do not assert dtype uniformity:
        # the previous spec check compared a constant global dtype and so never
        # fired, and real frames may legitimately mix dtypes.)
        backends = {backend_name(buf.data) for buf in self.buffers if not buf.is_gap}
        assert (
            len(backends) <= 1
        ), f"All buffers must have the same backend, got {backends}"

    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 _compute_spec(self) -> SeriesDataSpec | None:
        """The frame's backend identity, read from its live buffer data.

        Derived from the first non-gap buffer's actual array ("the array is the
        backend"). For an all-gap frame — which has no array to read — it uses a
        stamped hint (set by a source/Converter or inherited from upstream), and
        finally the numpy default. This is what lets an all-gap frame still know
        the backend it belongs to.
        """
        data_spec = SeriesDataSpec.from_first_data(self.buffers)
        if data_spec is not None:
            return data_spec
        if not self.buffers:
            return None
        # All-gap: take the sample rate from the frame itself (so rate-changing
        # transforms need no special handling) and inherit only the *backend
        # identity* (namespace/dtype/device) from upstream, else numpy.
        sr = self.buffers[0].sample_rate
        hint = self._spec_hint
        if hint is not None and hint.is_complete:
            return SeriesDataSpec(
                sample_rate=sr,
                data_type=hint.data_type,
                namespace=hint.namespace,
                device=hint.device,
            )
        return SeriesDataSpec.numpy_default(sr)

    def apply_spec_hint(self, spec_hint: SeriesDataSpec | None) -> None:
        """Adopt ``spec_hint`` as this frame's backend identity, recomputing ``spec``.

        An all-gap frame has no data to read its backend from; an element that builds
        such a frame by hand (a heartbeat or all-gap stretch) passes the upstream spec
        as a hint, and this stamps it and re-derives :attr:`spec`. The public entry
        point for that, so callers need not touch ``_spec_hint`` / ``_compute_spec``.
        """
        self._spec_hint = spec_hint
        self.spec = self._compute_spec()  # type: ignore[assignment]

    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()
        self.spec = self._compute_spec()  # type: ignore[assignment]

    @property
    @ensure_nonempty
    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
    @ensure_nonempty
    def samples(self) -> int:
        """The number of samples in the Frame.

        Return:
            int, the number of samples
        """
        return sum(buf.samples for buf in self.buffers)

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

    @property
    @ensure_nonempty
    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)])

    @ensure_nonempty
    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

    @ensure_nonempty
    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),
        )

    @property
    @ensure_nonempty
    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 numpy.arange(self.samples) / self.sample_rate + self.start

    @ensure_nonempty
    def filleddata(self) -> Array:
        """Combined frame data, with gaps materialized as zeros.

        Gap zeros are created in the **frame's** backend (``self.spec``), so even
        an all-gap frame materializes in the correct backend / dtype / device —
        the frame is the authority on "what would be here". This is the only
        sanctioned way to materialize gaps; bare buffers cannot (they have no
        backend of their own).

        Returns:
            Array, the filled data in ``self.spec``'s backend.
        """
        spec = cast(SeriesDataSpec, self.spec)
        parts = [b.data if not b.is_gap else spec.zeros(b.shape) for b in self.buffers]
        return spec.namespace.concat(parts, axis=-1)

    @ensure_nonempty
    def search(self, buf):
        out = []
        for b in self:
            intersects = b & buf
            if intersects is not None and intersects.isfinite():
                out.append(intersects)
        return out

    @ensure_nonempty
    def align(self, tsslices) -> "TSFrame":
        "Align buffers according to the TSSlices provided"
        assert (
            self.slice == tsslices.slice
        ), "The boundaries provided are not aligned with the frame boundaries"
        bufs = []
        for aligned_buf in [self[0].new(offslice) for offslice in tsslices]:
            searched_bufs = self.search(aligned_buf)
            # promote any gaps
            if any([b.is_gap for b in searched_bufs]):
                bufs.append(aligned_buf)
                continue
            # otherwise add the data from the found sub buffers
            for sb in searched_bufs:
                aligned_buf = aligned_buf + sb
            bufs.append(aligned_buf)
        return TSFrame(buffers=bufs)

    @ensure_nonempty
    def gap_mask(self) -> Array:
        """Mark which sections of a frame's data contained gap buffers,

        Returns a boolean array, the same shape as the frame's total data,
        with 1's where there were data values, and 0's where there were gaps
        """
        bool_stream: list[numpy.ndarray] = []
        for buf in self.buffers:
            if buf.is_gap:
                buf_bools = numpy.ones(buf.shape, dtype=bool)
            else:
                buf_bools = numpy.zeros_like(buf.data, dtype=bool)
            bool_stream.append(buf_bools)
        return numpy.concatenate(bool_stream, dtype=bool, axis=None)

    @ensure_nonempty
    def maskeddata(self) -> numpy.ma.MaskedArray:
        """Return a masked array of the frame's data

        Like TSFrame.filleddata() but returns a MaskedArray where the
        mask is given by TSFrame.gap_mask().

        """
        if any([buf.is_gap for buf in self.buffers]):
            mask = self.gap_mask()
        else:
            mask = False
        return numpy.ma.masked_array(
            self.filleddata(),
            mask=mask,
        )

    @classmethod
    def frame_from_ma(cls, masked_data, offset, sample_rate) -> TSFrame:
        """Return a TSFrame, based on a masked array
        (with masked values marking gaps), with the same
        offset and end_offset as the passed TSFrame"""

        masked = [(gap_slice, True) for gap_slice in numpy.ma.clump_masked(masked_data)]
        unmasked = [
            (nongap_slice, False)
            for nongap_slice in numpy.ma.clump_unmasked(masked_data)
        ]

        buffers = []
        for slice_, is_gap in sorted(itertools.chain(masked, unmasked)):
            arr_data = masked_data.data[slice_]
            buf = SeriesBuffer(
                offset=offset,
                sample_rate=sample_rate,
                data=None if is_gap else arr_data,
                shape=(len(arr_data),),
            )
            buffers.append(buf)
            offset = buf.end_offset

        return cls(buffers=buffers)

    def plot(
        self,
        ax=None,
        label: Optional[str] = None,
        channel: Optional[Union[int, tuple]] = None,
        gap_color: str = "red",
        gap_alpha: float = 0.3,
        show_gaps: bool = True,
        time_unit: Literal["s", "ms", "ns", "gps"] = "gps",
        multichannel: Literal["overlay", "subplots"] = "overlay",
        **kwargs,
    ):
        """Plot the frame's time-series data.

        Requires matplotlib. Install with: pip install sgn-ts[plot]

        Args:
            ax:
                matplotlib Axes, optional. If None, creates a new figure and axes.
                Ignored if multichannel='subplots' and channel is None.
            label:
                str, optional. Legend label for this frame's data line.
            channel:
                int or tuple[int, ...], optional. For multi-dimensional data,
                specifies which channel(s) to plot. If None and data is
                multi-dimensional, behavior depends on multichannel parameter.
            gap_color:
                str, color for gap region shading. Default 'red'.
            gap_alpha:
                float, alpha transparency for gap region shading. Default 0.3.
            show_gaps:
                bool, whether to show gap indicators. Default True.
            time_unit:
                str, time unit for x-axis: 's' (seconds since start), 'ms',
                'ns', or 'gps' (absolute GPS time). Default 'gps'.
            multichannel:
                str, how to handle multi-channel data when channel is None:
                'overlay' plots all channels on the same axes,
                'subplots' creates a subplot for each channel. Default 'overlay'.
            **kwargs:
                Additional keyword arguments passed to ax.plot().

        Returns:
            tuple: (fig, ax) matplotlib figure and axes objects. If
                   multichannel='subplots', ax is an array of axes.
        """
        from sgnts.plotting import plot_frame

        return plot_frame(
            self,
            ax=ax,
            label=label,
            channel=channel,
            gap_color=gap_color,
            gap_alpha=gap_alpha,
            show_gaps=show_gaps,
            time_unit=time_unit,
            multichannel=multichannel,
            **kwargs,
        )

sample_rate property

The sample rate of the TSFrame.

Returns:

Type	Description
int	int, the sample rate

sample_shape property

return the sample shape

samples property

The number of samples in the Frame.

Return

int, the number of samples

shape property

The shape of the TSFrame.

Returns:

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

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

__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 src/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
    )

align(tsslices)

Align buffers according to the TSSlices provided

Source code in src/sgnts/base/buffer.py

@ensure_nonempty
def align(self, tsslices) -> "TSFrame":
    "Align buffers according to the TSSlices provided"
    assert (
        self.slice == tsslices.slice
    ), "The boundaries provided are not aligned with the frame boundaries"
    bufs = []
    for aligned_buf in [self[0].new(offslice) for offslice in tsslices]:
        searched_bufs = self.search(aligned_buf)
        # promote any gaps
        if any([b.is_gap for b in searched_bufs]):
            bufs.append(aligned_buf)
            continue
        # otherwise add the data from the found sub buffers
        for sb in searched_bufs:
            aligned_buf = aligned_buf + sb
        bufs.append(aligned_buf)
    return TSFrame(buffers=bufs)

apply_spec_hint(spec_hint)

Adopt spec_hint as this frame's backend identity, recomputing spec.

An all-gap frame has no data to read its backend from; an element that builds such a frame by hand (a heartbeat or all-gap stretch) passes the upstream spec as a hint, and this stamps it and re-derives :attr:spec. The public entry point for that, so callers need not touch _spec_hint / _compute_spec.

Source code in src/sgnts/base/buffer.py

def apply_spec_hint(self, spec_hint: SeriesDataSpec | None) -> None:
    """Adopt ``spec_hint`` as this frame's backend identity, recomputing ``spec``.

    An all-gap frame has no data to read its backend from; an element that builds
    such a frame by hand (a heartbeat or all-gap stretch) passes the upstream spec
    as a hint, and this stamps it and re-derives :attr:`spec`. The public entry
    point for that, so callers need not touch ``_spec_hint`` / ``_compute_spec``.
    """
    self._spec_hint = spec_hint
    self.spec = self._compute_spec()  # type: ignore[assignment]

fill()

Create a TSCollectFrame for atomically populating this frame.

Returns a TSCollectFrame that can be used to incrementally add buffers with validation. The buffers are only committed to this frame when close() is called (or automatically via context manager).

Returns:

Name	Type	Description
TSCollectFrame	TSCollectFrame	A collector for building this frame

Usage

Context manager (recommended - automatic close)

frame = TSFrame(offset=0, noffset=1000) with frame.fill() as collector: collector.append(buf1) collector.append(buf2)

frame now has buffers

Manual (explicit control)

frame = TSFrame(offset=0, noffset=1000) collector = frame.fill() collector.append(buf1) collector.close() # commits to frame

Source code in src/sgnts/base/buffer.py

def fill(self) -> TSCollectFrame:
    """Create a TSCollectFrame for atomically populating this frame.

    Returns a TSCollectFrame that can be used to incrementally add buffers
    with validation. The buffers are only committed to this frame when
    close() is called (or automatically via context manager).

    Returns:
        TSCollectFrame: A collector for building this frame

    Usage:
        # Context manager (recommended - automatic close)
        frame = TSFrame(offset=0, noffset=1000)
        with frame.fill() as collector:
            collector.append(buf1)
            collector.append(buf2)
        # frame now has buffers

        # Manual (explicit control)
        frame = TSFrame(offset=0, noffset=1000)
        collector = frame.fill()
        collector.append(buf1)
        collector.close()  # commits to frame
    """
    return TSCollectFrame(parent_frame=self)

filleddata()

Combined frame data, with gaps materialized as zeros.

Gap zeros are created in the frame's backend (self.spec), so even an all-gap frame materializes in the correct backend / dtype / device — the frame is the authority on "what would be here". This is the only sanctioned way to materialize gaps; bare buffers cannot (they have no backend of their own).

Returns:

Type	Description
Array	Array, the filled data in self.spec's backend.

Source code in src/sgnts/base/buffer.py

@ensure_nonempty
def filleddata(self) -> Array:
    """Combined frame data, with gaps materialized as zeros.

    Gap zeros are created in the **frame's** backend (``self.spec``), so even
    an all-gap frame materializes in the correct backend / dtype / device —
    the frame is the authority on "what would be here". This is the only
    sanctioned way to materialize gaps; bare buffers cannot (they have no
    backend of their own).

    Returns:
        Array, the filled data in ``self.spec``'s backend.
    """
    spec = cast(SeriesDataSpec, self.spec)
    parts = [b.data if not b.is_gap else spec.zeros(b.shape) for b in self.buffers]
    return spec.namespace.concat(parts, axis=-1)

frame_from_ma(masked_data, offset, sample_rate) classmethod

Return a TSFrame, based on a masked array (with masked values marking gaps), with the same offset and end_offset as the passed TSFrame

Source code in src/sgnts/base/buffer.py

@classmethod
def frame_from_ma(cls, masked_data, offset, sample_rate) -> TSFrame:
    """Return a TSFrame, based on a masked array
    (with masked values marking gaps), with the same
    offset and end_offset as the passed TSFrame"""

    masked = [(gap_slice, True) for gap_slice in numpy.ma.clump_masked(masked_data)]
    unmasked = [
        (nongap_slice, False)
        for nongap_slice in numpy.ma.clump_unmasked(masked_data)
    ]

    buffers = []
    for slice_, is_gap in sorted(itertools.chain(masked, unmasked)):
        arr_data = masked_data.data[slice_]
        buf = SeriesBuffer(
            offset=offset,
            sample_rate=sample_rate,
            data=None if is_gap else arr_data,
            shape=(len(arr_data),),
        )
        buffers.append(buf)
        offset = buf.end_offset

    return cls(buffers=buffers)

from_buffer_kwargs(**kwargs) classmethod

A short hand for the following:

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

Source code in src/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)])

gap_mask()

Mark which sections of a frame's data contained gap buffers,

Returns a boolean array, the same shape as the frame's total data, with 1's where there were data values, and 0's where there were gaps

Source code in src/sgnts/base/buffer.py

@ensure_nonempty
def gap_mask(self) -> Array:
    """Mark which sections of a frame's data contained gap buffers,

    Returns a boolean array, the same shape as the frame's total data,
    with 1's where there were data values, and 0's where there were gaps
    """
    bool_stream: list[numpy.ndarray] = []
    for buf in self.buffers:
        if buf.is_gap:
            buf_bools = numpy.ones(buf.shape, dtype=bool)
        else:
            buf_bools = numpy.zeros_like(buf.data, dtype=bool)
        bool_stream.append(buf_bools)
    return numpy.concatenate(bool_stream, dtype=bool, axis=None)

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 src/sgnts/base/buffer.py

@ensure_nonempty
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),
    )

maskeddata()

Return a masked array of the frame's data

Like TSFrame.filleddata() but returns a MaskedArray where the mask is given by TSFrame.gap_mask().

Source code in src/sgnts/base/buffer.py

@ensure_nonempty
def maskeddata(self) -> numpy.ma.MaskedArray:
    """Return a masked array of the frame's data

    Like TSFrame.filleddata() but returns a MaskedArray where the
    mask is given by TSFrame.gap_mask().

    """
    if any([buf.is_gap for buf in self.buffers]):
        mask = self.gap_mask()
    else:
        mask = False
    return numpy.ma.masked_array(
        self.filleddata(),
        mask=mask,
    )

plot(ax=None, label=None, channel=None, gap_color='red', gap_alpha=0.3, show_gaps=True, time_unit='gps', multichannel='overlay', **kwargs)

Plot the frame's time-series data.

Requires matplotlib. Install with: pip install sgn-ts[plot]

Parameters:

Name	Type	Description	Default
ax		matplotlib Axes, optional. If None, creates a new figure and axes. Ignored if multichannel='subplots' and channel is None.	None
label	Optional[str]	str, optional. Legend label for this frame's data line.	None
channel	Optional[Union[int, tuple]]	int or tuple[int, ...], optional. For multi-dimensional data, specifies which channel(s) to plot. If None and data is multi-dimensional, behavior depends on multichannel parameter.	None
gap_color	str	str, color for gap region shading. Default 'red'.	'red'
gap_alpha	float	float, alpha transparency for gap region shading. Default 0.3.	0.3
show_gaps	bool	bool, whether to show gap indicators. Default True.	True
time_unit	Literal['s', 'ms', 'ns', 'gps']	str, time unit for x-axis: 's' (seconds since start), 'ms', 'ns', or 'gps' (absolute GPS time). Default 'gps'.	'gps'
multichannel	Literal['overlay', 'subplots']	str, how to handle multi-channel data when channel is None: 'overlay' plots all channels on the same axes, 'subplots' creates a subplot for each channel. Default 'overlay'.	'overlay'
**kwargs		Additional keyword arguments passed to ax.plot().	{}

Returns:

Name	Type	Description
tuple		(fig, ax) matplotlib figure and axes objects. If multichannel='subplots', ax is an array of axes.

Source code in src/sgnts/base/buffer.py

def plot(
    self,
    ax=None,
    label: Optional[str] = None,
    channel: Optional[Union[int, tuple]] = None,
    gap_color: str = "red",
    gap_alpha: float = 0.3,
    show_gaps: bool = True,
    time_unit: Literal["s", "ms", "ns", "gps"] = "gps",
    multichannel: Literal["overlay", "subplots"] = "overlay",
    **kwargs,
):
    """Plot the frame's time-series data.

    Requires matplotlib. Install with: pip install sgn-ts[plot]

    Args:
        ax:
            matplotlib Axes, optional. If None, creates a new figure and axes.
            Ignored if multichannel='subplots' and channel is None.
        label:
            str, optional. Legend label for this frame's data line.
        channel:
            int or tuple[int, ...], optional. For multi-dimensional data,
            specifies which channel(s) to plot. If None and data is
            multi-dimensional, behavior depends on multichannel parameter.
        gap_color:
            str, color for gap region shading. Default 'red'.
        gap_alpha:
            float, alpha transparency for gap region shading. Default 0.3.
        show_gaps:
            bool, whether to show gap indicators. Default True.
        time_unit:
            str, time unit for x-axis: 's' (seconds since start), 'ms',
            'ns', or 'gps' (absolute GPS time). Default 'gps'.
        multichannel:
            str, how to handle multi-channel data when channel is None:
            'overlay' plots all channels on the same axes,
            'subplots' creates a subplot for each channel. Default 'overlay'.
        **kwargs:
            Additional keyword arguments passed to ax.plot().

    Returns:
        tuple: (fig, ax) matplotlib figure and axes objects. If
               multichannel='subplots', ax is an array of axes.
    """
    from sgnts.plotting import plot_frame

    return plot_frame(
        self,
        ax=ax,
        label=label,
        channel=channel,
        gap_color=gap_color,
        gap_alpha=gap_alpha,
        show_gaps=show_gaps,
        time_unit=time_unit,
        multichannel=multichannel,
        **kwargs,
    )

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 src/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()
    self.spec = self._compute_spec()  # type: ignore[assignment]

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 src/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.

Source code in src/sgnts/base/buffer.py

def validate_buffers(self) -> None:
    """Sanity check that the buffers don't overlap nor have discontinuities."""
    # 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

    # All non-gap buffers must share one backend (the array is the backend);
    # gaps carry no array and are exempt. (We do not assert dtype uniformity:
    # the previous spec check compared a constant global dtype and so never
    # fired, and real frames may legitimately mix dtypes.)
    backends = {backend_name(buf.data) for buf in self.buffers if not buf.is_gap}
    assert (
        len(backends) <= 1
    ), f"All buffers must have the same backend, got {backends}"

validate_span()

Validate that buffers fully span the offset/noffset range.

Checks that: - First buffer starts at frame offset - Last buffer ends at frame offset+noffset (the frame's end_offset) - All buffers are contiguous

Source code in src/sgnts/base/buffer.py

def validate_span(self) -> None:
    """Validate that buffers fully span the offset/noffset range.

    Checks that:
    - First buffer starts at frame offset
    - Last buffer ends at frame offset+noffset (the frame's end_offset)
    - All buffers are contiguous
    """
    if self.buffers:
        frame_end_offset = self.offset + self.noffset

        assert self.buffers[0].offset == self.offset, (
            f"First buffer offset {self.buffers[0].offset:_} != "
            f"frame offset {self.offset:_}"
        )
        assert self.buffers[-1].end_offset == frame_end_offset, (
            f"Last buffer end_offset {self.buffers[-1].end_offset:_} != "
            f"frame end_offset {frame_end_offset:_}"
        )
        # validate_buffers checks contiguity
        self.validate_buffers()

TimeLike

Bases: Protocol

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

              

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

Source code in src/sgnts/base/buffer.py

@runtime_checkable
class TimeLike(Protocol):
    offset: int

    @property
    def time(self) -> float:
        """The reference time, in seconds.

        Returns:
            float, buffer time in seconds
        """
        return Offset.offset_ref_start / Time.SECONDS + Offset.tosec(self.offset)

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

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

    @property
    def start(self) -> float:
        """The start (reference) time, in seconds.

        Returns:
            float, buffer start time in seconds
        """
        return self.time

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

        Returns:
            int, buffer start time in nanoseconds
        """
        return self.time_ns

    @property
    def t0(self) -> float:
        """The start (reference) time, in seconds.

        Returns:
            float, buffer start time in seconds
        """
        return self.start

start property

The start (reference) time, in seconds.

Returns:

Type	Description
float	float, buffer start time in seconds

start_ns property

The start (reference) time, in integer nanoseconds.

Returns:

Type	Description
int	int, buffer start time in nanoseconds

t0 property

The start (reference) time, in seconds.

Returns:

Type	Description
float	float, buffer start time in seconds

time property

The reference time, in seconds.

Returns:

Type	Description
float	float, buffer time in seconds

time_ns property

The reference time, in integer nanoseconds.

Returns:

Type	Description
int	int, buffer time in nanoseconds

TimeSpanFrame

Bases: Frame, TimeSpanLike

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

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



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

Base class for frames with time span semantics.

TimeSpanFrame combines Frame's data-carrying capabilities with TimeSpanLike's temporal semantics (start/end offsets).

All TimeSpanFrame subclasses must be iterable.

Source code in src/sgnts/base/buffer.py

class TimeSpanFrame(Frame, TimeSpanLike):
    """Base class for frames with time span semantics.

    TimeSpanFrame combines Frame's data-carrying capabilities with
    TimeSpanLike's temporal semantics (start/end offsets).

    All TimeSpanFrame subclasses must be iterable.
    """

    @abstractmethod
    def __iter__(self):
        """Iterate over the frame's data elements."""
        ...

__iter__() abstractmethod

Iterate over the frame's data elements.

Source code in src/sgnts/base/buffer.py

@abstractmethod
def __iter__(self):
    """Iterate over the frame's data elements."""
    ...

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 src/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 offset_end(self) -> int:
        """The end offset.

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

    # FIXME: DEPRECATE ME
    @property
    def end_offset(self) -> int:
        """DEPRECATED: see `offset_end`"""
        return self.offset_end

    @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) -> float:
        """The duration of the buffer, in seconds.

        Returns:
            float, the buffer duration in seconds
        """
        return Offset.tosec(self.noffset)

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

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

    @property
    def end(self) -> float:
        """The end time of the buffer, in seconds.

        Returns:
            float, buffer end time in seconds
        """
        return self.start + self.duration

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

        Returns:
            int, buffer end time in nanoseconds
        """
        return self.start_ns + self.duration_ns

duration property

The duration of the buffer, in seconds.

Returns:

Type	Description
float	float, the buffer duration in seconds

duration_ns property

The duration of the buffer, in integer nanoseconds.

Returns:

Type	Description
int	int, the buffer duration in nanoseconds

end property

The end time of the buffer, in seconds.

Returns:

Type	Description
float	float, buffer end time in seconds

end_ns property

The end time of the buffer, in integer nanoseconds.

Returns:

Type	Description
int	int, buffer end time in nanoseconds

end_offset property

DEPRECATED: see offset_end

offset_end 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

ensure_nonempty(func)

Decorator to ensure TSFrame has buffers before accessing properties/methods.

Raises ValueError with a helpful message if the frame is empty.

Source code in src/sgnts/base/buffer.py

def ensure_nonempty(func):
    """Decorator to ensure TSFrame has buffers before accessing properties/methods.

    Raises ValueError with a helpful message if the frame is empty.
    """

    @wraps(func)
    def wrapper(self, *args, **kwargs):
        if len(self.buffers) == 0:
            raise ValueError(
                f"TSFrame.{func.__name__} cannot be used when there are no buffers "
                f"in the frame. Use TSFrame.fill() to populate the frame."
            )
        return func(self, *args, **kwargs)

    return wrapper