« prev ^ index » next coverage.py v7.13.1, created at 2026-01-10 00:11 -0800
1# =====================================================
2# Imports
3# =====================================================
4from typing import Annotated
6from loguru import logger
7from pydantic import computed_field, Field, field_validator, PrivateAttr, ValidationInfo
9from mt_metadata.base import MetadataBase
10from mt_metadata.common.units import get_unit_object
11from mt_metadata.timeseries import Auxiliary, Electric, Magnetic # noqa: F401
12from mt_metadata.transfer_functions.io.tools import _validate_str_with_equals
13from mt_metadata.utils.location_helpers import (
14 convert_position_float2str,
15 validate_position,
16)
18from . import EMeasurement, HMeasurement
21# =====================================================
22class DefineMeasurement(MetadataBase):
23 """
24 DefineMeasurement class holds information about the measurement. This
25 includes how each channel was setup. The main block contains information
26 on the reference location for the station. This is a bit of an archaic
27 part and was meant for a multiple station .edi file. This section is also
28 important if you did any forward modeling with Winglink cause it only gives
29 the station location in this section. The other parts are how each channel
30 was collected. An example define measurement section looks like::
32 >=DEFINEMEAS
34 MAXCHAN=7
35 MAXRUN=999
36 MAXMEAS=9999
37 UNITS=M
38 REFTYPE=CART
39 REFLAT=-30:12:49.4693
40 REFLONG=139:47:50.87
41 REFELEV=0
43 >HMEAS ID=1001.001 CHTYPE=HX X=0.0 Y=0.0 Z=0.0 AZM=0.0
44 >HMEAS ID=1002.001 CHTYPE=HY X=0.0 Y=0.0 Z=0.0 AZM=90.0
45 >HMEAS ID=1003.001 CHTYPE=HZ X=0.0 Y=0.0 Z=0.0 AZM=0.0
46 >EMEAS ID=1004.001 CHTYPE=EX X=0.0 Y=0.0 Z=0.0 X2=0.0 Y2=0.0
47 >EMEAS ID=1005.001 CHTYPE=EY X=0.0 Y=0.0 Z=0.0 X2=0.0 Y2=0.0
48 >HMEAS ID=1006.001 CHTYPE=HX X=0.0 Y=0.0 Z=0.0 AZM=0.0
49 >HMEAS ID=1007.001 CHTYPE=HY X=0.0 Y=0.0 Z=0.0 AZM=90.0
51 :param fn: full path to .edi file to read in.
52 :type fn: string
54 ================= ==================================== ======== ===========
55 Attributes Description Default In .edi
56 ================= ==================================== ======== ===========
57 fn Full path to edi file read in None no
58 maxchan Maximum number of channels measured None yes
59 maxmeas Maximum number of measurements 9999 yes
60 maxrun Maximum number of measurement runs 999 yes
61 meas_#### HMeasurement or EMEasurment object None yes
62 defining the measurement made [1]__
63 refelev Reference elevation (m) None yes
64 reflat Reference latitude [2]_ None yes
65 refloc Reference location None yes
66 reflon Reference longituted [2]__ None yes
67 reftype Reference coordinate system 'cart' yes
68 units Units of length m yes
69 _define_meas_keys Keys to include in define_measurment [3]__ no
70 section.
71 ================= ==================================== ======== ===========
73 .. [1] Each channel with have its own define measurement and depending on
74 whether it is an E or H channel the metadata will be different.
75 the #### correspond to the channel number.
76 .. [2] Internally everything is converted to decimal degrees. Output is
77 written as HH:MM:SS.ss so Winglink can read them in.
78 .. [3] If you want to change what metadata is written into the .edi file
79 change the items in _header_keys. Default attributes are:
80 * maxchan
81 * maxrun
82 * maxmeas
83 * reflat
84 * reflon
85 * refelev
86 * reftype
87 * units
89 """
91 maxchan: Annotated[
92 int,
93 Field(
94 default=999,
95 description="maximum number of channels",
96 alias=None,
97 json_schema_extra={
98 "units": None,
99 "required": True,
100 "examples": ["16"],
101 },
102 ),
103 ]
105 maxrun: Annotated[
106 int,
107 Field(
108 default=999,
109 description="maximum number of runs",
110 alias=None,
111 json_schema_extra={
112 "units": None,
113 "required": True,
114 "examples": ["999"],
115 },
116 ),
117 ]
119 maxmeas: Annotated[
120 int,
121 Field(
122 default=7,
123 description="maximum number of measurements",
124 alias=None,
125 json_schema_extra={
126 "units": None,
127 "required": True,
128 "examples": ["999"],
129 },
130 ),
131 ]
133 reftype: Annotated[
134 str | None,
135 Field(
136 default="cartesian",
137 description="Type of offset from reference center point.",
138 alias=None,
139 json_schema_extra={
140 "units": None,
141 "required": False,
142 "examples": ["cartesian", "cart"],
143 },
144 ),
145 ]
147 refloc: Annotated[
148 str | None,
149 Field(
150 default=None,
151 description="Description of location reference center point.",
152 alias=None,
153 json_schema_extra={
154 "units": None,
155 "required": False,
156 "examples": ["here"],
157 },
158 ),
159 ]
161 reflat: Annotated[
162 float,
163 Field(
164 default=0,
165 description="Latitude of reference center point.",
166 alias=None,
167 json_schema_extra={
168 "units": "degrees",
169 "required": False,
170 "examples": ["0"],
171 },
172 ),
173 ]
175 reflon: Annotated[
176 float,
177 Field(
178 default=0,
179 description="Longitude reference center point.",
180 alias=None,
181 json_schema_extra={
182 "units": "degrees",
183 "required": False,
184 "examples": ["0"],
185 },
186 ),
187 ]
189 refelev: Annotated[
190 float,
191 Field(
192 default=0,
193 description="Elevation reference center point.",
194 alias=None,
195 json_schema_extra={
196 "units": "meters",
197 "required": False,
198 "examples": ["0"],
199 },
200 ),
201 ]
203 units: Annotated[
204 str | None,
205 Field(
206 default="m",
207 description="In the EDI standards this is the elevation units.",
208 alias=None,
209 json_schema_extra={
210 "units": None,
211 "required": True,
212 "examples": ["m"],
213 },
214 ),
215 ]
217 measurements: Annotated[
218 dict[str, EMeasurement | HMeasurement],
219 Field(
220 default_factory=dict,
221 description="Dictionary of measurements with keys as channel types "
222 "(e.g., 'hx', 'hy', 'ex', 'ey', etc.) and values as "
223 "EMeasurement or HMeasurement objects.",
224 alias=None,
225 json_schema_extra={
226 "units": None,
227 "required": False,
228 "examples": ["{'hx': EMeasurement(...), 'hy': HMeasurement(...)}"],
229 },
230 ),
231 ]
233 _define_meas_keys: list[str] = PrivateAttr(
234 default=[
235 "maxchan",
236 "maxrun",
237 "maxmeas",
238 "refloc",
239 "reflat",
240 "reflon",
241 "refelev",
242 "reftype",
243 "units",
244 ]
245 )
247 @field_validator("units", mode="before")
248 @classmethod
249 def validate_units(cls, value: str) -> str:
250 if value in [None, ""]:
251 return ""
252 if value.lower() in ["m", "meters"]:
253 value = "m"
254 try:
255 unit_object = get_unit_object(value)
256 return unit_object.name
257 except ValueError as error:
258 raise KeyError(error)
259 except KeyError as error:
260 raise KeyError(error)
262 @field_validator("reflat", "reflon", mode="before")
263 @classmethod
264 def validate_position(cls, value, info: ValidationInfo):
265 if "lat" in info.field_name:
266 position_type = "latitude"
267 elif "lon" in info.field_name:
268 position_type = "longitude"
269 return validate_position(value, position_type)
271 def __str__(self):
272 return "".join(self.write_measurement())
274 def __repr__(self):
275 return self.__str__()
277 @computed_field
278 @property
279 def channel_ids(self) -> dict[str, str]:
280 ch_ids = {}
281 for comp in ["ex", "ey", "hx", "hy", "hz", "rrhx", "rrhy"]:
282 try:
283 m = self.measurements[comp]
284 # if there are remote references that are the same as the
285 # h channels skip them.
286 ch_ids[m.chtype] = m.id
287 except KeyError:
288 continue
290 return ch_ids
292 def get_measurement_lists(self, edi_lines: list[str]) -> None:
293 """
294 get measurement list including measurement setup
296 Attributes
297 ----------
298 edi_lines : str
299 lines from the edi file to parse
300 """
302 self._measurement_list = []
303 meas_find = False
304 count = 0
306 for line in edi_lines:
307 if ">=" in line and "definemeas" in line.lower():
308 meas_find = True
309 elif ">=" in line:
310 if meas_find is True:
311 return
312 elif meas_find is True and ">" not in line:
313 line = line.strip()
314 if len(line) > 2:
315 if count > 0:
316 line_list = _validate_str_with_equals(line)
317 for ll in line_list:
318 ll_list = ll.split("=")
319 key = ll_list[0].lower()
320 value = ll_list[1]
321 self._measurement_list[-1][key] = value
322 else:
323 self._measurement_list.append(line.strip())
325 # look for the >XMEAS parts
326 elif ">" in line and meas_find:
327 if line.find("!") > 0:
328 pass
329 elif "meas" in line.lower():
330 count += 1
331 line_list = _validate_str_with_equals(line)
332 m_dict = {}
333 for ll in line_list:
334 ll_list = ll.split("=")
335 key = ll_list[0].lower()
336 value = ll_list[1]
337 m_dict[key] = value
338 self._measurement_list.append(m_dict)
339 else:
340 return
342 def read_measurement(self, edi_lines: list[str]) -> None:
343 """
344 read the define measurment section of the edi file
346 should be a list with lines for:
348 - maxchan
349 - maxmeas
350 - maxrun
351 - refloc
352 - refelev
353 - reflat
354 - reflon
355 - reftype
356 - units
357 - dictionaries for >XMEAS with keys:
359 - id
360 - chtype
361 - x
362 - y
363 - axm
364 - acqchn
366 """
367 self.get_measurement_lists(edi_lines)
369 for line in self._measurement_list:
370 if isinstance(line, str):
371 line_list = line.split("=")
372 key = line_list[0].lower()
373 value = line_list[1].strip()
374 if key in "reflatitude":
375 key = "reflat"
376 value = value
377 elif key in "reflongitude":
378 key = "reflon"
379 value = value
380 elif key in "refelevation":
381 key = "refelev"
382 value = value
383 elif key in "maxchannels":
384 key = "maxchan"
385 try:
386 value = int(value)
387 except ValueError:
388 value = 0
389 elif key in "maxmeasurements":
390 key = "maxmeas"
391 try:
392 value = int(value)
393 except ValueError:
394 value = 0
395 elif key in "maxruns":
396 key = "maxrun"
397 try:
398 value = int(value)
399 except ValueError:
400 value = 0
401 setattr(self, key, value)
403 elif isinstance(line, dict):
404 ch_type = line["chtype"].lower()
405 key = f"{ch_type}"
406 if ch_type.find("h") >= 0:
407 value = HMeasurement(**line)
408 elif ch_type.find("e") >= 0:
409 value = EMeasurement(**line)
410 if value.azm == 0:
411 value.azm = value.azimuth
412 if key in self.measurements.keys():
413 existing_ch = self.measurements[key]
414 existing_line = existing_ch.write_meas_line()
415 value_line = value.write_meas_line()
416 if existing_line != value_line:
417 value.chtype = f"rr{ch_type}".upper()
418 key = f"rr{ch_type}"
419 else:
420 continue
421 self.measurements[key] = value
423 def _sort_measurements(self) -> list[str]:
424 """
425 Sort the measurements by channel type and return a list of sorted keys.
426 This is used to ensure that the measurements are written in a consistent order.
427 """
428 # need to write the >XMEAS type, but sort by channel number
429 m_key_list = []
430 count = 1.0
431 for key, meas in self.measurements.items():
432 value = meas.id
433 if value == 0.0:
434 value = count
435 count += 1
436 m_key_list.append((key, value))
438 return sorted(m_key_list, key=lambda x: x[1])
440 def write_measurement(
441 self,
442 longitude_format: str = "LON",
443 latlon_format: str = "degrees",
444 ) -> list[str]:
445 """
446 write_measurement writes the define measurement section of the edi file.
448 Parameters
449 ----------
450 longitude_format : str, optional
451 longitude format [ "LONG" | "LON" ] , by default "LON"
452 latlon_format : str, optional
453 position format [ "dd" | " degrees" ], by default "degrees" for decimal degrees
454 If you want to write the position in degrees, use " degrees" for the
455 latlon_format. This will write the position in the format of
456 HH:MM:SS.ss for the latitude and longitude. If you want to write
457 the position in decimal degrees, use "dd" for the latlon_format.
459 Returns
460 -------
461 list[str]
462 list of lines for the define measurement section or an empty list if no
463 measurements are defined.
465 Raises
466 ------
467 ValueError
468 If a value cannot be converted to a float or if the longitude format is not
469 recognized.
470 """
472 measurement_lines = ["\n>=DEFINEMEAS\n"]
473 for key in self._define_meas_keys:
474 value = getattr(self, key)
475 if key in ["reflat", "reflon", "reflong"]:
476 if latlon_format.lower() == "dd":
477 value = f"{float(value):.6f}"
478 else:
479 value = convert_position_float2str(value)
480 elif key == "refelev":
481 value = value
482 if key.upper() == "REFLON":
483 if longitude_format == "LONG":
484 key += "G"
485 if value is not None:
486 measurement_lines.append(f"{' '*4}{key.upper()}={value}\n")
487 measurement_lines.append("\n")
489 # need to write the >XMEAS type, but sort by channel number
490 m_key_list = self._sort_measurements()
492 if len(m_key_list) == 0:
493 logger.warning("No XMEAS information.")
494 else:
495 # need to sort the dictionary by chanel id
496 for meas in sorted(m_key_list, key=lambda x: x[1]):
497 x_key = meas[0]
498 m_obj = self.measurements[x_key]
499 if m_obj.id == 0.0:
500 m_obj.id = meas[1]
501 if m_obj.acqchan == "0":
502 m_obj.acqchan = meas[1]
504 measurement_lines.append(m_obj.write_meas_line())
506 return measurement_lines
508 def from_metadata(self, channel: Electric | Magnetic | Auxiliary) -> None:
509 """
511 from_metadata converts a channel object into a measurement object
512 and sets the attributes for the measurement object.
514 Parameters
515 ----------
516 channel : Electric | Magnetic | Auxiliary
517 The channel object to convert into a measurement object.
518 """
520 if channel.component is None:
521 return
523 azm = channel.measurement_azimuth
524 if azm != channel.translated_azimuth:
525 azm = channel.translated_azimuth
526 if azm is None:
527 azm = 0.0
528 if "e" in channel.component:
529 for attr in [
530 "negative.x",
531 "negative.y",
532 "positive.x2",
533 "positive.y2",
534 "measurement_azimuth",
535 "translated_azimuth",
536 ]:
537 if channel.get_attr_from_name(attr) is None:
538 channel.update_attribute(attr, 0)
539 meas = EMeasurement(
540 **{
541 "x": channel.negative.x,
542 "x2": channel.positive.x2,
543 "y": channel.negative.y,
544 "y2": channel.positive.y2,
545 "chtype": channel.component,
546 "id": channel.channel_id,
547 "azm": azm,
548 "acqchan": channel.channel_number,
549 }
550 )
551 self.measurements[channel.component.lower()] = meas
553 elif "h" in channel.component:
554 for attr in ["location.x", "location.y", "location.z"]:
555 if channel.get_attr_from_name(attr) is None:
556 channel.update_attribute(attr, 0)
557 meas = HMeasurement(
558 **{
559 "x": channel.location.x,
560 "y": channel.location.y,
561 "azm": azm,
562 "chtype": channel.component,
563 "id": channel.channel_id,
564 "acqchan": channel.channel_number,
565 "dip": channel.measurement_tilt,
566 }
567 )
568 self.measurements[channel.component.lower()] = meas
570 @computed_field(return_type=list[str])
571 @property
572 def channels_recorded(self) -> list[str]:
573 """Get the channels recorded"""
575 return [cc.lower() for cc in self.measurements.keys()]