Examples¶
This section contains working examples demonstrating various pymseed use cases. The example scripts are available in the examples directory of the repository.
Reading Examples¶
read_traces.py¶
Read miniSEED files and display trace information including channel ID, start/end times, and sample rate for each trace segment.
examples/read_traces.py¶
#!/usr/bin/env python3
"""
Read miniSEED files and display trace information.
Displays channel ID, start/end times, and sample rate for each trace segment.
Example:
> python read_traces.py example_data.mseed
This file is part of the pymseed package.
Copyright (c) 2026, EarthScope Data Services
"""
import argparse
import sys
from pathlib import Path
from pymseed import MS3TraceList, SubSecond
def read_traces(filename: str) -> None:
"""Read and display trace information from a miniSEED file."""
try:
traces = MS3TraceList.from_file(filename)
except Exception as e:
print(f"Error reading {filename}: {e}", file=sys.stderr)
return
# Print header
print(f"\nFile: {filename}")
print(f"{'Channel ID':<26} {'Start Time':<30} {'End Time':<30} {'Sample Rate'}")
print("-" * 95)
# Print trace information
for trace in traces:
for segment in trace:
start_time = segment.starttime_str(subsecond=SubSecond.NANO_MICRO)
end_time = segment.endtime_str(subsecond=SubSecond.NANO_MICRO)
print(f"{trace.sourceid:<26} {start_time:<30} {end_time:<30} {segment.samprate}")
def main():
"""Main function to parse arguments and process files."""
parser = argparse.ArgumentParser(
description="Read miniSEED files and display trace information"
)
parser.add_argument("input_files", nargs="+", help="One or more miniSEED files to read")
args = parser.parse_args()
# Validate files exist
for filename in args.input_files:
if not Path(filename).exists():
print(f"Error: File '{filename}' not found", file=sys.stderr)
sys.exit(1)
# Process each file
for filename in args.input_files:
read_traces(filename)
if __name__ == "__main__":
main()
read_numpy.py¶
Read miniSEED files and access data samples as NumPy arrays. Demonstrates how to extract data without duplicating memory by reading files twice - once to create the trace list and once to extract samples directly into pre-allocated NumPy arrays.
examples/read_numpy.py¶
#!/usr/bin/env python3
"""
Read miniSEED files and assemble independent traces using NumPy arrays.
This example demonstrates how to:
- Read miniSEED files using pymseed and create a trace list
- Extract data samples as NumPy arrays without copying data
- Access basic trace metadata
The result is a collection of trace data with no dependency
on any pymseed data structures.
Usage:
> python read_numpy.py [file1.mseed] [file2.mseed] ...
This file is part of the pymseed package.
Copyright (c) 2026, EarthScope Data Services
"""
import argparse
import sys
import numpy as np
from pymseed import MS3TraceList, sourceid2nslc
def read_traces(input_files):
"""Read miniSEED files and return list of trace data with NumPy arrays."""
trace_data = []
traces = MS3TraceList()
# Read all files
for filename in input_files:
print(f"Reading: {filename}")
try:
traces.add_file(filename, unpack_data=True)
except Exception as e:
print(f"Warning: Could not read {filename}: {e}")
continue
# Extract data for each trace segment
for trace_id in traces:
for segment in trace_id:
try:
# Take ownership of the data sample array
data_array = segment.take_np_datasamples()
# Organize trace information
trace_entry = {
"source_id": trace_id.sourceid,
"network_station_location_channel": sourceid2nslc(trace_id.sourceid),
"start_time": segment.starttime_str(),
"end_time": segment.endtime_str(),
"sample_rate_hz": segment.samprate,
"num_samples": len(data_array),
"data_samples": data_array,
}
trace_data.append(trace_entry)
except Exception as e:
print(f"Warning: Could not process segment for {trace_id.sourceid}: {e}")
continue
return trace_data
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="Read miniSEED files and assemble independent traces"
)
parser.add_argument("files", nargs="*", help="miniSEED files to read")
args = parser.parse_args()
# Check if files were provided
if not args.files:
parser.print_help()
sys.exit(1)
input_files = args.files
trace_data = read_traces(input_files)
if not trace_data:
sys.exit("No trace data found")
# Display trace information
print(f"\nFound {len(trace_data)} trace segments:")
print("-" * 80)
for trace in trace_data:
nslc = trace["network_station_location_channel"]
data = trace["data_samples"]
print(f"Trace {trace['source_id']}, NSLC: {nslc[0]}.{nslc[1]}.{nslc[2]}.{nslc[3]}")
print(f" Time: {trace['start_time']} to {trace['end_time']}")
print(f" Sample rate: {trace['sample_rate_hz']} Hz")
print(f" Samples: {trace['num_samples']:,}")
# Basic NumPy statistics
print(f" Data range: {np.min(data):.2f} to {np.max(data):.2f}")
print(f" Mean: {np.mean(data):.2f}, Std: {np.std(data):.2f}")
print()
read_selected_unpack.py¶
Read miniSEED files without unpacking data samples, then selectively unpack only the segments of interest into NumPy arrays. This limits the amount of data held in memory at any one time when processing large volumes.
examples/read_selected_unpack.py¶
#!/usr/bin/env python3
"""
Read miniSEED files and selectively unpack data samples.
This example demonstrates how to selectively unpack and processes
data samples.
When processing a large volume of data this strategy allows you
to limit the amount of data read into memory at any one time.
The strategy requires reading the data twice, with the first read
summarizing the data and the second read unpacking the data into
NumPy arrays when desired.
Usage:
> python read_selected_unpack.py [file1.mseed] [file2.mseed] ...
This file is part of the pymseed package.
Copyright (c) 2026, EarthScope Data Services
"""
import argparse
import sys
import numpy as np
from pymseed import MS3TraceList
def process_data(trace, segment, data_array):
"""Calculate min, max, and mean of the data array."""
if len(data_array) == 0:
raise ValueError(f"Data array is empty for segment {segment.sourceid}")
min_value = np.min(data_array)
max_value = np.max(data_array)
mean_value = np.mean(data_array)
return min_value, max_value, mean_value
if __name__ == "__main__":
# Simple argparse setup
parser = argparse.ArgumentParser(description="Read miniSEED files and convert to NumPy arrays")
parser.add_argument("files", nargs="*", help="miniSEED files to read")
args = parser.parse_args()
# Check if files were provided
if not args.files:
parser.print_help()
sys.exit(1)
input_files = args.files
# Read all files, explicitly not unpacking data samples and creating a record list
traces = MS3TraceList()
for filename in input_files:
print(f"Reading: {filename}")
try:
traces.add_file(filename, unpack_data=False, record_list=True)
except Exception as e:
print(f"Warning: Could not read {filename}: {e}")
continue
# Unpack and process each trace segment independently
for trace in traces:
for segment in trace:
# Skip segments with less than 100 samples
if segment.samplecnt < 100:
print(f"Skipping segment {trace.sourceid} with {segment.samplecnt} samples")
continue
# Unpack the data samples into a NumPy array
data_array = segment.create_numpy_array_from_recordlist()
# Process the data
try:
min_value, max_value, mean_value = process_data(trace, segment, data_array)
except Exception as e:
print(f"Warning: Could not process segment for {trace.sourceid}: {e}")
continue
# Report the processed results
print(f"Trace {trace.sourceid}")
print(f" Time: {segment.starttime_str()} to {segment.endtime_str()}")
print(f" Sample rate: {segment.samprate} Hz")
print(f" Samples: {len(data_array)}")
print(f" Data range: {min_value:.2f} to {max_value:.2f}")
print(f" Mean: {mean_value:.2f}")
print()
Writing Examples¶
generate_with_buffer.py¶
Illustrates how to write miniSEED using a rolling buffer of potentially multi-channel data. This pattern is useful for:
Generating miniSEED from continuous streams of unknown duration (real-time)
Processing large volumes of data while avoiding memory issues
examples/generate_with_buffer.py¶
#!/usr/bin/env python3
"""
This example illustrates how to write miniSEED using a rolling
buffer of potentially multi-channel data.
This pattern of usage is particularly useful for applications that need to:
a) generate miniSEED in a continuous stream for an unknown (long) duration,
i.e. real-time streams. See the `continuous_miniseed_creation.py` example.
b) generate miniSEED from a large volume of data while avoiding the need to
have it all in memory.
In this example, a sine wave generator is used to create synthetic data for 3
channels by producing 100 samples at a time.
The general pattern is (writing bytes to a file for example):
```python
traces = MS3TraceList() # Create an empty MS3TraceList object
Loop on input data:
traces.add_data() # Add data to the MS3TraceList object
# Generate filled records during regular data flow
for record in traces.generate(flush_data=False,
remove_packed=True):
# Write the record (bytes) to the output file
output_file.write(record)
# Flush any data remaining in the buffers
for record in traces.generate(flush_data=True,
remove_packed=True):
output_file.write(record)
```
This file is part of the the Python pymseed package.
Copyright (c) 2026, EarthScope Data Services
"""
import math
from pymseed import MS3TraceList, sample_time, timestr2nstime
output_file = "output.mseed"
def sine_generator(start_degree=0, yield_count=100, total=1000):
"""A generator returning a continuing sequence for a sine values."""
generated = 0
while generated < total:
bite_size = min(yield_count, total - generated)
# Yield a list of continuing sine values
yield [
int(math.sin(math.radians(x)) * 500)
for x in range(start_degree, start_degree + bite_size)
]
start_degree += bite_size
generated += bite_size
# Define 3 generators with offset starting degrees
generate_yield_count = 100
sine0 = sine_generator(start_degree=0, yield_count=generate_yield_count)
sine1 = sine_generator(start_degree=45, yield_count=generate_yield_count)
sine2 = sine_generator(start_degree=90, yield_count=generate_yield_count)
output_file = open(output_file, "wb")
traces = MS3TraceList()
total_records = 0
sample_rate = 40.0
starttime = timestr2nstime("2024-01-01T15:13:55.123456789Z")
format_version = 2
max_record_length = 512
# A loop that iteratively adds data to traces in the list.
#
# This could be any data collection operation that continually
# adds samples to the trace list.
for _ in range(10):
# Add new synthetic data to each trace using generators
traces.add_data(
sourceid="FDSN:XX_TEST__B_S_0",
data_samples=next(sine0),
sample_type="i",
sample_rate=sample_rate,
starttime=starttime,
)
traces.add_data(
sourceid="FDSN:XX_TEST__B_S_1",
data_samples=next(sine1),
sample_type="i",
sample_rate=sample_rate,
starttime=starttime,
)
traces.add_data(
sourceid="FDSN:XX_TEST__B_S_2",
data_samples=next(sine2),
sample_type="i",
sample_rate=sample_rate,
starttime=starttime,
)
# Update the start time for the next iteration of synthetic data
starttime = sample_time(starttime, generate_yield_count, sample_rate)
# Generate full records and do not flush the data buffers
for record in traces.generate(
format_version=format_version,
max_record_length=max_record_length,
flush_data=False,
remove_packed=True,
):
output_file.write(record)
total_records += 1
# Flush the data buffers and write any data to records
for record in traces.generate(
format_version=format_version,
max_record_length=max_record_length,
flush_data=True,
):
output_file.write(record)
total_records += 1
output_file.close()
print(f"Packed {total_records} records")
continuous_miniseed_creation.py¶
Generate miniSEED from a continuous stream of data using MS3TraceList as a
transient buffer. Demonstrates using MS3TraceList.generate() to continuously
generate miniSEED output with full records whenever possible.
examples/continuous_miniseed_creation.py¶
#!/usr/bin/env python3
"""
Generate miniSEED from a continuous stream of data using a rolling buffer.
This program illustrates the use of an MS3TraceList to function as a temporary,
or transient, data buffer for a continuous stream of data, and using
MS3TraceList.generate() to continuously generate miniSEED output.
This pattern is useful for generating miniSEED output from a continuous stream
of data from any arbitrary source in a manner that creates full miniSEED records
as much as possible. Data are either packed into records when a full record is
possible, when the data are idle for a specified number of seconds, or when the
program shuts down.
Usage: python continuous_miniseed_creation.py <output_file>
The output file will be continuously updated with new data.
For this example, a simple sine wave with is generated and used as the data
source.
Example usage:
> python continuous_miniseed_creation.py output.mseed
This file is part of the pymseed package.
Copyright (c) 2026, EarthScope Data Services
"""
import argparse
import math
import signal
import threading
from collections.abc import Generator
from typing import Any
from pymseed import NSTMODULUS, DataEncoding, MS3TraceList, system_time
# Global flag and event for handling shutdown signals
shutdown_requested = False
shutdown_event = threading.Event()
def signal_handler(signum: int, frame: Any) -> None:
"""Handle shutdown signals (SIGTERM, SIGINT) by setting shutdown flag and event."""
global shutdown_requested
print(f"\nReceived signal {signum}, initiating shutdown...")
shutdown_requested = True
shutdown_event.set() # Wake up any waiting threads immediately
def data_source(
start_time: int = None,
degree_offset: float = 0.0,
sample_rate: float = 100.0,
amplitude: float = 1.0,
integer_samples: bool = False,
) -> Generator[tuple[list[float], int]]:
"""
Generate a 1 Hertz sinusoidal signal for a time series starting at the
specified start time and degree offset with the specified sample rate. The
generator yields a list of new samples and the time of the first
sample in the list (as a nanosecond timestamp).
The number of samples returned is the number needed to fill the series since
the last invocation.
The start_time value is the time in nanosecond since the Unix epoch
(nstime_t in libmseed), such as returned by pymseed.system_time() or
pymseed.timestr2nstime(). If it is None, the current time is used.
The degree_offset value is the phase offset in degrees, allowing for
distinct sinusoids to be generated.
The amplitude value is the maximum amplitude of the sinusoid. The sinusoid
will range from -amplitude to +amplitude. Default is 1.0.
The integer_samples value is a boolean indicating whether the samples should
be integers or floats. Default is False (float samples).
"""
if start_time is None:
start_time = system_time()
# Calculate the sample interval in nanoseconds
sample_interval_ns = int(NSTMODULUS / sample_rate)
# Convert degree offset to radians
phase_offset = math.radians(degree_offset)
# Keep track of the next sample time
next_sample_time = start_time
while True:
current_time = system_time()
# Generate samples until we catch up to current time
samples = []
first_sample_time = next_sample_time
while next_sample_time <= current_time:
# Calculate time for this sample in seconds
sample_time_sec = (next_sample_time - start_time) / NSTMODULUS
# Generate sinusoid sample
sample_value = amplitude * math.sin(2 * math.pi * sample_time_sec + phase_offset)
samples.append(int(sample_value) if integer_samples else sample_value)
# Advance to next sample time
next_sample_time += sample_interval_ns
yield (samples, first_sample_time)
def create_continuous_miniseed(
output_file: str,
flush_idle_seconds: int,
record_length: int,
encoding: DataEncoding,
verbose: int = 0,
) -> None:
"""
Create a miniSEED file from a continuous stream of data.
Generates sinusoid data once per second for two source IDs until terminated
by SIGTERM or Control-C.
The data is written to the specified output file.
"""
global shutdown_requested
# Set up signal handlers for graceful shutdown
signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGINT, signal_handler)
# Parameters for the data series
sourceid_1 = "FDSN:XX_STA__H_X_1"
sourceid_2 = "FDSN:XX_STA__B_X_2"
sample_rate1 = 100.0
sample_rate2 = 40.0
# Determine sample type from encoding
sample_type = "f" if encoding in [DataEncoding.FLOAT32, DataEncoding.FLOAT64] else "i"
# Create data source generators for sinusoids with a 45 degree phase offset,
# +/-100 amplitude range, and integer or float samples depending on the encoding
sinusoid1 = data_source(
degree_offset=0, sample_rate=sample_rate1, amplitude=100, integer_samples=sample_type == "i"
)
sinusoid2 = data_source(
degree_offset=45,
sample_rate=sample_rate2,
amplitude=100,
integer_samples=sample_type == "i",
)
# Create a trace buffer to function as a temporary data buffer
# holding data until full-length records can be created or flushed
# when idle.
trace_buffer = MS3TraceList()
# Print some information about the data series
print(f"Starting continuous miniSEED creation to {output_file}")
print(f"Flush idle time: {flush_idle_seconds} seconds")
print(f"Record length: {record_length} bytes")
print(f"Encoding: {encoding.name}")
print(f"Verbosity level: {verbose}")
print("Press Control-C to stop...\n")
# Open the output file for writing
with open(output_file, "wb") as output_handle:
loop_count = 0
while not shutdown_requested:
try:
# Get new data from generators, could be an arbitrary source
new_data1, starttime1 = next(sinusoid1)
# Only generate data for sourceid_2 every 5th loop (every 5 seconds)
if loop_count % 5 == 0:
new_data2, starttime2 = next(sinusoid2)
else:
new_data2 = []
# Add any new data to trace buffer for sourceid_1
if new_data1:
trace_buffer.add_data(
sourceid=sourceid_1,
data_samples=new_data1,
sample_type=sample_type,
sample_rate=sample_rate1,
starttime=starttime1,
)
if verbose > 1:
print(f"Added {len(new_data1)} samples to {sourceid_1}")
# Add any new data to trace buffer for sourceid_2
if new_data2:
trace_buffer.add_data(
sourceid=sourceid_2,
data_samples=new_data2,
sample_type=sample_type,
sample_rate=sample_rate2,
starttime=starttime2,
)
if verbose > 1:
print(f"Added {len(new_data2)} samples to {sourceid_2}")
elif verbose > 2 and loop_count % 5 != 0:
print(f"Skipped data generation for {sourceid_2} (not 5th loop)")
# Pack traces with idle data flushing, but don't flush all data yet
packed_records = 0
for record in trace_buffer.generate(
max_record_length=record_length,
encoding=encoding,
flush_data=False,
flush_idle_seconds=flush_idle_seconds,
remove_packed=True,
):
output_handle.write(record)
packed_records += 1
if packed_records > 0 and verbose > 0:
print(f"Wrote {packed_records} records")
loop_count += 1
# Wait for 1 second or until shutdown is requested (interruptible)
# This allows the program to be terminated by SIGTERM/Control-C
shutdown_event.wait(1)
except KeyboardInterrupt:
# Handle Control-C gracefully
shutdown_requested = True
shutdown_event.set()
break
# Perform final packing with flush_data=True to ensure all data is written
print("\nPerforming final data creation to flush remaining data...")
packed_records = 0
for record in trace_buffer.generate(
max_record_length=record_length,
encoding=encoding,
flush_data=True,
remove_packed=True,
):
output_handle.write(record)
packed_records += 1
print(f"Final flush: wrote {packed_records} records")
print(f"miniSEED creation complete. Output written to {output_file}")
def main():
parser = argparse.ArgumentParser(
description="Generate miniSEED output from a continuous stream of data until the program is terminated"
)
parser.add_argument("output_file", help="The file to write the miniSEED output to")
parser.add_argument(
"-f",
"--flush_idle_seconds",
type=int,
default=10,
help="The number of seconds of idle time before flushing the data",
)
parser.add_argument(
"-r",
"--record_length",
type=int,
default=512,
help="Record length in bytes for miniSEED output (default: 512)",
)
parser.add_argument(
"-e",
"--encoding",
type=str,
choices=["STEIM1", "STEIM2", "FLOAT32", "FLOAT64", "INT16", "INT32"],
default="STEIM1",
help="Data encoding format for miniSEED output (default: STEIM1)",
)
parser.add_argument(
"-v",
"--verbose",
action="count",
default=0,
help="Increase verbosity level (can be used multiple times: -v, -vv, -vvv)",
)
args = parser.parse_args()
# Convert encoding string to DataEncoding enum
encoding_map = {
"STEIM1": DataEncoding.STEIM1,
"STEIM2": DataEncoding.STEIM2,
"FLOAT32": DataEncoding.FLOAT32,
"FLOAT64": DataEncoding.FLOAT64,
"INT16": DataEncoding.INT16,
"INT32": DataEncoding.INT32,
}
encoding = encoding_map[args.encoding.upper()]
# Create continuous miniSEED output until the program is terminated
create_continuous_miniseed(
args.output_file, args.flush_idle_seconds, args.record_length, encoding, args.verbose
)
if __name__ == "__main__":
main()
Streaming Examples¶
stream_stats.py¶
Read miniSEED from stdin, accumulate statistics (record count, sample count, bytes, per-source timing), write to stdout, and print statistics to stderr on completion.
examples/stream_stats.py¶
#!/usr/bin/env python3
"""
Read miniSEED file(s) from a stream, accumulate stats, and write to a stream.
For this illustration input is stdin, output is stdout, and stats are printed
to stderr on completion.
Example usage:
cat example_data.mseed | stream_stats.py > output.mseed
This file is part of the pymseed package.
Copyright (c) 2026, EarthScope Data Services
"""
import pprint
import sys
from pymseed import MS3Record, nstime2timestr
class StreamStats:
"""Accumulate statistics from a stream of miniSEED records."""
def __init__(self):
self.record_count = 0
self.sample_count = 0
self.bytes = 0
self.sourceids = {} # Per-sourceid statistics
def __str__(self):
"""Return a string representation of the statistics."""
printer = pprint.PrettyPrinter(indent=4, sort_dicts=False)
return printer.pformat(self.to_dict())
def to_dict(self):
"""Return a dict representation of the statistics."""
# Create copy to avoid modifying original
sourceids_copy = {}
for sid, stats in self.sourceids.items():
sid_copy = stats.copy()
if sid_copy["earliest"]:
sid_copy["earliest_str"] = nstime2timestr(sid_copy["earliest"])
if sid_copy["latest"]:
sid_copy["latest_str"] = nstime2timestr(sid_copy["latest"])
sourceids_copy[sid] = sid_copy
return {
"record_count": self.record_count,
"sample_count": self.sample_count,
"bytes": self.bytes,
"sourceids": sourceids_copy,
}
def update(self, msr):
"""Update statistics with data from a miniSEED record."""
# Update global statistics
self.record_count += 1
self.sample_count += msr.samplecnt
self.bytes += msr.reclen
# Track per-sourceid statistics
sid = msr.sourceid
if sid not in self.sourceids:
self.sourceids[sid] = {
"record_count": 0,
"sample_count": 0,
"bytes": 0,
"earliest": None,
"latest": None,
}
sid_stats = self.sourceids[sid]
sid_stats["record_count"] += 1
sid_stats["sample_count"] += msr.samplecnt
sid_stats["bytes"] += msr.reclen
if sid_stats["earliest"] is None or msr.starttime < sid_stats["earliest"]:
sid_stats["earliest"] = msr.starttime
if sid_stats["latest"] is None or msr.endtime > sid_stats["latest"]:
sid_stats["latest"] = msr.endtime
def main():
"""Main processing function."""
print("Reading miniSEED from stdin, writing to stdout", file=sys.stderr)
# Read miniSEED from stdin and accumulate stats for each record
stats = StreamStats()
for msr in MS3Record.from_file(sys.stdin.fileno()):
# Update statistics with data from the record
stats.update(msr)
# Write raw miniSEED record to stdout
sys.stdout.buffer.write(msr.record)
print(stats, file=sys.stderr)
if __name__ == "__main__":
main()
stream_timewindow.py¶
Read miniSEED from stdin, select records that fall within specified earliest and latest times, trim records at window boundaries, and write to stdout.
examples/stream_timewindow.py¶
#!/usr/bin/env python3
"""
Read miniSEED file(s) from a stream (stdin), select those that
fall within the selected earliest and latest times, and write out
to a stream (stdout). Records that contain the selected times are
trimmed to the selected times.
Example usage:
> stream_timewindow.py --earliest 2010-02-27T07:00:00 --latest 2010-02-27T07:10:00 < example_data.mseed > windowed.mseed
This file is part of the pymseed package.
Copyright (c) 2026, EarthScope Data Services
"""
import argparse
import sys
from pymseed import NSTMODULUS, MS3Record, timestr2nstime
def process_stream(args):
"""Process miniSEED records from stdin, applying time window selection."""
records_written = 0
bytes_written = 0
print("Reading miniSEED from stdin, writing to stdout", file=sys.stderr)
# Read miniSEED from stdin
for msr in MS3Record.from_file(sys.stdin.fileno()):
# Skip records completely outside the time window
if (args.earliest and msr.endtime < args.earliest) or (
args.latest and msr.starttime > args.latest
):
continue
# Trim if record overlaps with time window boundaries
output_record = msr.record
if (args.earliest and msr.starttime < args.earliest <= msr.endtime) or (
args.latest and msr.starttime <= args.latest < msr.endtime
):
trimmed_record = trim_record(msr, args.earliest, args.latest)
if trimmed_record:
output_record = trimmed_record
# Write record to stdout
sys.stdout.buffer.write(output_record)
records_written += 1
bytes_written += msr.reclen
print(f"Wrote {records_written} records, {bytes_written} bytes", file=sys.stderr)
def trim_record(msr, earliest, latest):
"""Trim a miniSEED record to the specified start and end times."""
# Cannot trim time coverage of a record with no coverage
if msr.samplecnt == 0 and msr.samprate == 0.0:
return None
# Re-parse the single miniSEED record and decode the data samples
msr_trimmed = MS3Record.parse(msr.record, unpack_data=True)
data_samples = msr_trimmed.datasamples[:]
start_time = msr_trimmed.starttime
end_time = msr_trimmed.endtime
sample_period_ns = int(NSTMODULUS / msr_trimmed.samprate)
# Trim early samples to the earliest time
if earliest and start_time < earliest <= end_time:
# Use ceiling division to ensure we skip enough samples
samples_to_skip = -((start_time - earliest) // sample_period_ns)
start_time += samples_to_skip * sample_period_ns
data_samples = data_samples[samples_to_skip:]
# Trim late samples to the latest time
if latest and start_time <= latest < end_time:
# Use ceiling division to ensure we remove enough samples
samples_to_remove = -((latest - end_time) // sample_period_ns)
data_samples = data_samples[:-samples_to_remove] if samples_to_remove > 0 else data_samples
if not data_samples:
return None
# Pack the trimmed record
msr_trimmed.starttime = start_time
record_buffer = b""
for packed_record in msr_trimmed.generate(
data_samples=data_samples, sample_type=msr_trimmed.sampletype
):
record_buffer += packed_record
return record_buffer
def parse_timestr(timestr):
"""
Helper for argparse to convert a time string to a nanosecond time value.
"""
try:
return timestr2nstime(timestr)
except ValueError:
raise argparse.ArgumentTypeError(f"Invalid time string: {timestr}") from None
def main():
"""Main entry point for the script."""
parser = argparse.ArgumentParser(
description="Stream miniSEED records with time window selection",
epilog="Reads from stdin and writes to stdout. Records overlapping the "
"time window boundaries are trimmed to fit within the window.",
)
parser.add_argument(
"--earliest",
"-e",
type=parse_timestr,
help="Earliest time to include (ISO format: YYYY-MM-DDTHH:MM:SS)",
)
parser.add_argument(
"--latest",
"-l",
type=parse_timestr,
help="Latest time to include (ISO format: YYYY-MM-DDTHH:MM:SS)",
)
args = parser.parse_args()
# Validate time arguments
if args.earliest and args.latest and args.earliest > args.latest:
parser.error("Earliest time cannot be after latest time")
if not args.earliest and not args.latest:
parser.error("At least one of --earliest or --latest must be specified")
try:
process_stream(args)
except BrokenPipeError:
# Handle broken pipe gracefully (e.g., when piping to head)
pass
except KeyboardInterrupt:
print("\nInterrupted by user", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()