Coverage for C: \ Users \ peaco \ OneDrive \ Documents \ GitHub \ mt_metadata \ mt_metadata \ transfer_functions \ io \ edi \ metadata \ header.py: 95%
150 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
6import numpy as np
7import pandas as pd
8from loguru import logger
9from pydantic import Field, field_validator
11from mt_metadata import __version__
12from mt_metadata.common import (
13 BasicLocation,
14 Declination,
15 GeographicLocation,
16 GeographicReferenceFrameEnum,
17 StdEDIversionsEnum,
18)
19from mt_metadata.common.mttime import get_now_utc, MTime
20from mt_metadata.common.units import get_unit_object
21from mt_metadata.utils.location_helpers import convert_position_float2str
22from mt_metadata.utils.validators import validate_station_name
25# =====================================================
28class Header(BasicLocation, GeographicLocation):
29 acqby: Annotated[
30 str | None,
31 Field(
32 default=None,
33 description="person, group, company, university that collected the data",
34 alias=None,
35 json_schema_extra={
36 "units": None,
37 "required": False,
38 "examples": ["mt experts"],
39 },
40 ),
41 ]
43 acqdate: Annotated[
44 MTime | str | float | int | np.datetime64 | pd.Timestamp,
45 Field(
46 default_factory=lambda: MTime(time_stamp=None),
47 description="Start date the time series data were collected",
48 alias=None,
49 json_schema_extra={
50 "units": None,
51 "required": True,
52 "examples": ["2020-01-01"],
53 },
54 ),
55 ]
57 coordinate_system: Annotated[
58 GeographicReferenceFrameEnum,
59 Field(
60 default="geographic",
61 description="coordinate system the transfer function is currently in. Its preferred the transfer function be in a geographic coordinate system for archiving and sharing.",
62 alias=None,
63 json_schema_extra={
64 "units": None,
65 "required": True,
66 "examples": ["geopgraphic"],
67 },
68 ),
69 ]
71 dataid: Annotated[
72 str,
73 Field(
74 default="",
75 description="station ID.",
76 alias=None,
77 json_schema_extra={
78 "units": None,
79 "required": True,
80 "examples": ["mt001"],
81 },
82 ),
83 ]
85 enddate: Annotated[
86 MTime | str | float | int | np.datetime64 | pd.Timestamp | None,
87 Field(
88 default_factory=lambda: MTime(time_stamp=None),
89 description="End date the time series data were collected",
90 alias=None,
91 json_schema_extra={
92 "units": None,
93 "required": False,
94 "examples": ["2020-01-01"],
95 },
96 ),
97 ]
99 empty: Annotated[
100 float,
101 Field(
102 default=1e32,
103 description="null data values, usually a large number",
104 alias=None,
105 json_schema_extra={
106 "units": None,
107 "required": True,
108 "examples": ["1E+32"],
109 },
110 ),
111 ]
113 fileby: Annotated[
114 str,
115 Field(
116 default="",
117 description="person, group, company, university that made the file",
118 alias=None,
119 json_schema_extra={
120 "units": None,
121 "required": True,
122 "examples": ["mt experts"],
123 },
124 ),
125 ]
127 filedate: Annotated[
128 MTime | str | float | int | np.datetime64 | pd.Timestamp,
129 Field(
130 default_factory=lambda: MTime(time_stamp=None),
131 description="Date the file was made",
132 alias=None,
133 json_schema_extra={
134 "units": None,
135 "required": True,
136 "examples": ["2020-01-01"],
137 },
138 ),
139 ]
141 progdate: Annotated[
142 MTime | str | float | int | np.datetime64 | pd.Timestamp,
143 Field(
144 default_factory=lambda: MTime(time_stamp=None),
145 description="Date of the most recent update of the program used to make the file",
146 alias=None,
147 json_schema_extra={
148 "units": None,
149 "required": True,
150 "examples": ["2020-01-01"],
151 },
152 ),
153 ]
155 progname: Annotated[
156 str,
157 Field(
158 default="mt_metadata",
159 description="Name of the program used to make the file.",
160 alias=None,
161 json_schema_extra={
162 "units": None,
163 "required": True,
164 "examples": ["mt_metadata"],
165 },
166 ),
167 ]
169 progvers: Annotated[
170 str,
171 Field(
172 default="0.1.6",
173 description="Version of the program used to make the file.",
174 alias=None,
175 json_schema_extra={
176 "units": None,
177 "required": True,
178 "examples": ["0.1.6"],
179 },
180 ),
181 ]
183 project: Annotated[
184 str | None,
185 Field(
186 default=None,
187 description="Name of the project the data was collected for, usually a short description or acronym of the project name.",
188 alias=None,
189 json_schema_extra={
190 "units": None,
191 "required": False,
192 "examples": ["iMUSH"],
193 },
194 ),
195 ]
197 prospect: Annotated[
198 str | None,
199 Field(
200 default=None,
201 description="Name of the prospect the data was collected for, usually a short description of the location",
202 alias=None,
203 json_schema_extra={
204 "units": None,
205 "required": False,
206 "examples": ["Benton"],
207 },
208 ),
209 ]
211 loc: Annotated[
212 str | None,
213 Field(
214 default=None,
215 description="Usually a short description of the location",
216 alias=None,
217 json_schema_extra={
218 "units": None,
219 "required": False,
220 "examples": ["Benton, CA"],
221 },
222 ),
223 ]
225 declination: Annotated[
226 Declination,
227 Field(
228 default_factory=lambda: Declination(value=0.0), # type: ignore
229 description="Declination of the station in degrees",
230 alias=None,
231 json_schema_extra={
232 "units": "degrees",
233 "required": True,
234 "examples": ["Declination(10.0)"],
235 },
236 ),
237 ]
239 stdvers: Annotated[
240 StdEDIversionsEnum,
241 Field(
242 default="SEG 1.0",
243 description="EDI standards version SEG 1.0",
244 alias=None,
245 json_schema_extra={
246 "units": None,
247 "required": True,
248 "examples": ["SEG 1.0"],
249 },
250 ),
251 ]
253 survey: Annotated[
254 str | None,
255 Field(
256 default=None,
257 description="Name of the survey",
258 alias=None,
259 json_schema_extra={
260 "units": None,
261 "required": False,
262 "examples": ["CONUS"],
263 },
264 ),
265 ]
267 units: Annotated[
268 str | None,
269 Field(
270 default="milliVolt per kilometer per nanoTesla",
271 description="In the EDI standards this is the elevation units, in newer versions this should be units of the transfer function.",
272 alias=None,
273 json_schema_extra={
274 "units": None,
275 "required": True,
276 "examples": ["milliVolt per kilometer per nanoTesla"],
277 },
278 ),
279 ]
281 @field_validator("acqdate", "enddate", "filedate", "progdate", mode="before")
282 @classmethod
283 def validate_acqdate(
284 cls, field_value: MTime | float | int | np.datetime64 | pd.Timestamp | str
285 ):
286 if isinstance(field_value, MTime):
287 return field_value
288 return MTime(time_stamp=field_value)
290 @field_validator("units", mode="before")
291 @classmethod
292 def validate_units(cls, value: str) -> str:
293 if value in [None, ""]:
294 return ""
295 try:
296 unit_object = get_unit_object(value)
297 return unit_object.name
298 except ValueError as error:
299 raise KeyError(error)
300 except KeyError as error:
301 raise KeyError(error)
303 def __str__(self):
304 return "".join(self.write_header())
306 def __repr__(self):
307 return self.__str__()
309 def get_header_list(self, edi_lines):
310 """
311 Get the header information from the .edi file in the form of a list,
312 where each item is a line in the header section.
314 :param edi_lines: DESCRIPTION
315 :type edi_lines: TYPE
316 :return: DESCRIPTION
317 :rtype: TYPE
319 """
321 header_list = []
322 head_find = False
324 # read in list line by line and then truncate
325 for line in edi_lines:
326 # check for header label
327 if ">" in line and "head" in line.lower():
328 head_find = True
329 # if the header line has been found then the next >
330 # should be the next section so stop
331 elif ">" in line:
332 if head_find is True:
333 break
334 else:
335 pass
336 # get the header information into a list
337 elif head_find:
338 # skip any blank lines
339 if len(line.strip()) > 2:
340 line = line.strip().replace('"', "")
341 h_list = line.split("=")
342 if len(h_list) == 2:
343 key = h_list[0].strip()
344 value = h_list[1].strip()
345 header_list.append("{0}={1}".format(key, value))
346 return header_list
348 def read_header(self, edi_lines):
349 """
350 read a header information from a list of lines
351 containing header information.
353 :param edi_lines: DESCRIPTION
354 :type edi_lines: TYPE
355 :return: DESCRIPTION
356 :rtype: TYPE
358 """
360 for h_line in self.get_header_list(edi_lines):
361 h_list = h_line.split("=")
362 key = h_list[0].lower()
363 value = h_list[1]
364 # test if its a phoenix formated .edi file
365 if key in ["progvers"]:
366 if value.lower().find("mt-editor") != -1:
367 self.phoenix_edi = True
368 elif key in ["coordinate_system"]:
369 value = value.lower()
370 if "geomagnetic" in value:
371 value = "geomagnetic"
372 elif "geographic" in value:
373 value = "geographic"
374 elif "station" in value:
375 value = "station"
376 elif key in ["stdvers"]:
377 if value in ["N/A", "None", "null"]:
378 value = "SEG 1.0"
379 elif key in ["units"]:
380 if value in ["m", "M"]:
381 value = "m"
383 if key == "declination":
384 setattr(self.declination, "value", value)
385 continue
386 elif key in ["long", "lon", "lonigutde"]:
387 key = "longitude"
388 elif key in ["lat", "latitude"]:
389 key = "latitude"
390 elif key in ["elev", "elevation"]:
391 key = "elevation"
392 else:
393 if key in ["dataid"]:
394 value = validate_station_name(value)
396 setattr(self, key, value)
398 def write_header(
399 self,
400 longitude_format="LON",
401 latlon_format="dms",
402 required=False,
403 ):
404 """
405 Write header information to a list of lines.
408 :param header_list: should be read from an .edi file or input as
409 ['key_01=value_01', 'key_02=value_02']
410 :type header_list: list
411 :param longitude_format: whether to write longitude as LON or LONG.
412 options are 'LON' or 'LONG', default 'LON'
413 :type longitude_format: string
414 :param latlon_format: format of latitude and longitude in output edi,
415 degrees minutes seconds ('dms') or decimal
416 degrees ('dd')
417 :type latlon_format: string
419 :returns header_lines: list of lines containing header information
421 """
423 self.filedate = get_now_utc()
424 self.progvers = __version__
425 self.progname = "mt_metadata"
426 self.progdate = "2021-12-01"
428 header_lines = [">HEAD\n"]
429 for key, value in self.to_dict(single=True, required=required).items():
430 if key in ["x", "x2", "y", "y2", "z", "z2"]:
431 continue
432 if value in [None, "None"]:
433 continue
434 if key in ["latitude"]:
435 key = "lat"
436 elif key in ["longitude"]:
437 key = longitude_format.lower()
438 elif key in ["elevation"]:
439 key = "elev"
440 if "declination" in key:
441 if self.declination.value == 0.0:
442 continue
443 if key in ["lat", "lon", "long"] and value is not None:
444 if latlon_format.lower() == "dd":
445 value = f"{value:.6f}"
446 else:
447 value = convert_position_float2str(value)
448 if key in ["elev"] and value is not None:
449 value = "{0:.3f}".format(value)
450 if isinstance(value, list):
451 value = ",".join(value)
452 header_lines.append(f"\t{key.upper()}={value}\n")
453 header_lines.append("\n")
454 return header_lines
456 def _validate_header_list(self, header_list):
457 """
458 make sure the input header list is valid
460 returns a validated header list
461 """
463 if header_list is None:
464 logger.info("No header information to read")
465 return None
466 new_header_list = []
467 for h_line in header_list:
468 h_line = h_line.strip().replace('"', "")
469 if len(h_line) > 1:
470 h_list = h_line.split("=")
471 if len(h_list) == 2:
472 key = h_list[0].strip().lower()
473 value = h_list[1].strip()
474 new_header_list.append("{0}={1}".format(key, value))
475 return new_header_list