"""
Core trace list implementation for pymseed
"""
from __future__ import annotations
import itertools
import os
import sys
from collections.abc import Iterator
from typing import Any
from .clib import (
buffer_pointer,
cdata_to_string,
clibmseed,
ffi,
owned_memoryview,
owning_memoryview,
)
from .definitions import DataEncoding, SubSecond, TimeFormat
from .exceptions import MiniSEEDError
from .logging import begin_operation
from .msrecord import MS3Record
from .selections import build_selections
from .util import (
SAMPLE_FORMATS,
check_encoding,
check_format_version,
check_path,
check_str,
encoding_sizetype,
format_nstime,
numpy_dtype,
parse_flags,
require_numpy,
sample_preview,
)
def _indent(text: str) -> str:
"""Indent each line of `text` by two spaces, for repr()/str() nesting."""
return "\n".join(" " + line for line in text.split("\n"))
def _summary_lines(seq: Any, render: Any) -> list[str]:
"""Render up to 5 items of `seq` with `render` (repr or str), otherwise the
first two, an "... N more" marker, and the last two."""
items = list(seq)
if len(items) <= 5:
return [_indent(render(item)) for item in items]
return [
_indent(render(items[0])),
_indent(render(items[1])),
f" ... {len(items) - 4} more",
_indent(render(items[-2])),
_indent(render(items[-1])),
]
def _linked_getitem(seq: Any, key: int | slice) -> Any:
"""Shared `__getitem__` for the linked-list-backed containers below:
MS3RecordList, MS3TraceID, and MS3TraceList."""
if isinstance(key, slice):
return list(seq)[key]
if not isinstance(key, int):
raise TypeError("indices must be integers or slices")
length = len(seq)
if key < 0:
key += length
if key < 0 or key >= length:
raise IndexError("list index out of range")
return next(itertools.islice(seq, key, key + 1))
[docs]
class MS3RecordPtr:
"""A record that contributed samples to a trace segment
Entries of an :class:`MS3RecordList`, present only when the trace list was
read with ``record_list=True``. Each locates its record in the source it
was read from, by :attr:`filename` and :attr:`fileoffset` for a file or by
an internal pointer for a buffer, and exposes the parsed header as
:attr:`record` without decoding the data samples.
Invalidated when the owning :class:`MS3TraceList` is closed; using it
afterward raises :class:`ValueError` instead of reading freed memory.
"""
def __init__(self, cffi_ptr: Any, parent_tracelist: Any) -> None:
self._ptr_raw = cffi_ptr
# The referenced structure is owned by the trace list; hold a reference
# so it cannot be freed while this wrapper is in use.
self._parent_tracelist = parent_tracelist
@property
def _ptr(self) -> Any:
self._parent_tracelist._check_open()
return self._ptr_raw
def __repr__(self) -> str:
return (
f"MS3RecordPtr(sourceid: {self.record.sourceid}\n"
f" filename: {cdata_to_string(self._ptr.filename)}\n"
f" fileoffset: {self._ptr.fileoffset}\n"
f" bufferptr: {self._ptr.bufferptr}\n"
f" reclen: {self._ptr.msr.reclen}\n"
f" starttime: {self.record.starttime_str(timeformat=TimeFormat.ISOMONTHDAY_Z)}\n"
f" endtime: {self.record.endtime_str(timeformat=TimeFormat.ISOMONTHDAY_Z)}\n"
")"
)
def __str__(self) -> str:
return (
f"{self.record.sourceid}, "
f"{cdata_to_string(self._ptr.filename)}, "
f"byte offset: {self._ptr.fileoffset}"
)
@property
def record(self) -> MS3Record:
"""Return a constructed MS3Record.
Borrowed from this entry: reading it after the trace list is closed
raises :class:`ValueError` (see ``_msr_for_borrower()``).
"""
if not hasattr(self, "_msrecord"):
# libmseed leaves msr->record unset unless the source bytes outlive the
# read, as they do for a buffer-sourced entry held by the trace list.
self._msrecord = MS3Record._borrow(self._ptr.msr, self)
return self._msrecord
def _msr_for_borrower(self) -> Any:
"""Return the record struct, for the MS3Record built by `record`."""
return self._ptr.msr
@property
def filename(self) -> str | None:
"""Return filename, or None if the record came from a buffer rather
than a file."""
return cdata_to_string(self._ptr.filename)
@property
def fileoffset(self) -> int:
"""Return file offset"""
return self._ptr.fileoffset
@property
def endtime(self) -> int:
"""Return end time"""
return self._ptr.endtime
@property
def dataoffset(self) -> int:
"""Return data offset"""
return self._ptr.dataoffset
[docs]
class MS3RecordList:
"""Wrapper around CFFI MS3RecordList structure
This class supports list-like access to the record pointers:
- len(record_list) returns the number of records
- record_list[i] returns the i-th record pointer
- record_list[start:end] returns a slice of record pointers
- for record_ptr in record_list: iterates over all record pointers
Invalidated when the owning :class:`MS3TraceList` is closed; using it
afterward raises :class:`ValueError` instead of reading freed memory.
"""
def __init__(self, cffi_ptr: Any, parent_tracelist: Any) -> None:
self._list_raw = cffi_ptr
# The referenced structure is owned by the trace list; hold a reference
# so it cannot be freed while this wrapper is in use.
self._parent_tracelist = parent_tracelist
@property
def _list(self) -> Any:
self._parent_tracelist._check_open()
return self._list_raw
def __repr__(self) -> str:
lines = "\n".join(_summary_lines(self, repr))
return f"MS3RecordList(recordcnt: {len(self)}\n{lines}\n)"
def __str__(self) -> str:
lines = "\n".join(_summary_lines(self, str))
return f"Record list with {len(self)} records\n{lines}"
[docs]
def __len__(self) -> int:
"""Return number of records"""
return self._list.recordcnt
[docs]
def __getitem__(self, key: int | slice) -> Any:
"""Enable indexing and slicing access to record pointers"""
return _linked_getitem(self, key)
[docs]
def __iter__(self) -> Iterator[MS3RecordPtr]:
"""Return iterator over record pointers"""
current_record = self._list.first
while current_record != ffi.NULL:
yield MS3RecordPtr(current_record, self._parent_tracelist)
current_record = current_record.next
[docs]
def records(self) -> Iterator[MS3RecordPtr]:
"""Alias for __iter__()"""
return iter(self)
@property
def recordcnt(self) -> int:
"""Return record count"""
return self._list.recordcnt
[docs]
class MS3TraceSeg:
"""A continuous span of samples for a single trace ID
Segments of an :class:`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 :attr:`datasamples` when the trace list was read
with ``unpack_data=True``, or after :meth:`unpack_recordlist`.
Invalidated when the owning :class:`MS3TraceList` is closed; using it
afterward raises :class:`ValueError` instead of reading freed memory.
"""
def __init__(self, cffi_ptr: Any, parent_traceid: Any, parent_tracelist: Any) -> None:
self._seg_raw = cffi_ptr
self._parent_traceid = parent_traceid # Reference to parent MS3TraceID
self._parent_tracelist = parent_tracelist # Reference to parent MS3TraceList
@property
def _seg(self) -> Any:
self._parent_tracelist._check_open()
return self._seg_raw
def __repr__(self) -> str:
preview = sample_preview(self.datasamples) if self.numsamples > 0 else "[]"
return (
f"MS3TraceSeg(start: {self.starttime_str(timeformat=TimeFormat.ISOMONTHDAY_DOY_Z)}\n"
f" end: {self.endtime_str(timeformat=TimeFormat.ISOMONTHDAY_DOY_Z)}\n"
f" samprate: {self.samprate}\n"
f" samplecnt: {self.samplecnt}\n"
f" datasamples: {preview}\n"
f" datasize: {self.datasize}\n"
f" numsamples: {self.numsamples}\n"
f" sampletype: {self.sampletype}\n"
f" recordlist: {'Record list of ' + str(len(self.recordlist)) + ' records' if self.recordlist else 'None'}\n"
")"
)
def __str__(self) -> str:
return (
f"start: {self.starttime_str(timeformat=TimeFormat.ISOMONTHDAY_DOY_Z)}, "
f"end: {self.endtime_str(timeformat=TimeFormat.ISOMONTHDAY_DOY_Z)}, "
f"samprate: {self.samprate}, "
f"samples: {self.samplecnt}"
)
@property
def starttime(self) -> int:
"""Return start time as nanoseconds since Unix/POSIX epoch"""
return self._seg.starttime
@property
def starttime_seconds(self) -> float:
"""Return start time as seconds since Unix/POSIX epoch"""
return self._seg.starttime / clibmseed.NSTMODULUS
[docs]
def starttime_str(
self,
timeformat: TimeFormat = TimeFormat.ISOMONTHDAY_Z,
subsecond: SubSecond = SubSecond.NANO_MICRO_NONE,
) -> str:
"""Return start time as formatted string
Returns the sentinel strings ``"ERROR"`` or ``"UNSET"`` when the
underlying nanosecond timestamp is the corresponding libmseed
sentinel, mirroring :meth:`~pymseed.MS3Record.starttime_str`.
"""
return format_nstime(self._seg.starttime, timeformat, subsecond)
@property
def endtime(self) -> int:
"""Return end time as nanoseconds since Unix/POSIX epoch"""
return self._seg.endtime
@property
def endtime_seconds(self) -> float:
"""Return end time as seconds since Unix/POSIX epoch"""
return self._seg.endtime / clibmseed.NSTMODULUS
[docs]
def endtime_str(
self,
timeformat: TimeFormat = TimeFormat.ISOMONTHDAY_Z,
subsecond: SubSecond = SubSecond.NANO_MICRO_NONE,
) -> str:
"""Return end time as formatted string
Returns the sentinel strings ``"ERROR"`` or ``"UNSET"`` when the
underlying nanosecond timestamp is the corresponding libmseed
sentinel, mirroring :meth:`~pymseed.MS3Record.endtime_str`.
"""
return format_nstime(self._seg.endtime, timeformat, subsecond)
@property
def samprate(self) -> float:
"""Return sample rate in samples/second (Hz)"""
return self._seg.samprate
@property
def samplecnt(self) -> int:
"""Return sample count"""
return self._seg.samplecnt
@property
def update_time(self) -> 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 :class:`MS3TraceList`,
and :meth:`~pymseed.MS3TraceList.generate` compares it against the system clock
to decide which segments ``flush_idle_seconds`` flushes. Compare with
:func:`pymseed.system_time` to measure how long a segment has been
idle, e.g. in a rolling buffer.
"""
if self._seg.prvtptr == ffi.NULL:
return None
return int(ffi.cast("nstime_t *", self._seg.prvtptr)[0])
@property
def update_time_seconds(self) -> 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 :attr:`update_time`.
"""
update_time = self.update_time
return None if update_time is None else update_time / clibmseed.NSTMODULUS
@property
def recordlist(self) -> MS3RecordList | None:
"""Return the record list structure"""
if not self._seg.recordlist:
return None
return MS3RecordList(self._seg.recordlist, self._parent_tracelist)
@property
def datasamples(self) -> 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 :meth:`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[:]
"""
if self._seg.numsamples <= 0:
return memoryview(b"") # Empty memoryview
sampletype = self.sampletype
if sampletype not in SAMPLE_FORMATS:
raise ValueError(f"Unknown sample type: {sampletype}")
fmt, itemsize = SAMPLE_FORMATS[sampletype]
ptr = ffi.cast("char *", self._seg.datasamples)
nbytes = self._seg.numsamples * itemsize
return owned_memoryview(ptr, nbytes, fmt, self)
@property
def sampletype(self) -> str | None:
"""Return sample type code if available, otherwise None"""
if self._seg.sampletype == b"\x00":
return None
return self._seg.sampletype.decode("ascii")
@property
def numsamples(self) -> int:
"""Return number of samples"""
return self._seg.numsamples
@property
def datasize(self) -> int:
"""Return data size in bytes"""
return self._seg.datasize
@property
def sample_size_type(self) -> 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.
"""
if not self._seg.recordlist:
raise ValueError("No record list available to determine sample size and type")
# Get the first record
first_record_ptr = self._seg.recordlist.first
if not first_record_ptr:
raise ValueError("No records in record list")
return encoding_sizetype(first_record_ptr.msr.encoding)
@property
def np_datasamples(self) -> 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 :attr:`datasamples`: the trace list
is held by the view, but the samples are only valid until the trace list
is next changed. See :meth:`take_np_datasamples` for a numpy array that
outlives the trace list instead.
"""
np = require_numpy()
sampletype = self.sampletype
if self._seg.numsamples <= 0:
# Use the known sample type's dtype where available, rather than
# numpy's float64 default, so a caller checking dtype isn't misled.
return np.array([], dtype=numpy_dtype(np, sampletype) if sampletype else None)
dtype = numpy_dtype(np, sampletype)
# Create numpy array view from CFFI buffer
return np.frombuffer(self.datasamples, dtype=dtype)
[docs]
def take_np_datasamples(self) -> Any:
"""Return data samples as a numpy array, taking ownership of the buffer (no copy)
Unlike :attr:`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
:meth:`unpack_recordlist` can still decode fresh samples if a record
list is available.
Any view taken earlier from :attr:`datasamples` or :attr:`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.
"""
np = require_numpy()
if self._seg.numsamples <= 0:
# Use the known sample type's dtype where available, rather than
# numpy's float64 default, so a caller checking dtype isn't misled.
sampletype = self.sampletype
return np.array([], dtype=numpy_dtype(np, sampletype) if sampletype else None)
numsamples = self._seg.numsamples
dtype = numpy_dtype(np, self.sampletype)
nbytes = numsamples * dtype.itemsize
# Windows preallocates buffer growth in blocks, so the segment's
# buffer can be larger than its samples; shrink it to size so the
# array doesn't retain the unused tail for its entire lifetime.
if self._seg.datasize > nbytes:
shrunk = clibmseed.libmseed_memory.realloc(self._seg.datasamples, nbytes)
if shrunk:
self._seg.datasamples = shrunk
ptr = ffi.cast("char *", self._seg.datasamples)
view = owning_memoryview(ptr, nbytes)
# Detach the buffer from the segment now that the array owns it
self._seg.datasamples = ffi.NULL
self._seg.datasize = 0
self._seg.numsamples = 0
return np.frombuffer(view, dtype=dtype)
[docs]
def create_numpy_array_from_recordlist(self) -> Any:
"""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
:meth:`take_np_datasamples` is usually faster. The record list is still
preferable to decode only some segments, or into a caller's own buffer.
"""
np = require_numpy()
if self.recordlist is None:
raise ValueError(
"Record list required, use record_list=True when populating MS3TraceList"
)
if self.samplecnt <= 0:
# Use the known sample type's dtype where available, rather than
# numpy's float64 default, so a caller checking dtype isn't misled.
try:
_, sample_type = self.sample_size_type
dtype = numpy_dtype(np, sample_type)
except ValueError:
dtype = None
return np.array([], dtype=dtype)
(_, sample_type) = self.sample_size_type
dtype = numpy_dtype(np, sample_type)
# Create numpy array of the correct type and size
array = np.empty(self.samplecnt, dtype=dtype)
# Unpack data samples into the array
self.unpack_recordlist(buffer=array)
return array
[docs]
def unpack_recordlist(self, buffer: Any = None, *, verbose: int = 0) -> int:
"""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
:meth:`take_np_datasamples` is usually faster when every segment is wanted.
Args:
buffer: 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: 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
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.
"""
if not self.recordlist:
raise ValueError("No record list available to unpack")
if self._seg.numsamples > 0 and buffer is not None:
raise ValueError("Data samples already unpacked")
buffer_ptr = ffi.NULL
buffer_size = 0
if buffer is not None:
# libmseed memcpys the decoded samples into this buffer.
buffer_ptr = buffer_pointer(
buffer, writable=True, context="Cannot unpack into the provided buffer"
)
buffer_size = len(buffer_ptr)
status = clibmseed.mstl3_unpack_recordlist(
self._parent_traceid._id,
self._seg,
buffer_ptr,
buffer_size,
verbose,
)
if status < 0:
raise MiniSEEDError(status, "Error unpacking record list")
return status
[docs]
def has_same_data(self, other: object) -> bool:
"""Compare trace segments for equivalent data
Args:
other: Another MS3TraceSeg to compare with
Returns:
True if segments have equivalent data, False otherwise
"""
if not isinstance(other, MS3TraceSeg):
return False
return (
self.sampletype == other.sampletype
and self.starttime == other.starttime
and self.endtime == other.endtime
and self.samprate == other.samprate
and self.samplecnt == other.samplecnt
and self.datasize == other.datasize
and self.numsamples == other.numsamples
and (self.numsamples == 0 or self.datasamples == other.datasamples)
)
[docs]
class MS3TraceID:
"""Wrapper around CFFI MS3TraceID structure
This class supports list-like access to the trace segments:
- len(traceid) returns the number of segments
- traceid[i] returns the i-th segment
- traceid[start:end] returns a slice of segments
- for segment in traceid: iterates over all segments
Invalidated when the owning :class:`MS3TraceList` is closed; using it
afterward raises :class:`ValueError` instead of reading freed memory.
"""
def __init__(self, cffi_ptr: Any, parent_tracelist: Any) -> None:
self._id_raw = cffi_ptr
self._parent_tracelist = parent_tracelist
@property
def _id(self) -> Any:
self._parent_tracelist._check_open()
return self._id_raw
def __repr__(self) -> str:
lines = "\n".join(_summary_lines(self, repr))
return (
f"MS3TraceID(sourceid: {self.sourceid}\n"
f" pubversion: {self.pubversion}\n"
f" earliest: {self.earliest_str(timeformat=TimeFormat.ISOMONTHDAY_DOY_Z)}\n"
f" latest: {self.latest_str(timeformat=TimeFormat.ISOMONTHDAY_DOY_Z)}\n"
f" numsegments: {len(self)}\n"
f"{lines}"
"\n)"
)
def __str__(self) -> str:
return (
f"{self.sourceid}, "
f"v{self.pubversion}, "
f"earliest: {self.earliest_str(timeformat=TimeFormat.ISOMONTHDAY_DOY_Z)}, "
f"latest: {self.latest_str(timeformat=TimeFormat.ISOMONTHDAY_DOY_Z)}, "
f"{len(self)} segments"
)
[docs]
def __len__(self) -> int:
"""Return number of segments"""
return self._id.numsegments
[docs]
def __iter__(self) -> Iterator[MS3TraceSeg]:
"""Return iterator over segments"""
current_segment = self._id.first
while current_segment != ffi.NULL:
yield MS3TraceSeg(current_segment, self, self._parent_tracelist)
current_segment = current_segment.next
[docs]
def __getitem__(self, key: int | slice) -> Any:
"""Enable indexing and slicing access to segments"""
return _linked_getitem(self, key)
@property
def sourceid(self) -> str:
"""Return source ID as string"""
return ffi.string(self._id.sid).decode("utf-8")
@property
def pubversion(self) -> int:
"""Return publication version"""
return self._id.pubversion
@property
def earliest(self) -> int:
"""Return earliest time as nanoseconds since Unix/POSIX epoch"""
return self._id.earliest
@property
def earliest_seconds(self) -> float:
"""Return earliest time as seconds since Unix/POSIX epoch"""
return self._id.earliest / clibmseed.NSTMODULUS
[docs]
def earliest_str(
self,
timeformat: TimeFormat = TimeFormat.ISOMONTHDAY_Z,
subsecond: SubSecond = SubSecond.NANO_MICRO_NONE,
) -> str:
"""Return earliest time as formatted string
Returns the sentinel strings ``"ERROR"`` or ``"UNSET"`` when the
underlying nanosecond timestamp is the corresponding libmseed
sentinel.
"""
return format_nstime(self._id.earliest, timeformat, subsecond)
@property
def latest(self) -> int:
"""Return latest time as nanoseconds since Unix/POSIX epoch"""
return self._id.latest
@property
def latest_seconds(self) -> float:
"""Return latest time as seconds since Unix/POSIX epoch"""
return self._id.latest / clibmseed.NSTMODULUS
[docs]
def latest_str(
self,
timeformat: TimeFormat = TimeFormat.ISOMONTHDAY_Z,
subsecond: SubSecond = SubSecond.NANO_MICRO_NONE,
) -> str:
"""Return latest time as formatted string
Returns the sentinel strings ``"ERROR"`` or ``"UNSET"`` when the
underlying nanosecond timestamp is the corresponding libmseed
sentinel.
"""
return format_nstime(self._id.latest, timeformat, subsecond)
[docs]
class MS3TraceList:
"""A container for a list of traces read from miniSEED
If ``file_name`` is specified miniSEED will be read from the file.
If ``unpack_data`` is True, the data samples will be decoded.
If ``sourceid``, ``starttime``, or ``endtime`` are specified, only records
matching those criteria will be included in the trace list. ``sourceid``
is a glob pattern matched against the record source ID (e.g.
``"FDSN:IU_COLA_*"``); set to ``None`` to match all source IDs.
``starttime`` and ``endtime`` are formatted date-time strings
(e.g. ``"2024-01-01T00:00:00Z"``); set either to ``None`` for an
open-ended time window, set both to ``None`` to match all time.
If ``skip_not_data`` is True, bytes from the input stream will be skipped
until a record is found.
If ``validate_crc`` is True, the CRC will be validated if contained in
the record (legacy miniSEED v2 contains no CRCs). The CRC provides an
internal integrity check of the record contents.
The overall structure of the trace list is a list of trace IDs, each of
which contains a list of trace segments illustrated as follows:
- TraceList
- TraceID
- Trace Segment
- Trace Segment
- Trace Segment
- ...
- TraceID
- Trace Segment
- Trace Segment
- ...
- ...
TraceIDs can be accessed via indexing and slicing:
- ``traces[0]`` returns the first TraceID
- ``traces[1:3]`` returns a slice of the TraceIDs
- ``for traceid in traces:`` iterates over all TraceIDs
- ``sourceid in traces`` tests for a source ID, see ``get_traceid()``
Trace Segments can be accessed via indexing and slicing:
- ``traceid[0]`` returns the first Trace Segment
- ``traceid[1:3]`` returns a slice of the Trace Segments
- ``for segment in traceid:`` iterates over all Trace Segments
Example usage iterating over the trace list:
>>> from pymseed import MS3TraceList
>>> for traceid in MS3TraceList.from_file("examples/example_data.mseed"):
... print(f"{traceid.sourceid}, {traceid.pubversion}")
... for segment in traceid:
... print(
... f" {segment.starttime_str()} - {segment.endtime_str()}, "
... f"{segment.samprate} sps, {segment.samplecnt} samples"
... )
FDSN:IU_COLA_00_L_H_1, 4
2010-02-27T06:50:00.069539Z - 2010-02-27T07:59:59.069538Z, 1.0 sps, 4200 samples
FDSN:IU_COLA_00_L_H_2, 4
2010-02-27T06:50:00.069539Z - 2010-02-27T07:59:59.069538Z, 1.0 sps, 4200 samples
FDSN:IU_COLA_00_L_H_Z, 4
2010-02-27T06:50:00.069539Z - 2010-02-27T07:59:59.069538Z, 1.0 sps, 4200 samples
Example using source ID and time-window selection:
>>> traces = MS3TraceList.from_file(
... "examples/example_data.mseed",
... sourceid="FDSN:IU_COLA_00_L_H_Z",
... starttime="2010-02-27T07:00:00Z",
... endtime="2010-02-27T07:30:00Z",
... )
>>> len(traces)
1
>>> traces[0].sourceid
'FDSN:IU_COLA_00_L_H_Z'
>>> traces[0][0].samplecnt
1963
The trace list, its samples, and the sources held for any record lists are
released when it is garbage collected. Call :meth:`close`, or use the
trace list as a context manager, to release them at a known point instead:
>>> with MS3TraceList.from_file("examples/example_data.mseed") as traces:
... print(len(traces))
3
"""
def __init__(
self,
file_name: str | os.PathLike[str] | None = None,
buffer: Any = None,
*,
unpack_data: bool = False,
sourceid: str | None = None,
starttime: str | None = None,
endtime: str | None = None,
record_list: bool = False,
skip_not_data: bool = False,
validate_crc: bool = True,
split_version: bool = False,
verbose: int = 0,
) -> None:
# Initialize trace list - mstl3_init() returns an initialized pointer
self._mstl = clibmseed.mstl3_init(ffi.NULL)
if self._mstl == ffi.NULL:
raise MiniSEEDError(clibmseed.MS_GENERROR, "Error initializing trace list")
# One C-compatible file name buffer per encoded path. Record list
# entries hold the pointer itself, so a buffer must outlive them.
self._c_file_names: dict[bytes, Any] = {}
# Source buffers behind record list entries, which point into them
# rather than copying. Held, not copied, so they cannot be released
# while the record lists refer to them.
self._buffer_refs: list[Any] = []
# Read specified file
if file_name is not None:
self.add_file(
file_name,
unpack_data=unpack_data,
sourceid=sourceid,
starttime=starttime,
endtime=endtime,
record_list=record_list,
skip_not_data=skip_not_data,
validate_crc=validate_crc,
split_version=split_version,
verbose=verbose,
)
if buffer is not None:
self.add_buffer(
buffer,
unpack_data=unpack_data,
sourceid=sourceid,
starttime=starttime,
endtime=endtime,
record_list=record_list,
skip_not_data=skip_not_data,
validate_crc=validate_crc,
split_version=split_version,
verbose=verbose,
)
[docs]
def __enter__(self) -> MS3TraceList:
"""Context manager entry point - returns self for use in 'with' statements."""
return self
[docs]
def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
"""Context manager exit point - ensures proper cleanup by calling close()."""
self.close()
[docs]
def __del__(self) -> None:
"""Destructor to ensure proper cleanup"""
if sys.is_finalizing():
return
try:
self.close()
except (AttributeError, TypeError):
# Module-teardown race: clibmseed/ffi/cdata fields may have
# been nulled out by Python before sys.is_finalizing() flipped.
# Nothing actionable; let any other exception propagate via
# Python's "Exception ignored in" mechanism so real bugs
# surface.
pass
[docs]
def close(self) -> None:
"""Free the trace list and release the sources held for it.
Waiting for garbage collection is otherwise the only way to release
the samples, the file name buffers, and the source buffers that record
lists point into.
Idempotent: safe to call multiple times.
.. warning::
Every :class:`~pymseed.mstracelist.MS3TraceID`,
:class:`~pymseed.mstracelist.MS3TraceSeg`, record list, and
data sample view obtained from this trace list is invalidated.
Reading one after this call reads freed memory. The trace list
itself reports the closure, raising :class:`ValueError` for any
further operation on it.
"""
if self._mstl != ffi.NULL:
mstl_ptr = ffi.new("MS3TraceList **")
mstl_ptr[0] = self._mstl
clibmseed.mstl3_free(mstl_ptr, 1)
self._mstl = ffi.NULL
# Nothing refers to the file names or source buffers now
self._c_file_names.clear()
self._buffer_refs.clear()
def _check_open(self) -> None:
"""Raise if the trace list has been closed"""
if self._mstl == ffi.NULL:
raise ValueError("operation on closed MS3TraceList")
def __repr__(self) -> str:
if self._mstl == ffi.NULL:
return "MS3TraceList(closed)"
lines = "\n".join(_summary_lines(self, repr))
return f"MS3TraceList(numtraceids: {len(self)}\n{lines}\n)"
def __str__(self) -> str:
if self._mstl == ffi.NULL:
return "Closed trace list"
lines = "\n".join(_summary_lines(self, str))
return f"Trace list with {len(self)} trace IDs\n{lines}\n"
[docs]
def __len__(self) -> int:
"""Return number of trace IDs in the list"""
if self._mstl == ffi.NULL:
return 0
return int(self._mstl.numtraceids)
[docs]
def __iter__(self) -> Iterator[MS3TraceID]:
"""Return iterator over trace IDs"""
self._check_open()
current_traceid = self._mstl.traces.next[0]
while current_traceid != ffi.NULL:
yield MS3TraceID(current_traceid, self)
current_traceid = current_traceid.next[0]
[docs]
def __contains__(self, item: object) -> bool:
"""Test for a source ID, as a str or MS3TraceID, in the list
A str matches any publication version, an MS3TraceID matches only its own.
Any other type is not in the list.
"""
if isinstance(item, MS3TraceID):
return self.get_traceid(item.sourceid, item.pubversion) is not None
elif isinstance(item, str):
return self.get_traceid(item) is not None
return False
[docs]
def __getitem__(self, key: int | slice) -> Any:
"""Enable indexing and slicing access to trace IDs"""
return _linked_getitem(self, key)
@property
def numtraceids(self) -> int:
"""Return number of trace IDs in the list"""
return len(self)
[docs]
def get_traceid(self, sourceid: str, version: int = 0) -> MS3TraceID | None:
"""Get a specific trace ID from the list, or None if not present
Raises:
TypeError: If sourceid is not a str
"""
check_str("sourceid", sourceid)
self._check_open()
c_sourceid = ffi.new("char[]", sourceid.encode("utf-8"))
traceid_ptr = clibmseed.mstl3_findID(self._mstl, c_sourceid, version, ffi.NULL)
if traceid_ptr == ffi.NULL:
return None
return MS3TraceID(traceid_ptr, self)
[docs]
def sourceids(self) -> Iterator[str]:
"""Return source IDs via a generator iterator"""
for traceid in self:
yield traceid.sourceid
[docs]
def print(
self,
details: int = 0,
gaps: bool = False,
versions: bool = False,
timeformat: TimeFormat = TimeFormat.ISOMONTHDAY_Z,
) -> None:
"""Print a summary of the trace list, one line per segment
Each line has the source ID, start time, and end time. The summary is
written to standard output by the C library, so it is not captured by
redirecting :data:`sys.stdout` and does not interleave with the output
of Python's :func:`print`.
Args:
details: If greater than 0, add the sample rate and sample count
to each line, and a total trace and segment count.
Default: 0
gaps: If True, add the gap or overlap between each segment and the
previous segment of the same source ID and sample rate.
Default: False
versions: If True, append the publication version to each source
ID as ``"#<version>"``. Default: False
timeformat: Format of the start and end times, a
:class:`TimeFormat` value.
Default: TimeFormat.ISOMONTHDAY_Z
Raises:
ValueError: If the trace list has been closed.
"""
self._check_open()
clibmseed.mstl3_printtracelist(self._mstl, timeformat, details, gaps, versions)
[docs]
def add_file(
self,
file_name: str | os.PathLike[str],
*,
unpack_data: bool = False,
sourceid: str | None = None,
starttime: str | None = None,
endtime: str | None = None,
record_list: bool = False,
skip_not_data: bool = False,
validate_crc: bool = True,
split_version: bool = False,
verbose: int = 0,
) -> None:
"""Read miniSEED data from file and add to existing trace list
This method reads miniSEED records from a file and adds the data they contain
to the current trace list. Data are organized by source ID and time, with
overlapping or adjacent data automatically merged into continuous segments.
Args:
file_name: Path to the miniSEED file to read. Accepts ``str`` or
any :class:`os.PathLike` (e.g. :class:`pathlib.Path`). Other
types raise :class:`TypeError`.
unpack_data: If True, decode data samples immediately. If False, data
samples remain packed and must be unpacked later with
:meth:`~pymseed.mstracelist.MS3TraceSeg.unpack_recordlist`. Default: False
sourceid: Source ID glob pattern to select matching records
(e.g. ``"FDSN:IU_COLA_*"``). None matches all source IDs.
Default: None
starttime: Start of time window as a formatted date-time string
(e.g. ``"2024-01-01T00:00:00Z"``). Only records containing data
after this time are included. None means open start.
Default: None
endtime: End of time window as a formatted date-time string.
Only records containing data before this time are included.
None means open end. Default: None
record_list: If True, maintain a list of original records for each
trace segment. Required for `unpack_recordlist()` and allows
access to individual record metadata. Default: False
NOTE: the files must remain accessible to unpack data with a record list
skip_not_data: If True, skip non-data records in the file instead
of raising an error. Useful for files with mixed content. Default: False
validate_crc: If True, validate CRC checksums if present in records
(miniSEED v3 only). Provides integrity verification. Default: True
split_version: If True, treat different publication versions as
separate trace IDs. Default: False (merge by source ID only)
verbose: Verbosity level for diagnostic output (0=quiet, 1-3=increasing
detail). Default: 0
Raises:
ValueError: If starttime or endtime is not a valid date-time string
MiniSEEDError: If file cannot be read or contains invalid data
Note:
This method adds data to the existing trace list. It does not replace
existing data. To start fresh, create a new MS3TraceList instance.
Examples:
Basic usage examples:
>>> from pymseed import MS3TraceList
>>> traces = MS3TraceList()
>>> traces.add_file("examples/example_data.mseed")
>>> len(traces)
3
Unpacking data while reading for immediate access
>>> traces = MS3TraceList()
>>> traces.add_file("examples/example_data.mseed", unpack_data=True)
>>> traces[0].sourceid
'FDSN:IU_COLA_00_L_H_1'
>>> segment1 = traces[0][0]
>>> segment1.starttime_str()
'2010-02-27T06:50:00.069539Z'
>>> segment1.endtime_str()
'2010-02-27T07:59:59.069538Z'
>>> segment1.samprate
1.0
>>> segment1.samplecnt
4200
>>> segment1.numsamples
4200
>>> segment1.sampletype
'i'
Read with record list for later unpacking:
>>> traces = MS3TraceList()
>>> traces.add_file("examples/example_data.mseed", record_list=True)
>>> total_samples = 0
>>> for traceid in traces:
... for segment in traceid:
... # Unpack when desired, can unpacked to designated buffer
... samples_count = segment.unpack_recordlist()
... total_samples += samples_count
>>> total_samples
12600
Add multiple files to same trace list, in this trivial example the
same file is read twice, so the data are duplicated in the trace list:
>>> traces = MS3TraceList()
>>> traces.add_file("examples/example_data.mseed")
>>> traces.add_file("examples/example_data.mseed") # Appends to existing data
>>> len(traces)
3
>>> traceid = traces[0]
>>> traceid[0].samplecnt
4200
>>> len(traceid) # Two segments (duplicate in this case)
2
>>> # Compare using explicit data comparison (fast)
>>> traceid[0].has_same_data(traceid[1])
True
"""
file_name = check_path("file_name", file_name)
begin_operation()
# Store file name for reference and use in record lists. Sharing one
# buffer per path also lets unpack_recordlist() match entries by
# pointer and open the file once.
encoded_file_name = os.fsencode(file_name)
if record_list:
c_file_name = self._c_file_names.get(encoded_file_name)
if c_file_name is None:
c_file_name = ffi.new("char[]", encoded_file_name)
self._c_file_names[encoded_file_name] = c_file_name
else:
c_file_name = ffi.new("char[]", encoded_file_name)
# Request storing time of update in the trace list segment
# This stores the update time as an nstime_t in the segment's private pointer (seg.prvtptr)
flags = clibmseed.MSF_PPUPDATETIME | parse_flags(
unpack_data=unpack_data,
validate_crc=validate_crc,
skip_not_data=skip_not_data,
record_list=record_list,
)
# Create a reference to the current trace list pointer
self._check_open()
mstl_ptr = ffi.new("MS3TraceList **")
mstl_ptr[0] = self._mstl
# An empty regular file has no records to add, matching add_buffer()
# on an empty buffer; skip the read rather than let it raise
# MS_NOTSEED. A stat failure (e.g. a nonexistent path) is left for
# the read call below to report as it normally would.
try:
if os.stat(file_name).st_size == 0:
return
except OSError:
pass
# Build selections, if sourceid, starttime, or endtime are specified
selections_ptr, free_selections = build_selections(sourceid, starttime, endtime)
try:
status = clibmseed.ms3_readtracelist_selection(
mstl_ptr,
c_file_name,
ffi.NULL, # tolerance
selections_ptr,
int(split_version),
flags,
verbose,
)
finally:
if free_selections is not None:
free_selections()
if status != clibmseed.MS_NOERROR:
raise MiniSEEDError(status, f"Error reading file: {file_name}")
[docs]
def add_buffer(
self,
buffer: Any,
*,
unpack_data: bool = False,
sourceid: str | None = None,
starttime: str | None = None,
endtime: str | None = None,
record_list: bool = False,
skip_not_data: bool = False,
validate_crc: bool = True,
split_version: bool = False,
verbose: int = 0,
) -> None:
"""Read miniSEED data from a buffer and add to existing trace list
This method reads miniSEED records from a bytes-like object and adds the
data they contain to the current trace list. Data are organized by
source ID and time, with overlapping or adjacent data automatically
merged into continuous segments.
Args:
buffer: Bytes-like object containing miniSEED data
unpack_data: If True, decode data samples immediately. If False, data
samples remain packed and must be unpacked later with
:meth:`~pymseed.mstracelist.MS3TraceSeg.unpack_recordlist`. Default: False
sourceid: Source ID glob pattern to select matching records
(e.g. ``"FDSN:IU_COLA_*"``). None matches all source IDs.
Default: None
starttime: Start of time window as a formatted date-time string
(e.g. ``"2024-01-01T00:00:00Z"``). Only records containing data
after this time are included. None means open start.
Default: None
endtime: End of time window as a formatted date-time string.
Only records containing data before this time are included.
None means open end. Default: None
record_list: If True, maintain a list of original records for each
trace segment. Required for `unpack_recordlist()` and allows
access to individual record metadata. Default: False
NOTE: the record list entries point into the buffer, which the
trace list holds a reference to until it is discarded
skip_not_data: If True, skip non-data records in the buffer instead
of raising an error. Useful for files with mixed content. Default: False
validate_crc: If True, validate CRC checksums if present in records
(miniSEED v3 only). Provides integrity verification. Default: True
split_version: If True, treat different publication versions as
separate trace IDs. Default: False (merge by source ID only)
verbose: Verbosity level for diagnostic output (0=quiet, 1-3=increasing
detail). Default: 0
Raises:
ValueError: If buffer does not support the buffer protocol, or if
starttime or endtime is not a valid date-time string
BufferError: If buffer is not C-contiguous
MiniSEEDError: If buffer cannot be read or contains invalid data
Note:
This method adds data to the existing trace list. It does not replace
existing data. To start fresh, create a new MS3TraceList instance.
To add data from a file, use `add_file()`.
Examples:
Read miniSEED data from a file into a buffer:
>>> with open("examples/example_data.mseed", "rb") as f:
... buffer = f.read()
Basic usage examples:
>>> from pymseed import MS3TraceList
>>> traces = MS3TraceList()
>>> traces.add_buffer(buffer)
>>> len(traces)
3
Unpacking data while reading for immediate access
>>> traces = MS3TraceList()
>>> traces.add_buffer(buffer, unpack_data=True)
>>> traces[0].sourceid
'FDSN:IU_COLA_00_L_H_1'
>>> segment1 = traces[0][0]
>>> segment1.starttime_str()
'2010-02-27T06:50:00.069539Z'
>>> segment1.endtime_str()
'2010-02-27T07:59:59.069538Z'
>>> segment1.samprate
1.0
>>> segment1.samplecnt
4200
>>> segment1.numsamples
4200
>>> segment1.sampletype
'i'
Read with record list for later unpacking:
>>> traces = MS3TraceList()
>>> traces.add_buffer(buffer, record_list=True)
>>> total_samples = 0
>>> for traceid in traces:
... for segment in traceid:
... # Unpack when desired, can unpacked to designated buffer
... samples_count = segment.unpack_recordlist()
... total_samples += samples_count
>>> total_samples
12600
Add multiple buffers to same trace list, in this trivial example the
same file is read twice, so the data are duplicated in the trace list:
>>> traces = MS3TraceList()
>>> traces.add_buffer(buffer)
>>> traces.add_buffer(buffer) # Appends to existing data
>>> len(traces)
3
>>> traceid = traces[0]
>>> traceid[0].samplecnt
4200
>>> len(traceid) # Two segments (duplicate in this case)
2
"""
# Request storing time of update in the trace list segment
# This stores the update time as an nstime_t in the segment's private pointer (seg.prvtptr)
flags = clibmseed.MSF_PPUPDATETIME | parse_flags(
unpack_data=unpack_data,
validate_crc=validate_crc,
skip_not_data=skip_not_data,
record_list=record_list,
)
# Create a reference to the current trace list pointer
self._check_open()
mstl_ptr = ffi.new("MS3TraceList **")
mstl_ptr[0] = self._mstl
begin_operation()
buffer_ptr = buffer_pointer(buffer)
buffer_length = len(buffer_ptr)
# Record list entries point into the buffer instead of copying it, for
# both the raw records and unpacking. Hold it before reading so the
# records a partial read added remain valid.
if record_list:
self._buffer_refs.append(buffer_ptr)
# Build selections, if sourceid, starttime, or endtime are specified
selections_ptr, free_selections = build_selections(sourceid, starttime, endtime)
try:
status = clibmseed.mstl3_readbuffer_selection(
mstl_ptr,
buffer_ptr,
buffer_length,
int(split_version),
flags,
ffi.NULL, # tolerance
selections_ptr,
verbose,
)
finally:
if free_selections is not None:
free_selections()
if status < 0:
raise MiniSEEDError(status, f"Error reading buffer (status: {status})")
[docs]
def add_filelike(
self,
fh: Any,
*,
chunk_size: int = 65536,
unpack_data: bool = False,
sourceid: str | None = None,
starttime: str | None = None,
endtime: str | None = None,
record_list: bool = False,
validate_crc: bool = True,
split_version: bool = False,
verbose: int = 0,
) -> None:
"""Read miniSEED data from a file-like stream and add to the trace list.
This method reads miniSEED records from any object exposing
``.read(n) -> bytes`` (e.g. ``io.BytesIO``, ``sys.stdin.buffer``, an
HTTP response body, a network socket file) using a chunked sliding
buffer. The stream is **not** required to be seekable and the full
contents do not need to fit in memory. The caller retains ownership
of ``fh`` and is responsible for closing it.
Performance note:
This method is intended for streaming sources that cannot be
represented as a file path or in-memory buffer. For files on
disk, prefer :meth:`add_file` (or :meth:`from_file`). For data
already resident in memory, prefer :meth:`add_buffer` (or
:meth:`from_buffer`). Those routines run a tight loop inside
libmseed and apply selections at parse time; this method
round-trips each record through Python, which is typically much
slower. Use it as a last resort when the other methods cannot be used.
Record list limitation:
``record_list=True`` is supported and produces the same
per-record metadata as :meth:`add_file` / :meth:`add_buffer`
(source ID, start/end times, record length, encoding, etc.),
**but** :meth:`~pymseed.mstracelist.MS3TraceSeg.unpack_recordlist` cannot be used on
the resulting record list as the original source bytes do not persist.
The per-record references to those bytes are cleared, so
:attr:`MS3Record.record` and :attr:`MS3Record.record_mv` also raise
for entries in this list. If you need either, read the data with
:meth:`add_file` or :meth:`add_buffer` instead.
Args:
fh: A file-like object with a ``.read(n)`` method returning bytes.
chunk_size: Number of bytes to read per ``.read()`` call, greater
than 0 and less than 1 GiB. Default: 65536.
unpack_data: If True, decode data samples immediately.
Default: False.
sourceid: Source ID glob pattern to select matching records
(e.g. ``"FDSN:IU_COLA_*"``). None matches all source IDs.
Default: None.
starttime: Start of time window as a formatted date-time string
(e.g. ``"2024-01-01T00:00:00Z"``). Only records overlapping
this window are included. None means open start. Default: None.
endtime: End of time window as a formatted date-time string.
None means open end. Default: None.
record_list: If True, maintain a per-segment list of original
records (source ID, times, reclen, encoding, etc.).
See the "Record list limitation" note above:
:meth:`~pymseed.mstracelist.MS3TraceSeg.unpack_recordlist` cannot be used on the resulting
list because the source bytes do not persist. Default: False.
validate_crc: If True, validate CRC checksums when present
(miniSEED v3 only). Default: True.
split_version: If True, treat different publication versions as
separate trace IDs. Default: False.
verbose: Verbosity level for libmseed diagnostics. Default: 0.
Raises:
TypeError: If ``fh`` has no callable ``.read`` method.
ValueError: If ``chunk_size`` is not greater than 0 and less than
1 GiB, or if ``starttime`` or ``endtime`` is not a valid
date-time string.
MiniSEEDError: If a record cannot be parsed or cannot be added
to the trace list.
Examples:
Read miniSEED from a BytesIO stream:
>>> import io
>>> from pymseed import MS3TraceList
>>> with open("examples/example_data.mseed", "rb") as f:
... stream = io.BytesIO(f.read())
>>> traces = MS3TraceList()
>>> traces.add_filelike(stream)
>>> len(traces)
3
With source ID and time-window filtering:
>>> with open("examples/example_data.mseed", "rb") as f:
... stream = io.BytesIO(f.read())
>>> traces = MS3TraceList()
>>> traces.add_filelike(
... stream,
... sourceid="FDSN:IU_COLA_00_L_H_Z",
... starttime="2010-02-27T07:00:00Z",
... endtime="2010-02-27T07:30:00Z",
... )
>>> len(traces)
1
>>> traces[0].sourceid
'FDSN:IU_COLA_00_L_H_Z'
"""
# Built before _check_open() so an invalid fh or chunk_size raises
# before the open check.
records = MS3Record.from_filelike(
fh,
chunk_size=chunk_size,
unpack_data=unpack_data,
sourceid=sourceid,
starttime=starttime,
endtime=endtime,
validate_crc=validate_crc,
verbose=verbose,
)
self._check_open()
flags = clibmseed.MSF_PPUPDATETIME | parse_flags(
validate_crc=validate_crc, record_list=record_list
)
begin_operation()
# A handle for the record entries in a record list, reused for each record
pprecptr = ffi.new("MS3RecordPtr **") if record_list else ffi.NULL
# Selection matching and deferred data unpacking happen in `records`.
for msr in records:
# A record added directly carries no source reference, msr->record
# included, matching source bytes that do not outlive the read.
seg = clibmseed.mstl3_addmsr_recordptr(
self._mstl,
msr._msr,
pprecptr,
int(split_version),
1, # autoheal
flags,
ffi.NULL, # tolerance
)
if seg == ffi.NULL:
raise MiniSEEDError(
clibmseed.MS_GENERROR,
"Error adding record from file-like stream",
)
[docs]
def add_data(
self,
sourceid: str,
data_samples: Any,
sample_type: str,
sample_rate: float,
*,
starttime_str: str | None = None,
starttime: int | None = None,
starttime_seconds: float | None = None,
publication_version: int = 0,
) -> None:
"""Add data samples to the trace list
A segment of regularly sampled data values for the given source ID of
the specific type and sample rate are added to the trace list.
Args:
sourceid: Source identifier for the trace (e.g., "FDSN:XX_STA__B_H_Z").
Should follow FDSN Source Identifier format.
data_samples: Sequence of data samples. Can be a Python list, numpy array,
or any buffer-like object. Data type must match `sample_type`.
sample_type: Data sample type code:
- "i": 32-bit signed integers (int32)
- "f": 32-bit floating point (float32)
- "d": 64-bit floating point (float64)
- "t": Text/character data (single bytes)
sample_rate: Sample rate in samples per second (Hz) or period (seconds).
Use positive values for samples/second, and negative values for sample period in seconds.
starttime_str: Start time as formatted string (e.g., "2023-01-01T12:00:00.000Z").
Mutually exclusive with starttime and starttime_seconds.
starttime: Start time as nanoseconds since Unix epoch.
Mutually exclusive with starttime_str and starttime_seconds.
starttime_seconds: Start time as seconds since Unix epoch (float).
Mutually exclusive with starttime_str and starttime.
publication_version: Publication version number for the trace. Default: 0
Raises:
ValueError: If sample_type is invalid, time parameters are conflicting,
or data_samples format is incompatible with sample_type
MiniSEEDError: If the data cannot be added to the trace list
Note:
Data is automatically merged with existing segments based on source ID,
time continuity, and sample rate similarity. Adjacent or overlapping
segments are combined when possible.
Performance:
The method attempts zero-copy optimization when data_samples is a compatible
buffer (correct type and format). Otherwise, data is converted with a copy.
Use arrays (or any buffer with memoryviews) with matching types for best performance.
Examples:
Basic usage with integer data:
>>> from pymseed import MS3TraceList
>>> traces = MS3TraceList()
>>> data_series = [100, 105, 98, 102, 99, 103, 97]
>>> traces.add_data(
... sourceid="FDSN:XX_STA__B_H_Z",
... data_samples=data_series,
... sample_type="i",
... sample_rate=20.0,
... starttime_str="2023-01-01T00:00:00.000Z"
... )
>>> len(traces)
1
>>> traces[0].sourceid
'FDSN:XX_STA__B_H_Z'
Multiple segments that get merged:
>>> traces = MS3TraceList()
>>> traces.add_data("FDSN:XX_STA__B_H_1", [1, 2, 3], "i", 10.0,
... starttime_str="2023-01-01T00:00:00.000Z")
>>> traces.add_data("FDSN:XX_STA__B_H_1", [4, 5, 6], "i", 10.0,
... starttime_str="2023-01-01T00:00:00.300Z")
>>> len(traces) # One traceID
1
>>> len(traces[0]) # One trace segment
1
"""
self._check_open()
begin_operation()
# Create an MS3Record to hold the data
msr = MS3Record()
msr.sourceid = sourceid
msr.samprate = sample_rate
msr.pubversion = publication_version
# Ensure that start time definitions are mutually exclusive
provided_count = sum(
value is not None for value in (starttime_str, starttime, starttime_seconds)
)
if provided_count != 1:
raise ValueError(
"Specify exactly one of starttime_str, starttime, or "
f"starttime_seconds; got {provided_count}"
)
if starttime_str is not None:
msr.set_starttime_str(starttime_str)
elif starttime is not None:
msr.starttime = starttime
elif starttime_seconds is not None:
msr.starttime_seconds = starttime_seconds
# Request storing time of update in the trace list segment
# This stores the update time as an nstime_t in the segment's private pointer (seg.prvtptr)
flags = clibmseed.MSF_PPUPDATETIME
# Set data samples array, type, and counts temporarily for potential zero-copy operations
with msr.with_datasamples(data_samples, sample_type):
# Add the MS3Record to the trace list, setting auto-heal flag to 1 (true)
segptr = clibmseed.mstl3_addmsr_recordptr(
self._mstl, msr._msr, ffi.NULL, 0, 1, flags, ffi.NULL
)
if segptr == ffi.NULL:
raise MiniSEEDError(clibmseed.MS_GENERROR, "Error adding data samples")
[docs]
def generate(
self,
*,
max_record_length: int = 4096,
encoding: DataEncoding = DataEncoding.STEIM1,
format_version: int | None = None,
extra_headers: str | None = None,
flush_data: bool = True,
flush_idle_seconds: int = 0,
remove_packed: bool = False,
verbose: int = 0,
) -> Iterator[bytes]:
"""Create miniSEED record(s) for data in the trace list.
This method creates, or packs, miniSEED record(s) for the time series
data from all traces in the trace list using the provided parameters
(encoding, record length, etc.).
Args:
max_record_length: Maximum length of each miniSEED record in bytes.
For miniSEED format version 3, this is the maximum record length.
For miniSEED format version 2, this must be a power of 2 between
128 and 65536. Common values are 512 and 4096.
Default is 4096.
encoding: Data encoding format for compression. Options include:
- DataEncoding.STEIM1: Steim-1 compression (default, good
general purpose for 32-bit ints)
- DataEncoding.STEIM2: Steim-2 compression
- DataEncoding.INT16: 16-bit integers (no compression)
- DataEncoding.INT32: 32-bit integers (no compression)
- DataEncoding.FLOAT32: 32-bit IEEE floats
- DataEncoding.FLOAT64: 64-bit IEEE doubles
- DataEncoding.TEXT: Text encoding (UTF-8)
format_version: miniSEED format version (2 or 3). If None, uses
library default. Version 2 is legacy format, version 3 is the
latest standard.
extra_headers: Optional extra header fields to include.
Must be valid JSON string.
flush_data: If True, forces creation of records for all
data, even if it doesn't fill a complete record. If False, data
samples at the end of traces may be held in internal buffers.
Default is True.
flush_idle_seconds: If > 0, forces flushing of data segments that
have not been updated within the specified number of seconds.
Default is 0 (disabled), which holds idle data in the trace
list until a call with `flush_data=True`.
remove_packed: If True, data samples packed into records will be
removed from the trace list. See "Rolling buffer" section below
for more details. Default is False.
verbose: Verbosity level for libmseed output (0=quiet, 1=info,
2=detailed). Default is 0 (quiet).
Yields:
bytes: Each miniSEED record as it is created
Raises:
ValueError: If format_version is not 2 or 3, or encoding is outside
the range 0..255. Raised by this call, before the first record
is created.
MiniSEEDError: If the underlying libmseed library encounters an
error during creation of miniSEED records, such as a
max_record_length the format or encoding cannot use. Raised
while iterating, by the record it applies to.
Examples:
Simple example creating miniSEED records:
>>> # Create a trace list with some data
>>> traces = MS3TraceList()
>>> traces.add_data("FDSN:XX_STA__H_H_Z", [1, 2, 3, 4, 5], "i", 100.0, starttime_str="2023-01-01T00:00:00.000Z")
>>> traces.add_data("FDSN:XX_STA__H_H_1", [6, 7, 8, 9, 9], "i", 100.0, starttime_str="2023-01-01T00:00:00.000Z")
>>> traces.add_data("FDSN:XX_STA__H_H_2", [9, 9, 8, 7, 6], "i", 100.0, starttime_str="2023-01-01T00:00:00.000Z")
>>> record_count = 0
>>> for record in traces.generate():
... record_count += 1
>>> print(f"Created {record_count} records")
Created 3 records
Rolling buffer:
A common pattern is to use a MS3TraceList as a rolling buffer to
generate miniSEED records from an arbitrary number of continuous
data streams. In particular this pattern allows creating filled
miniSEED records as much as possible during regular data flow, and
then flushing any remaining data at the end. This is particularly
useful for converting real-time data streams into miniSEED or for
converting large data sets into miniSEED without loading all the
source data into memory.
A few options are needed to properly configure the rolling buffer.
For creating filled records during regular data flow:
* `flush_data=False` to keep data in the trace list
* `flush_idle_seconds=N` to flush data segments that have not been updated within N seconds
* `remove_packed=True` to remove packed data from the trace list
To flush the all data from the buffer on termination:
* `flush_data=True` to flush all data from the trace list
* `remove_packed=True` to remove packed data from the trace list
With `flush_data=False`, a source that stops delivering data keeps
its remaining samples in the trace list until `flush_idle_seconds`
elapses, or until a later call sets `flush_data=True`. Set
`flush_idle_seconds=N` whenever the set of source IDs can change
over the life of the buffer: at the default of 0 those samples, and
the trace ID holding them, are retained for as long as the trace
list lives.
Choose N larger than the time it takes the slowest source to fill a
record, otherwise idle flushing creates partial records where full
ones are still possible: 1 sample/second fills a 4096-byte STEIM2
record in about 110 minutes, a 512-byte one in about 12 minutes.
Choose a smaller N when bounding output latency matters more than
filling records. :attr:`~pymseed.mstracelist.MS3TraceSeg.update_time` reports the time
a segment was last updated, which is the value N is compared
against.
.. warning::
With `remove_packed=True`, samples are removed from the trace list
only when the segment they belong to finishes packing. Abandoning
this generator early — a ``break``, a ``return``, or an exception
raised while handling a record — leaves the samples of the records
already yielded in the trace list, and a later `generate()` creates
records for them again. Iterate to completion to keep the buffer
and the output consistent, handling per-record failures inside the
loop.
See Also:
- to_file()
"""
if format_version is not None:
check_format_version(format_version)
check_encoding(encoding)
self._check_open()
return self._generate(
max_record_length=max_record_length,
encoding=encoding,
format_version=format_version,
extra_headers=extra_headers,
flush_data=flush_data,
flush_idle_seconds=flush_idle_seconds,
remove_packed=remove_packed,
verbose=verbose,
)
def _generate(
self,
max_record_length: int,
encoding: DataEncoding,
format_version: int | None,
extra_headers: str | None,
flush_data: bool,
flush_idle_seconds: int,
remove_packed: bool,
verbose: int,
) -> Iterator[bytes]:
"""Generator body for :meth:`generate`. Kept private so the public
wrapper can validate arguments eagerly before the first yield."""
begin_operation()
flags = 0
if flush_data:
flags |= clibmseed.MSF_FLUSHDATA
if not remove_packed:
flags |= clibmseed.MSF_MAINTAINMSTL
# Validated by generate(), the public wrapper
if format_version == 2:
flags |= clibmseed.MSF_PACKVER2
c_extra = ffi.new("char[]", extra_headers.encode("utf-8")) if extra_headers else ffi.NULL
packer = clibmseed.mstl3_pack_init(
self._mstl,
max_record_length,
encoding,
flags,
verbose,
c_extra,
flush_idle_seconds,
)
if not packer:
raise MiniSEEDError(clibmseed.MS_GENERROR, "Error initializing packer")
record_pp = ffi.new("char **")
reclen_p = ffi.new("int32_t *")
try:
while True:
# 1 = record available, 0 = finished, < 0 = error
status = clibmseed.mstl3_pack_next(packer, flags, record_pp, reclen_p)
if status < 0:
raise MiniSEEDError(status, "Error packing miniSEED record(s)")
if status != 1:
break
yield ffi.buffer(record_pp[0], reclen_p[0])[:]
finally:
packer_pp = ffi.new("MS3TraceListPacker **")
packer_pp[0] = packer
clibmseed.mstl3_pack_free(packer_pp, ffi.NULL)
[docs]
def to_file(
self,
filename: str | os.PathLike[str],
*,
overwrite: bool = False,
max_record_length: int = 4096,
encoding: DataEncoding = DataEncoding.STEIM1,
format_version: int | None = None,
verbose: int = 0,
) -> int:
"""Write trace list data to a miniSEED file.
This method packages the time series data from all traces in the trace
list into miniSEED format and writes them directly to a file. This is a
convenience method that combines packing and file writing in a single
operation.
Args:
filename: Path to the output miniSEED file. Accepts ``str`` or any
:class:`os.PathLike` (e.g. :class:`pathlib.Path`); other types
raise :class:`TypeError`. The file will be created if it
doesn't exist. Directory must already exist.
overwrite: If True, overwrites any existing file. If False and file
exists, append data to the end of the file. Default is False for
safety.
max_record_length: Maximum length of each miniSEED record in bytes.
For miniSEED format version 3, this is the maximum record length.
For miniSEED format version 2, this must be a power of 2
between 128 and 65536. Common values are 512 and 4096.
Default is 4096.
encoding: Data encoding format for compression. Options include:
- DataEncoding.STEIM1: Steim-1 compression (default, good
general purpose for 32-bit ints)
- DataEncoding.STEIM2: Steim-2 compression
- DataEncoding.INT16: 16-bit integers (no compression)
- DataEncoding.INT32: 32-bit integers (no compression)
- DataEncoding.FLOAT32: 32-bit IEEE floats
- DataEncoding.FLOAT64: 64-bit IEEE doubles
- DataEncoding.TEXT: Text encoding (UTF-8)
format_version: miniSEED format version (2 or 3). If None, uses
library default. Version 2 is legacy format, version 3 is the
latest standard.
verbose: Verbosity level for libmseed output (0=quiet, 1=info,
2=detailed). Default is 0 (quiet).
Returns:
int: Number of miniSEED records written to the file.
Raises:
TypeError: If filename is not a str or os.PathLike.
ValueError: If format_version is not 2 or 3, or encoding is outside
the range 0..255.
MiniSEEDError: If the underlying libmseed library
encounters an error during file writing (e.g., permission
denied, disk full, invalid data, or a max_record_length the
format or encoding cannot use).
Examples:
Simple file writing:
>>> # Create a trace list with some data
>>> traces = MS3TraceList()
>>> traces.add_data("FDSN:XX_STA__B_H_Z", [1, 2, 3, 4, 5], "i", 100.0, starttime_str="2023-01-01T00:00:00Z")
>>> # Write to file (basic usage)
>>> records_written = traces.to_file("output.mseed") # doctest: +SKIP
>>> print(f"Wrote {records_written} records to output.mseed") # doctest: +SKIP
Wrote 1 records to output.mseed
Writing with specific options:
>>> # Write as miniSEED v2 with Steim-2 compression
>>> records_written = traces.to_file( # doctest: +SKIP
... "output.mseed",
... overwrite=True,
... format_version=2,
... encoding=DataEncoding.STEIM2,
... max_record_length=512
... )
>>> print(f"Wrote {records_written} records to output.mseed") # doctest: +SKIP
Wrote 1 records to output.mseed
Note:
This method is more convenient and efficient than using generate()
and writing to a file with a file handler.
See Also:
- generate(): Lower-level method to create miniSEED records
- add_data(): Add time series data to the trace list
- from_file(): Read miniSEED data from file
"""
filename = check_path("filename", filename)
check_encoding(encoding)
self._check_open()
begin_operation()
# Convert filename to bytes (C string).
c_filename = ffi.new("char[]", os.fsencode(filename))
pack_flags = 0
if format_version is not None:
check_format_version(format_version)
if format_version == 2:
pack_flags |= clibmseed.MSF_PACKVER2
# Call the C function
packed_records = clibmseed.mstl3_writemseed(
self._mstl,
c_filename,
overwrite,
max_record_length,
encoding,
pack_flags,
verbose,
)
if packed_records < 0:
raise MiniSEEDError(packed_records, "Error writing miniSEED file")
return packed_records
[docs]
@classmethod
def from_file(
cls,
filename: str | os.PathLike[str],
**kwargs: Any,
) -> MS3TraceList:
"""Create MS3TraceList from a specified miniSEED file"""
return cls(file_name=filename, **kwargs)
[docs]
@classmethod
def from_buffer(cls, buffer: Any, **kwargs: Any) -> MS3TraceList:
"""Create an MS3TraceList from miniSEED data in a memory buffer"""
return cls(buffer=buffer, **kwargs)
[docs]
@classmethod
def from_filelike(cls, fh: Any, **kwargs: Any) -> MS3TraceList:
"""Create an MS3TraceList from a miniSEED file-like stream.
See :meth:`add_filelike` for the full parameter list, supported
filters, the record-list limitation (``unpack_recordlist()`` is
not available on the resulting record list), and performance
guidance (prefer :meth:`from_file` / :meth:`from_buffer` when
applicable).
"""
traces = cls()
traces.add_filelike(fh, **kwargs)
return traces