mth5.timeseries.run_ts

lists and arrays that goes on, seems easiest to convert all lists to strings and then convert them back if read in.

copyright:: Jared Peacock (jpeacock@usgs.gov)
license:: MIT

Attributes

meta_classes

Classes

RunTS

Container for MT time series data from a single run.

Module Contents

timeseries.meta_classes

Container for MT time series data from a single run.

Holds all run time series in one aligned xarray Dataset with channels as data variables and time as the coordinate. Manages metadata for survey, station, and run levels.

Parameters:

array_list (list[ChannelTS] | list[xr.DataArray] | xr.Dataset | None, optional) – List of ChannelTS objects, xarray DataArrays, or an xarray Dataset containing the time series data. All channels will be aligned to a common time index.
run_metadata (timeseries.Run | dict | None, optional) – Metadata for the run. Can be a Run object or dictionary.
station_metadata (timeseries.Station | dict | None, optional) – Metadata for the station. Can be a Station object or dictionary.
survey_metadata (timeseries.Survey | dict | None, optional) – Metadata for the survey. Can be a Survey object or dictionary.

dataset

xarray Dataset containing all channel data with time coordinate

Type:: xr.Dataset

survey_metadata

Survey-level metadata

Type:: timeseries.Survey

station_metadata

Station-level metadata

Type:: timeseries.Station

run_metadata

Run-level metadata

Type:: timeseries.Run

filters

Dictionary of channel response filters keyed by filter name

Type:: dict[str, Filter]

sample_rate

Sample rate in samples per second

Type:: float

channels

List of channel names in the dataset

Type:: list[str]

Examples

Create an empty RunTS:

>>> from mth5.timeseries import RunTS
>>> run = RunTS()

Create RunTS from ChannelTS objects:

>>> from mth5.timeseries import ChannelTS, RunTS
>>> ex = ChannelTS('electric', data=ex_data,
...                channel_metadata={'component': 'ex'})
>>> ey = ChannelTS('electric', data=ey_data,
...                channel_metadata={'component': 'ey'})
>>> run = RunTS(array_list=[ex, ey])
>>> print(run.channels)
['ex', 'ey']

Access individual channels:

>>> ex_channel = run.ex  # Returns ChannelTS object
>>> print(ex_channel.sample_rate)
256.0

See also

ChannelTS: Individual channel time series container

Notes

When multiple channels are provided with different start/end times, they will be automatically aligned using the earliest start and latest end times, with NaN values filling gaps.

logger

property survey_metadata: mt_metadata.timeseries.Survey

Survey timeseries.

Returns:: Survey-level metadata object.
Return type:: timeseries.Survey

property station_metadata: mt_metadata.timeseries.Station

Station timeseries.

Returns:: Station-level metadata object (first station in survey).
Return type:: timeseries.Station

property run_metadata: mt_metadata.timeseries.Run

Run timeseries.

Returns:: Run-level metadata object (first run in first station).
Return type:: timeseries.Run

copy(data: bool = True) → RunTS

Create a copy of the RunTS object.

Parameters:: data (bool, optional) – If True, copy the data along with timeseries. If False, only copy the metadata (default is True).
Returns:: A copy of the RunTS object.
Return type:: RunTS

Examples

Create a copy with data:

>>> run_copy = run.copy()

Create a metadata-only copy:

>>> run_meta = run.copy(data=False)
>>> print(run_meta.has_data())
False

has_data() → bool

Check if the RunTS contains any data.

Returns:: True if channels with data exist, False otherwise.
Return type:: bool

Examples

>>> run = RunTS()
>>> print(run.has_data())
False
>>> run.add_channel(ex_channel)
>>> print(run.has_data())
True

property summarize_metadata: dict[str, any]

Get a summary of all channel timeseries.

Flattens the metadata from all channels into a single dictionary with keys in the format ‘channel.attribute’.

Returns:: Dictionary with flattened metadata from all channels.
Return type:: dict[str, any]

Examples

>>> meta_summary = run.summarize_metadata
>>> print(meta_summary.keys())
dict_keys(['ex.time_period.start', 'ex.sample_rate', ...])

validate_metadata() → None

Check to make sure that the metadata matches what is in the data set.

updates metadata from the data.

Check the start and end times, channels recorded

set_dataset(array_list: list[mth5.timeseries.channel_ts.ChannelTS] | list[xarray.DataArray] | xarray.Dataset, align_type: str = 'outer') → None

Set the dataset from a list of channels or existing dataset.

Creates an xarray Dataset from the input channels or dataset, validates metadata consistency, and updates dataset attributes with run metadata.

Parameters:

array_list (list[ChannelTS] | list[xr.DataArray] | xr.Dataset) – Input data as a list of ChannelTS objects, list of xarray DataArrays, or an existing xarray Dataset.
align_type (str, optional) –
Method for aligning channels with different time indexes:
- ’outer’ - use union of all time indexes (default)
- ’inner’ - use intersection of time indexes
- ’left’ - use time index from first channel
- ’right’ - use time index from last channel
- ’exact’ - raise ValueError if indexes don’t match exactly
- ’override’ - rewrite indexes to match first channel (requires same size)

Notes

This method performs the following operations:

Validates the input array_list
Converts ChannelTS objects to xarray format
Combines channels into a single Dataset
Validates metadata consistency
Updates dataset attributes with run metadata

When providing ChannelTS objects, their metadata and filters are automatically extracted and merged into the run’s metadata structure.

Examples

Set dataset from ChannelTS objects:

>>> ex = ChannelTS('electric', data=ex_data,
...                channel_metadata={'component': 'ex'})
>>> ey = ChannelTS('electric', data=ey_data,
...                channel_metadata={'component': 'ey'})
>>> run.set_dataset([ex, ey])

Set dataset with custom alignment:

>>> run.set_dataset([ex, ey, hx], align_type='inner')

Set dataset from existing xarray Dataset:

>>> import xarray as xr
>>> ds = xr.Dataset({'ex': ex_da, 'ey': ey_da})
>>> run.set_dataset(ds)

See also

dataset: Property for setting dataset with default alignment
add_channel: Add a single channel to existing dataset
_validate_array_list: Validation and conversion of array list

add_channel(channel: xarray.DataArray | mth5.timeseries.channel_ts.ChannelTS) → None

Add a channel to the dataset.

The channel must have the same sample rate and time coordinates that are compatible with the existing dataset. If start times don’t match, NaN values will be placed where timing doesn’t align.

Parameters:: channel (xr.DataArray | ChannelTS) – A channel as an xarray DataArray or ChannelTS object to add.
Raises:: ValueError – If the channel has a different sample rate than the run, or if the input is not a DataArray or ChannelTS.

Examples

Add a ChannelTS:

>>> hz = ChannelTS('magnetic', data=hz_data,
...                channel_metadata={'component': 'hz'})
>>> run.add_channel(hz)
>>> print(run.channels)
['ex', 'ey', 'hx', 'hy', 'hz']

Add an xarray DataArray:

>>> import xarray as xr
>>> data_array = xr.DataArray(data, coords={'time': times})
>>> run.add_channel(data_array)

property dataset: xarray.Dataset

The xarray Dataset containing all channel data.

Returns:: Dataset with channels as data variables and time as coordinate.
Return type:: xr.Dataset

Examples

>>> print(run.dataset)
<xarray.Dataset>
Dimensions:  (time: 4096)
Coordinates:
  * time     (time) datetime64[ns] ...
Data variables:
    ex       (time) float64 ...
    ey       (time) float64 ...

property start: mt_metadata.common.mttime.MTime

Start time of the run in UTC.

Returns:: Start time from the dataset if data exists, otherwise from run_metadata.
Return type:: MTime

Examples

>>> print(run.start)
2020-01-01T00:00:00+00:00

property end: mt_metadata.common.mttime.MTime

End time of the run in UTC.

Returns:: End time from the dataset if data exists, otherwise from run_metadata.
Return type:: MTime

Examples

>>> print(run.end)
2020-01-01T01:00:00+00:00

property sample_rate: float

Sample rate in samples per second.

Returns:: Sample rate estimated from time differences if data exists, otherwise from timeseries.
Return type:: float

Examples

>>> print(run.sample_rate)
256.0

property sample_interval: float

Sample interval in seconds (inverse of sample_rate).

Returns:: Sample interval = 1 / sample_rate, or 0.0 if sample_rate is 0.
Return type:: float

Examples

>>> print(run.sample_interval)
0.00390625  # for 256 Hz

property channels: list[str]

List of channel names in the dataset.

Returns:: List of channel component names (e.g., [‘ex’, ‘ey’, ‘hx’]).
Return type:: list[str]

Examples

>>> print(run.channels)
['ex', 'ey', 'hx', 'hy', 'hz']

property filters: dict[str, mt_metadata.timeseries.filters.ChannelResponse]

Dictionary of channel response filters.

Returns:: Dictionary keyed by filter name containing ChannelResponse objects.
Return type:: dict[str, ChannelResponse]

Examples

>>> print(run.filters.keys())
dict_keys(['v_to_counts', 'dipole_100m'])

to_obspy_stream(network_code: str | None = None, encoding: str | None = None) → obspy.core.Stream

Convert time series to an ObsPy Stream object.

Creates an ObsPy Stream containing a Trace for each channel in the run.

Parameters:

network_code (str | None, optional) – Two-letter network code provided by FDSN DMC. If None, uses station timeseries.
encoding (str | None, optional) – Data encoding format (e.g., ‘STEIM2’, ‘FLOAT64’). If None, uses default encoding.

Returns:

Stream object containing Trace objects for all channels.

Return type:

obspy.core.Stream

Examples

>>> stream = run.to_obspy_stream(network_code='MT')
>>> print(stream)
3 Trace(s) in Stream:
MT.MT001..EX | 2020-01-01T00:00:00 - ... | 256.0 Hz, 4096 samples

See also

from_obspy_stream: Create RunTS from ObsPy Stream
ChannelTS.to_obspy_trace: Convert single channel

wrangle_leap_seconds_from_obspy(array_list: list[mth5.timeseries.channel_ts.ChannelTS]) → list[mth5.timeseries.channel_ts.ChannelTS]

Handle potential leap second issues from ObsPy streams.

Removes runs with only one sample that are numerically identical to adjacent samples, which may be artifacts of leap second handling.

Parameters:: array_list (list[ChannelTS]) – List of ChannelTS objects from ObsPy conversion.
Returns:: Filtered list with single-sample runs removed.
Return type:: list[ChannelTS]

Notes

This is experimental handling for issue #169. The exact behavior of ObsPy’s leap second handling is not fully documented.

from_obspy_stream(obspy_stream: obspy.core.Stream, run_metadata: mt_metadata.timeseries.Run | None = None) → None

Get a run from an obspy.core.stream which is a list of obspy.core.Trace objects.

Parameters:: obspy_stream (obspy.core.Stream) – Obspy Stream object

Development Notes:

There is a baked in assumption here that the channel nomenclature in obspy is e1,e2,h1,h2,h3 and we want to convert to mth5 conventions ex,ey,hx,hy,hz. This should be made more flexible in the future.
A bug was found that was creating channels e1, ex, ey in the same run when reading from obspy – this is fixed here by renaming the components and a workaround to reset the station’s channels_recorded list.

get_slice(start: str | mt_metadata.common.mttime.MTime, end: str | mt_metadata.common.mttime.MTime | None = None, n_samples: int | None = None) → RunTS

Extract a time slice from the run.

Gets a chunk of data from the run, finding the closest points to the given parameters. Uses pandas slice_indexer for robust slicing.

Parameters:

start (str | MTime) – Start time of the slice (ISO format string or MTime object).
end (str | MTime | None, optional) – End time of the slice. Required if n_samples not provided.
n_samples (int | None, optional) – Number of samples to get. Required if end not provided.

Returns:

New RunTS object containing the requested slice with copies of metadata and filters.

Return type:

RunTS

Raises:

ValueError – If neither end nor n_samples is provided.

Examples

Get slice by start and end times:

>>> slice1 = run.get_slice('2020-01-01T00:00:00',
...                         '2020-01-01T00:01:00')
>>> print(slice1.start, slice1.end)

Get slice by start time and number of samples:

>>> slice2 = run.get_slice('2020-01-01T00:00:00', n_samples=1024)
>>> print(len(slice2.dataset.time))
1024

Notes

Uses pandas slice_indexer which handles near-matches better than xarray’s native slicing. The actual slice may be slightly adjusted to match available data points.

calibrate(**kwargs) → RunTS

Remove instrument response from all channels.

Applies the channel response filters to calibrate each channel, creating a new run with calibrated data.

Parameters:: **kwargs – Additional keyword arguments passed to each channel’s remove_instrument_response method.
Returns:: New RunTS object with calibrated channels.
Return type:: RunTS

Examples

>>> calibrated_run = run.calibrate()
>>> # Calibration typically converts from counts to physical units

See also

ChannelTS.remove_instrument_response: Calibrate single channel

decimate(new_sample_rate: float, inplace: bool = False, max_decimation: int = 8) → RunTS | None

Decimate data to a new sample rate using multi-stage decimation.

Applies FIR filtering and downsampling in multiple stages to achieve the target sample rate while preventing aliasing.

Parameters:

new_sample_rate (float) – Target sample rate in samples per second.
inplace (bool, optional) – If True, modify the current run. If False, return a new run (default is False).
max_decimation (int, optional) – Maximum decimation factor for each stage (default is 8).

Returns:

If inplace is False, returns new decimated RunTS. Otherwise None.

Return type:

RunTS | None

Examples

Decimate from 256 Hz to 1 Hz:

>>> decimated_run = run.decimate(1.0)
>>> print(decimated_run.sample_rate)
1.0

Decimate in place:

>>> run.decimate(16.0, inplace=True)
>>> print(run.sample_rate)
16.0

Notes

NaN values are filled with 0 before decimation to prevent NaN propagation. Multi-stage decimation is used to maintain signal quality and prevent aliasing.

See also

resample_poly: Alternative resampling using polyphase filtering
resample: Simple resampling without anti-aliasing

resample_poly(new_sample_rate: float, pad_type: str = 'mean', inplace: bool = False) → RunTS | None

Resample data using polyphase filtering.

Uses scipy.signal.resample_poly to resample while applying an FIR filter to remove aliasing. Generally more accurate than simple resampling but slower than decimation.

Parameters:

new_sample_rate (float) – Target sample rate in samples per second.
pad_type (str, optional) – Padding method for edge effects: ‘mean’, ‘median’, ‘zero’ (default is ‘mean’).
inplace (bool, optional) – If True, modify current run. If False, return new run (default is False).

Returns:

If inplace is False, returns new resampled RunTS. Otherwise None.

Return type:

RunTS | None

Examples

Resample from 256 Hz to 100 Hz:

>>> resampled_run = run.resample_poly(100.0)
>>> print(resampled_run.sample_rate)
100.0

Notes

NaN values are filled with 0 before resampling. The polyphase method is particularly good for arbitrary sample rate ratios.

See also

decimate: Multi-stage decimation for downsampling
resample: Simple nearest-neighbor resampling

resample(new_sample_rate: float, inplace: bool = False) → RunTS | None

Resample data to a new sample rate using nearest-neighbor method.

Simple resampling without anti-aliasing filtering. Use decimate or resample_poly for better quality when downsampling.

Parameters:

new_sample_rate (float) – Target sample rate in samples per second.
inplace (bool, optional) – If True, modify current run. If False, return new run (default is False).

Returns:

If inplace is False, returns new resampled RunTS. Otherwise None.

Return type:

RunTS | None

Examples

>>> resampled_run = run.resample(128.0)
>>> print(resampled_run.sample_rate)
128.0

Warning

This method does not apply anti-aliasing filtering. For downsampling, consider using decimate() or resample_poly() instead.

See also

decimate: Proper downsampling with anti-aliasing
resample_poly: High-quality resampling with polyphase filtering

merge(other: RunTS | list[RunTS], gap_method: str = 'slinear', new_sample_rate: float | None = None, resample_method: str = 'poly') → RunTS

Merge multiple runs into a single run.

Combines this run with one or more other runs, optionally resampling to a common sample rate and filling gaps with interpolation.

Parameters:

other (RunTS | list[RunTS]) – Another RunTS object or list of RunTS objects to merge.
gap_method (str, optional) – Interpolation method for filling gaps: ‘linear’, ‘nearest’, ‘zero’, ‘slinear’, ‘quadratic’, ‘cubic’ (default is ‘slinear’).
new_sample_rate (float | None, optional) – If provided, all runs will be resampled to this rate before merging. If None, uses the sample rate of the first run.
resample_method (str, optional) – Resampling method if new_sample_rate is provided: ‘decimate’ or ‘poly’ (default is ‘poly’).

Returns:

New merged RunTS object with monotonic time index.

Return type:

RunTS

Raises:

TypeError – If other is not a RunTS or list of RunTS objects.

Examples

Merge two runs:

>>> run1 = RunTS(array_list=[ex1, ey1])
>>> run2 = RunTS(array_list=[ex2, ey2])
>>> merged = run1.merge(run2)

Merge multiple runs with resampling:

>>> runs = [run1, run2, run3]
>>> merged = run1.merge(runs, new_sample_rate=1.0,
...                     resample_method='poly')

Notes

The merge process:

Optionally resample all runs to common sample rate
Combine datasets using xr.combine_by_coords
Create monotonic time index spanning full range
Interpolate to new index filling gaps
Merge all filter dictionaries

Metadata is taken from the first run (self).

See also

__add__: Simple merging with + operator

plot(color_map: dict[str, tuple[float, float, float]] | None = None, channel_order: list[str] | None = None) → matplotlib.figure.Figure

Plot all channels as time series.

Creates a multi-panel figure with each channel in its own subplot, sharing a common time axis.

Parameters:

color_map (dict[str, tuple[float, float, float]] | None, optional) –
Dictionary mapping channel names to RGB color tuples (values 0-1). Default colors:
- ex: (1, 0.2, 0.2) - red
- ey: (1, 0.5, 0) - orange
- hx: (0, 0.5, 1) - blue
- hy: (0.5, 0.2, 1) - purple
- hz: (0.2, 1, 1) - cyan
channel_order (list[str] | None, optional) – Order of channels from top to bottom. If None, uses order from self.channels.

Returns:

Figure object containing the plot.

Return type:

matplotlib.figure.Figure

Examples

Plot with default settings:

>>> fig = run.plot()
>>> fig.savefig('timeseries.png')

Plot with custom colors and order:

>>> colors = {'ex': (1, 0, 0), 'ey': (0, 1, 0)}
>>> fig = run.plot(color_map=colors, channel_order=['ey', 'ex'])

Warning

May be slow for large datasets (millions of points). Consider using get_slice() first to plot a subset.

plot_spectra(spectra_type: str = 'welch', color_map: dict[str, tuple[float, float, float]] | None = None, **kwargs) → matplotlib.figure.Figure

Plot power spectral density for all channels.

Computes and plots the power spectrum of each channel on a single log-log plot with period on x-axis.

Parameters:

spectra_type (str, optional) – Type of spectral estimate to compute. Currently only ‘welch’ is supported (default is ‘welch’).
color_map (dict[str, tuple[float, float, float]] | None, optional) – Dictionary mapping channel names to RGB color tuples (values 0-1). Uses same defaults as plot().
**kwargs – Additional keyword arguments passed to the spectra computation method (e.g., nperseg, window for Welch’s method).

Returns:

Figure object containing the spectra plot.

Return type:

matplotlib.figure.Figure

Examples

Plot spectra with default settings:

>>> fig = run.plot_spectra()

Plot with custom Welch parameters:

>>> fig = run.plot_spectra(nperseg=1024, window='hann')

Notes

The plot shows:

Period (seconds) on bottom x-axis
Frequency (Hz) on top x-axis
Power (dB) on y-axis

See also

ChannelTS.welch_spectra: Compute Welch power spectrum