Coverage for C: \ Users \ peaco \ OneDrive \ Documents \ GitHub \ mt_metadata \ mt_metadata \ transfer_functions \ tf \ transfer_function.py: 88%

40 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 

5 

6import numpy as np 

7import pandas as pd 

8from pydantic import Field, field_validator 

9 

10from mt_metadata.base import MetadataBase 

11from mt_metadata.common import ( 

12 AuthorPerson, 

13 DataQuality, 

14 GeographicReferenceFrameEnum, 

15 SignConventionEnum, 

16 Software, 

17) 

18from mt_metadata.common.mttime import MTime 

19from mt_metadata.common.units import get_unit_object 

20 

21 

22# ===================================================== 

23 

24 

25class TransferFunction(MetadataBase): 

26 id: Annotated[ 

27 str, 

28 Field( 

29 default="", 

30 description="transfer function id", 

31 alias=None, 

32 json_schema_extra={ 

33 "units": None, 

34 "required": True, 

35 "examples": ["mt01_256"], 

36 }, 

37 ), 

38 ] 

39 

40 sign_convention: Annotated[ 

41 SignConventionEnum, 

42 Field( 

43 default="+", 

44 description="sign of the transfer function estimates", 

45 alias=None, 

46 json_schema_extra={ 

47 "units": None, 

48 "required": True, 

49 "examples": ["+"], 

50 }, 

51 ), 

52 ] 

53 

54 units: Annotated[ 

55 str, 

56 Field( 

57 default="milliVolt per kilometer per nanoTesla", 

58 description="units of the impedance tensor estimates", 

59 alias=None, 

60 json_schema_extra={ 

61 "units": None, 

62 "required": True, 

63 "examples": ["milliVolt per kilometer per nanoTesla"], 

64 }, 

65 ), 

66 ] 

67 

68 runs_processed: Annotated[ 

69 list[str], 

70 Field( 

71 default=[], 

72 description="list of runs used in the processing", 

73 alias=None, 

74 json_schema_extra={ 

75 "units": None, 

76 "required": True, 

77 "examples": [["MT001a", "MT001c"]], 

78 }, 

79 ), 

80 ] 

81 

82 remote_references: Annotated[ 

83 list[str], 

84 Field( 

85 default_factory=list, 

86 description="list of remote references", 

87 alias=None, 

88 json_schema_extra={ 

89 "units": None, 

90 "required": True, 

91 "examples": [["MT002b", "MT002c"]], 

92 }, 

93 ), 

94 ] 

95 

96 processed_date: Annotated[ 

97 MTime | str | float | int | np.datetime64 | pd.Timestamp, 

98 Field( 

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

100 description="date the data were processed", 

101 alias=None, 

102 json_schema_extra={ 

103 "units": None, 

104 "required": True, 

105 "examples": ["2020-01-01T12:00:00"], 

106 }, 

107 ), 

108 ] 

109 

110 processing_parameters: Annotated[ 

111 list[str], 

112 Field( 

113 default_factory=list, 

114 description="list of processing parameters with structure name = value", 

115 alias=None, 

116 json_schema_extra={ 

117 "units": None, 

118 "required": True, 

119 "examples": [["nfft=4096", "n_windows=16"]], 

120 }, 

121 ), 

122 ] 

123 

124 processed_by: Annotated[ 

125 AuthorPerson, 

126 Field( 

127 default_factory=AuthorPerson, # type: ignore 

128 description="person who processed the data", 

129 alias=None, 

130 json_schema_extra={ 

131 "units": None, 

132 "required": True, 

133 "examples": ["Person(name='John Doe', email='john.doe@example.com')"], 

134 }, 

135 ), 

136 ] 

137 

138 processing_type: Annotated[ 

139 str, 

140 Field( 

141 default="", 

142 description="Type of processing", 

143 alias=None, 

144 json_schema_extra={ 

145 "units": None, 

146 "required": True, 

147 "examples": ["robust remote reference"], 

148 }, 

149 ), 

150 ] 

151 

152 software: Annotated[ 

153 Software, 

154 Field( 

155 default_factory=Software, # type: ignore 

156 description="software used to process the data", 

157 alias=None, 

158 json_schema_extra={ 

159 "units": None, 

160 "required": True, 

161 "examples": ["Software(name='Aurora', version='1.0.0')"], 

162 }, 

163 ), 

164 ] 

165 

166 data_quality: Annotated[ 

167 DataQuality, 

168 Field( 

169 default_factory=DataQuality, # type: ignore 

170 description="data quality information", 

171 alias=None, 

172 json_schema_extra={ 

173 "units": None, 

174 "required": True, 

175 "examples": ["DataQuality()"], 

176 }, 

177 ), 

178 ] 

179 

180 coordinate_system: Annotated[ 

181 GeographicReferenceFrameEnum, 

182 Field( 

183 default=GeographicReferenceFrameEnum.geographic, 

184 description="coordinate system that the transfer function is in. It is strongly recommended that the transfer functions be rotated to align with geographic coordinates with geographic north as 0 and east as 90.", 

185 alias=None, 

186 json_schema_extra={ 

187 "units": None, 

188 "required": True, 

189 "examples": ["geographic"], 

190 }, 

191 ), 

192 ] 

193 

194 # this will evenutally be config objects for the various processing programs 

195 # e.g. aurora, razorback, resistics, etc. 

196 # for now it is just a string 

197 processing_config: Annotated[ 

198 str | None, 

199 Field( 

200 default=None, 

201 description="processing configuration", 

202 alias=None, 

203 json_schema_extra={ 

204 "units": None, 

205 "required": False, 

206 "examples": ["aurora.processing"], 

207 }, 

208 ), 

209 ] 

210 

211 @field_validator("processed_date", mode="before") 

212 @classmethod 

213 def validate_processed_date( 

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

215 ): 

216 if isinstance(field_value, MTime): 

217 return field_value 

218 return MTime(time_stamp=field_value) 

219 

220 @field_validator("units", mode="before") 

221 @classmethod 

222 def validate_units(cls, value: str) -> str: 

223 if value in [None, ""]: 

224 return "" 

225 try: 

226 unit_object = get_unit_object(value) 

227 return unit_object.name 

228 except ValueError as error: 

229 raise KeyError(error) 

230 except KeyError as error: 

231 raise KeyError(error)