mth5.timeseries

Submodules

Classes

`ChannelTS`	Time series container for a single MT channel with full metadata.
`RunTS`	Container for MT time series data from a single run.

Package Contents

Time series container for a single MT channel with full metadata.

Stores equally-spaced time series data in an xarray.DataArray with a time coordinate index. Integrates comprehensive metadata from Survey/Station/Run/Channel hierarchy and supports calibration, resampling, merging, and format conversions.

Parameters:

channel_type ({'electric', 'magnetic', 'auxiliary'}, default 'auxiliary') – Type of the channel.
data (array-like, optional) – Time series data (numpy array, pandas DataFrame/Series, xarray.DataArray).
channel_metadata (mt_metadata.timeseries.Electric | Magnetic | Auxiliary | dict, optional) – Channel-specific metadata.
station_metadata (mt_metadata.timeseries.Station | dict, optional) – Station metadata.
run_metadata (mt_metadata.timeseries.Run | dict, optional) – Run metadata.
survey_metadata (mt_metadata.timeseries.Survey | dict, optional) – Survey metadata.
**kwargs – Additional attributes to set on the object.

ts

The time series data array.

Type:: numpy.ndarray

sample_rate

Sample rate in samples per second.

Type:: float

start

Start time (UTC).

Type:: MTime

end

End time (UTC), derived from start + duration.

Type:: MTime

n_samples

Number of samples.

Type:: int

component

Component name (e.g., ‘ex’, ‘hy’, ‘temperature’).

Type:: str

channel_response

Full instrument response filter chain.

Type:: ChannelResponse

Notes

End time is a derived property and cannot be set directly.
Leverages xarray for efficient interpolation, resampling, and groupby operations.
Metadata follows mt_metadata standards with automatic time period updates.

Examples

Create an auxiliary channel with synthetic data:

>>> from mth5.timeseries import ChannelTS
>>> import numpy as np
>>> ts_obj = ChannelTS('auxiliary')
>>> ts_obj.sample_rate = 8
>>> ts_obj.start = '2020-01-01T12:00:00+00:00'
>>> ts_obj.ts = np.random.randn(4096)
>>> ts_obj.station_metadata.id = 'MT001'
>>> ts_obj.run_metadata.id = 'MT001a'
>>> ts_obj.component = 'temperature'
>>> print(ts_obj)

Calibrate and remove instrument response:

>>> calibrated = ts_obj.remove_instrument_response()
>>> calibrated.channel_metadata.units

logger

data_array

property survey_metadata: mt_metadata.timeseries.Survey

Survey metadata.

Returns:: Survey metadata with updated keys.
Return type:: mt_metadata.timeseries.Survey

property station_metadata: mt_metadata.timeseries.Station

Station metadata.

Returns:: Station metadata from the first station in the survey.
Return type:: mt_metadata.timeseries.Station

property run_metadata: mt_metadata.timeseries.Run

Run metadata.

Returns:: Run metadata from the first run in the station.
Return type:: mt_metadata.timeseries.Run

property channel_metadata: mt_metadata.timeseries.Electric | mt_metadata.timeseries.Magnetic | mt_metadata.timeseries.Auxiliary

Channel metadata.

Returns:: Channel metadata from the first channel in the run.
Return type:: mt_metadata.timeseries.Electric | Magnetic | Auxiliary

Extract sample_rate from channel_metadata if available.

Parameters:: channel_metadata (mt_metadata.timeseries.Electric | Magnetic | Auxiliary | dict | None) – Metadata that may contain a sample_rate field.
Returns:: Sample rate if found, otherwise None.
Return type:: float | None

Notes

Supports nested dict structures like {"electric": {"sample_rate": 8.0}}.

copy(data: bool = True) → ChannelTS[source]

Create a copy of the ChannelTS object.

Parameters:: data (bool, default True) – Include data in the copy (True) or only metadata (False).
Returns:: Copy of the channel.
Return type:: ChannelTS

Examples

Copy metadata structure without data:

>>> ch_copy = ts_obj.copy(data=False)

property ts: numpy.ndarray

Time series data as a numpy array.

Returns:: The time series data.
Return type:: numpy.ndarray

property time_index: numpy.ndarray

Time index as a numpy array.

Returns:: Array of datetime64[ns] timestamps.
Return type:: numpy.ndarray

property channel_type: str

Channel type.

Returns:: Channel type: ‘Electric’, ‘Magnetic’, or ‘Auxiliary’.
Return type:: str

property component: component

property n_samples: number of samples

has_data()[source]: check to see if there is an index in the time series

is_high_frequency(threshold_dt=0.0001)[source]: Quasi hard-coded condition to check if data are logged at more than 10kHz can be parameterized in future

compute_sample_rate()[source]

Two cases, high_frequency (HF) data and not HF data.

# Original comment about the HF case: Taking the median(diff(timestamps)) is more accurate for high sample rates, the way pandas.date_range rounds nanoseconds is not consistent between samples, therefore taking the median provides better results if the time series is long this can be inefficient so test first

property sample_rate: sample rate in samples/second

property sample_interval

Sample interval = 1 / sample_rate

Returns:: sample interval as time distance between time samples
Return type:: float

property start: MTime object

property end: MTime object

property channel_response

Full channel response filter

Returns:: full channel response filter
Return type:: mt_metadata.timeseries.filters.ChannelResponse

get_calibration_operation()[source]

get_calibrated_units()[source]

Follows the FDSN standard which has the filter stages starting with physical units to digital counts.

The channel_response is expected to have a list of filter “stages” of which the first stage has input units corresponding to the the physical quantity that the instrument measures, and the last is normally counts.

channel_response can be viewed as the chaining together of all of these filters.

Thus it is normal for channel_response.units_out will be in the same units as the archived raw time series, and for the units after the response is corrected for will be the units_in of

The units of the channel metadata are compared to the input and output units of the channel_response.

Returns:: tuple, calibration_operation, either “mulitply” or divide”, and a string for calibrated units
Return type:: tuple (of two strings_

remove_instrument_response(include_decimation=False, include_delay=False, **kwargs)[source]

Remove instrument response from the given channel response filter

The order of operations is important (if applied):

detrend

zero mean

zero pad

time window

frequency window

remove response

undo time window

bandpass

Parameters:

include_decimation (bool, optional) – Include decimation in response, defaults to True
include_delay (bool, optional) – include delay in complex response, defaults to False

kwargs

Parameters:

plot (boolean, default True) – to plot the calibration process [ False | True ]
detrend (boolean, default True) – Remove linar trend of the time series
zero_mean (boolean, default True) – Remove the mean of the time series
zero_pad (boolean, default True) – pad the time series to the next power of 2 for efficiency
t_window (string, default None) – Time domain windown name see scipy.signal.windows for options
t_window_params – Time domain window parameters, parameters can be

found in scipy.signal.windows :type t_window_params: dictionary :param f_window: Frequency domain windown name see scipy.signal.windows for options :type f_window: string, defualt None :param f_window_params: Frequency window parameters, parameters can be found in scipy.signal.windows :type f_window_params: dictionary :param bandpass: bandpass freequency and order {“low”:, “high”:, “order”:,} :type bandpass: dictionary

get_slice(start, end=None, n_samples=None)[source]

Get a slice from the time series given a start and end time.

Looks for >= start & <= end

Uses loc to be exact with milliseconds

Parameters:

start (string, MTime) – start time of the slice
end (string, MTime) – end time of the slice
n_samples (integer) – number of sample to get after start time

Returns:

slice of the channel requested

Return type:

ChannelTS

decimate(new_sample_rate, inplace=False, max_decimation=8)[source]

decimate the data by using scipy.signal.decimate

Parameters:: dec_factor (int) – decimation factor

refills ts.data with decimated data and replaces sample_rate

resample_poly(new_sample_rate, pad_type='mean', inplace=False)[source]

Use scipy.signal.resample_poly to resample data while using an FIR filter to remove aliasing.

Parameters:

new_sample_rate (TYPE) – DESCRIPTION
pad_type (TYPE, optional) – DESCRIPTION, defaults to “mean”

Returns:

DESCRIPTION

Return type:

TYPE

merge(other, gap_method='slinear', new_sample_rate=None, resample_method='poly')[source]

merg two channels or list of channels together in the following steps

xr.combine_by_coords([original, other])
compute monotonic time index
reindex(new_time_index, method=gap_method)

If you want a different method or more control use merge

Parameters:

other (mth5.timeseries.ChannelTS) – Another channel

Raises:

TypeError – If input is not a ChannelTS
ValueError – if the components are different

Returns:

Combined channel with monotonic time index and same metadata

Return type:

mth5.timeseries.ChannelTS

to_xarray()[source]

Returns a xarray.DataArray object of the channel timeseries this way metadata from the metadata class is updated upon return.

Returns:: Returns a xarray.DataArray object of the channel timeseries

this way metadata from the metadata class is updated upon return. :rtype: xarray.DataArray

>>> import numpy as np
>>> from mth5.timeseries import ChannelTS
>>> ex = ChannelTS("electric")
>>> ex.start = "2020-01-01T12:00:00"
>>> ex.sample_rate = 16
>>> ex.ts = np.random.rand(4096)

to_obspy_trace(network_code=None, encoding=None)[source]

Convert the time series to an obspy.core.trace.Trace object. This will be helpful for converting between data pulled from IRIS and data going into IRIS.

Parameters:: network_code (string) – two letter code provided by FDSN DMC
Returns:: DESCRIPTION
Return type:: TYPE

from_obspy_trace(obspy_trace)[source]

Fill data from an obspy.core.Trace

Parameters:: obspy_trace (obspy.core.trace) – Obspy trace object

plot()[source]

Simple plot of the data

Returns:: figure object
Return type:: matplotlib.figure

welch_spectra(window_length=2**12, **kwargs)[source]

get welch spectra

Parameters:

window_length (TYPE) – DESCRIPTION
**kwargs –
DESCRIPTION

Returns:

DESCRIPTION

Return type:

TYPE

plot_spectra(spectra_type='welch', window_length=2**12, **kwargs)[source]

Parameters:

spectra_type (string, optional) – spectra type, defaults to “welch”
window_length (int, optional) – window length of the welch method should be a power of 2, defaults to 2 ** 12
**kwargs –
DESCRIPTION

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[source]

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[source]

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[source]

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[source]

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[source]

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[source]

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][source]

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[source]

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[source]

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[source]

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[source]

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[source]

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[source]

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[source]

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[source]

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[source]

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