Coverage for C: \ Users \ peaco \ OneDrive \ Documents \ GitHub \ mt_metadata \ mt_metadata \ timeseries \ stationxml \ xml_equipment_mt_run.py: 88%

1# -*- coding: utf-8 -*- 

2""" 

3Created on Thu Feb 18 12:49:13 2021 

4 

5:copyright: 

6 Jared Peacock (jpeacock@usgs.gov) 

7 

8:license: MIT 

9 

10""" 

11# ============================================================================= 

12# Imports 

13# ============================================================================= 

14import enum 

15 

16from mt_metadata import timeseries as metadata 

17from mt_metadata.base.helpers import requires 

18from mt_metadata.timeseries.stationxml.utils import BaseTranslator 

19 

20 

21try: 

22 from obspy.core import inventory 

23except ImportError: 

24 inventory = None 

25 

26# ============================================================================= 

27 

28 

29@requires(obspy=inventory) 

30class XMLEquipmentMTRun(BaseTranslator): 

31 """ 

32 translate back and forth between StationXML Station and MT Station 

33 """ 

34 

35 def __init__(self): 

36 super().__init__() 

37 

38 self.xml_translator = { 

39 "type": "data_type", 

40 "manufacturer": "data_logger.manufacturer", 

41 "model": "data_logger.model", 

42 "serial_number": "data_logger.id", 

43 "installation_date": "time_period.start", 

44 "removal_date": "time_period.end", 

45 "description": "special", 

46 "resource_id": "id", 

47 } 

48 

49 # StationXML to MT Survey 

50 self.mt_translator = self.flip_dict(self.xml_translator) 

51 self.mt_translator["notes"] = "description" 

52 

53 self.mt_comments_list = [ 

54 {"acquired_by": ["acquired_by.author", "acquired_by.comments"]}, 

55 {"metadata_by": ["metadata_by.author", "metadata_by.comments"]}, 

56 "comments", 

57 ] 

58 

59 self.mt_description_list = [ 

60 "data_logger.firmware.author", 

61 "data_logger.firmware.name", 

62 "data_logger.firmware.version", 

63 "data_logger.power_source.comments.value", 

64 "data_logger.power_source.id", 

65 "data_logger.power_source.type", 

66 "data_logger.power_source.voltage.end", 

67 "data_logger.power_source.voltage.start", 

68 "data_logger.timing_system.comments.value", 

69 "data_logger.timing_system.drift", 

70 "data_logger.timing_system.type", 

71 "data_logger.timing_system.uncertainty", 

72 "data_logger.type", 

73 ] 

74 

75 def xml_to_mt(self, equipment): 

76 """ 

77 Read in an equipment block. 

78 

79 :param equipment: an Equipment element 

80 :type equipment: :class:`obspy.core.inventory.Equipment` 

81 

82 """ 

83 

84 if not isinstance(equipment, inventory.Equipment): 

85 msg = f"Input must be obspy.core.inventory.Equipment object not {type(equipment)}" 

86 self.logger.error(msg) 

87 raise TypeError(msg) 

88 

89 mt_run = metadata.Run() 

90 for xml_key, mt_key in self.xml_translator.items(): 

91 value = getattr(equipment, xml_key) 

92 if xml_key in ["description"]: 

93 mt_run = self._parse_description(value, mt_run) 

94 elif xml_key in ["resource_id"]: 

95 mt_run.id = value.split(":")[1] 

96 elif xml_key in ["installation_date", "removal_date"]: 

97 # Handle time fields - convert from ObsPy UTCDateTime to string 

98 if value is not None: 

99 # Convert ObsPy UTCDateTime to ISO format string for MTime objects 

100 value = str(value) 

101 mt_run.update_attribute(mt_key, value) 

102 else: 

103 mt_run.update_attribute(mt_key, value) 

104 

105 return mt_run 

106 

107 def mt_to_xml(self, mt_run): 

108 """ 

109 Convert an :class:mt_metadata.timeseries.Run` to XML equipment and comments 

110 

111 :param mt_run: DESCRIPTION 

112 :type mt_run: TYPE 

113 :return: DESCRIPTION 

114 :rtype: TYPE 

115 

116 """ 

117 

118 equipment = inventory.Equipment() 

119 for mt_key, xml_key in self.mt_translator.items(): 

120 if mt_key == "notes": 

121 value = self._make_description(mt_run) 

122 elif mt_key == "id": 

123 value = f"mt.run.id:{mt_run.id}" 

124 elif "date" in xml_key: 

125 time_obj = mt_run.get_attr_from_name(mt_key) 

126 # Convert MTime object to ISO format string for ObsPy UTCDateTime 

127 value = ( 

128 time_obj.isoformat() 

129 if hasattr(time_obj, "isoformat") 

130 else time_obj.time_stamp 

131 ) 

132 else: 

133 value = mt_run.get_attr_from_name(mt_key) 

134 # Handle enum values by getting their actual value instead of string representation 

135 if isinstance(value, enum.Enum): 

136 value = value.value 

137 setattr(equipment, xml_key, value) 

138 

139 return equipment 

140 

141 def _parse_description(self, description, run_obj): 

142 """ 

143 Parse a run description into run 

144 

145 :param description: DESCRIPTION 

146 :type description: TYPE 

147 :return: DESCRIPTION 

148 :rtype: TYPE 

149 

150 """ 

151 

152 for d_str in description.split(","): 

153 d_key, d_value = d_str.split(":") 

154 d_value = d_value.strip() 

155 if not d_value in ["", "None", "null"]: 

156 run_obj.update_attribute( 

157 f"data_logger.{d_key.strip()}", d_value.strip() 

158 ) 

159 

160 return run_obj 

161 

162 def _make_description(self, run_obj): 

163 """ 

164 Make an Equipment description from a run object 

165 

166 :param run_obj: DESCRIPTION 

167 :type run_obj: TYPE 

168 :return: DESCRIPTION 

169 :rtype: TYPE 

170 

171 """ 

172 if not isinstance(run_obj, metadata.Run): 

173 msg = ( 

174 f"Input must be a mt_metadta.timeseries.Run object not {type(run_obj)}" 

175 ) 

176 self.logger.error(msg) 

177 raise TypeError(msg) 

178 

179 lines = [] 

180 for key in self.mt_description_list: 

181 value = run_obj.get_attr_from_name(key) 

182 if value: 

183 lines.append(f"{key.split('data_logger.')[1]}: {value}") 

184 

185 return ", ".join(lines)