1from __future__ import annotations
3from datetime import datetime
4from typing import Any, Iterator
6from pyrocko import trace
8META_KEYS = {
9 'measure_length': 'MeasureLength',
10 'start_position': 'StartPosition',
11 'spatial_resolution': 'SpatialResolution',
12 'fibre_index': 'FibreIndex',
13 'fibre_length_multiplier': 'FibreLengthMultiplier',
14 # 'unit_calibration': 'Unit Calibration (nm)',
15 'start_distance': 'StartDistance',
16 'stop_distance': 'StopDistance',
17 'normalization': 'Normalization',
18 'decimation_filter': 'DecimationFilter',
19 'gauge_length': 'GaugeLength',
20 'norm_offset': 'NormOffset',
21 'source_mode': 'SourceMode',
22 'time_decimation': 'TimeDecimation',
23 'zero_offset': 'ZeroOffset',
24 'p_parameter': 'P',
25 'p_coefficients': 'P_Coefficients',
26 'idas_version': 'Version',
27 'precice_sampling_freq': 'PreciseSamplingFrequency',
28 'receiver_gain': 'ReceiverGain',
29 # 'continuous_mode': 'Continuous Mode',
30 'geo_lat': 'Latitude',
31 'geo_lon': 'Longitude',
32 'geo_elevation': 'Altitude',
34 'unit': 'RawDataUnit'
35}
38def get_meta(h5file) -> dict[str, Any]:
39 '''Get metadata from HDF5 file using the same fields as for the TDMS files.
41 Parameters
42 ----------
43 h5file : HDF5 file object
44 The file to extract metadata from
45 Returns
46 -------
47 Dictionary containing the metadata
48 '''
50 key_list = list(META_KEYS.keys())
51 val_list = list(META_KEYS.values())
53 meta = dict()
54 # the following is based on the PRODML format
55 for field in ['Acquisition',
56 'Acquisition/Custom/AdvancedUserSettings',
57 'Acquisition/Custom/SystemSettings',
58 'Acquisition/Custom/SystemInformation/GPS',
59 'Acquisition/Custom/SystemInformation/OSVersion',
60 'Acquisition/Custom/UserSettings',
61 'Acquisition/Raw[0]']:
62 try:
63 field_keys = h5file[field].attrs.keys()
64 for val in val_list:
65 if val not in field_keys:
66 continue
67 meta[key_list[val_list.index(val)]] = h5file[field].attrs[val]
68 except KeyError:
69 raise KeyError(f"Key '{val}' not found in PRODML H5 file.")
71 # some type conversions
72 for val in (
73 'p_coefficients',
74 'receiver_gain',
75 'source_mode',
76 'unit',
77 'idas_version'):
78 meta[val] = meta[val].decode('utf-8')
79 for val in ['decimation_filter', 'normalization']:
80 meta[val] = bool(meta[val])
81 for val in ['receiver_gain']:
82 meta[val] = tuple([float(item) for item in meta[val].split(';')])
84 return meta
87def iload(filename, load_data=True) -> Iterator[trace.Trace]:
89 # prevent hard dependency on h5py
90 try:
91 import h5py
92 except ImportError as exc:
93 raise ImportError(
94 "Please install 'h5py' to proceed,"
95 "e.g. by running 'pip install h5py'") from exc
97 with h5py.File(filename, 'r') as file:
98 # get the meta data
99 meta = get_meta(file)
100 # get the actual time series if load_data
101 data = file['Acquisition/Raw[0]/RawData'][:] if load_data else None
102 # get the sampling rate, starttime, number of samples in space and time
103 deltat = 1.0 / file['Acquisition/Raw[0]'].attrs['OutputDataRate']
104 tmin = datetime.fromisoformat(
105 file['Acquisition/Raw[0]/RawData'].attrs['PartStartTime']
106 .decode('ascii')).timestamp()
107 nchannels = int(file['Acquisition/Raw[0]'].attrs['NumberOfLoci'])
108 nsamples = int(file['Acquisition/Raw[0]/RawDataTime'].attrs['Count'])
110 for icha in range(nchannels):
111 station = '%05i' % icha
112 meta_icha = meta.copy()
113 meta_icha['channel'] = icha
115 tr = trace.Trace(
116 network='DA',
117 station=station,
118 ydata=None,
119 deltat=deltat,
120 tmin=tmin,
121 tmax=tmin + (nsamples - 1) * deltat,
122 meta=meta_icha)
124 if data:
125 tr.set_ydata(data[:, icha])
127 yield tr
130def detect(first512) -> bool:
131 return first512.startswith(b'\x89HDF')