mt_metadata.processing.aurora.decimation_level

This module contains the DecimationLevel class. TODO: Factor or rename. The decimation level class here has information about the entire processing.

Classes

SaveFcsTypeEnum	str(object='') -> str
DecimationLevel	Base class for all metadata objects with Pydantic validation.

Functions

get_fft_harmonics(samples_per_window, sample_rate)

Works for odd and even number of points.

Module Contents

class mt_metadata.processing.aurora.decimation_level.SaveFcsTypeEnum

Bases: mt_metadata.common.enumerations.StrEnumerationBase

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to ‘utf-8’. errors defaults to ‘strict’.

h5 = 'h5'

csv = 'csv'

class mt_metadata.processing.aurora.decimation_level.DecimationLevel(**data)

Bases: mt_metadata.base.MetadataBase

Base class for all metadata objects with Pydantic validation.

MetadataBase extends DotNotationBaseModel (which inherits from Pydantic’s BaseModel) to provide automatic validation according to metadata standards. It adds functionality beyond dictionaries, supporting JSON, XML, pandas Series, and other formats for metadata interchange.

_skip_equals

Private attribute listing fields to skip in equality comparisons

Type:

list[str]

_fields

Private attribute caching field information

Type:

dict[str, Any]

Notes

• All field assignments are validated automatically via Pydantic

• None values are converted to appropriate defaults (empty string or 0.0)

• Supports nested attribute access via dot notation

• Thread-safe for read operations after initialization

bands: Annotated[list[mt_metadata.common.band.Band], Field(default_factory=list, description='List of bands', json_schema_extra={'units': None, 'required': True, 'examples': ['[]']})]

channel_weight_specs: Annotated[List[mt_metadata.features.weights.ChannelWeightSpec], Field(default_factory=list, description='List of weighting schemes to use for TF processing for each output channel', alias=None, json_schema_extra={'units': None, 'required': True, 'examples': ['[]']})]

input_channels: Annotated[list[str], Field(default_factory=list, description='list of input channels (sources)', alias=None, json_schema_extra={'units': None, 'required': True, 'examples': ['hx, hy']})]

output_channels: Annotated[list[str], Field(default_factory=list, description='list of output channels (responses)', alias=None, json_schema_extra={'units': None, 'required': True, 'examples': ['ex, ey, hz']})]

reference_channels: Annotated[list[str], Field(default_factory=list, description='list of reference channels (remote sources)', alias=None, json_schema_extra={'units': None, 'required': True, 'examples': ['hx, hy']})]

save_fcs: Annotated[bool, Field(default=False, description='Whether the Fourier coefficients are saved [True] or not [False].', alias=None, json_schema_extra={'units': None, 'required': True, 'examples': [True]})]

save_fcs_type: Annotated[SaveFcsTypeEnum | None, Field(default=None, description='Format to use for fc storage', alias=None, json_schema_extra={'units': None, 'required': False, 'examples': ['h5']})]

decimation: Annotated[mt_metadata.processing.TimeSeriesDecimation, Field(default_factory=Decimation, description='Decimation settings', alias=None, json_schema_extra={'units': None, 'required': False, 'examples': ['Decimation()']})]

estimator: Annotated[mt_metadata.processing.aurora.estimator.Estimator, Field(default_factory=Estimator, description='Estimator settings', alias=None, json_schema_extra={'units': None, 'required': False, 'examples': ['Estimator()']})]

regression: Annotated[mt_metadata.processing.aurora.regression.Regression, Field(default_factory=Regression, description='Regression settings', alias=None, json_schema_extra={'units': None, 'required': False, 'examples': ['Regression()']})]

stft: Annotated[mt_metadata.processing.ShortTimeFourierTransform, Field(default_factory=STFT, description='Short-time Fourier transform settings', alias=None, json_schema_extra={'units': None, 'required': False, 'examples': ['STFT()']})]

classmethod validate_channel_weight_specs(value, info)

Validator for channel_weight_specs field.

classmethod validate_bands(value, info)

add_band(band)

add a band

property lower_bounds: numpy.ndarray

get lower bounds index values into an array.

property upper_bounds: numpy.ndarray

get upper bounds index values into an array.

property bands_dataframe: pandas.DataFrame

Utility function that transforms a list of bands into a dataframe

See notes in _df_from_bands.

Returns:

bands_df – Same format as that generated by EMTFBandSetupFile.get_decimation_level()

Return type:

pd.Dataframe

property frequency_sample_interval: float

Returns the delta_f in frequency domain df = 1 / (N * dt) Here dt is the sample interval after decimation

Returns:

frequency_sample_interval – The frequency sample interval after decimation.

Return type:

float

property band_edges: numpy.ndarray

Returns the band edges as a numpy array

Returns:

band_edges

Return type:

2D numpy array, one row per frequency band and two columns

frequency_bands_obj()

Gets a FrequencyBands object that is used as input to processing.

Used by Aurora.

TODO: consider adding .to_frequency_bands() method directly to self.bands

Returns:

frequency_bands – A FrequencyBands object that can be used as an iterator for processing.

Return type:

FrequencyBands

property fft_frequencies: numpy.ndarray

Gets the harmonics of the STFT.

Returns:

freqs – The frequencies at which the stft will be available.

Return type:

np.ndarray

property harmonic_indices: List[int]

Loops over all bands and returns a list of the harminic indices. TODO: Distinguish the bands which are a processing construction vs harmonic indices which are FFT info.

Returns:

return_list – The indices of the harmonics that are needed for processing.

Return type:

list of integers

property local_channels

is_consistent_with_archived_fc_parameters(fc_decimation, remote)

Usage: For an already existing spectrogram stored in an MTH5 archive, this compares the metadata within the archive (fc_decimation) with an aurora decimation level (self), and tells whether the parameters are in agreement. If True, this allows aurora to skip the calculation of FCs and instead read them from the archive.

TODO: Merge all checks of TimeSeriesDecimation parameters into a single check. - e.g. Compress all decimation checks to: assert fc_decimation.decimation == self.decimation

decimation_level: FCDecimation

metadata describing the parameters used to compute an archived spectrogram

remote: bool

If True, we are looking for reference channels, not local channels in the FCGroup.

Iterates over FCDecimation attributes:

“channels_estimated”: to ensure all expected channels are in the group “decimation.anti_alias_filter”: check that the expected AAF was applied “decimation.sample_rate, “decimation.method”, “stft.prewhitening_type”, “stft.recoloring”, “stft.pre_fft_detrend_type”, “stft.min_num_stft_windows”, “stft.window”, “stft.harmonic_indices”,

Return type:

return:

to_fc_decimation(remote=False, ignore_harmonic_indices=True)

Generates a FC Decimation() object for use with FC Layer in mth5.

TODO: this is being tested only in aurora – move a test to mt_metadata or move the method. Ignoring for now these properties “time_period.end”: “1980-01-01T00:00:00+00:00”, “time_period.start”: “1980-01-01T00:00:00+00:00”,

TODO: FIXME: Assignment of TSDecimation can be done in one shot once #235 is addressed.

Parameters:

• remote (bool) – If True, use reference channels, if False, use local_channels. We may wish to not pass remote=True when _building_ FCs however, because then not all channels will get built.

• ignore_harmonic_indices (bool) – If True, leave harmonic indices at default [-1,], which means all indices. If False, only the specific harmonic indices needed for processing will be stored. Thus, when building FCs, it maybe best to leave this as True, that way all FCs will be stored, so if the band setup is changed, the FCs will still be there.

• Returns – fc_dec_obj:mt_metadata.transfer_functions.processing.fourier_coefficients.decimation.Decimation A decimation object configured for STFT processing

mt_metadata.processing.aurora.decimation_level.get_fft_harmonics(samples_per_window, sample_rate)

Works for odd and even number of points.

Development notes: Could be modified with kwargs to support one_sided, two_sided, ignore_dc ignore_nyquist, and etc. Consider taking FrequencyBands as an argument.

Parameters:

• samples_per_window (integer) – Number of samples in a window that will be Fourier transformed.

• sample_rate (float) – Inverse of time step between samples, Samples per second

Returns:

harmonic_frequencies – The frequencies that the fft will be computed. These are one-sided (positive frequencies only) Does not return Nyquist Does return DC component

Return type:

numpy array