Coverage for C: \ Users \ peaco \ OneDrive \ Documents \ GitHub \ mth5 \ mth5 \ io \ metronix \ metronix_metadata.py: 99%

1# -*- coding: utf-8 -*- 

2""" 

3Metronix metadata parsing utilities. 

4 

5This module provides classes for parsing and managing metadata from Metronix 

6ATSS (Audio Time Series System) files and associated JSON metadata files. 

7 

8Classes 

9------- 

10MetronixFileNameMetadata 

11 Parse metadata from Metronix filename conventions 

12MetronixChannelJSON 

13 Read and parse Metronix JSON metadata files 

14 

15Created on Fri Nov 22 13:23:42 2024 

16 

17@author: jpeacock 

18""" 

19 

20# ============================================================================= 

21# Imports 

22# ============================================================================= 

23import json 

24from pathlib import Path 

25from types import SimpleNamespace 

26from typing import Any, Union 

27 

28import numpy as np 

29from loguru import logger 

30from mt_metadata.timeseries import AppliedFilter, Electric, Magnetic 

31from mt_metadata.timeseries.filters import ChannelResponse, FrequencyResponseTableFilter 

32 

33 

34# ============================================================================= 

35 

36 

37class MetronixFileNameMetadata: 

38 """ 

39 Parse and manage metadata from Metronix filename conventions. 

40 

41 This class extracts metadata information from Metronix ATSS filenames 

42 including system information, channel details, and file properties. 

43 

44 Parameters 

45 ---------- 

46 fn : Union[str, Path, None], optional 

47 Path to Metronix file, by default None 

48 **kwargs 

49 Additional keyword arguments (currently unused) 

50 

51 Attributes 

52 ---------- 

53 system_number : str or None 

54 System identification number 

55 system_name : str or None 

56 Name of the system 

57 channel_number : int or None 

58 Channel number (parsed from C## format) 

59 component : str or None 

60 Component designation (e.g., 'ex', 'ey', 'hx', 'hy', 'hz') 

61 sample_rate : float or None 

62 Sampling rate in Hz 

63 file_type : str or None 

64 Type of file ('metadata' or 'timeseries') 

65 """ 

66 

67 def __init__(self, fn: Union[str, Path, None] = None, **kwargs: Any) -> None: 

68 self.system_number: str | None = None 

69 self.system_name: str | None = None 

70 self.channel_number: int | None = None 

71 self.component: str | None = None 

72 self.sample_rate: float | None = None 

73 self.file_type: str | None = None 

74 

75 self.fn = fn 

76 

77 def __str__(self) -> str: 

78 """ 

79 Return string representation of the metadata. 

80 

81 Returns 

82 ------- 

83 str 

84 Formatted string showing Metronix file information 

85 """ 

86 if self.fn is not None: 

87 lines = [f"Metronix ATSS {self.file_type.upper()}:"] 

88 lines.append(f"\tSystem Name: {self.system_name}") 

89 lines.append(f"\tSystem Number: {self.system_number}") 

90 lines.append(f"\tChannel Number: {self.channel_number}") 

91 lines.append(f"\tComponent: {self.component}") 

92 lines.append(f"\tSample Rate: {self.sample_rate}") 

93 return "\n".join(lines) 

94 

95 def __repr__(self) -> str: 

96 """ 

97 Return string representation for debugging. 

98 

99 Returns 

100 ------- 

101 str 

102 String representation of the object 

103 """ 

104 return self.__str__() 

105 

106 @property 

107 def fn(self) -> Path | None: 

108 """ 

109 Get the file path. 

110 

111 Returns 

112 ------- 

113 Path or None 

114 File path object or None if not set 

115 """ 

116 return self._fn 

117 

118 @fn.setter 

119 def fn(self, value: Union[str, Path, None]) -> None: 

120 """ 

121 Set the file path and parse metadata from filename. 

122 

123 Parameters 

124 ---------- 

125 value : Union[str, Path, None] 

126 File path to set 

127 """ 

128 if value is None: 

129 self._fn = None 

130 else: 

131 self._fn = Path(value) 

132 self._parse_fn(self._fn) 

133 

134 @property 

135 def fn_exists(self) -> bool: 

136 """ 

137 Check if the file exists. 

138 

139 Returns 

140 ------- 

141 bool 

142 True if file exists, False otherwise 

143 """ 

144 if self.fn is not None: 

145 return self.fn.exists() 

146 return False 

147 

148 def _parse_fn(self, fn: Path | None) -> None: 

149 """ 

150 Parse metadata from Metronix filename. 

151 

152 Extracts system number, system name, channel number, component, 

153 sample rate, and file type from the filename following Metronix 

154 conventions. 

155 

156 Parameters 

157 ---------- 

158 fn : Path or None 

159 File path to parse 

160 """ 

161 if fn is None: 

162 return 

163 

164 fn_list = fn.stem.split("_") 

165 self.system_number = fn_list[0] 

166 self.system_name = fn_list[1] 

167 self.channel_number = self._parse_channel_number(fn_list[2]) 

168 self.component = self._parse_component(fn_list[3]) 

169 self.sample_rate = self._parse_sample_rate(fn_list[4]) 

170 self.file_type = self._get_file_type(fn) 

171 

172 def _parse_channel_number(self, value: str) -> int: 

173 """ 

174 Parse channel number from filename component. 

175 

176 Channel number is in format C## where ## is the channel number. 

177 

178 Parameters 

179 ---------- 

180 value : str 

181 Channel string in format 'C##' 

182 

183 Returns 

184 ------- 

185 int 

186 Channel number 

187 """ 

188 return int(value.replace("C", "0")) 

189 

190 def _parse_component(self, value: str) -> str: 

191 """ 

192 Parse component designation from filename. 

193 

194 Component is in format T{comp} where {comp} is the component name 

195 (e.g., 'ex', 'ey', 'hx', 'hy', 'hz'). 

196 

197 Parameters 

198 ---------- 

199 value : str 

200 Component string in format 'T{comp}' 

201 

202 Returns 

203 ------- 

204 str 

205 Component name in lowercase 

206 """ 

207 return value.replace("T", "").lower() 

208 

209 def _parse_sample_rate(self, value: str) -> float: 

210 """ 

211 Parse sample rate from filename component. 

212 

213 Sample rate can be in format {sr}Hz (frequency) or {sr}s (period). 

214 For period format, returns 1/period to get frequency. 

215 

216 Parameters 

217 ---------- 

218 value : str 

219 Sample rate string (e.g., '100Hz' or '0.01s') 

220 

221 Returns 

222 ------- 

223 float 

224 Sample rate in Hz 

225 """ 

226 if "hz" in value.lower(): 

227 return float(value.lower().replace("hz", "")) 

228 elif "s" in value.lower(): 

229 return 1.0 / float(value.lower().replace("s", "")) 

230 

231 def _get_file_type(self, value: Path) -> str: 

232 """ 

233 Determine file type from file extension. 

234 

235 Parameters 

236 ---------- 

237 value : Path 

238 File path object 

239 

240 Returns 

241 ------- 

242 str 

243 File type ('metadata' for .json, 'timeseries' for .atss) 

244 

245 Raises 

246 ------ 

247 ValueError 

248 If file type is not supported 

249 """ 

250 if value.suffix in [".json"]: 

251 return "metadata" 

252 elif value.suffix in [".atss"]: 

253 return "timeseries" 

254 else: 

255 raise ValueError(f"Metronix file type {value} not supported.") 

256 

257 @property 

258 def file_size(self) -> int: 

259 """ 

260 Get file size in bytes. 

261 

262 Returns 

263 ------- 

264 int 

265 File size in bytes, 0 if file is None 

266 """ 

267 if self.fn is not None: 

268 return self.fn.stat().st_size 

269 return 0 

270 

271 @property 

272 def n_samples(self) -> float: 

273 """ 

274 Get estimated number of samples in file. 

275 

276 Assumes 8 bytes per sample (double precision). 

277 

278 Returns 

279 ------- 

280 float 

281 Estimated number of samples 

282 """ 

283 return self.file_size / 8 

284 

285 @property 

286 def duration(self) -> float: 

287 """ 

288 Get estimated duration of the file in seconds. 

289 

290 Returns 

291 ------- 

292 float 

293 Duration in seconds 

294 """ 

295 return self.n_samples / self.sample_rate 

296 

297 

298class MetronixChannelJSON(MetronixFileNameMetadata): 

299 """ 

300 Read and parse Metronix JSON metadata files. 

301 

302 This class extends MetronixFileNameMetadata to handle JSON metadata 

303 files containing channel configuration and calibration information. 

304 

305 Parameters 

306 ---------- 

307 fn : Union[str, Path, None], optional 

308 Path to Metronix JSON file, by default None 

309 **kwargs 

310 Additional keyword arguments passed to parent class 

311 

312 Attributes 

313 ---------- 

314 metadata : SimpleNamespace or None 

315 Parsed JSON metadata as a SimpleNamespace object 

316 """ 

317 

318 def __init__(self, fn: Union[str, Path, None] = None, **kwargs: Any) -> None: 

319 super().__init__(fn=fn, **kwargs) 

320 self.metadata: SimpleNamespace | None = None 

321 if self.fn is not None: 

322 self.read(self.fn) 

323 

324 def _has_metadata(self) -> bool: 

325 """ 

326 Check if metadata has been loaded. 

327 

328 Returns 

329 ------- 

330 bool 

331 True if metadata is loaded, False otherwise 

332 """ 

333 if self.metadata is None: 

334 return False 

335 return True 

336 

337 @MetronixFileNameMetadata.fn.setter 

338 def fn(self, value: Union[str, Path, None]) -> None: 

339 """ 

340 Set the file path and read JSON metadata. 

341 

342 Parameters 

343 ---------- 

344 value : Union[str, Path, None] 

345 Path to JSON file 

346 

347 Raises 

348 ------ 

349 IOError 

350 If JSON file cannot be found 

351 """ 

352 if value is None: 

353 self._fn = None 

354 

355 else: 

356 value = Path(value) 

357 if not value.exists(): 

358 raise IOError(f"Cannot find Metronix JSON file {value}") 

359 self._fn = value 

360 self._parse_fn(self._fn) 

361 self.read() 

362 

363 def read(self, fn: Union[str, Path, None] = None) -> None: 

364 """ 

365 Read JSON metadata from file. 

366 

367 Parameters 

368 ---------- 

369 fn : Union[str, Path, None], optional 

370 Path to JSON file, by default None (uses self.fn) 

371 

372 Raises 

373 ------ 

374 IOError 

375 If JSON file cannot be found 

376 """ 

377 if fn is not None: 

378 self.fn = fn 

379 

380 if not self.fn_exists: 

381 raise IOError(f"Cannot find Metronix JSON file {self.fn}") 

382 

383 with open(self.fn, "r") as fid: 

384 self.metadata = json.load(fid, object_hook=lambda d: SimpleNamespace(**d)) 

385 

386 def get_channel_metadata(self) -> Union[Electric, Magnetic, None]: 

387 """ 

388 Translate to mt_metadata.timeseries.Channel object. 

389 

390 Creates either Electric or Magnetic metadata objects based on the 

391 component type and applies calibration filters. 

392 

393 Returns 

394 ------- 

395 Union[Electric, Magnetic, None] 

396 mt_metadata object based on component type, or None if no metadata 

397 

398 Raises 

399 ------ 

400 ValueError 

401 If component type is not recognized 

402 """ 

403 if not self._has_metadata(): 

404 return 

405 

406 sensor_response_filter = self.get_sensor_response_filter() 

407 

408 if self.component.startswith("e"): 

409 metadata_object = Electric( 

410 component=self.component, 

411 channel_number=self.channel_number, 

412 measurement_azimuth=self.metadata.angle, 

413 measurement_tilt=self.metadata.tilt, 

414 sample_rate=self.sample_rate, 

415 type="electric", 

416 ) 

417 metadata_object.positive.latitude = self.metadata.latitude 

418 metadata_object.positive.longitude = self.metadata.longitude 

419 metadata_object.positive.elevation = self.metadata.elevation 

420 metadata_object.contact_resistance.start = self.metadata.resistance 

421 elif self.component.startswith("h"): 

422 metadata_object = Magnetic( 

423 component=self.component, 

424 channel_number=self.channel_number, 

425 measurement_azimuth=self.metadata.angle, 

426 measurement_tilt=self.metadata.tilt, 

427 sample_rate=self.sample_rate, 

428 type="magnetic", 

429 ) 

430 metadata_object.location.latitude = self.metadata.latitude 

431 metadata_object.location.longitude = self.metadata.longitude 

432 metadata_object.location.elevation = self.metadata.elevation 

433 metadata_object.sensor.id = self.metadata.sensor_calibration.serial 

434 metadata_object.sensor.manufacturer = "Metronix Geophysics" 

435 metadata_object.sensor.type = "induction coil" 

436 metadata_object.sensor.model = self.metadata.sensor_calibration.sensor 

437 

438 else: 

439 msg = f"Do not understand channel component {self.component}" 

440 logger.error(msg) 

441 raise ValueError(msg) 

442 

443 metadata_object.time_period.start = self.metadata.datetime 

444 metadata_object.time_period.end = ( 

445 metadata_object.time_period.start + self.duration 

446 ) 

447 

448 metadata_object.units = self.metadata.units 

449 

450 count = 0 

451 for f in self.metadata.filter.split(","): 

452 f = f.strip() 

453 if not f: 

454 continue 

455 count += 1 

456 metadata_object.add_filter( 

457 AppliedFilter(name=f, applied=True, stage=count) 

458 ) 

459 if sensor_response_filter is not None: 

460 metadata_object.add_filter( 

461 AppliedFilter( 

462 name=sensor_response_filter.name, applied=True, stage=count + 1 

463 ) 

464 ) 

465 # metadata_object.filter.name = self.metadata.filter.split(",") + [ 

466 # sensor_response_filter.name 

467 # ] 

468 # else: 

469 # metadata_object.filter.name = self.metadata.filter.split(",") 

470 # metadata_object.filter.applied = [True] * len(metadata_object.filter.name) 

471 

472 return metadata_object 

473 

474 def get_sensor_response_filter(self) -> FrequencyResponseTableFilter | None: 

475 """ 

476 Get the sensor response frequency-amplitude-phase filter. 

477 

478 Creates a FrequencyResponseTableFilter from the sensor calibration 

479 data stored in the JSON metadata. 

480 

481 Returns 

482 ------- 

483 FrequencyResponseTableFilter or None 

484 Sensor response filter if calibration data exists, None otherwise 

485 """ 

486 if not self._has_metadata(): 

487 return 

488 

489 fap = FrequencyResponseTableFilter( 

490 calibration_date=self.metadata.sensor_calibration.datetime, 

491 name=f"{self.metadata.sensor_calibration.sensor}_chopper_{self.metadata.sensor_calibration.chopper}".lower(), 

492 frequencies=self.metadata.sensor_calibration.f, 

493 amplitudes=self.metadata.sensor_calibration.a, 

494 units_out=self.metadata.units, 

495 units_in=self.metadata.sensor_calibration.units_amplitude.split("/")[-1], 

496 ) 

497 

498 if self.metadata.sensor_calibration.units_phase in ["degrees", "deg"]: 

499 fap.phases = np.deg2rad(self.metadata.sensor_calibration.p) 

500 else: 

501 fap.phases = self.metadata.sensor_calibration.p 

502 

503 if len(fap.frequencies) > 0: 

504 return fap 

505 return None 

506 

507 def get_channel_response(self) -> ChannelResponse: 

508 """ 

509 Get all filters needed to calibrate the data. 

510 

511 Returns 

512 ------- 

513 ChannelResponse 

514 Channel response object containing all calibration filters 

515 """ 

516 filter_list = [] 

517 fap = self.get_sensor_response_filter() 

518 if fap is not None: 

519 filter_list.append(fap) 

520 return ChannelResponse(filters_list=filter_list)