Coverage for C: \ Users \ peaco \ OneDrive \ Documents \ GitHub \ mth5 \ mth5 \ io \ zen \ z3d_metadata.py: 79%
177 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-27 20:09 -0800
1# -*- coding: utf-8 -*-
2"""
3Created on Wed Aug 24 11:35:59 2022
5@author: jpeacock
6"""
8# =============================================================================
9# Imports
10# =============================================================================
11from __future__ import annotations
13from pathlib import Path
14from typing import Any, BinaryIO
16import numpy as np
17from loguru import logger
20# =============================================================================
21class Z3DMetadata:
22 """
23 Read metadata information from a Z3D file and make each metadata entry an attribute.
25 The attributes are left in capitalization of the Z3D file format.
27 Parameters
28 ----------
29 fn : str | pathlib.Path, optional
30 Full path to Z3D file.
31 fid : BinaryIO, optional
32 File object (e.g., open(Z3Dfile, 'rb')).
33 **kwargs : dict
34 Additional keyword arguments to set as attributes.
36 Attributes
37 ----------
38 _header_length : int
39 Length of header in bits (512).
40 _metadata_length : int
41 Length of metadata blocks (512).
42 _schedule_metadata_len : int
43 Length of schedule meta data (512).
44 board_cal : np.ndarray | None
45 Board calibration array with frequency, rate, amplitude, phase.
46 cal_ant : str | None
47 Antenna calibration information.
48 cal_board : dict | None
49 Board calibration dictionary.
50 cal_ver : str | None
51 Calibration version.
52 ch_azimuth : str | None
53 Channel azimuth.
54 ch_cmp : str | None
55 Channel component.
56 ch_length : str | None
57 Channel length (or number of coils).
58 ch_number : str | None
59 Channel number on the ZEN board.
60 ch_xyz1 : str | None
61 Channel xyz location.
62 ch_xyz2 : str | None
63 Channel xyz location.
64 ch_cres : str | None
65 Channel resistance.
66 coil_cal : np.ndarray | None
67 Coil calibration array (frequency, amplitude, phase).
68 fid : BinaryIO | None
69 File object.
70 find_metadata : bool
71 Boolean flag for finding metadata.
72 fn : str | pathlib.Path | None
73 Full path to Z3D file.
74 gdp_operator : str | None
75 Operator of the survey.
76 gdp_progver : str | None
77 Program version.
78 gdp_temp : str | None
79 GDP temperature.
80 gdp_volt : str | None
81 GDP voltage.
82 job_by : str | None
83 Job performed by.
84 job_for : str | None
85 Job for.
86 job_name : str | None
87 Job name.
88 job_number : str | None
89 Job number.
90 line_name : str | None
91 Survey line name.
92 m_tell : int
93 Location in the file where the last metadata block was found.
94 notes : str | None
95 Additional notes from metadata.
96 rx_aspace : str | None
97 Electrode spacing.
98 rx_sspace : str | None
99 Receiver spacing.
100 rx_xazimuth : str | None
101 X azimuth of electrode.
102 rx_xyz0 : str | None
103 Receiver xyz coordinates.
104 rx_yazimuth : str | None
105 Y azimuth of electrode.
106 rx_zpositive : str
107 Z positive direction (default 'down').
108 station : str | None
109 Station name.
110 survey_type : str | None
111 Type of survey.
112 unit_length : str | None
113 Length units (m).
114 count : int
115 Counter for metadata blocks read.
117 Examples
118 --------
119 >>> from mth5.io.zen import Z3DMetadata
120 >>> Z3Dfn = r"/home/mt/mt01/mt01_20150522_080000_256_EX.Z3D"
121 >>> header_obj = Z3DMetadata(fn=Z3Dfn)
122 >>> header_obj.read_metadata()
124 """
126 def __init__(
127 self,
128 fn: str | Path | None = None,
129 fid: BinaryIO | None = None,
130 **kwargs: Any,
131 ) -> None:
132 self.logger = logger
133 self.fn: str | Path | None = fn
134 self.fid: BinaryIO | None = fid
135 self.find_metadata: bool = True
136 self.board_cal: list | np.ndarray | None = None
137 self.coil_cal: list | np.ndarray | None = None
138 self._metadata_length: int = 512
139 self._header_length: int = 512
140 self._schedule_metadata_len: int = 512
141 self.m_tell: int = 0
143 self.cal_ant: str | None = None
144 self.cal_board: dict[str, Any] | None = None
145 self.cal_ver: str | None = None
146 self.ch_azimuth: str | None = None
147 self.ch_cmp: str | None = None
148 self.ch_length: str | None = None
149 self.ch_number: str | None = None
150 self.ch_xyz1: str | None = None
151 self.ch_xyz2: str | None = None
152 self.ch_cres: str | None = None
153 self.gdp_operator: str | None = None
154 self.gdp_progver: str | None = None
155 self.gdp_volt: str | None = None
156 self.gdp_temp: str | None = None
157 self.job_by: str | None = None
158 self.job_for: str | None = None
159 self.job_name: str | None = None
160 self.job_number: str | None = None
161 self.rx_aspace: str | None = None
162 self.rx_sspace: str | None = None
163 self.rx_xazimuth: str | None = None
164 self.rx_xyz0: str | None = None
165 self.rx_yazimuth: str | None = None
166 self.rx_zpositive: str = "down"
167 self.line_name: str | None = None
168 self.survey_type: str | None = None
169 self.unit_length: str | None = None
170 self.station: str | None = None
171 self.count: int = 0
172 self.notes: str | None = None
174 for key in kwargs:
175 setattr(self, key, kwargs[key])
177 def read_metadata(
178 self, fn: str | Path | None = None, fid: BinaryIO | None = None
179 ) -> None:
180 """
181 Read metadata from Z3D file.
183 Parses the metadata blocks in a Z3D file and populates the object's
184 attributes with the extracted values. Also reads calibration data
185 for both board and coil calibrations.
187 Parameters
188 ----------
189 fn : str | pathlib.Path, optional
190 Full path to file. If None, uses the instance's fn attribute.
191 fid : BinaryIO, optional
192 Open file object. If None, uses the instance's fid attribute or
193 opens the file specified by fn.
195 Raises
196 ------
197 UnicodeDecodeError
198 If metadata blocks cannot be decoded as text.
200 Notes
201 -----
202 This method reads metadata blocks sequentially from the Z3D file,
203 starting after the header and schedule metadata sections. It processes:
205 - Standard metadata records with key=value pairs
206 - Board calibration data (cal.brd format)
207 - Coil calibration data (cal.ant format)
208 - Calibration data blocks (caldata format)
210 The method automatically determines the station name from available
211 metadata fields in the following priority:
212 1. line_name + rx_xyz0 (first coordinate)
213 2. rx_stn
214 3. ch_stn
215 """
216 if fn is not None:
217 self.fn = fn
218 if fid is not None:
219 self.fid = fid
220 if self.fn is None and self.fid is None:
221 self.logger.warning("No Z3D file to read")
222 elif self.fn is None:
223 if self.fid is not None:
224 self.fid.seek(self._header_length + self._schedule_metadata_len)
225 elif self.fn is not None:
226 if self.fid is None:
227 self.fid = open(self.fn, "rb")
228 self.fid.seek(self._header_length + self._schedule_metadata_len)
229 else:
230 self.fid.seek(self._header_length + self._schedule_metadata_len)
231 # read in calibration and meta data
232 self.find_metadata = True
233 self.board_cal = []
234 self.coil_cal = []
235 self.count = 0
236 cal_find = False
237 while self.find_metadata == True:
238 try:
239 test_str = self.fid.read(self._metadata_length).decode().lower()
240 except UnicodeDecodeError:
241 self.find_metadata = False
242 self.m_tell = self.fid.tell() + self._metadata_length
243 break
244 if "metadata" in test_str:
245 self.count += 1
246 test_str = test_str.strip().split("record")[1].strip()
248 # split the metadata records with key=value style
249 if test_str.count("|") > 1:
250 for t_str in test_str.split("|"):
251 # get metadata name and value
252 if (
253 t_str.find("=") == -1
254 and t_str.lower().find("line.name") == -1
255 ):
256 # get metadata for older versions of z3d files
257 if len(t_str.split(",")) == 2:
258 t_list = t_str.lower().split(",")
259 t_key = t_list[0].strip().replace(".", "_")
260 if t_key == "ch_varasp":
261 t_key = "ch_length"
262 t_value = t_list[1].strip()
263 setattr(self, t_key, t_value)
264 if t_str.count(" ") > 1:
265 self.notes = t_str
266 # get metadata for just the line that has line name
267 # because for some reason that is still comma separated
268 elif t_str.lower().find("line.name") >= 0:
269 t_list = t_str.split(",")
270 t_key = t_list[0].strip().replace(".", "_")
271 t_value = t_list[1].strip()
272 setattr(self, t_key.lower(), t_value)
273 # get metadata for newer z3d files
274 else:
275 t_list = t_str.split("=")
276 t_key = t_list[0].strip().replace(".", "_")
277 t_value = t_list[1].strip()
278 setattr(self, t_key.lower(), t_value)
279 elif "cal.brd" in test_str:
280 t_list = test_str.split(",")
281 t_key = t_list[0].strip().replace(".", "_")
282 setattr(self, t_key.lower(), t_list[1])
283 for t_str in t_list[2:]:
284 t_str = t_str.replace("\x00", "").replace("|", "")
285 try:
286 self.board_cal.append(
287 [float(tt.strip()) for tt in t_str.strip().split(":")]
288 )
289 except ValueError:
290 self.board_cal.append(
291 [tt.strip() for tt in t_str.strip().split(":")]
292 )
293 # some times the coil calibration does not start on its own line
294 # so need to parse the line up and I'm not sure what the calibration
295 # version is for so I have named it odd
296 elif "cal.ant" in test_str:
297 # check to see if the coil calibration exists
298 cal_find = True
299 test_list = test_str.split(",")
300 coil_num = test_list[1].split("|")[1]
301 coil_key, coil_value = coil_num.split("=")
302 setattr(
303 self,
304 coil_key.replace(".", "_").lower(),
305 coil_value.strip(),
306 )
307 for t_str in test_list[2:]:
308 if "\x00" in t_str:
309 break
310 for tt in t_str.split(":"):
311 try:
312 self.coil_cal.append(float(tt.strip()))
313 except ValueError:
314 pass
316 elif cal_find and self.count > 3:
317 t_list = test_str.replace("|", ",").split(",")
318 for t_str in t_list:
319 if "\x00" in t_str:
320 break
321 else:
322 for tt in t_str.split(":"):
323 try:
324 self.coil_cal.append(float(tt.strip()))
325 except ValueError:
326 pass
327 elif "caldata" in test_str:
328 self.cal_board = {}
329 sr = 256
331 t_list = test_str.lower().split("|")
332 for t_str in t_list:
333 if "\x00" in t_str:
334 continue
335 else:
336 if "cal.brd" in t_str:
337 values = [
338 float(tt) for tt in t_str.split(",")[-1].split(":")
339 ]
340 self.cal_board[sr] = dict(
341 [
342 (tkey, tvalue)
343 for tkey, tvalue in zip(
344 ["frequency", "amplitude", "phase"],
345 values,
346 )
347 ]
348 )
349 elif "cal.adfreq" in t_str:
350 sr = int(t_str.split("=")[-1])
351 elif "caldata" in t_str:
352 continue
353 else:
354 try:
355 cal_key, cal_value = t_str.split("=")
356 try:
357 cal_value = float(cal_value)
358 except ValueError:
359 pass
360 self.cal_board[cal_key] = cal_value
361 except ValueError:
362 self.logger.info("Could not read Calibration Data")
363 else:
364 self.find_metadata = False
365 # need to go back to where the meta data was found so
366 # we don't skip a gps time stamp
367 self.m_tell = self.fid.tell() - self._metadata_length
368 # make coil calibration and board calibration structured arrays
369 if len(self.coil_cal) > 0:
370 a = np.array(self.coil_cal)
371 a = a.reshape((int(a.size / 3), 3))
372 self.coil_cal = np.rec.fromrecords(a, names="frequency, amplitude, phase")
373 if len(self.board_cal) > 0:
374 try:
375 self.board_cal = np.rec.fromrecords(
376 self.board_cal, names="frequency, rate, amplitude, phase"
377 )
378 except ValueError:
379 self.board_cal = None
380 try:
381 self.station = "{0}{1}".format(self.line_name, self.rx_xyz0.split(":")[0])
382 except AttributeError:
383 if hasattr(self, "rx_stn"):
384 self.station = f"{self.rx_stn}"
385 elif hasattr(self, "ch_stn"):
386 self.station = f"{self.ch_stn}"
387 else:
388 self.station = None
389 self.logger.warning("Need to input station name")