Coverage for C: \ Users \ peaco \ OneDrive \ Documents \ GitHub \ mt_metadata \ mt_metadata \ timeseries \ stationxml \ xml_network_mt_survey.py: 91%
106 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-10 00:11 -0800
1# -*- coding: utf-8 -*-
2"""
3Created on Tue Feb 16 10:18:29 2021
5:copyright:
6 Jared Peacock (jpeacock@usgs.gov)
8:license: MIT
10"""
11from mt_metadata import timeseries as metadata
12from mt_metadata.base.helpers import requires
14# =============================================================================
15# Imports
16# =============================================================================
17from mt_metadata.timeseries.stationxml.fdsn_tools import release_dict
18from mt_metadata.timeseries.stationxml.utils import BaseTranslator
21try:
22 from obspy.core import inventory
23except ImportError:
24 inventory = None
26# =============================================================================
29@requires(obspy=inventory)
30class XMLNetworkMTSurvey(BaseTranslator):
31 """
32 translate back and forth between StationXML Network and MT Survey
33 """
35 def __init__(self):
36 super().__init__()
38 self.xml_translator.update(
39 {
40 "description": "summary",
41 "comments": "comments",
42 "start_date": "time_period.start_date",
43 "end_date": "time_period.end_date",
44 "restricted_status": "release_license",
45 "operators": "project_lead",
46 "code": "fdsn.network",
47 "identifiers": "citation_dataset.doi",
48 "alternate_code": "id",
49 }
50 )
52 # StationXML to MT Survey
53 self.mt_translator = self.flip_dict(self.xml_translator)
54 self.mt_translator["project_lead"] = "operators"
55 self.mt_translator["name"] = "alternate_code"
57 self.mt_comments_list = [
58 "country",
59 "geographic_name",
60 "citation_journal.doi",
61 "id",
62 "project",
63 "acquired_by.author",
64 "acquired_by.comments.value",
65 ]
67 def xml_to_mt(self, network):
68 """
69 Translate a StationXML Network object to MT Survey object
71 :param network: StationXML network element
72 :type network: :class:`obspy.core.inventory.Network`
74 """
76 if not isinstance(network, inventory.Network):
77 msg = (
78 f"Input must be obspy.core.inventory.Network object not {type(network)}"
79 )
80 self.logger.error(msg)
81 raise ValueError(msg)
82 mt_survey = metadata.Survey()
84 for mt_key, xml_key in self.mt_translator.items():
85 if mt_key == "project_lead":
86 name = []
87 email = []
88 org = []
89 for operator in network.operators:
90 org.append(operator.agency)
91 for person in operator.contacts:
92 name.append(", ".join(person.names))
93 email.append(", ".join(person.emails))
94 if name:
95 mt_survey.update_attribute("project_lead.author", ", ".join(name))
96 if email:
97 mt_survey.update_attribute("project_lead.email", ", ".join(email))
98 if org:
99 mt_survey.update_attribute(
100 "project_lead.organization", ", ".join(org)
101 )
102 elif mt_key in ["citation_dataset.doi"]:
103 mt_survey.update_attribute(
104 mt_key, self.read_xml_identifier(network.identifiers)
105 )
106 elif mt_key in ["comments"]:
107 for comment in network.comments:
108 key, value = self.read_xml_comment(comment)
109 if "doi" in key:
110 # need to make a doi a proper web location.
111 if not value.startswith("http"):
112 if "doi.org" not in value:
113 value = f"https://doi.org/{value}"
114 else:
115 value = f"https://{value}"
116 if "mt.survey" in key:
117 key = key.split("mt.survey.")[1]
118 if "summary" in key:
119 key = key.replace("summary", "comments")
120 if key in ["comments"]:
121 if mt_survey.comments.value:
122 mt_survey.comments.value += f", {value}"
123 else:
124 mt_survey.comments.value = value
125 else:
126 mt_survey.update_attribute(key, value)
127 else:
128 if mt_survey.comments.value:
129 if not value in mt_survey.comments:
130 mt_survey.comments.value += f", {value}"
131 else:
132 mt_survey.comments.value = value
133 else:
134 value = getattr(network, xml_key)
135 if value is None:
136 continue
137 if isinstance(value, (list, tuple)):
138 for k, v in zip(mt_key, value):
139 mt_survey.update_attribute(k, v)
140 else:
141 if xml_key == "restricted_status":
142 value = self.flip_dict(release_dict)[value]
143 mt_survey.update_attribute(mt_key, value)
144 if mt_survey.id is None:
145 mt_survey.id = mt_survey.fdsn.network
147 mt_survey.update_all()
148 return mt_survey
150 def mt_to_xml(self, survey, code="ZU"):
151 """
152 Convert MT Survey to Obspy Network
154 .. note:: For now the default code is ZU which is an IRIS catch-all network
156 """
158 if not isinstance(survey, metadata.Survey):
159 msg = (
160 f"Input must be mt_metadata.timeseries.Survey object not {type(survey)}"
161 )
162 self.logger.error(msg)
163 raise ValueError(msg)
164 network = inventory.Network(code)
165 for inv_key, mt_key in self.xml_translator.items():
166 if mt_key is None:
167 msg = "cannot currently map Survey to network.{0}".format(inv_key)
168 self.logger.debug(msg)
169 continue
170 if inv_key == "operators":
171 if survey.project_lead.organization:
172 operator = inventory.Operator(
173 agency=survey.project_lead.organization
174 )
175 if survey.project_lead.author:
176 person = inventory.Person(
177 names=[survey.project_lead.author],
178 emails=[survey.project_lead.email],
179 )
180 operator.contacts = [person]
181 network.operators = [operator]
182 elif inv_key == "comments":
183 if survey.comments is not None:
184 comment = inventory.Comment(survey.comments)
185 network.comments.append(comment)
186 elif inv_key == "restricted_status":
187 network.restricted_status = release_dict[
188 survey.release_license.replace("-", "_")
189 .replace(" ", "_")
190 .replace(".", "_")
191 .replace("(", "_")
192 .replace(")", "_")
193 .replace("/", "_")
194 .replace(":", "_")
195 ]
196 elif inv_key == "identifiers":
197 doi = survey.get_attr_from_name(mt_key)
198 network.identifiers.append(f"DOI:{doi}")
199 else:
200 setattr(network, inv_key, survey.get_attr_from_name(mt_key))
201 comments = self.make_mt_comments(survey, mt_key_base="mt.survey")
202 network.comments = comments
204 return network