Coverage for C: \ Users \ peaco \ OneDrive \ Documents \ GitHub \ mth5 \ mth5 \ clients \ phoenix.py: 93%
91 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-27 20:09 -0800
1#!/usr/bin/env python
2# coding: utf-8
4# # Make an MTH5 from Phoenix Data
5#
6# This example demonstrates how to read Phoenix data into an MTH5 file. The data comes from example data in [PhoenixGeoPy](https://github.com/torresolmx/PhoenixGeoPy). Here I downloaded those data into a local folder on my computer by forking the main branch.
8# ## Imports
10# =============================================================================
11# Imports
12# =============================================================================
13from pathlib import Path
15from mt_metadata.timeseries import AppliedFilter
17from mth5 import read_file
18from mth5.clients.base import ClientBase
19from mth5.io.phoenix import PhoenixCollection
20from mth5.io.phoenix.readers.calibrations import PhoenixCalibration
21from mth5.mth5 import MTH5
24# =============================================================================
27class PhoenixClient(ClientBase):
28 def __init__(
29 self,
30 data_path: str | Path,
31 sample_rates: list[int] = [150, 24000],
32 save_path: str | Path | None = None,
33 receiver_calibration_dict: dict | str | Path = {},
34 sensor_calibration_dict: dict | str | Path = {},
35 mth5_filename: str = "from_phoenix.h5",
36 **kwargs: dict,
37 ) -> None:
38 super().__init__(
39 data_path,
40 save_path=save_path,
41 sample_rates=sample_rates,
42 mth5_filename=mth5_filename,
43 **kwargs,
44 )
46 self.receiver_calibration_dict = receiver_calibration_dict
47 self.sensor_calibration_dict = sensor_calibration_dict
49 self.collection = PhoenixCollection(self.data_path)
51 @property
52 def receiver_calibration_dict(self) -> dict:
53 """Receiver calibrations.
55 Returns
56 -------
57 dict
58 Dictionary mapping receiver IDs to calibration file paths.
60 Examples
61 --------
62 >>> client = PhoenixClient('data/path')
63 >>> client.receiver_calibration_dict = {'RX001': Path('RX001_rxcal.json')}
64 >>> client.receiver_calibration_dict
65 {'RX001': Path('RX001_rxcal.json')}
66 """
67 return self._receiver_calibration_dict
69 @receiver_calibration_dict.setter
70 def receiver_calibration_dict(self, value: dict | str | Path) -> None:
71 """Set receiver calibration dictionary from dict or path.
73 Parameters
74 ----------
75 value : dict or str or Path
76 Dictionary of calibrations or path to calibration files.
78 Raises
79 ------
80 TypeError
81 If value is not a dict, str, or Path.
83 Examples
84 --------
85 >>> client = PhoenixClient('data/path')
86 >>> client.receiver_calibration_dict = 'calibrations/'
87 >>> client.receiver_calibration_dict # doctest: +SKIP
88 {'RX001': Path('calibrations/RX001_rxcal.json'), ...}
89 """
90 self._receiver_calibration_dict = {}
91 if isinstance(value, dict):
92 self._receiver_calibration_dict = value
93 elif isinstance(value, (str, Path)):
94 receiver_path = Path(value)
95 if receiver_path.is_dir():
96 self._receiver_calibration_dict = {}
97 for fn in list(receiver_path.glob("*.rxcal.json")) + list(
98 receiver_path.glob("*.rx_cal.json")
99 ):
100 self._receiver_calibration_dict[fn.stem.split("_")[0]] = fn
101 elif receiver_path.is_file():
102 self._receiver_calibration_dict = {}
103 key = receiver_path.stem.split("_")[0]
104 self._receiver_calibration_dict[key] = receiver_path
105 else:
106 raise TypeError(f"type {type(value)} not supported.")
108 @property
109 def sensor_calibration_dict(self) -> dict:
110 """Sensor calibration dictionary.
112 Returns
113 -------
114 dict
115 Dictionary mapping sensor IDs to PhoenixCalibration objects.
117 Examples
118 --------
119 >>> client = PhoenixClient('data/path')
120 >>> client.sensor_calibration_dict = {'H001': PhoenixCalibration('H001_scal.json')}
121 >>> client.sensor_calibration_dict['H001'] # doctest: +SKIP
122 <PhoenixCalibration object>
123 """
124 return self._sensor_calibration_dict
126 @sensor_calibration_dict.setter
127 def sensor_calibration_dict(self, value: dict | str | Path) -> None:
128 """Set sensor calibration dictionary from dict or path.
130 Parameters
131 ----------
132 value : dict or str or Path
133 Dictionary of calibrations or path to calibration files.
135 Raises
136 ------
137 ValueError
138 If value is not a dict, str, or Path.
140 Examples
141 --------
142 >>> client = PhoenixClient('data/path')
143 >>> client.sensor_calibration_dict = 'calibrations/'
144 >>> client.sensor_calibration_dict # doctest: +SKIP
145 {'H001': <PhoenixCalibration object>, ...}
146 """
147 if isinstance(value, dict):
148 self._sensor_calibration_dict = value
149 elif isinstance(value, (str, Path)):
150 self._sensor_calibration_dict = {}
151 cal_path = Path(value)
152 if cal_path.is_dir():
153 for fn in cal_path.glob("*scal.json"):
154 self._sensor_calibration_dict[
155 fn.stem.split("_")[0]
156 ] = PhoenixCalibration(fn)
157 elif cal_path.is_file():
158 if not cal_path.exists():
159 raise IOError(f"Could not find {cal_path}")
160 key = cal_path.stem.split("_")[0]
161 self._sensor_calibration_dict[key] = PhoenixCalibration(cal_path)
162 else:
163 raise ValueError("calibration_path cannot be None")
165 def make_mth5_from_phoenix(self, **kwargs: dict) -> str | Path | None:
166 """Make an MTH5 from Phoenix files.
168 Split into runs, account for filters. Updates the MTH5 file with Phoenix data.
170 Parameters
171 ----------
172 **kwargs : dict
173 Optional keyword arguments to override instance attributes.
175 Returns
176 -------
177 str, Path, or None
178 Path to the saved MTH5 file.
180 Examples
181 --------
182 >>> client = PhoenixClient('data/path', save_path='output.h5')
183 >>> client.make_mth5_from_phoenix()
184 'output.h5'
185 """
186 for key, value in kwargs.items():
187 if value is not None:
188 setattr(self, key, value)
190 run_dict = self.get_run_dict()
192 with MTH5(**self.h5_kwargs) as m:
193 m.open_mth5(self.save_path, self.mth5_file_mode)
195 for station_id, station_dict in run_dict.items():
196 collection_metadata = self.collection.metadata_dict[station_id]
197 survey_metadata = collection_metadata.survey_metadata
198 survey_group = m.add_survey(survey_metadata.id)
200 station_metadata = self.collection.metadata_dict[
201 station_id
202 ].station_metadata
203 station_group = survey_group.stations_group.add_station(
204 station_metadata.id, station_metadata=station_metadata
205 )
206 for run_id, run_df in station_dict.items():
207 run_metadata = collection_metadata.run_metadata
208 run_metadata.id = run_id
209 run_metadata.sample_rate = float(run_df.sample_rate.unique()[0])
211 run_group = station_group.add_run(
212 run_metadata.id, run_metadata=run_metadata
213 )
214 for row in run_df.itertuples():
215 try:
216 ch_ts = read_file(
217 row.fn,
218 **{
219 "channel_map": collection_metadata.channel_map,
220 "rxcal_fn": self.receiver_calibration_dict[
221 collection_metadata.instrument_id
222 ],
223 },
224 )
225 except OSError:
226 print(f"OSError: skipping {row.fn.name} likely too small")
227 continue
229 if ch_ts.component in ["h1", "h2", "h3"]:
230 # for phx coils from generic response curves
231 if (
232 ch_ts.channel_metadata.sensor.id
233 in self.sensor_calibration_dict.keys()
234 ):
235 pc = self.sensor_calibration_dict[
236 ch_ts.channel_metadata.sensor.id
237 ]
238 for key in pc.__dict__.keys():
239 if key.startswith("h"):
240 break
241 coil_fap = getattr(pc, key)
243 # add filter
244 applied_filter = AppliedFilter(
245 name=coil_fap.name, applied=True, stage=1
246 )
247 ch_ts.channel_metadata.add_filter(applied_filter)
248 ch_ts.channel_response.filters_list.append(coil_fap)
249 else:
250 self.logger.warning(
251 f"Could not find coil {ch_ts.channel_metadata.sensor.id} in sensor calibrations."
252 )
254 # add channel to the run group
255 run_group.from_channel_ts(ch_ts)
257 run_group.update_metadata()
259 station_group.update_metadata()
260 survey_group.update_metadata()
262 return self.save_path