tessl/pypi-obspy
ObsPy is a Python toolbox for seismology providing parsers for seismological data formats, clients for data centers, and signal processing routines for seismological time series analysis.
—
Pending
core-data-structures.mddocs/
Core Data Structures
Fundamental data objects for seismological time series, event catalogs, and station metadata. These objects provide the foundation for all ObsPy functionality with automatic format detection and seamless integration across the seismological data ecosystem.
Capabilities
UTCDateTime Class
UTC-based datetime handling specifically designed for seismological timestamps with microsecond precision and specialized formatting for seismological standards.
class UTCDateTime:
def __init__(self, *args, **kwargs):
"""
Create UTC datetime object.
Args:
*args: Various input formats (string, timestamp, datetime components)
**kwargs: Additional parameters
Examples:
UTCDateTime() # Current time
UTCDateTime("2023-01-01T00:00:00")
UTCDateTime(2023, 1, 1, 0, 0, 0)
UTCDateTime(1672531200.0) # Unix timestamp
"""
def strftime(self, fmt: str) -> str:
"""Format datetime as string using strftime format codes."""
def strptime(cls, date_string: str, format: str) -> 'UTCDateTime':
"""Parse string to UTCDateTime using format codes."""
def format_fissures(self) -> str:
"""Format for FISSURES/DHI web services."""
def format_arclink(self) -> str:
"""Format for ArcLink protocol."""
def format_seedlink(self) -> str:
"""Format for SeedLink protocol."""
def format_seed(self) -> str:
"""Format for SEED standard."""
def format_iris_web_service(self) -> str:
"""Format for IRIS web services."""
def now(cls) -> 'UTCDateTime':
"""Get current UTC time."""
def utcnow(cls) -> 'UTCDateTime':
"""Get current UTC time (alias for now())."""
@property
def matplotlib_date(self) -> float:
"""Get matplotlib-compatible date number."""
def replace(self, **kwargs) -> 'UTCDateTime':
"""Return datetime with specified components replaced."""
def timetuple(self) -> tuple:
"""Return time tuple compatible with time module."""
def utctimetuple(self) -> tuple:
"""Return UTC time tuple."""
def toordinal(self) -> int:
"""Return proleptic Gregorian ordinal."""
def isoformat(self) -> str:
"""Return ISO format string."""
def isoweekday(self) -> int:
"""Return day of week (1=Monday, 7=Sunday)."""
def isocalendar(self) -> tuple:
"""Return ISO calendar tuple (year, week, weekday)."""Trace Class
Single seismogram trace container with data array and metadata, providing the fundamental building block for seismological time series analysis.
class Trace:
def __init__(self, data=None, header=None):
"""
Create seismogram trace.
Args:
data: NumPy array containing seismic data
header: Dictionary or Stats object with metadata
"""
@property
def data(self):
"""NumPy array containing seismic data."""
@property
def stats(self) -> 'Stats':
"""Trace metadata (Stats object)."""
@property
def id(self) -> str:
"""Unique trace identifier (network.station.location.channel)."""
def plot(self, **kwargs):
"""
Plot trace waveform.
Args:
**kwargs: Matplotlib plotting options
"""
def spectrogram(self, **kwargs):
"""
Plot spectrogram of trace data.
Args:
**kwargs: Spectrogram parameters and plotting options
"""
def write(self, filename: str, format: str = None, **kwargs):
"""
Write trace to file.
Args:
filename: Output filename
format: Format (auto-detected if None)
**kwargs: Format-specific options
"""
def trim(self, starttime=None, endtime=None, **kwargs) -> 'Trace':
"""
Trim trace to time window.
Args:
starttime: Start time (UTCDateTime or None)
endtime: End time (UTCDateTime or None)
**kwargs: Additional trim options
Returns:
Modified trace (in-place)
"""
def slice(self, starttime=None, endtime=None) -> 'Trace':
"""
Return new trace with data sliced to time window.
Args:
starttime: Start time (UTCDateTime or None)
endtime: End time (UTCDateTime or None)
Returns:
New Trace object with sliced data
"""
def filter(self, type: str, **options) -> 'Trace':
"""
Apply digital filter to trace data.
Args:
type: Filter type ('bandpass', 'lowpass', 'highpass', etc.)
**options: Filter-specific parameters (frequencies, etc.)
Returns:
Modified trace (in-place)
"""
def trigger(self, type: str, **options):
"""
Apply triggering algorithm to trace data.
Args:
type: Trigger type ('recstalta', 'classicstalta', etc.)
**options: Trigger-specific parameters
Returns:
Trigger function or onset times
"""
def resample(self, sampling_rate: float, **kwargs) -> 'Trace':
"""
Resample trace to new sampling rate.
Args:
sampling_rate: Target sampling rate in Hz
**kwargs: Resampling options
Returns:
Modified trace (in-place)
"""
def decimate(self, factor: int, **kwargs) -> 'Trace':
"""
Decimate trace by integer factor.
Args:
factor: Decimation factor
**kwargs: Decimation options
Returns:
Modified trace (in-place)
"""
def simulate(self, paz_remove=None, paz_simulate=None, **kwargs) -> 'Trace':
"""
Simulate instrument response.
Args:
paz_remove: Poles and zeros to remove
paz_simulate: Poles and zeros to simulate
**kwargs: Simulation options
Returns:
Modified trace (in-place)
"""
def detrend(self, type: str = 'linear', **kwargs) -> 'Trace':
"""
Remove trend from trace data.
Args:
type: Detrend type ('linear', 'constant', 'polynomial', etc.)
**kwargs: Detrend-specific options
Returns:
Modified trace (in-place)
"""
def taper(self, max_percentage: float = 0.05, type: str = 'hann', **kwargs) -> 'Trace':
"""
Apply taper to trace data.
Args:
max_percentage: Taper length as fraction of trace
type: Taper type ('hann', 'cosine', etc.)
**kwargs: Taper options
Returns:
Modified trace (in-place)
"""
def normalize(self, norm: float = None) -> 'Trace':
"""
Normalize trace data.
Args:
norm: Normalization value (max absolute value if None)
Returns:
Modified trace (in-place)
"""
def differentiate(self, method: str = 'gradient') -> 'Trace':
"""
Differentiate trace data.
Args:
method: Differentiation method
Returns:
Modified trace (in-place)
"""
def integrate(self, method: str = 'cumtrapz') -> 'Trace':
"""
Integrate trace data.
Args:
method: Integration method
Returns:
Modified trace (in-place)
"""
def attach_response(self, response):
"""
Attach instrument response to trace.
Args:
response: Response object or filename
"""
def remove_response(self, **kwargs) -> 'Trace':
"""
Remove instrument response from trace data.
Args:
**kwargs: Response removal options
Returns:
Modified trace (in-place)
"""
def remove_sensitivity(self, **kwargs) -> 'Trace':
"""
Remove instrument sensitivity from trace data.
Args:
**kwargs: Sensitivity removal options
Returns:
Modified trace (in-place)
"""
def copy(self) -> 'Trace':
"""Return deep copy of trace."""
def split(self) -> list['Trace']:
"""Split trace at gaps, returning list of traces."""
def times(self, type: str = 'matplotlib') -> np.ndarray:
"""
Generate time array for trace data.
Args:
type: Time array type ('matplotlib', 'utcdatetime', 'timestamp')
Returns:
NumPy array of time values
"""
def max(self) -> float:
"""Return maximum absolute value of trace data."""
def std(self) -> float:
"""Return standard deviation of trace data."""
def verify(self) -> bool:
"""Verify trace integrity and consistency."""Stats Class
Metadata container for trace information extending dictionary functionality with attribute access and seismological standards compliance.
class Stats(AttribDict):
def __init__(self, header=None):
"""
Create trace metadata container.
Args:
header: Dictionary of metadata values
"""
# Standard SEED metadata attributes
sampling_rate: float # Sampling rate in Hz
delta: float # Sample interval in seconds
npts: int # Number of data points
network: str # Network code (SEED standard)
station: str # Station code (SEED standard)
location: str # Location code (SEED standard)
channel: str # Channel code (SEED standard)
starttime: UTCDateTime # Start time of trace
endtime: UTCDateTime # End time of trace
calib: float # Calibration factorStream Class
Collection of Trace objects with ensemble processing capabilities for multi-channel and multi-station seismological data analysis.
class Stream:
def __init__(self, traces=None):
"""
Create stream from traces.
Args:
traces: List of Trace objects or None for empty stream
"""
def __len__(self) -> int:
"""Return number of traces in stream."""
def __getitem__(self, index) -> Trace:
"""Get trace by index or slice."""
def __setitem__(self, index, trace: Trace):
"""Set trace at index."""
def append(self, trace: Trace):
"""Add trace to end of stream."""
def extend(self, trace_list: list[Trace]):
"""Extend stream with list of traces."""
def insert(self, index: int, trace: Trace):
"""Insert trace at specified index."""
def remove(self, trace: Trace):
"""Remove trace from stream."""
def pop(self, index: int = -1) -> Trace:
"""Remove and return trace at index."""
def reverse(self):
"""Reverse order of traces in stream."""
def sort(self, keys: list[str] = ['network', 'station', 'location', 'channel', 'starttime']):
"""
Sort traces by metadata keys.
Args:
keys: List of metadata keys for sorting priority
"""
def select(self, **kwargs) -> 'Stream':
"""
Select traces matching criteria.
Args:
**kwargs: Selection criteria (network, station, channel, etc.)
Returns:
New Stream with matching traces
"""
def plot(self, **kwargs):
"""
Plot all traces in stream.
Args:
**kwargs: Plotting options
"""
def spectrogram(self, **kwargs):
"""
Plot spectrograms of all traces.
Args:
**kwargs: Spectrogram parameters
"""
def write(self, filename: str, format: str = None, **kwargs):
"""
Write stream to file.
Args:
filename: Output filename
format: Format (auto-detected if None)
**kwargs: Format-specific options
"""
def trim(self, starttime=None, endtime=None, **kwargs) -> 'Stream':
"""
Trim all traces to time window.
Args:
starttime: Start time (UTCDateTime or None)
endtime: End time (UTCDateTime or None)
**kwargs: Trim options
Returns:
Modified stream (in-place)
"""
def cutout(self, starttime, endtime) -> 'Stream':
"""
Remove time window from all traces.
Args:
starttime: Start time of cutout window
endtime: End time of cutout window
Returns:
Modified stream (in-place)
"""
def slice(self, starttime=None, endtime=None) -> 'Stream':
"""
Return new stream with traces sliced to time window.
Args:
starttime: Start time (UTCDateTime or None)
endtime: End time (UTCDateTime or None)
Returns:
New Stream with sliced traces
"""
def slide(self, window_length: float, step: float, **kwargs):
"""
Generator yielding sliding time windows.
Args:
window_length: Window length in seconds
step: Step size in seconds
**kwargs: Additional options
Yields:
Stream objects for each time window
"""
def merge(self, method: int = -1, interpolation_samples: int = -1, **kwargs) -> 'Stream':
"""
Merge overlapping or adjacent traces.
Args:
method: Merge method (-1=auto, 0=sum, 1=overlap)
interpolation_samples: Interpolation gap limit
**kwargs: Merge options
Returns:
Modified stream (in-place)
"""
def get_gaps(self) -> list:
"""
Get list of gaps between traces.
Returns:
List of gap information tuples
"""
def print_gaps(self):
"""Print gap information to stdout."""
def verify(self) -> bool:
"""Verify stream integrity and trace consistency."""
def filter(self, type: str, **options) -> 'Stream':
"""
Apply filter to all traces.
Args:
type: Filter type
**options: Filter parameters
Returns:
Modified stream (in-place)
"""
def trigger(self, type: str, **options):
"""
Apply trigger to all traces.
Args:
type: Trigger type
**options: Trigger parameters
Returns:
Trigger results for all traces
"""
def resample(self, sampling_rate: float, **kwargs) -> 'Stream':
"""
Resample all traces.
Args:
sampling_rate: Target sampling rate
**kwargs: Resampling options
Returns:
Modified stream (in-place)
"""
def decimate(self, factor: int, **kwargs) -> 'Stream':
"""
Decimate all traces.
Args:
factor: Decimation factor
**kwargs: Decimation options
Returns:
Modified stream (in-place)
"""
def simulate(self, **kwargs) -> 'Stream':
"""
Simulate instrument response for all traces.
Args:
**kwargs: Simulation options
Returns:
Modified stream (in-place)
"""
def detrend(self, type: str = 'linear', **kwargs) -> 'Stream':
"""
Detrend all traces.
Args:
type: Detrend type
**kwargs: Detrend options
Returns:
Modified stream (in-place)
"""
def taper(self, max_percentage: float = 0.05, **kwargs) -> 'Stream':
"""
Taper all traces.
Args:
max_percentage: Taper length fraction
**kwargs: Taper options
Returns:
Modified stream (in-place)
"""
def normalize(self, **kwargs) -> 'Stream':
"""
Normalize all traces.
Args:
**kwargs: Normalization options
Returns:
Modified stream (in-place)
"""
def differentiate(self, **kwargs) -> 'Stream':
"""
Differentiate all traces.
Args:
**kwargs: Differentiation options
Returns:
Modified stream (in-place)
"""
def integrate(self, **kwargs) -> 'Stream':
"""
Integrate all traces.
Args:
**kwargs: Integration options
Returns:
Modified stream (in-place)
"""
def max(self) -> float:
"""Return maximum absolute value across all traces."""
def std(self) -> float:
"""Return standard deviation across all traces."""
def rotate(self, method: str, **kwargs) -> 'Stream':
"""
Rotate coordinate components.
Args:
method: Rotation method ('NE->RT', 'ZNE->LQT', etc.)
**kwargs: Rotation parameters (back_azimuth, inclination, etc.)
Returns:
Modified stream (in-place)
"""
def attach_response(self, response):
"""
Attach instrument response to all traces.
Args:
response: Response object or inventory
"""
def remove_response(self, **kwargs) -> 'Stream':
"""
Remove instrument response from all traces.
Args:
**kwargs: Response removal options
Returns:
Modified stream (in-place)
"""
def remove_sensitivity(self, **kwargs) -> 'Stream':
"""
Remove sensitivity from all traces.
Args:
**kwargs: Sensitivity removal options
Returns:
Modified stream (in-place)
"""
def stack(self, **kwargs) -> list[Trace]:
"""
Stack traces by metadata groups.
Args:
**kwargs: Stacking options
Returns:
List of stacked traces
"""
def copy(self, deep: bool = True) -> 'Stream':
"""
Return copy of stream.
Args:
deep: Create deep copy if True
Returns:
Copied Stream object
"""
def clear(self):
"""Remove all traces from stream."""
def split(self) -> 'Stream':
"""
Split traces at gaps.
Returns:
Modified stream with split traces (in-place)
"""Universal Read Functions
Format-agnostic functions for reading seismological data files with automatic format detection and unified interfaces.
def read(pathname_or_url, format=None, headonly=False, starttime=None,
endtime=None, nearest_sample=True, dtype=None, apply_calib=False,
check_compression=True, **kwargs) -> Stream:
"""
Read waveform files into Stream object.
Args:
pathname_or_url: File path, URL, or file-like object
format: Format hint (auto-detected if None)
headonly: Read metadata only, skip data
starttime: Start time for reading window
endtime: End time for reading window
nearest_sample: Align times to nearest sample
dtype: Data type for conversion
apply_calib: Apply calibration factor
check_compression: Verify compressed data integrity
**kwargs: Format-specific options
Returns:
Stream object containing traces
Supported formats: MiniSEED, SAC, GSE2, SEG-Y, WIN, CSS, SEISAN,
AH, WAV, GCF, RefTek, PDAS, Y, SEG-2, SH,
Kinemetrics, NIED, RG16, DMX, ALSEP, and others
"""
def read_events(pathname_or_url, format=None, **kwargs):
"""
Read earthquake event files into Catalog object.
Args:
pathname_or_url: File path, URL, or file-like object
format: Format hint (auto-detected if None)
**kwargs: Format-specific options
Returns:
Catalog object containing events
Supported formats: QuakeML, NDK, CMTSOLUTION, Nordic, NonLinLoc,
SC3ML, ZMAP, JSON, MCHEDR, CNV, FOCMEC,
HypoDD, SCARDEC, GSE2, IMS1.0, and others
"""
def read_inventory(pathname_or_url, format=None, **kwargs):
"""
Read station metadata files into Inventory object.
Args:
pathname_or_url: File path, URL, or file-like object
format: Format hint (auto-detected if None)
**kwargs: Format-specific options
Returns:
Inventory object containing station metadata
Supported formats: StationXML, SEED/XSEED, SACPZ, CSS,
Station text, SC3ML, ArcLink XML, and others
"""Usage Examples
Basic Trace Operations
from obspy import Trace, UTCDateTime
import numpy as np
# Create trace with synthetic data
data = np.random.randn(1000)
trace = Trace(data=data)
trace.stats.sampling_rate = 100.0
trace.stats.starttime = UTCDateTime("2023-01-01T00:00:00")
trace.stats.network = "XX"
trace.stats.station = "TEST"
trace.stats.channel = "HHZ"
# Basic processing
trace.detrend('linear')
trace.filter('bandpass', freqmin=1.0, freqmax=10.0)
trace.taper(0.05)
# Get trace information
print(f"Trace ID: {trace.id}")
print(f"Duration: {trace.stats.endtime - trace.stats.starttime} seconds")
print(f"Max amplitude: {trace.max()}")Stream Processing Workflow
from obspy import read
# Read seismic data file
st = read('seismic_data.mseed')
# Pre-processing
st.detrend('linear')
st.taper(0.05)
st.filter('bandpass', freqmin=1.0, freqmax=20.0)
# Merge overlapping traces and remove gaps
st.merge(method=1, interpolation_samples=10)
# Select specific components
st_z = st.select(channel="*Z") # Vertical components
st_n = st.select(channel="*N") # North components
st_e = st.select(channel="*E") # East components
# Rotate horizontal components to radial-transverse
st_horizontal = st_n + st_e # Combine N and E components
st_horizontal.rotate('NE->RT', back_azimuth=45.0)
# Time window extraction
event_time = UTCDateTime("2023-01-15T10:30:00")
st_event = st.slice(event_time, event_time + 300) # 5-minute windowTypes
class AttribDict(dict):
"""Dictionary with attribute-style access to keys."""
def __init__(self, *args, **kwargs):
"""Initialize with dictionary data."""
def __getattr__(self, name):
"""Get dictionary value as attribute."""
def __setattr__(self, name, value):
"""Set dictionary value as attribute."""
def __delattr__(self, name):
"""Delete dictionary key as attribute."""Install with Tessl CLI
npx tessl i tessl/pypi-obspy