time_series#

class arte.time_series.time_series.TimeSeries(axes=None)#

Bases: object

Base class for time series analysis of adaptive optics telemetry data.

This class provides a framework for analyzing time-evolving data where multiple “sister” quantities (ensemble elements) are sampled simultaneously over time. Examples include:

Camera pixel intensities evolving over multiple frames
Deformable mirror actuator commands over time
Wavefront sensor slopes for multiple subapertures
Modal coefficients (Zernike, KL modes) during AO operation

The data structure is always (n_time_samples, ...ensemble_shape...) where the first dimension is time and remaining dimensions define the ensemble geometry.

Subclasses must implement:

_get_not_indexed_data(): Return raw data as numpy array with time as first axis
get_index_of(): Define ensemble indexing logic (select specific elements)

The class provides a chainable fluent API for data analysis:

Ensemble reductions: ensemble_rms, ensemble_mean, ensemble_std, ensemble_median, ensemble_ptp
Time reductions: time_mean, time_std, time_rms, time_median, time_ptp
Filtering: filter(modes=..., times=...) or with_times(...)
Value extraction: value property (like pandas)
Chain operations: ts.filter(modes=[2,3]).ensemble_rms.time_mean.value

Additional features:

Power spectral density analysis with Welch method
Physical units support via astropy.units
Masked arrays for missing/invalid data (e.g., points outside circular pupil)

Parameters:: axes (sequence of str, optional) – Named axes for the ensemble dimensions, enabling axis transposition. If None, no axis reordering is possible.

Notes

MaskedArray support: Use numpy.ma.MaskedArray to handle missing data in the ensemble dimension (e.g., wavefront defined only inside a circular pupil). Temporal masking is possible but may cause issues with FFT-based operations if sampling becomes non-uniform.

Originally implemented as part of the ARGOS codebase.

See also

MultiTimeSeries: Combine multiple TimeSeries with different sampling rates

Examples

>>> class MyTimeSeries(TimeSeries):
...     def __init__(self, data):
...         super().__init__()
...         self._data = data  # shape: (n_times, n_elements)
...
...     def _get_not_indexed_data(self):
...         return self._data
...
...     def get_index_of(self, *args, **kwargs):
...         if len(args) == 0:
...             return None  # Return all elements
...         return args[0]   # Return specific index
>>>
>>> # Analyze DM commands for 1000 time steps, 100 actuators
>>> dm_commands = np.random.randn(1000, 100)
>>> ts = MyTimeSeries(dm_commands)
>>>
>>> # Fluent API (chainable)
>>> mean_rms = ts.ensemble_rms.time_mean.value  # Scalar
>>> rms_time_series = ts.filter(elements=[0,1,2]).ensemble_rms.value  # Array
>>> long_exposure_ptp = ts.time_mean.ensemble_ptp.value  # Peak-to-valley

Attributes:

axes

Data series shape

delta_time

Property with the interval between samples.

If no interval can be determined (time vector too short), returns 1 with the correct unit if applicable.

ensemble_mean

Mean over ensemble dimensions (chainable property).

ensemble_median

Median over ensemble dimensions (chainable property).

ensemble_ptp

Peak-to-peak over ensemble dimensions (chainable property).

ensemble_rms

RMS over ensemble dimensions (chainable property).

ensemble_std

Standard deviation over ensemble dimensions (chainable property).

shape

Shape of the underlying data array.

time_mean

Mean over time dimension (chainable property).

time_median

Median over time dimension (chainable property).

time_ptp

Peak-to-peak over time dimension (chainable property).

time_rms

RMS over time dimension (chainable property).

time_std

Standard deviation over time dimension (chainable property).

value

Extract final value(s) from TimeSeries (convenience accessor).

Methods

`data_label`()	Override to return a string with a readable unit name
`data_unit`()	Override to return a string with a compact unit notation
`ensemble_size`()	Number of distinct series in this time ensemble
`filter`(args, *kwargs)	Create filtered TimeSeries by applying ensemble and/or time selection.
`get_data`(*args[, times, axes])	Retrieve time series data with optional filtering and indexing.
`get_ensemble_average`(*args[, times])	Average across series at each sampling time
`get_ensemble_median`(*args[, times])	Median across series at each sampling time
`get_ensemble_ptp`(*args[, times])	Peak-to-peak (max - min) across series at each sampling time
`get_ensemble_rms`(*args[, times])	Root-Mean-Square across series at each sampling time (legacy method).
`get_ensemble_std`(*args[, times])	Standard deviation across series at each sampling time
`get_time_average`(*args[, times])	Average value over time for each series
`get_time_median`(*args[, times])	Median over time for each series
`get_time_ptp`(*args[, times])	Peak-to-peak (max - min) over time for each series
`get_time_rms`(*args[, times])	Root-Mean-Square value over time for each series
`get_time_std`(*args[, times])	Standard deviation over time for each series
`get_time_vector`()	Return the series time vector
`help`([search, prefix])	Interactive help
`power`(*args[, from_freq, to_freq, ...])	Compute Power Spectral Density using Welch's method.
`time_size`()	Number of time samples in this time ensemble
`with_times`(times)	Filter to specific time interval (convenience alias for filter(times=...)).

frequency
get_index_of
last_cut_frequency

property axes#: Data series shape

data_label()#: Override to return a string with a readable unit name

data_unit()#: Override to return a string with a compact unit notation

property delta_time#

Property with the interval between samples.

If no interval can be determined (time vector too short), returns 1 with the correct unit if applicable.

property ensemble_mean#

Mean over ensemble dimensions (chainable property).

Returns:: Mean across ensemble at each time step (chainable)
Return type:: TimeSeries

Examples

>>> mean_series = ts.ensemble_mean.value  # Shape: (n_times,)
>>> overall_mean = ts.ensemble_mean.time_mean.value  # Scalar

property ensemble_median#

Median over ensemble dimensions (chainable property).

Returns:: Median across ensemble at each time step (chainable)
Return type:: TimeSeries

Examples

>>> median_series = ts.ensemble_median.value
>>> time_avg_median = ts.ensemble_median.time_mean.value

property ensemble_ptp#

Peak-to-peak over ensemble dimensions (chainable property).

Returns:: Peak-to-peak across ensemble at each time step (chainable)
Return type:: TimeSeries

Examples

>>> ptp_series = ts.ensemble_ptp.value
>>> mean_ptp = ts.ensemble_ptp.time_mean.value

property ensemble_rms#

RMS over ensemble dimensions (chainable property).

Returns a TimeSeries containing the RMS computed across all ensemble (non-time) dimensions at each time step. The result can be chained with time aggregation operations.

Returns:: Time series of ensemble RMS values (numpy-compatible via __array__)
Return type:: TimeSeries

Notes

This property: - Works on ALL data (no filtering) - Returns TimeSeries (chainable with .time_mean, .time_std, etc.) - Is numpy-compatible via __array__ protocol

For filtered data, use upstream filtering: ts.filter(modes=[2,3,4]).with_times([1,2]).ensemble_rms

For legacy array return, use: get_ensemble_rms() (deprecated)

Examples

>>> # Chainable operations
>>> ts = WavefrontSeries(...)  # shape (n_frames, ny, nx)
>>> rms_series = ts.ensemble_rms  # shape (n_frames,) - chainable!
>>> mean_rms = rms_series.time_mean  # scalar - fluent API!
>>>
>>> # Numpy/matplotlib compatible:
>>> plt.plot(time, rms_series)  # Works via __array__()
>>> np.mean(rms_series)  # Works!
>>>
>>> # With filtering (upstream):
>>> mean_rms = ts.filter(modes=[2,3,4]).ensemble_rms.time_mean

ensemble_size()#: Number of distinct series in this time ensemble

property ensemble_std#

Standard deviation over ensemble dimensions (chainable property).

Returns:: Std across ensemble at each time step (chainable)
Return type:: TimeSeries

Examples

>>> std_series = ts.ensemble_std.value
>>> mean_std = ts.ensemble_std.time_mean.value

filter(*args, **kwargs)#

Create filtered TimeSeries by applying ensemble and/or time selection.

Accepts same arguments as get_data(), creating a new TimeSeries with pre-filtered data for fluent chaining. Works with any Indexer implementation (ModeIndexer, RowColIndexer, DefaultIndexer, custom).

Parameters:

*args – Passed to get_index_of() for ensemble filtering. Common kwargs depend on the Indexer: - modes= : for ModeIndexer (mode numbers) - elements= : for DefaultIndexer (element indices) - rows=, cols= : for RowColIndexer (spatial coordinates) - coord=, axis= : for interleaved/sequential xy layouts
**kwargs – Passed to get_index_of() for ensemble filtering. Common kwargs depend on the Indexer: - modes= : for ModeIndexer (mode numbers) - elements= : for DefaultIndexer (element indices) - rows=, cols= : for RowColIndexer (spatial coordinates) - coord=, axis= : for interleaved/sequential xy layouts
times (sequence of 2 elements, optional) – Time range [start, stop] for temporal filtering. Can contain None for one-sided filtering.

Returns:

New TimeSeries with filtered data (chainable)

Return type:

TimeSeries

Examples

>>> # Modal selection (uses ModeIndexer in derived classes)
>>> ts.filter(modes=[2, 3, 4]).ensemble_rms.time_mean
>>>
>>> # Time + modal selection
>>> ts.filter(modes=[2, 3, 4], times=[0.5, 1.0]).ensemble_rms.time_mean
>>>
>>> # Element selection (DefaultIndexer)
>>> ts.filter(elements=[0, 2, 4, 6]).time_std
>>>
>>> # Row/col selection (RowColIndexer for 2D data)
>>> ts.filter(rows=slice(0, 10), cols=[5, 10]).ensemble_rms
>>>
>>> # Chainable (filters are accumulated)
>>> ts.filter(modes=[2, 3, 4]).filter(times=[0.5, 1.0]).ensemble_rms.time_mean
>>>
>>> # Custom indexer kwargs work automatically
>>> ts.filter(coord='x').time_mean  # for interleaved xy data

frequency()#

get_data(*args, times=None, axes=None, **kwargs)#

Retrieve time series data with optional filtering and indexing.

Parameters:

*args – Passed to get_index_of() for ensemble element selection.
**kwargs – Passed to get_index_of() for ensemble element selection.
times (sequence of 2 elements, optional) – Time range [start, stop] for filtering. Can be None (no filtering), or contain None elements for one-sided filtering. Units must match the time vector if using astropy quantities.
axes (sequence of str, optional) – Reorder axes to specified order. Requires axis names to be defined during initialization.

Returns:

Data array with shape (n_times, ...ensemble_shape...) after applying time filtering, ensemble indexing, and axis transposition.

Return type:

ndarray

Examples

>>> # Get all data
>>> data = ts.get_data()
>>>
>>> # Get data for specific time range (closed-loop only)
>>> data = ts.get_data(times=[1.0 * u.s, 5.0 * u.s])
>>>
>>> # Get data for specific ensemble elements
>>> data = ts.get_data(modes=[2, 3, 4])  # Assuming mode-based indexing

get_ensemble_average(*args, times=None, **kwargs)#: Average across series at each sampling time

Deprecated since version Use: the ensemble_mean property instead. This method will be removed in a future version.

get_ensemble_median(*args, times=None, **kwargs)#: Median across series at each sampling time

Deprecated since version Use: the ensemble_median property instead. This method will be removed in a future version.

get_ensemble_ptp(*args, times=None, **kwargs)#: Peak-to-peak (max - min) across series at each sampling time

Deprecated since version Use: the ensemble_ptp property instead. This method will be removed in a future version.

get_ensemble_rms(*args, times=None, **kwargs)#

Root-Mean-Square across series at each sampling time (legacy method).

Deprecated since version Use: the chainable property ensemble_rms for fluent API: ts.ensemble_rms or ts.filter(modes=[2,3,4]).ensemble_rms This method will be removed in a future version.

Returns:: RMS values (not chainable)
Return type:: ndarray

get_ensemble_std(*args, times=None, **kwargs)#: Standard deviation across series at each sampling time

Deprecated since version Use: the ensemble_std property instead. This method will be removed in a future version.

abstractmethod get_index_of(*args, **kwargs)#

get_time_average(*args, times=None, **kwargs)#: Average value over time for each series

Deprecated since version Use: the time_mean property instead. This method will be removed in a future version.

get_time_median(*args, times=None, **kwargs)#: Median over time for each series

Deprecated since version Use: the time_median property instead. This method will be removed in a future version.

get_time_ptp(*args, times=None, **kwargs)#: Peak-to-peak (max - min) over time for each series

Deprecated since version Use: the time_ptp property instead. This method will be removed in a future version.

get_time_rms(*args, times=None, **kwargs)#: Root-Mean-Square value over time for each series

Deprecated since version Use: the time_rms property instead. This method will be removed in a future version.

get_time_std(*args, times=None, **kwargs)#: Standard deviation over time for each series

Deprecated since version Use: the time_std property instead. This method will be removed in a future version.

get_time_vector()#: Return the series time vector

help(search='', prefix='')#

Interactive help

Prints on stdout a list of methods that match the search substring or all of them if search is left to the default value of an empty string, together with a one-line help taken from the first line of their docstring, if any.

The prefix argument is prepended to the method name and is used for recursive help of every class member.

last_cut_frequency()#

power(*args, from_freq=None, to_freq=None, segment_factor=None, window='boxcar', **kwargs)#

Compute Power Spectral Density using Welch’s method.

Parameters:

*args – Passed to get_index_of() for ensemble element selection.
**kwargs – Passed to get_index_of() for ensemble element selection.
from_freq (float, optional) – Lower frequency bound for output. If None, starts from DC.
to_freq (float, optional) – Upper frequency bound for output. If None, goes to Nyquist.
segment_factor (float, optional) – Segment length factor for Welch method. Larger values give better frequency resolution but worse variance. Default is 1.0 (one segment).
window (str, optional) – Window function name for Welch method. Default is ‘boxcar’ (no windowing). See scipy.signal.welch for available windows.

Returns:

Power spectral density with shape (n_frequencies, ...ensemble_shape...). Units are data_unit^2 / Hz (standard PSD units) if applicable.

Return type:

ndarray

Notes

The PSD is computed using scipy.signal.welch in standard units [data_unit^2/Hz]. Results are cached unless segment_factor or window changes.

For MaskedArrays, masked values are filled with zeros before FFT computation.

Examples

>>> # Compute full PSD
>>> psd = ts.power()
>>> freq = ts.frequency()
>>>
>>> # Analyze only low frequencies (< 100 Hz)
>>> psd = ts.power(from_freq=0, to_freq=100)
>>>
>>> # Use Hanning window with 4 segments for variance reduction
>>> psd = ts.power(segment_factor=4, window='hann')

property shape#

Shape of the underlying data array.

Returns the shape tuple of the TimeSeries data, similar to numpy arrays. This provides convenient access without requiring explicit conversion.

Returns:: Shape of the data array (time, ensemble…)
Return type:: tuple of int

Examples

>>> ts.shape
(1000, 10)  # 1000 time steps, 10 modes
>>> ts.ensemble_rms.shape
(1000,)  # Collapsed to time dimension only

property time_mean#

Mean over time dimension (chainable property).

Computes the average across time (axis 0), returning a TimeSeries with time_size=1 containing the temporal mean. Further operations (like ensemble_rms, ensemble_ptp) can be chained.

Returns:: TimeSeries with shape (1, …ensemble_shape…) (chainable) Use .value to extract the final array/scalar.
Return type:: TimeSeries

Notes

Chainable operation: returns TimeSeries (not array). For convenient array extraction, use .value: ts.time_mean.value → ndarray or scalar

Examples

>>> ts = WavefrontSeries(...)  # shape (n_frames, ny, nx)
>>> mean_ts = ts.time_mean  # TimeSeries(1, ny, nx) - chainable!
>>> mean_wf = ts.time_mean.value  # ndarray(ny, nx) - extracted
>>>
>>> # Chaining works both ways now:
>>> rms_series = ts.ensemble_rms  # (n_frames,)
>>> mean_rms = rms_series.time_mean.value  # scalar
>>>
>>> # NEW: time-then-ensemble operations
>>> ptp_map = ts.time_mean.ensemble_ptp.value  # peak-to-valley of long-exposure

property time_median#

Median over time dimension (chainable property).

Returns:: TimeSeries with time_size=1 containing temporal median (chainable)
Return type:: TimeSeries

Examples

>>> median_wf = ts.time_median.value
>>> rms_of_median = ts.time_median.ensemble_rms.value

property time_ptp#

Peak-to-peak over time dimension (chainable property).

Returns:: TimeSeries with time_size=1 containing temporal peak-to-peak (chainable)
Return type:: TimeSeries

Examples

>>> ptp_wf = ts.time_ptp.value
>>> mean_ptp = ts.time_ptp.ensemble_mean.value

property time_rms#

RMS over time dimension (chainable property).

Returns:: TimeSeries with time_size=1 containing temporal RMS (chainable)
Return type:: TimeSeries

Examples

>>> rms_wf = ts.time_rms.value
>>> max_rms = ts.time_rms.ensemble_ptp.value

time_size()#: Number of time samples in this time ensemble

property time_std#

Standard deviation over time dimension (chainable property).

Computes the std across time (axis 0), returning a TimeSeries with time_size=1 containing the temporal std. Further operations (like ensemble_rms, ensemble_ptp) can be chained.

Returns:: TimeSeries with shape (1, …ensemble_shape…) (chainable) Use .value to extract the final array/scalar.
Return type:: TimeSeries

Notes

Chainable operation: returns TimeSeries (not array). For convenient array extraction, use .value: ts.time_std.value → ndarray or scalar

Examples

>>> ts = WavefrontSeries(...)  # shape (n_frames, ny, nx)
>>> std_ts = ts.time_std  # TimeSeries(1, ny, nx) - chainable!
>>> std_wf = ts.time_std.value  # ndarray(ny, nx) - extracted
>>>
>>> # Chaining:
>>> rms_series = ts.ensemble_rms  # (n_frames,)
>>> std_rms = rms_series.time_std.value  # scalar
>>>
>>> # NEW: time-then-ensemble operations
>>> ptp_std = ts.time_std.ensemble_ptp.value  # peak-to-peak temporal variability

property value#

Extract final value(s) from TimeSeries (convenience accessor).

Returns the underlying data as a numpy array or scalar, similar to pandas .values. Useful for getting final results from chained operations without explicit numpy conversion.

Returns:

If shape is (1,): returns scalar
If shape is (1, n): returns 1D array of length n
Otherwise: returns the full data array

Return type:

float, int, or ndarray

Examples

>>> # Extract scalar from fully reduced series
>>> scalar = ts.ensemble_rms.time_mean.value  # → float
>>>
>>> # Extract array from partially reduced series
>>> array = ts.time_mean.value  # → ndarray(ensemble_shape)
>>>
>>> # Works like pandas .values
>>> values = ts.value  # → full data array

with_times(times)#

Filter to specific time interval (convenience alias for filter(times=…)).

Returns a new chainable TimeSeries containing only data within the specified time range [start, stop].

Note: This is an alias for filter(times=…). For filtering by both time and ensemble elements, use filter() directly.

Parameters:: times (sequence of 2 elements) – Time range [start, stop] (can contain None for one-sided filtering)
Returns:: Time-filtered TimeSeries (chainable)
Return type:: TimeSeries

Examples

>>> # Filter time range and compute statistics
>>> ts.with_times([1, 2]).ensemble_rms.time_mean
>>>
>>> # Chain with filter() for ensemble selection
>>> ts.filter(modes=[2, 3, 4]).with_times([1, 2]).ensemble_rms
>>>
>>> # Equivalent using filter() directly (preferred)
>>> ts.filter(modes=[2, 3, 4], times=[1, 2]).ensemble_rms

exception arte.time_series.time_series.TimeSeriesException#

Bases: Exception

Exception raised by TimeSeries