Coverage for C: \ Users \ peaco \ OneDrive \ Documents \ GitHub \ mt_metadata \ mt_metadata \ timeseries \ stationxml \ xml_station_mt_station.py: 77%
155 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-10 00:11 -0800
1# -*- coding: utf-8 -*-
2"""
3Created on Thu Feb 18 12:49:13 2021
5:copyright:
6:copyright:
7 Jared Peacock (jpeacock@usgs.gov)
9:license: MIT
11"""
12from mt_metadata import timeseries as metadata
13from mt_metadata.base.helpers import requires
14from mt_metadata.timeseries.stationxml import XMLEquipmentMTRun
16# =============================================================================
17# Imports
18# =============================================================================
19from mt_metadata.timeseries.stationxml.fdsn_tools import release_dict
20from mt_metadata.timeseries.stationxml.utils import BaseTranslator
23try:
24 from obspy.core import inventory
25except ImportError:
26 inventory = None
28# =============================================================================
31@requires(obspy=inventory)
32class XMLStationMTStation(BaseTranslator):
33 """
34 translate back and forth between StationXML Station and MT Station
35 """
37 def __init__(self):
38 """
39 Initialize the XMLStationMTStation converter.
41 Sets up translation dictionaries between StationXML and MT metadata
42 attributes, and defines which MT attributes should be stored as comments.
43 """
44 super().__init__()
46 self.xml_translator.update(
47 {
48 "alternate_code": "id",
49 "channels": None,
50 "code": "fdsn.id",
51 "comments": "provenance.comments.value",
52 "creation_date": "time_period.start",
53 "data_availability": None,
54 "description": "comments.value",
55 "elevation": "location.elevation",
56 "end_date": "time_period.end",
57 "equipments": None,
58 "external_references": None,
59 "geology": None,
60 "identifiers": None,
61 "latitude": "location.latitude",
62 "longitude": "location.longitude",
63 "operators": "special",
64 "site": "special",
65 "start_date": "time_period.start",
66 "termination_date": "time_period.end",
67 "vault": None,
68 "water_level": None,
69 "restricted_status": "special",
70 }
71 )
73 # StationXML to MT Survey
74 self.mt_translator = self.flip_dict(self.xml_translator)
75 self.mt_translator["geographic_name"] = "site"
76 self.mt_translator["provenance.comments.value"] = None
77 self.mt_translator["time_period.start"] = "start_date"
78 self.mt_translator["time_period.end"] = "end_date"
80 self.mt_comments_list = [
81 "run_list",
82 "orientation.method",
83 "orientation.reference_frame",
84 "location.declination.value",
85 "location.declination.model",
86 "location.declination.comments",
87 "provenance.software.author",
88 "provenance.software.name",
89 "provenance.software.version",
90 "data_type",
91 ]
93 def xml_to_mt(self, xml_station) -> metadata.Station:
94 """
95 Translate a StationXML station object to MT Station object.
97 Parameters
98 ----------
99 xml_station : obspy.core.inventory.Station
100 StationXML station element to convert.
102 Returns
103 -------
104 mt_metadata.timeseries.Station
105 MT Station object with attributes populated from the XML station.
107 Raises
108 ------
109 ValueError
110 If input is not an obspy.core.inventory.Station object.
111 """
113 if not isinstance(xml_station, inventory.Station):
114 msg = f"Input must be obspy.core.inventory.station object not {type(xml_station)}"
115 self.logger.error(msg)
116 raise ValueError(msg)
118 mt_station = metadata.Station()
120 ## read comments
121 run_comments = []
122 for comment in xml_station.comments:
123 key, value = self.read_xml_comment(comment)
124 if "mt.run" in key:
125 run_comments.append({key: value})
126 continue
128 try:
129 key = key.split("mt.station.")[1]
130 except IndexError:
131 pass
133 if "summary" in key:
134 key = key.replace("summary", "comments")
135 if key in ["comments"]:
136 if mt_station.comments.value:
137 mt_station.comments.value += value
138 else:
139 mt_station.comments.value = value
140 else:
141 mt_station.update_attribute(key, value)
143 for mt_key, xml_key in self.mt_translator.items():
144 if xml_key is None:
145 continue
146 if xml_key in ["site"]:
147 site = xml_station.site
148 mt_station.geographic_name = site.name
149 else:
150 value = getattr(xml_station, xml_key)
151 if value is None:
152 continue
153 if isinstance(value, (list, tuple)):
154 for k, v in zip(mt_key, value):
155 mt_station.update_attribute(k, v)
156 else:
157 if xml_key == "restricted_status":
158 value = self.flip_dict(release_dict)[value]
160 mt_station.update_attribute(mt_key, value)
162 if mt_station.id is None:
163 if mt_station.fdsn.id is not None:
164 mt_station.id = mt_station.fdsn.id
166 # read in equipment information
167 mt_station = self._equipments_to_runs(xml_station.equipments, mt_station)
168 mt_station = self._equipments_to_runs(xml_station.equipments, mt_station)
169 mt_station = self._add_run_comments(run_comments, mt_station)
171 return mt_station
173 def mt_to_xml(self, mt_station: metadata.Station):
174 """
175 Convert MT Station to ObsPy StationXML Station object.
177 Parameters
178 ----------
179 mt_station : mt_metadata.timeseries.Station
180 MT Station object to convert.
182 Returns
183 -------
184 obspy.core.inventory.Station
185 StationXML Station object with attributes populated from MT Station.
187 Raises
188 ------
189 ValueError
190 If input is not an mt_metadata.timeseries.Station object,
191 or if both id and fdsn.id attributes are None.
193 Notes
194 -----
195 Station code is set to uppercase in the resulting StationXML object.
196 """
198 if not isinstance(mt_station, metadata.Station):
199 msg = f"Input must be mt_metadata.timeseries.Station object not {type(mt_station)}"
200 self.logger.error(msg)
201 raise ValueError(msg)
203 if mt_station.id is None:
204 if mt_station.fdsn.id is None:
205 msg = "Need to input id or fdsn.id, both cannot be None"
206 self.logger.error(msg)
207 raise ValueError(msg)
208 else:
209 code = mt_station.fdsn.id
210 else:
211 code = mt_station.id
213 if mt_station.fdsn.id is None:
214 mt_station.fdsn.id = mt_station.id
216 xml_station = inventory.Station(
217 code.upper(),
218 mt_station.location.latitude,
219 mt_station.location.longitude,
220 mt_station.location.elevation,
221 )
223 for xml_key, mt_key in self.xml_translator.items():
224 # need to skip code because we just set it above and it needs to be upper.
226 if mt_key is None:
227 self.logger.debug(
228 f"Cannot currently map mt_key.station to inventory.station.{xml_key}"
229 )
230 continue
232 if xml_key in ["code"]:
233 continue
234 elif xml_key in ["alternate_code"]:
235 xml_station.alternate_code = mt_station.id.upper()
237 elif xml_key == "operators":
238 if mt_station.acquired_by.author:
239 if mt_station.acquired_by.organization is None:
240 mt_station.acquired_by.organization = " "
241 operator = inventory.Operator(
242 agency=mt_station.acquired_by.organization
243 )
244 person = inventory.Person(names=[mt_station.acquired_by.author])
245 person = inventory.Person(names=[mt_station.acquired_by.author])
246 operator.contacts = [person]
247 xml_station.operators = [operator]
249 elif xml_key == "site":
250 if mt_station.geographic_name is None:
251 xml_station.site.name = mt_station.id.upper()
252 self.logger.warning(
253 f"Station.geographic_name is None, using Station.id = {mt_station.id}."
254 "Check StationXML site.name."
255 )
256 else:
257 xml_station.site.name = mt_station.geographic_name
259 elif xml_key == "comments":
260 if mt_station.comments is not None:
261 comment = inventory.Comment(mt_station.comments.value)
262 xml_station.comments.append(comment)
263 elif xml_key == "restricted_status":
264 xml_station.restricted_status = release_dict[
265 xml_station.restricted_status
266 ]
267 elif "time_period" in mt_key:
268 value = mt_station.get_attr_from_name(mt_key).time_stamp
269 setattr(xml_station, xml_key, value)
270 else:
271 setattr(xml_station, xml_key, mt_station.get_attr_from_name(mt_key))
272 setattr(xml_station, xml_key, mt_station.get_attr_from_name(mt_key))
274 # add mt comments
275 xml_station.comments = self.make_mt_comments(mt_station, "mt.station")
277 # add run information
278 for mt_run in mt_station.runs:
279 run_converter = XMLEquipmentMTRun()
280 xml_station.equipments.append(run_converter.mt_to_xml(mt_run))
281 xml_station.comments += run_converter.make_mt_comments(
282 mt_run, f"mt.run:{mt_run.id}"
283 )
285 return xml_station
287 def _equipments_to_runs(
288 self, equipments, station_obj: metadata.Station
289 ) -> metadata.Station:
290 """
291 Convert equipment list to station runs.
293 Parameters
294 ----------
295 equipments : list
296 List of StationXML Equipment objects. (inventory.Equipement)
297 station_obj : mt_metadata.timeseries.Station
298 MT Station object to add runs to.
300 Returns
301 -------
302 mt_metadata.timeseries.Station
303 Updated MT Station object with runs added.
305 Raises
306 ------
307 TypeError
308 If equipments parameter is not a list.
309 """
310 if not isinstance(equipments, list):
311 msg = f"Input must be a list not {type(equipments)}"
312 self.logger.error(msg)
313 raise TypeError(msg)
315 for equipment in equipments:
316 run_translator = XMLEquipmentMTRun()
317 run_item = run_translator.xml_to_mt(equipment)
318 run_index = station_obj.run_index(run_item.id)
319 if run_index:
320 station_obj.runs[run_index].from_dict(run_item.to_dict())
321 else:
322 station_obj.add_run(run_item)
324 return station_obj
326 def _add_run_comments(
327 self, run_comments: list[dict], station_obj: metadata.Station
328 ) -> metadata.Station:
329 """
330 Add StationXML comments to MT Station run objects.
332 Parameters
333 ----------
334 run_comments : list of dict
335 List of dictionaries containing run comments.
336 Each dict should be in the format {key: value} where key
337 includes the run ID and attribute name.
338 station_obj : mt_metadata.timeseries.Station
339 MT Station object with runs to update.
341 Returns
342 -------
343 mt_metadata.timeseries.Station
344 Updated MT Station object with comments added to runs.
346 Notes
347 -----
348 Comment keys should be in the format "mt.run:run_id" or
349 "mt.run:run_id.attribute" to specify which run and attribute
350 to update.
351 """
352 for comment in run_comments:
354 for rkey, rvalue in comment.items():
355 run_id = rkey.split(":", 1)[1]
356 run_attr = None
357 if run_id.count(".") > 0:
358 run_id, run_attr = run_id.split(".", 1)
359 run_index = station_obj.run_index(run_id)
360 if run_index is None:
361 continue
362 if isinstance(rvalue, dict):
363 for ckey, cvalue in rvalue.items():
364 if run_attr:
365 if run_attr == "comments":
366 value = f"{ckey}: {cvalue}"
367 try:
368 station_obj.runs[
369 run_index
370 ].comments.value += f", {value}"
371 except TypeError:
372 station_obj.runs[run_index].comments.value = value
373 else:
374 c_attr = f"{run_attr}.{ckey}"
376 station_obj.runs[run_index].update_attribute(
377 c_attr, cvalue
378 )
379 else:
380 station_obj.runs[run_index].update_attribute(ckey, cvalue)
382 return station_obj