Skip to content

Buffers and Frames

TSFrame

A TSFrame is the top-level data container in SGN-TS — analogous to SGN's Frame, but constrained to hold uniformly sampled time-series data. It wraps one or more contiguous SeriesBuffer objects:

import numpy as np
from sgnts.base import SeriesBuffer, TSFrame

buf1 = SeriesBuffer(offset=0, sample_rate=2048, data=np.ones(2048))
buf2 = SeriesBuffer(offset=16384, sample_rate=2048, data=np.ones(2048))

frame = TSFrame(buffers=[buf1, buf2])

Buffers must be contiguous — buf2.offset must equal buf1.offset_end. Non-contiguous buffers raise an AssertionError.

Frame Properties

Property Description
offset Offset of the first buffer
end_offset End offset of the last buffer
sample_rate Sample rate (same for all buffers)
EOS End-of-stream flag
is_gap True if all buffers are gaps
buffers The list of SeriesBuffer objects

Iterating

for buf in frame:
    print(buf.offset, buf.shape)

# Index into specific buffers
first = frame[0]

Factory Methods

Create an empty frame from parameters:

from sgnts.base import TSFrame

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

Generate sequential frames with next():

frame2 = next(frame)  # Contiguous frame at the next offset

SeriesBuffer

A SeriesBuffer holds a single chunk of uniformly sampled data with precise offset-based timing. This is the core data unit inside a TSFrame.

import numpy as np
from sgnts.base import SeriesBuffer

buf = SeriesBuffer(offset=0, sample_rate=2048, data=np.random.randn(2048))

Key properties:

Property Description
offset Start position in offset units (samples at max rate)
offset_end End position in offset units
sample_rate Samples per second (must be power of 2)
shape Full array shape including time dimension
data The numpy array (or None for gap buffers)
duration Duration in seconds
duration_ns Duration in nanoseconds
is_gap True when data is None

Multi-dimensional Data

The last dimension is always time. For multi-channel data, add leading dimensions:

import numpy as np
from sgnts.base import SeriesBuffer

# Stereo audio: 2 channels x 2048 samples
buf = SeriesBuffer(offset=0, sample_rate=2048, data=np.random.randn(2, 2048))
print(buf.shape)  # (2, 2048)

Gap Buffers

A gap buffer signals missing data. Set data=None and provide shape so downstream elements know the expected dimensions:

from sgnts.base import SeriesBuffer

gap = SeriesBuffer(offset=0, sample_rate=2048, shape=(2048,), data=None)
print(gap.is_gap)  # True

Buffer Operations

Sub-buffer — extract a portion by offset range:

import numpy as np
from sgnts.base import SeriesBuffer, TSSlice

buf = SeriesBuffer(offset=0, sample_rate=2048, data=np.zeros(2048))
sub = buf.sub_buffer(TSSlice(0, 8192))  # First half second

Split — split a buffer at offset boundaries:

import numpy as np
from sgnts.base import SeriesBuffer, TSSlice, TSSlices

buf = SeriesBuffer(offset=0, sample_rate=2048, data=np.zeros(2048))
pieces = buf.split(TSSlices([TSSlice(0, 8192), TSSlice(8192, 16384)]))

Pad — create a gap buffer that extends backward from the start of a buffer:

pad = buf.pad_buffer(target_offset)  # Gap buffer from target_offset to buf.offset

Create from existing — create an empty buffer with the same metadata using new(), or from an offset slice:

import numpy as np
from sgnts.base import SeriesBuffer, TSSlice

buf = SeriesBuffer(offset=0, sample_rate=2048, data=np.zeros(2048))
empty = buf.new()  # Same offset, rate, shape — data=None

gap = SeriesBuffer.fromoffsetslice(TSSlice(0, 16384), sample_rate=2048)

Copy with modified fields:

import numpy as np
from sgnts.base import SeriesBuffer

buf = SeriesBuffer(offset=0, sample_rate=2048, data=np.ones(2048))
scaled = buf.copy(data=buf.data * 2.0)

Operators

Buffers support several operators:

# Addition — pads to the longer shape, gaps treated as zeros
combined = buf1 + buf2

# Equality — compares offset, shape, sample rate, and data
buf1 == buf2

# Containment — check if an offset or buffer falls within another
16384 in buf          # True if offset is within the buffer's range
small_buf in big_buf  # True if small_buf's span is within big_buf

# Comparison — based on end offsets
buf1 < buf2

# Truth / length
bool(buf)  # True if data is not None
len(buf)   # Number of samples (0 for gap buffers)

EventBuffer and EventFrame

For event-based (non-uniform) data, use EventBuffer and EventFrame. Event buffers carry offset ranges but have no sample rate — they represent discrete events within a time span (triggers, detections, annotations).

Creating Event Buffers

from sgnts.base import EventBuffer

# From raw offsets — offset and noffset (duration in offset units)
eb = EventBuffer(offset=0, noffset=16384, data=[{"snr": 10.5}])

# From seconds (convenience factory)
eb = EventBuffer.from_span(start=0.0, end=1.0, data=[{"snr": 10.5}])

# From nanoseconds
eb = EventBuffer.from_span_ns(start=0, end=1_000_000_000, data=[{"snr": 10.5}])

An EventBuffer with no data (empty list) is considered a gap:

from sgnts.base import EventBuffer

gap = EventBuffer(offset=0, noffset=16384)
print(gap.is_gap)  # True

Creating Event Frames

An EventFrame holds a list of EventBuffer objects. When created with data, offset and duration are computed from the buffers automatically. Buffers must be contiguous — this is validated internally.

from sgnts.base import EventBuffer, EventFrame

eb1 = EventBuffer.from_span(0.0, 1.0, data=[{"snr": 8.0}])
eb2 = EventBuffer.from_span(1.0, 2.0, data=[{"snr": 12.0}, {"snr": 9.5}])

frame = EventFrame(data=[eb1, eb2])

For incremental construction, create an empty frame with explicit bounds and append buffers:

from sgnts.base import EventBuffer, EventFrame, Offset

eb1 = EventBuffer.from_span(0.0, 1.0, data=[{"snr": 8.0}])
eb2 = EventBuffer.from_span(1.0, 2.0, data=[{"snr": 12.0}, {"snr": 9.5}])

frame = EventFrame(offset=0, noffset=Offset.fromsec(2.0))
frame.append(eb1)
frame.append(eb2)

Iterating Events

Iterate over buffers, or use .events to get a flat list of all events across all buffers:

# Iterate over EventBuffers
for eb in frame:
    print(eb.offset, eb.data)

# Flat list of all events across all buffers
for event in frame.events:
    print(event)

Individual EventBuffer objects are also iterable:

for event in eb2:
    print(event)  # {"snr": 12.0}, then {"snr": 9.5}