Coverage for C: \ Users \ peaco \ OneDrive \ Documents \ GitHub \ mt_metadata \ mt_metadata \ transfer_functions \ io \ edi \ metadata \ emeasurement.py: 88%

73 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 

7from pydantic import ( 

8 computed_field, 

9 Field, 

10 field_validator, 

11 model_validator, 

12 PrivateAttr, 

13) 

14 

15from mt_metadata.base import MetadataBase 

16 

17 

18# ===================================================== 

19 

20 

21class EMeasurement(MetadataBase): 

22 id: Annotated[ 

23 float | None, 

24 Field( 

25 default=0.0, 

26 description="Channel number, could be location.channel_number.", 

27 alias=None, 

28 json_schema_extra={ 

29 "units": None, 

30 "required": True, 

31 "examples": ["1"], 

32 }, 

33 ), 

34 ] 

35 

36 chtype: Annotated[ 

37 str, 

38 Field( 

39 default="", 

40 description="channel type, should start with an 'e'", 

41 alias=None, 

42 pattern=r"^(RR|rr|[eE])[a-zA-Z0-9_]+$", 

43 json_schema_extra={ 

44 "units": None, 

45 "required": True, 

46 "examples": ["ex"], 

47 }, 

48 ), 

49 ] 

50 

51 x: Annotated[ 

52 float, 

53 Field( 

54 default=0.0, 

55 description="location of negative sensor relative center point in north direction", 

56 alias=None, 

57 json_schema_extra={ 

58 "units": "meters", 

59 "required": True, 

60 "examples": ["100.0"], 

61 }, 

62 ), 

63 ] 

64 

65 x2: Annotated[ 

66 float, 

67 Field( 

68 default=0.0, 

69 description="location of positive sensor relative center point in north direction", 

70 alias=None, 

71 json_schema_extra={ 

72 "units": "meters", 

73 "required": True, 

74 "examples": ["100.0"], 

75 }, 

76 ), 

77 ] 

78 

79 y: Annotated[ 

80 float, 

81 Field( 

82 default=0.0, 

83 description="location of negative sensor relative center point in east direction", 

84 alias=None, 

85 json_schema_extra={ 

86 "units": "meters", 

87 "required": True, 

88 "examples": ["100.0"], 

89 }, 

90 ), 

91 ] 

92 

93 y2: Annotated[ 

94 float, 

95 Field( 

96 default=0.0, 

97 description="location of positive sensor relative center point in east direction", 

98 alias=None, 

99 json_schema_extra={ 

100 "units": "meters", 

101 "required": True, 

102 "examples": ["100.0"], 

103 }, 

104 ), 

105 ] 

106 

107 z: Annotated[ 

108 float, 

109 Field( 

110 default=0.0, 

111 description="location of negative sensor relative center point in depth", 

112 alias=None, 

113 json_schema_extra={ 

114 "units": "meters", 

115 "required": True, 

116 "examples": ["100.0"], 

117 }, 

118 ), 

119 ] 

120 

121 z2: Annotated[ 

122 float, 

123 Field( 

124 default=0.0, 

125 description="location of positive sensor relative center point in depth", 

126 alias=None, 

127 json_schema_extra={ 

128 "units": "meters", 

129 "required": True, 

130 "examples": ["100.0"], 

131 }, 

132 ), 

133 ] 

134 

135 azm: Annotated[ 

136 float, 

137 Field( 

138 default=0.0, 

139 description="orientation of the sensor relative to coordinate system, clockwise positive.", 

140 alias=None, 

141 json_schema_extra={ 

142 "units": "degrees", 

143 "required": True, 

144 "examples": ["100.0"], 

145 }, 

146 ), 

147 ] 

148 

149 acqchan: Annotated[ 

150 str, 

151 Field( 

152 default="", 

153 description="description of acquired channel", 

154 alias=None, 

155 json_schema_extra={ 

156 "units": None, 

157 "required": True, 

158 "examples": ["100.0"], 

159 }, 

160 ), 

161 ] 

162 

163 _fmt_dict: dict[str, str] = PrivateAttr( 

164 default={ 

165 "id": "<", 

166 "chtype": "<", 

167 "x": "<.2f", 

168 "y": "<.2f", 

169 "z": "<.2f", 

170 "x2": "<.2f", 

171 "y2": "<.2f", 

172 "z2": "<.2f", 

173 "azm": "<.2f", 

174 "acqchan": "<", 

175 } 

176 ) 

177 

178 @field_validator("id", mode="before") 

179 @classmethod 

180 def validate_id(cls, value: float | str | None) -> float: 

181 """Ensure id is a float or None, convert if necessary""" 

182 if isinstance(value, str): 

183 try: 

184 value = float(value) 

185 except ValueError: 

186 raise ValueError("id must be a number or convertible to float") 

187 elif not isinstance(value, (float, int, type(None))): 

188 raise TypeError("id must be a number or None") 

189 if value is None: 

190 value = 0.0 # Default to 0.0 if None 

191 return value 

192 

193 @model_validator(mode="after") 

194 def update_azimuth_from_coords(self): 

195 """Update azm based on coordinates after validation""" 

196 # Only update if coordinates have been explicitly set 

197 if any(value != 0 for value in [self.x, self.y, self.x2, self.y2]): 

198 computed_azimuth = self.azimuth 

199 if self.azm == 0: # Only update if azm wasn't explicitly set 

200 # Bypass validation by using object.__setattr__ 

201 object.__setattr__(self, "azm", computed_azimuth) 

202 return self 

203 

204 def __str__(self): 

205 return "\n".join([f"{k} = {v}" for k, v in self.to_dict(single=True).items()]) 

206 

207 def __repr__(self): 

208 return self.__str__() 

209 

210 @computed_field 

211 @property 

212 def dipole_length(self) -> float: 

213 """dipole length based on x, y, z coordinates""" 

214 try: 

215 return ( 

216 (self.x2 - self.x) ** 2 

217 + (self.y2 - self.y) ** 2 

218 + (self.z2 - self.z) ** 2 

219 ) ** 0.5 

220 except TypeError: 

221 return 0 

222 

223 @computed_field 

224 @property 

225 def azimuth(self) -> float: 

226 """aximuth based on x, y coordinates""" 

227 try: 

228 return np.rad2deg(np.arctan2((self.y2 - self.y), (self.x2 - self.x))) 

229 except (ZeroDivisionError, TypeError): 

230 return 0.0 

231 

232 @computed_field 

233 @property 

234 def channel_number(self) -> int: 

235 """Extract channel number from acqchan.""" 

236 if self.acqchan is not None: 

237 if not isinstance(self.acqchan, (int, float)): 

238 try: 

239 return int("".join(i for i in self.acqchan if i.isdigit())) 

240 except (IndexError, ValueError): 

241 return 0 

242 return int(self.acqchan) 

243 return 0 

244 

245 def write_meas_line(self): 

246 """ 

247 write string 

248 :return: DESCRIPTION 

249 :rtype: TYPE 

250 

251 """ 

252 

253 line = [">emeas".upper()] 

254 

255 for mkey, mfmt in self._fmt_dict.items(): 

256 try: 

257 line.append(f"{mkey.upper()}={getattr(self, mkey):{mfmt}}") 

258 except (ValueError, TypeError): 

259 line.append(f"{mkey.upper()}={0.0:{mfmt}}") 

260 

261 return f"{' '.join(line)}\n"