mth5.groups.fourier_coefficients

Fourier Coefficient group management for MTH5 format.

This module provides classes for organizing and managing Fourier Coefficient data at multiple decimation levels, including utilities for data import/export with different formats (numpy, xarray, pandas).

copyright:

Jared Peacock (jpeacock@usgs.gov)

Classes

MasterFCGroup

Master container for all Fourier Coefficient estimations of time series data.

FCDecimationGroup

Container for a single decimation level of Fourier Coefficient data.

FCGroup

Manage a set of Fourier Coefficients from a single processing run.

Module Contents

class mth5.groups.fourier_coefficients.MasterFCGroup(group: h5py.Group, **kwargs)[source]

Bases: mth5.groups.BaseGroup

Master container for all Fourier Coefficient estimations of time series data.

This class manages multiple Fourier Coefficient processing runs, each containing different decimation levels. No metadata is required at the master level.

Hierarchy

MasterFCGroup -> FCGroup (processing runs) -> FCDecimationGroup (decimation levels) -> FCChannelDataset (individual channels)

param group:

HDF5 group object for the master FC container.

type group:

h5py.Group

param **kwargs:

Additional keyword arguments passed to BaseGroup.

Examples

>>> import h5py
>>> from mth5.groups.fourier_coefficients import MasterFCGroup
>>> with h5py.File('data.h5', 'r') as f:
...     master = MasterFCGroup(f['FC'])
...     fc_group = master.add_fc_group('processing_run_1')
property fc_summary: pandas.DataFrame[source]

Get a summary of all Fourier Coefficient processing runs.

Returns:

Summary information for all FC groups including names and metadata.

Return type:

pd.DataFrame

Examples

>>> master = MasterFCGroup(h5_group)
>>> summary = master.fc_summary
add_fc_group(fc_name: str, fc_metadata: mt_metadata.processing.fourier_coefficients.Decimation | None = None) → FCGroup[source]

Add a Fourier Coefficient processing run group.

Parameters:

  • fc_name (str) – Name for the FC group (usually identifies the processing run).

  • fc_metadata (fc.Decimation, optional) – Metadata for the FC group. Default is None.

Returns:

Newly created Fourier Coefficient group.

Return type:

FCGroup

Examples

>>> master = MasterFCGroup(h5_group)
>>> fc_group = master.add_fc_group('processing_run_1')
>>> print(fc_group.name)
'processing_run_1'
get_fc_group(fc_name: str) → FCGroup[source]

Retrieve a Fourier Coefficient group by name.

Parameters:

fc_name (str) – Name of the FC group to retrieve.

Returns:

The requested Fourier Coefficient group.

Return type:

FCGroup

Raises:

MTH5Error – If the FC group does not exist.

Examples

>>> master = MasterFCGroup(h5_group)
>>> fc_group = master.get_fc_group('processing_run_1')
remove_fc_group(fc_name: str) → None[source]

Remove a Fourier Coefficient group.

Deletes the specified FC group and all associated decimation levels and channels.

Parameters:

fc_name (str) – Name of the FC group to remove.

Raises:

MTH5Error – If the FC group does not exist.

Examples

>>> master = MasterFCGroup(h5_group)
>>> master.remove_fc_group('processing_run_1')
class mth5.groups.fourier_coefficients.FCDecimationGroup(group: h5py.Group, decimation_level_metadata: mt_metadata.processing.fourier_coefficients.Decimation | None = None, **kwargs)[source]

Bases: mth5.groups.BaseGroup

Container for a single decimation level of Fourier Coefficient data.

This class manages all channels at a specific decimation level, assuming uniform sampling in both frequency and time domains.

Data Assumptions

  1. Data uniformly sampled in frequency domain

  2. Data uniformly sampled in time domain

  3. FFT moving window has uniform step size

start_time

Start time of the decimation level

Type:

datetime

end_time

End time of the decimation level

Type:

datetime

channels

List of channel names in this decimation level

Type:

list

decimation_factor

Factor by which data was decimated

Type:

int

decimation_level

Level index in decimation hierarchy

Type:

int

sample_rate

Sample rate after decimation (Hz)

Type:

float

method

Method used (FFT, wavelet, etc.)

Type:

str

window

Window parameters (length, overlap, type, sample rate)

Type:

dict

param group:

HDF5 group object for this decimation level.

type group:

h5py.Group

param decimation_level_metadata:

Metadata for the decimation level. Default is None.

type decimation_level_metadata:

optional

param **kwargs:

Additional keyword arguments passed to BaseGroup.

Examples

>>> decimation = FCDecimationGroup(h5_group, decimation_level_metadata=metadata)
>>> channel = decimation.add_channel('Ex', fc_data=fc_array)
metadata()[source]

Overwrite get metadata to include channel information in the runs

property channel_summary: pandas.DataFrame[source]

Get a summary of all channels in this decimation level.

Returns a pandas DataFrame with detailed information about each Fourier Coefficient channel including time ranges, dimensions, and sampling rates.

Returns:

DataFrame with columns:

  • componentstr

    Channel component name (e.g., ‘Ex’, ‘Hy’)

  • startdatetime64[ns]

    Start time of the channel data

  • enddatetime64[ns]

    End time of the channel data

  • n_frequencyint64

    Number of frequency bins

  • n_windowsint64

    Number of time windows

  • sample_rate_decimation_levelfloat64

    Decimation level sample rate (Hz)

  • sample_rate_window_stepfloat64

    Sample rate of window stepping (Hz)

  • unitsstr

    Physical units of the data

  • hdf5_referenceh5py.ref_dtype

    HDF5 reference to the channel dataset

Return type:

pd.DataFrame

Examples

>>> decimation = FCDecimationGroup(h5_group)
>>> summary = decimation.channel_summary
>>> print(summary[['component', 'n_frequency', 'n_windows']])
from_dataframe(df: pandas.DataFrame, channel_key: str, time_key: str = 'time', frequency_key: str = 'frequency') → None[source]

Load Fourier Coefficient data from a pandas DataFrame.

Assumes the channel_key column contains complex coefficient values organized with time and frequency dimensions.

Parameters:

  • df (pd.DataFrame) – Input DataFrame containing the coefficient data.

  • channel_key (str) – Name of the column containing coefficient values.

  • time_key (str, default='time') – Name of the time coordinate column.

  • frequency_key (str, default='frequency') – Name of the frequency coordinate column.

Raises:

TypeError – If df is not a pandas DataFrame.

Examples

>>> decimation = FCDecimationGroup(h5_group)
>>> decimation.from_dataframe(df, channel_key='Ex', time_key='time')
from_xarray(data_array: xarray.Dataset | xarray.DataArray, sample_rate_decimation_level: float) → None[source]

Load Fourier Coefficient data from an xarray DataArray or Dataset.

Automatically extracts metadata (time, frequency, units) from the xarray object and creates appropriate FCChannelDataset instances for each variable or the single DataArray.

Parameters:

  • data_array (xr.DataArray or xr.Dataset) – Input xarray object with ‘time’ and ‘frequency’ coordinates and dimensions [‘time’, ‘frequency’] (or transposed variant).

  • sample_rate_decimation_level (float) – Sample rate of the decimation level (Hz).

Raises:

TypeError – If data_array is not an xarray Dataset or DataArray.

Notes

Automatically handles both (time, frequency) and (frequency, time) dimension ordering. Units are extracted from xarray attributes if available.

Examples

>>> import xarray as xr
>>> import numpy as np
>>> decimation = FCDecimationGroup(h5_group)

Create sample xarray data:

>>> times = np.arange('2023-01-01', '2023-01-02', dtype='datetime64[s]')
>>> freqs = np.linspace(0.01, 100, 256)
>>> data_array = np.random.randn(len(times), len(freqs)) + \
...              1j * np.random.randn(len(times), len(freqs))
>>> xr_data = xr.DataArray(
...     data_array,
...     dims=['time', 'frequency'],
...     coords={'time': times, 'frequency': freqs},
...     name='Ex'
... )

Load into decimation group:

>>> decimation.from_xarray(xr_data, sample_rate_decimation_level=0.5)
to_xarray(channels: list[str] | None = None) → xarray.Dataset[source]

Create an xarray Dataset from Fourier Coefficient channels.

If no channels are specified, all channels in the decimation level are included. Each channel becomes a data variable in the resulting Dataset.

Parameters:

channels (list[str], optional) – List of channel names to include. If None, all channels are used. Default is None.

Returns:

xarray Dataset with channels as data variables and ‘time’ and ‘frequency’ as shared coordinates.

Return type:

xr.Dataset

Examples

>>> decimation = FCDecimationGroup(h5_group)
>>> xr_data = decimation.to_xarray()
>>> print(xr_data.data_vars)
Data variables:
    Ex  (time, frequency) complex128
    Ey  (time, frequency) complex128

Get specific channels:

>>> subset = decimation.to_xarray(channels=['Ex', 'Ey'])
from_numpy_array(nd_array: numpy.ndarray, ch_name: str | list[str]) → None[source]

Load Fourier Coefficient data from a numpy array.

Assumes array shape is either (n_frequencies, n_windows) for a single channel or (n_channels, n_frequencies, n_windows) for multiple channels.

Parameters:

  • nd_array (np.ndarray) – Input numpy array containing coefficient data.

  • ch_name (str or list[str]) – Channel name (for 2D array) or list of channel names (for 3D array).

Raises:

  • TypeError – If nd_array is not a numpy ndarray.

  • ValueError – If array shape is not (n_frequencies, n_windows) or (n_channels, n_frequencies, n_windows).

Examples

>>> decimation = FCDecimationGroup(h5_group)

Load single channel:

>>> data_2d = np.random.randn(256, 100) + 1j * np.random.randn(256, 100)
>>> decimation.from_numpy_array(data_2d, ch_name='Ex')

Load multiple channels:

>>> data_3d = np.random.randn(2, 256, 100) + 1j * np.random.randn(2, 256, 100)
>>> decimation.from_numpy_array(data_3d, ch_name=['Ex', 'Ey'])
add_channel(fc_name: str, fc_data: numpy.ndarray | None = None, fc_metadata: mt_metadata.processing.fourier_coefficients.FCChannel | None = None, max_shape: tuple = (None, None), chunks: bool = True, dtype: type = complex, **kwargs) → mth5.groups.FCChannelDataset[source]

Add a Fourier Coefficient channel to the decimation level.

Creates a new FCChannelDataset for a single channel at a single decimation level. Input data can be provided as numpy array or created empty.

Parameters:

  • fc_name (str) – Name for the Fourier Coefficient channel (usually component name like ‘Ex’).

  • fc_data (np.ndarray, optional) – Input data with shape (n_frequencies, n_windows). Default is None (creates empty).

  • fc_metadata (fc.FCChannel, optional) – Metadata for the channel. Default is None.

  • max_shape (tuple, default=(None, None)) – Maximum shape for HDF5 dataset dimensions (expandable if None).

  • chunks (bool, default=True) – Whether to use HDF5 chunking.

  • dtype (type, default=complex) – Data type for the dataset.

  • **kwargs – Additional keyword arguments for HDF5 dataset creation.

Returns:

Newly created FCChannelDataset object.

Return type:

FCChannelDataset

Raises:

TypeError – If fc_data type is not supported.

Notes

Data layout assumes (time, frequency) organization:

  • time index: window start times

  • frequency index: harmonic indices or float values

  • data: complex Fourier coefficients

If a channel with the same name already exists, the existing channel is returned instead of creating a duplicate.

Examples

>>> decimation = FCDecimationGroup(h5_group)
>>> metadata = fc.FCChannel(component='Ex')

Create from numpy array:

>>> fc_data = np.random.randn(100, 256) + 1j * np.random.randn(100, 256)
>>> channel = decimation.add_channel('Ex', fc_data=fc_data, fc_metadata=metadata)

Create empty channel (expandable):

>>> channel = decimation.add_channel('Ex', fc_metadata=metadata)
get_channel(fc_name: str) → mth5.groups.FCChannelDataset[source]

Retrieve a Fourier Coefficient channel by name.

Parameters:

fc_name (str) – Name of the Fourier Coefficient channel to retrieve.

Returns:

The requested Fourier Coefficient channel dataset.

Return type:

FCChannelDataset

Raises:

  • KeyError – If the channel does not exist in this decimation level.

  • MTH5Error – If unable to retrieve the channel from HDF5.

Examples

>>> decimation = FCDecimationGroup(h5_group)
>>> channel = decimation.get_channel('Ex')
>>> print(channel.shape)
(100, 256)
remove_channel(fc_name: str) → None[source]

Remove a Fourier Coefficient channel from the decimation level.

Deletes the HDF5 dataset associated with the channel. Note that this removes the reference but does not reduce the HDF5 file size.

Parameters:

fc_name (str) – Name of the Fourier Coefficient channel to remove.

Raises:

MTH5Error – If the channel does not exist.

Notes

Deleting a channel does not reduce the HDF5 file size; it simply removes the reference to the data. To truly reduce file size, copy the desired data to a new file.

Examples

>>> decimation = FCDecimationGroup(h5_group)
>>> decimation.remove_channel('Ex')
update_metadata() → None[source]

Update decimation level metadata from all channels.

Aggregates metadata from all FC channels in the decimation level including time period, sample rates, and window step information. Updates the internal metadata object and writes to HDF5.

Notes

Collects the following information from channels:

  • Time period start/end from channel data

  • Sample rate decimation level

  • Sample rate window step

Should be called after adding or modifying channels to keep metadata synchronized.

Examples

>>> decimation = FCDecimationGroup(h5_group)
>>> decimation.add_channel('Ex', fc_data=data_ex)
>>> decimation.add_channel('Ey', fc_data=data_ey)
>>> decimation.update_metadata()
add_feature(feature_name: str, feature_data: numpy.ndarray | None = None, feature_metadata: dict | None = None, max_shape: tuple = (None, None, None), chunks: bool = True, **kwargs) → None[source]

Add a feature dataset to the decimation level.

Creates a new dataset for auxiliary features or derived quantities related to Fourier Coefficients (e.g., SNR, coherency, power, etc.).

Parameters:

  • feature_name (str) – Name for the feature dataset.

  • feature_data (np.ndarray, optional) – Input data for the feature. Default is None (creates empty).

  • feature_metadata (dict, optional) – Metadata dictionary for the feature. Default is None.

  • max_shape (tuple, default=(None, None, None)) – Maximum shape for HDF5 dataset dimensions (expandable if None).

  • chunks (bool, default=True) – Whether to use HDF5 chunking.

  • **kwargs – Additional keyword arguments for HDF5 dataset creation.

Notes

Feature types may include:

  • Power: Total power in Fourier coefficients

  • SNR: Signal-to-noise ratio

  • Coherency: Cross-component coherence

  • Weights: Channel-specific weights

  • Flags: Data quality or processing flags

Examples

>>> decimation = FCDecimationGroup(h5_group)
>>> snr_data = np.random.randn(100, 256)
>>> decimation.add_feature('snr', feature_data=snr_data)

Or create empty feature for later population:

>>> decimation.add_feature('power_Ex')
class mth5.groups.fourier_coefficients.FCGroup(group: h5py.Group, decimation_level_metadata: mt_metadata.processing.fourier_coefficients.Decimation | None = None, **kwargs)[source]

Bases: mth5.groups.BaseGroup

Manage a set of Fourier Coefficients from a single processing run.

Holds Fourier Coefficient estimations organized by decimation level. Each decimation level contains channels (Ex, Ey, Hz, etc.) with complex frequency or time-frequency representations of the input signal.

All channels must use the same calibration. Recalibration requires rerunning the Fourier Coefficient estimation.

hdf5_group

The HDF5 group containing decimation levels

Type:

h5py.Group

metadata[source]

Decimation metadata including time period, sample rates, and channels

Type:

fc.Decimation

Notes

Processing run structure:

  • Multiple decimation levels at different sample rates

  • Each decimation level contains multiple channels

  • Each channel contains complex Fourier coefficients

  • Time period and sample rates define the estimation window

Examples

>>> with h5py.File('data.h5', 'r') as f:
...     fc_run = FCGroup(f['Fourier_Coefficients/run_1'])
...     print(fc_run.decimation_level_summary)
metadata() → mt_metadata.processing.fourier_coefficients.Decimation[source]

Get processing run metadata including all decimation levels.

Collects metadata from all decimation level groups and aggregates into a single Decimation metadata object.

Returns:

Metadata containing time period, sample rates, and all decimation level information.

Return type:

fc.Decimation

Notes

This getter automatically populates:

  • Time period (start and end)

  • List of all decimation levels and their metadata

  • HDF5 reference to this group

Examples

>>> fc_run = FCGroup(h5_group)
>>> metadata = fc_run.metadata
>>> print(metadata.time_period.start)
2023-01-01T00:00:00
property decimation_level_summary: pandas.DataFrame[source]

Get a summary of all decimation levels in this processing run.

Returns information about each decimation level including sample rate, decimation level value, and time span.

Returns:

Summary with columns:

  • decimation_level: Integer decimation level identifier

  • start: ISO format start time of this decimation level

  • end: ISO format end time of this decimation level

  • hdf5_reference: Reference to the HDF5 group

Return type:

pd.DataFrame

Notes

Each row represents a single decimation level containing multiple channels with Fourier coefficients at different sample rates.

Examples

>>> fc_run = FCGroup(h5_group)
>>> summary = fc_run.decimation_level_summary
>>> print(summary[['decimation_level', 'start', 'end']])
   decimation_level                start                  end
0              0     2023-01-01T00:00:00.000000  2023-01-01T01:00:00.000000
1              1     2023-01-01T00:00:00.000000  2023-01-01T02:00:00.000000
add_decimation_level(decimation_level_name: str, decimation_level_metadata: dict | mt_metadata.processing.fourier_coefficients.Decimation | None = None) → FCDecimationGroup[source]

Add a new decimation level to the processing run.

Creates a new FCDecimationGroup for a single decimation level containing Fourier Coefficient channels at a specific sample rate.

Parameters:

  • decimation_level_name (str) – Identifier for the decimation level.

  • decimation_level_metadata (dict | fc.Decimation, optional) – Metadata for the decimation level. Can be a dictionary or fc.Decimation object. Default is None.

Returns:

Newly created decimation level group.

Return type:

FCDecimationGroup

Examples

>>> fc_run = FCGroup(h5_group)
>>> metadata = fc.Decimation(decimation_level=0)
>>> decimation = fc_run.add_decimation_level('0', metadata)
get_decimation_level(decimation_level_name: str) → FCDecimationGroup[source]

Retrieve a decimation level by name.

Parameters:

decimation_level_name (str) – Name or identifier of the decimation level.

Returns:

The requested decimation level group.

Return type:

FCDecimationGroup

Examples

>>> fc_run = FCGroup(h5_group)
>>> decimation = fc_run.get_decimation_level('0')
>>> channels = decimation.groups_list
remove_decimation_level(decimation_level_name: str) → None[source]

Remove a decimation level from the processing run.

Deletes the HDF5 group and all its channels (FCChannelDataset objects).

Parameters:

decimation_level_name (str) – Name or identifier of the decimation level to remove.

Notes

This removes the entire decimation level and all channels within it. To remove individual channels, use FCDecimationGroup.remove_channel() instead.

Examples

>>> fc_run = FCGroup(h5_group)
>>> fc_run.remove_decimation_level('0')
update_metadata() → None[source]

Update processing run metadata from all decimation levels.

Aggregates time period information from all decimation levels and writes updated metadata to HDF5.

Notes

Collects:

  • Earliest start time across all decimation levels

  • Latest end time across all decimation levels

Should be called after adding or removing decimation levels.

Examples

>>> fc_run = FCGroup(h5_group)
>>> fc_run.add_decimation_level('0', metadata0)
>>> fc_run.add_decimation_level('1', metadata1)
>>> fc_run.update_metadata()
supports_aurora_processing_config(processing_config: aurora.config.metadata.processing.Processing, remote: bool) → bool[source]

Check if all required decimation levels exist for Aurora processing.

Performs an all-or-nothing check: returns True only if every decimation level required by the processing config is available in this FCGroup.

Uses sequential logic to short-circuit: if any required decimation level is missing, immediately returns False without checking remaining levels.

Parameters:

  • processing_config (aurora.config.metadata.processing.Processing) – Aurora processing configuration containing required decimation levels.

  • remote (bool) – Whether to check for remote processing compatibility.

Returns:

True if all required decimation levels are available and consistent, False otherwise.

Return type:

bool

Notes

Validation logic:

  1. Extract list of decimation levels from processing config

  2. Iterate through each required level in sequence

  3. For each level, find a matching FCDecimation in this group

  4. Check consistency using Aurora’s validation method

  5. If any level is missing or inconsistent, return False immediately

  6. Return True only if all levels pass validation

Examples

>>> fc_run = FCGroup(h5_group)
>>> config = aurora.config.metadata.processing.Processing(...)
>>> if fc_run.supports_aurora_processing_config(config, remote=False):
...     # All decimation levels are available
...     pass