Coverage for C: \ Users \ peaco \ OneDrive \ Documents \ GitHub \ mt_metadata \ mt_metadata \ transfer_functions \ io \ emtfxml \ metadata \ processing_info.py: 97%

39 statements  

« prev     ^ index     » next       coverage.py v7.13.1, created at 2026-01-10 00:11 -0800

1# ===================================================== 

2# Imports 

3# ===================================================== 

4from typing import Annotated 

5from xml.etree import ElementTree as et 

6 

7import numpy as np 

8import pandas as pd 

9from loguru import logger 

10from pydantic import Field, field_validator 

11 

12from mt_metadata.base import MetadataBase 

13from mt_metadata.common.enumerations import SignConventionEnum 

14from mt_metadata.common.mttime import MTime 

15 

16from . import helpers, ProcessingSoftware, RemoteInfo, RemoteRef 

17 

18 

19# ===================================================== 

20 

21 

22class ProcessingInfo(MetadataBase): 

23 sign_convention: Annotated[ 

24 SignConventionEnum, 

25 Field( 

26 default=r"exp(+ i\omega t)", 

27 description="Sign convention of the processing software output", 

28 alias=None, 

29 json_schema_extra={ 

30 "units": None, 

31 "required": True, 

32 "examples": ["exp(+ i\\omega t)"], 

33 }, 

34 ), 

35 ] 

36 

37 processed_by: Annotated[ 

38 str | None, 

39 Field( 

40 default=None, 

41 description="Names of people who processed the data", 

42 alias=None, 

43 json_schema_extra={ 

44 "units": None, 

45 "required": False, 

46 "examples": ["MT Guru"], 

47 }, 

48 ), 

49 ] 

50 

51 process_date: Annotated[ 

52 MTime | str | float | int | np.datetime64 | pd.Timestamp | None, 

53 Field( 

54 default_factory=lambda: MTime(time_stamp=None), 

55 description="Date the data were processed", 

56 alias=None, 

57 json_schema_extra={ 

58 "units": None, 

59 "required": False, 

60 "examples": ["2020-01-01"], 

61 }, 

62 ), 

63 ] 

64 

65 processing_tag: Annotated[ 

66 str | None, 

67 Field( 

68 default=None, 

69 description="List of remote references", 

70 alias=None, 

71 json_schema_extra={ 

72 "units": None, 

73 "required": False, 

74 "examples": ["mt001-mt002"], 

75 }, 

76 ), 

77 ] 

78 

79 processing_software: Annotated[ 

80 ProcessingSoftware, 

81 Field( 

82 default_factory=lambda: ProcessingSoftware(), # type: ignore 

83 description="Information about the processing software used", 

84 alias=None, 

85 json_schema_extra={ 

86 "units": None, 

87 "required": False, 

88 "examples": [ 

89 "ProcessingSoftware(name='MT Processing Software', version='1.0')" 

90 ], 

91 }, 

92 ), 

93 ] 

94 

95 remote_info: Annotated[ 

96 RemoteInfo, 

97 Field( 

98 default_factory=RemoteInfo, # type: ignore 

99 description="Information about remote data sources", 

100 alias=None, 

101 json_schema_extra={ 

102 "units": None, 

103 "required": False, 

104 "examples": ["RemoteInfo(name='MT Remote')"], 

105 }, 

106 ), 

107 ] 

108 

109 remote_ref: Annotated[ 

110 RemoteRef, 

111 Field( 

112 default_factory=RemoteRef, # type: ignore 

113 description="List of remote references", 

114 alias=None, 

115 json_schema_extra={ 

116 "units": None, 

117 "required": False, 

118 "examples": [["MT001a", "MT001b"]], 

119 }, 

120 ), 

121 ] 

122 

123 _order: list[str] = [ 

124 "sign_convention", 

125 "remote_ref", 

126 "remote_info", 

127 "processed_by", 

128 "process_date", 

129 "processing_software", 

130 "processing_tag", 

131 ] 

132 

133 @field_validator("process_date", mode="before") 

134 @classmethod 

135 def validate_process_date( 

136 cls, field_value: MTime | float | int | np.datetime64 | pd.Timestamp | str 

137 ): 

138 return MTime(time_stamp=field_value) 

139 

140 def read_dict(self, input_dict: dict) -> None: 

141 """ 

142 Read processing information from a dictionary. 

143 

144 Parameters 

145 ---------- 

146 input_dict : dict 

147 A dictionary containing processing information. 

148 """ 

149 try: 

150 processing_dict = input_dict["processing_info"] 

151 except KeyError: 

152 return 

153 

154 if "field_notes" in processing_dict.keys(): 

155 processing_dict.pop("field_notes") 

156 

157 for key in ["remote_ref", "remote_info", "processing_software"]: 

158 try: 

159 pop_dict = {key: processing_dict.pop(key)} 

160 getattr(self, key).read_dict(pop_dict) 

161 except KeyError: 

162 logger.debug(f"No {key} information in xml.") 

163 

164 helpers._read_element(self, input_dict, "processing_info") 

165 

166 def to_xml(self, string: bool = False, required: bool = True) -> str | et.Element: 

167 """ 

168 Convert the processing information to XML format. 

169 

170 Parameters 

171 ---------- 

172 string : bool, optional 

173 Whether to return the XML as a string, by default False 

174 required : bool, optional 

175 Whether the XML is required, by default True 

176 

177 Returns 

178 ------- 

179 str | et.Element 

180 The XML representation of the processing information 

181 """ 

182 

183 return helpers.to_xml( 

184 self, 

185 string=string, 

186 required=required, 

187 order=self._order, 

188 )