MS3TraceSeg

class pymseed.mstracelist.MS3TraceSeg(cffi_ptr, parent_traceid, parent_tracelist)

Bases: object

A continuous span of samples for a single trace ID

Segments of an MS3TraceID, one per run of contiguous data; a time gap or overlap, or a change of sample rate, starts a new segment. Data samples are available as datasamples when the trace list was read with unpack_data=True, or after unpack_recordlist().

Invalidated when the owning MS3TraceList is closed; using it afterward raises ValueError instead of reading freed memory.

Parameters:

• cffi_ptr (Any)

• parent_traceid (Any)

• parent_tracelist (Any)

property starttime: int

Return start time as nanoseconds since Unix/POSIX epoch

property starttime_seconds: float

Return start time as seconds since Unix/POSIX epoch

starttime_str(timeformat=TimeFormat.ISOMONTHDAY_Z, subsecond=SubSecond.NANO_MICRO_NONE)

Return start time as formatted string

Returns the sentinel strings "ERROR" or "UNSET" when the underlying nanosecond timestamp is the corresponding libmseed sentinel, mirroring starttime_str().

Parameters:

• timeformat (TimeFormat)

• subsecond (SubSecond)

Return type:

str

property endtime: int

Return end time as nanoseconds since Unix/POSIX epoch

property endtime_seconds: float

Return end time as seconds since Unix/POSIX epoch

endtime_str(timeformat=TimeFormat.ISOMONTHDAY_Z, subsecond=SubSecond.NANO_MICRO_NONE)

Return end time as formatted string

Returns the sentinel strings "ERROR" or "UNSET" when the underlying nanosecond timestamp is the corresponding libmseed sentinel, mirroring endtime_str().

Parameters:

• timeformat (TimeFormat)

• subsecond (SubSecond)

Return type:

str

property samprate: float

Return sample rate in samples/second (Hz)

property samplecnt: int

Return sample count

property update_time: int | None

Return time of last update as nanoseconds since Unix/POSIX epoch, or None if no update time is recorded for this segment.

libmseed records this for segments added through MS3TraceList, and generate() compares it against the system clock to decide which segments flush_idle_seconds flushes. Compare with pymseed.system_time() to measure how long a segment has been idle, e.g. in a rolling buffer.

property update_time_seconds: float | None

Return time of last update as seconds since Unix/POSIX epoch, or None if no update time is recorded for this segment.

See update_time.

property recordlist: MS3RecordList | None

Return the record list structure

property datasamples: memoryview

Return data samples as a memoryview (no copy)

A view of the data samples in a buffer owned by the trace list is returned. The view holds the trace list, so it cannot be freed while the view exists, but the samples are only valid until the trace list is next changed: adding data can move a segment’s buffer, and packing with remove_packed=True releases it. Copy the samples to keep them across such calls, or use take_np_datasamples() to detach the buffer itself.

The returned view can be used directly with slicing and indexing from 0 to MS3TraceSeg.numsamples - 1.

The view can efficiently be copied to a Python list using:

data_samples = MS3TraceSeg.datasamples[:]

property sampletype: str | None

Return sample type code if available, otherwise None

property numsamples: int

Return number of samples

property datasize: int

Return data size in bytes

property sample_size_type: tuple[int, str]

Return data sample size and type code from first record in list

NOTE: This is a guesstimate based on the first record in the record list. It is not guaranteed to be correct for any other records in the list.

property np_datasamples: Any

Return data samples as a numpy array (no copy)

A view of the data samples in a buffer owned by the trace list is returned, with the same lifetime as datasamples: the trace list is held by the view, but the samples are only valid until the trace list is next changed. See take_np_datasamples() for a numpy array that outlives the trace list instead.

take_np_datasamples()

Return data samples as a numpy array, taking ownership of the buffer (no copy)

Unlike np_datasamples, the returned array has no dependency on the trace list: the segment’s data buffer is detached and handed to the array’s memoryview, which is freed once the array is garbage collected. Afterwards numsamples is 0 while samplecnt is unchanged, so a second call returns an empty array rather than the same buffer again, and unpack_recordlist() can still decode fresh samples if a record list is available.

Any view taken earlier from datasamples or np_datasamples still addresses this buffer, so it stays valid only as long as the array returned here does; dropping the array while such a view is still around leaves that view referring to freed memory.

Return type:

Any

create_numpy_array_from_recordlist()

Return data samples as a numpy array unpacked from the record list

The numpy array returned is an independent copy of the data samples.

This makes a second pass over the source, re-reading each record; for a file, reading with unpack_data=True and calling take_np_datasamples() is usually faster. The record list is still preferable to decode only some segments, or into a caller’s own buffer.

Return type:

Any

unpack_recordlist(buffer=None, *, verbose=0)

Unpack data samples from miniSEED record list into accessible format

This method decodes data samples from the original miniSEED records that were stored when reading with record_list=True. It’s used for memory-efficient workflows where you delay data unpacking until needed.

Decoding here makes a second pass over the source, re-reading each record; for a file, reading with unpack_data=True and taking the samples with take_np_datasamples() is usually faster when every segment is wanted.

Parameters:

• buffer (Any) – Optional destination buffer for unpacked data. Must be writable, C-contiguous and support the buffer protocol (e.g., numpy array, bytearray, writable memoryview). If provided, must be large enough to hold self.samplecnt samples of the appropriate data type. If None, data is unpacked into internal memory owned by this segment instance.

• verbose (int) – Verbosity level for diagnostic output (0=quiet, 1-3=increasing detail). Default: 0

Returns:

Number of samples successfully unpacked

Raises:

• ValueError – If no record list is available (requires record_list=True when reading), if data is already unpacked and a buffer is provided, or if the provided buffer doesn’t support the buffer protocol

• BufferError – If the provided buffer is read-only or not C-contiguous, and so cannot be written to

• MiniSEEDError – If unpacking fails due to corrupted or invalid record data

Return type:

int

Note

• Requires the segment to have been created with record_list=True

• Can only be called once per segment if using internal memory (buffer=None)

• If using a provided buffer, the buffer format must match the segment’s sample type (int32 for “i”, float32 for “f”, float64 for “d”, bytes for “t”)

• For performance, use memoryviews with matching dtype when providing buffers

Examples

Basic workflow illustrating how to use unpack_recordlist():

>>> from pymseed import MS3TraceList
>>> traces = MS3TraceList.from_file("examples/example_data.mseed", record_list=True)
>>> len(traces)
3
>>> # Before unpacking, the data samples are not available
>>> for traceid in traces:
...     for segment in traceid:
...         assert segment.datasamples == memoryview(b'')
...         assert segment.numsamples == 0
>>> # After unpacking, the data samples are available
>>> for traceid in traces:
...     for segment in traceid:
...         count = segment.unpack_recordlist()
...         assert segment.numsamples == segment.samplecnt
...         assert len(segment.datasamples) == segment.numsamples

Advanced example of unpacking data to a numpy array. This is for illustration only; if numpy arrays are desired, use create_numpy_array_from_recordlist() or np_datasamples() instead.

>>> # For doctest conditional skipping, unneeded for real code
>>> try:
...     import numpy as np
...     HAS_NUMPY = True
... except ImportError:
...     HAS_NUMPY = False
>>> if HAS_NUMPY:
...     traces = MS3TraceList.from_file("examples/example_data.mseed", record_list=True)
...     for traceid in traces:
...         for segment in traceid:
...             # Get the sample size and type from the first record in the record list
...             (size, sample_type) = segment.sample_size_type
...             if sample_type == "i":
...                 # Create a numpy array to hold the unpacked data
...                 numpy_array = np.zeros(segment.samplecnt, dtype=np.int32)
...                 # Unpack the data directly into our array
...                 count = segment.unpack_recordlist(numpy_array)
...                 # Check that the array is not all zeros
...                 assert not np.all(numpy_array == 0), "Numpy array is all zeros (should not happen)"
...             # Other sample types would need different numpy array types

Advanced example of unpacking data to an Apache Arrow array using pyarrow, illustrating the same caller-provided-buffer pattern:

>>> # For doctest conditional skipping, unneeded for real code
>>> try:
...     import pyarrow as pa
...     HAS_PYARROW = True
... except ImportError:
...     HAS_PYARROW = False
>>> if HAS_PYARROW:
...     traces = MS3TraceList.from_file("examples/example_data.mseed", record_list=True)
...     for traceid in traces:
...         for segment in traceid:
...             # Get the sample size and type from the first record in the record list
...             (size, sample_type) = segment.sample_size_type
...             if sample_type == "i":
...                 # Create an arrow array to hold the unpacked data
...                 arrow_array = pa.array([0] * segment.samplecnt, type=pa.int32())
...                 # Get the data buffer for direct writing. Buffers that
...                 # arrow allocated itself are mutable; an imported one
...                 # (e.g. pa.py_buffer(bytes)) is not and is rejected.
...                 array_bitmap, array_buffer = arrow_array.buffers()
...                 # Unpack the data directly into our array buffer
...                 count = segment.unpack_recordlist(array_buffer)
...                 # Check that the array has no nulls (all values are valid)
...                 assert arrow_array.null_count == 0, "Arrow array has nulls (should not happen)"
...             # Other sample types would need different pyarrow array types

The point of the advanced examples is to illustrate how to use the pattern to unpack data into a buffer provided by the caller in order to avoid copying the data.

has_same_data(other)

Compare trace segments for equivalent data

Parameters:

other (object) – Another MS3TraceSeg to compare with

Returns:

True if segments have equivalent data, False otherwise

Return type:

bool