mt_metadata.features.coherence ============================== .. py:module:: mt_metadata.features.coherence .. autoapi-nested-parse:: This module contains the simplest coherence feature. The feature is computed with scipy.signal coherence. Note that this coherence is one number for the entire time-series (per frequency), i.e. The Window object is used to taper the time series before FFT. Development Notes: Coherence extends the Feature class. This means that it should have all the attrs that a Feature instance does, as well as its own unique ones. When setting up the attr_dict, one is confronted with the question of adding BaseFeatures attrs one of two ways: - To add the features directly, use: attr_dict.add_dict(get_schema("base_feature", SCHEMA_FN_PATHS)) { "coherence": { "ch1": "ex", "ch2": "hy", "description": "Simple coherence between two channels derived directly from scipy.signal.coherence applied to time domain data", "domain": "frequency", "name": "coherence", "window.clock_zero_type": "ignore", "window.normalized": true, "window.num_samples": 512, "window.overlap": 128, "window.type": "hamming" } } - To nest the features use: attr_dict.add_dict(BaseFeature()._attr_dict, "base_feature") { "coherence": { "base_feature.description": null, "base_feature.domain": null, "base_feature.name": null, "ch1": "ex", "ch2": "hy", "window.clock_zero_type": "ignore", "window.normalized": true, "window.num_samples": 512, "window.overlap": 128, "window.type": "hamming" } } Devlopment Notes: To specify a channel in the context of tf processing we need station and channel names. I have been fighting the use of `rx` and `ry` for several reasons, including that the [ex, ey, hx, hy, hz, rx, ry] convention forces the assumption that remote channels are remote magnetics, and are overly specific to the remote reference processing convention. Hoever, for a feature like this, it could seem to be a hassle to update the processing config with the station name all over the feature definations. So, it seems that we should have a station field, and a channel field. If the user wishes to specify station and channel, fine. If the user prefers the more general, but less well defined [ex, ey, hx, hy, hz, rx, ry] nomenclature, then we can ddeduce this for them. Development Note (2025-05-24): Note that the simple coherence as computed here, just returns one number per frequency. It is the average coherence over the entire run, and is not innately a "per-time-window feature". To make it a per-time-window feature, we need to apply the transform on individual windows (not the whole run). i.e. chunk a run into sub-windows, and then compute coherence on each of those individually. To accomplish this we must shorten the window.num_samples to be smaller than the sub-window size, otherwise, coherence degenerates to 1 everywhere. (Recall coherenc is th average cross-power over average sqrt auto-powers, and having only one spectral estimate means there is no averaging). Selection of an appropriate "window-within-the-sub-window" for spectral esimation comes with some caveats; The length of the window-within-the-window must be small enough to get at least a few) spectral estimates, meaning that the frequency content will not mirror that of the FFT That said, we can know our lowest frequency of TF estimation (usually no fewer than 5 cycles), so we could set the window-within-window width to be, say 1/5 the FFT window length, and then we'll get something we can use, although it will be somwhat unrealiable at long period (but so is everything else:/). Note that when we are using long FFT windows (such as for HF data processing) this is not such a concern Way Forward: A "StridingWindowCoherence" (effectively a spectrogram of coherence) can be an extension of the Cohernece feature. It will have the same properties, but will also have a "SubWindow". The SubWindow will be another window function object, but it can be parameterized, for example, as a fraction of the "Spectrogram Sliding Window". The compute function could possibly be done by computing Coherence on each sub-window (kinda elegant but may wind up being a bit slow with all the for-looping) Classes ------- .. autoapisummary:: mt_metadata.features.coherence.DetrendEnum mt_metadata.features.coherence.Coherence Module Contents --------------- .. py:class:: DetrendEnum Bases: :py:obj:`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'. .. py:attribute:: linear :value: 'linear' .. py:attribute:: constant :value: 'constant' .. py:class:: Coherence(**data) Bases: :py:obj:`mt_metadata.features.feature.Feature` 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. .. attribute:: _skip_equals Private attribute listing fields to skip in equality comparisons :type: list[str] .. attribute:: _fields Private attribute caching field information :type: dict[str, Any] .. rubric:: 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 .. py:attribute:: channel_1 :type: Annotated[str, Field(default='', description='The first channel of two channels in the coherence calculation.', alias=None, json_schema_extra={'units': None, 'required': True, 'examples': ['ex']})] .. py:attribute:: channel_2 :type: Annotated[str, Field(default='', description='The second channel of two channels in the coherence calculation.', alias=None, json_schema_extra={'units': None, 'required': True, 'examples': ['hy']})] .. py:attribute:: detrend :type: Annotated[DetrendEnum, Field(default=DetrendEnum.linear, description='How to detrend the data segments before fft.', alias=None, json_schema_extra={'units': None, 'required': True, 'examples': ['constant']})] .. py:attribute:: station_1 :type: Annotated[str | None, Field(default=None, description='The station associated with the first channel in the coherence calculation.', alias=None, json_schema_extra={'units': None, 'required': True, 'examples': ['PKD']})] .. py:attribute:: station_2 :type: Annotated[str | None, Field(default=None, description='The station associated with the second channel in the coherence calculation.', alias=None, json_schema_extra={'units': None, 'required': True, 'examples': ['SAO']})] .. py:attribute:: window :type: Annotated[mt_metadata.processing.window.Window, Field(default=Window(num_samples=256, overlap=128, type='hamming'), description='The window function to apply to the data segments before fft.', alias=None, json_schema_extra={'units': None, 'required': True, 'examples': [{'type': 'hamming', 'num_samples': 256, 'overlap': 128}]})] .. py:method:: set_defaults(data) :classmethod: .. py:property:: channel_pair_str :type: str .. py:method:: validate_station_ids(local_station_id, remote_station_id = None) Make sure that ch1, ch2 are unambiguous. Ideally the station for each channel is specified, but if not, try deducing the channel. :param local_station_id: The name of the local station for a TF calculation :type local_station_id: str :param remote_station_id: The name of the remote station for a TF calculation :type remote_station_id: Optional[str] .. py:method:: compute(ts_1, ts_2) Calls scipy's coherence function. TODO: Consider making this return an xarray indexed by frequency. :param ts_1: :param ts_2: