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"