Coverage for C: \ Users \ peaco \ OneDrive \ Documents \ GitHub \ mt_metadata \ mt_metadata \ timeseries \ filters \ helper_functions.py: 70%
50 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-10 00:11 -0800
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-10 00:11 -0800
1from loguru import logger
3from mt_metadata.timeseries.filters import (
4 CoefficientFilter,
5 FrequencyResponseTableFilter,
6)
9def make_coefficient_filter(gain=1.0, name="generic coefficient filter", **kwargs):
10 """
12 Parameters
13 ----------
14 gain: float
15 name: string
16 units_in : string
17 A supported unit or "unknown"
18 one of "digital counts", "millivolts", etc.
19 A complete list of units can be found in mt_metadata/mt_metadata/util/units.py
20 and is accessible as a table via:
21 from mt_metadata.common.units import UNITS_DF
23 Returns
24 -------
26 """
27 # in general, you need to add all required fields from the standards.json
28 default_units_in = "unknown"
29 default_units_out = "unknown"
31 cf = CoefficientFilter()
32 cf.gain = gain
33 cf.name = name
35 cf.units_in = kwargs.get("units_in", default_units_in)
36 cf.units_out = kwargs.get("units_out", default_units_out)
38 return cf
41def make_frequency_response_table_filter(file_path, case="bf4"):
42 """
43 Parameters
44 ----------
45 filepath: pathlib.Path or string
46 case : string, placeholder for handling different fap table formats.
48 Returns
49 -------
50 fap_filter: FrequencyResponseTableFilter
51 """
52 fap_filter = FrequencyResponseTableFilter()
54 if case == "bf4":
55 import numpy as np
56 import pandas as pd
58 df = pd.read_csv(file_path) # , skiprows=1)
59 # Hz, V/nT, degrees
60 fap_filter.frequencies = df["Frequency [Hz]"].values
61 fap_filter.amplitudes = df["Amplitude [V/nT]"].values
62 fap_filter.phases = np.deg2rad(df["Phase [degrees]"].values)
63 fap_filter.units_in = "volt"
64 fap_filter.units_out = "nanotesla"
65 fap_filter.gain = 1.0
66 fap_filter.name = "bf4"
67 return fap_filter
68 else:
69 msg = f"Case {case} not supported for FAP Table"
70 logger.error(msg)
71 raise NotImplementedError(msg)
74def make_volt_per_meter_to_millivolt_per_km_converter():
75 """
76 This represents a filter that converts from mV/km to V/m.
78 Returns
79 -------
81 """
82 coeff_filter = make_coefficient_filter(
83 gain=1e-6,
84 units_in="millivolt per kilometer",
85 units_out="volt per meter",
86 name="MT to SI electric field conversion",
87 )
88 return coeff_filter
91def make_tesla_to_nanotesla_converter():
92 """
93 This represents a filter that converts from nt to T.
95 Returns
96 -------
98 """
99 coeff_filter = make_coefficient_filter(
100 gain=1e-9,
101 units_in="nanotesla",
102 units_out="tesla",
103 name="MT to SI magnetic field conversion",
104 )
105 return coeff_filter
108def decimation_info_is_degenerate(obspy_stage):
109 """
110 Check a few condtions that may apply to an obspy stage which if true
111 imply that the decimation information can be stripped out as it bears
112 no information about aany data transformation;
113 Case 1: All these attrs are None decimation has no information:
114 {'decimation_input_sample_rate', 'decimation_factor',
115 'decimation_offset', 'decimation_delay', 'decimation_correction'}
116 Case 2:
118 """
119 cond1 = obspy_stage.stage_gain in [1.0, None]
120 cond2 = obspy_stage.decimation_factor in [1, None]
122 if cond1 & cond2:
123 return True
124 else:
125 return False
128def decimation_info_is_pure_delay(stage):
129 cond1 = stage.stage_gain == 1.0
130 cond2 = stage.decimation_factor == 1
131 cond3 = stage.decimation_delay != 0.0
132 cond4 = stage.decimation_correction == 0.0
134 if cond1 & cond2 & cond3 & cond4:
135 return True
136 else:
137 return False
140# def stage_gain_is_degenerate():
141# # if gain is 1.0 ignore it
142# pass
145MT2SI_ELECTRIC_FIELD_FILTER = make_volt_per_meter_to_millivolt_per_km_converter()
146MT2SI_MAGNETIC_FIELD_FILTER = make_tesla_to_nanotesla_converter()