Coverage for /usr/local/lib/python3.11/dist-packages/pyrocko/io/hdf5_idas.py: 20%

1# http://pyrocko.org - GPLv3 

2# 

3# The Pyrocko Developers, 21st Century 

4# ---|P------/S----------~Lg---------- 

5 

6''' 

7Reader for `Silixia iDAS 

8<https://silixa.com/technology/idas-intelligent-distributed-acoustic-sensor/>`_ 

9HDF5 files. 

10''' 

11 

12from __future__ import annotations 

13 

14from datetime import datetime 

15from typing import Any, Iterator 

16 

17from pyrocko import trace 

18 

19META_KEYS = { 

20 'measure_length': 'MeasureLength', 

21 'start_position': 'StartPosition', 

22 'spatial_resolution': 'SpatialResolution', 

23 'fibre_index': 'FibreIndex', 

24 'fibre_length_multiplier': 'FibreLengthMultiplier', 

25 # 'unit_calibration': 'Unit Calibration (nm)', 

26 'start_distance': 'StartDistance', 

27 'stop_distance': 'StopDistance', 

28 'normalization': 'Normalization', 

29 'decimation_filter': 'DecimationFilter', 

30 'gauge_length': 'GaugeLength', 

31 'norm_offset': 'NormOffset', 

32 'source_mode': 'SourceMode', 

33 'time_decimation': 'TimeDecimation', 

34 'zero_offset': 'ZeroOffset', 

35 'p_parameter': 'P', 

36 'p_coefficients': 'P_Coefficients', 

37 'idas_version': 'Version', 

38 'precice_sampling_freq': 'PreciseSamplingFrequency', 

39 'receiver_gain': 'ReceiverGain', 

40 # 'continuous_mode': 'Continuous Mode', 

41 'geo_lat': 'Latitude', 

42 'geo_lon': 'Longitude', 

43 'geo_elevation': 'Altitude', 

44 

45 'unit': 'RawDataUnit' 

46} 

47 

48 

49def get_meta(h5file) -> dict[str, Any]: 

50 ''' 

51 Get metadata from HDF5 file using the same fields as for the TDMS files. 

52 

53 :param h5file: 

54 The file to extract metadata from. 

55 :type h5file: 

56 `HDF5 file object <https://docs.h5py.org/en/stable/high/file.html>`_ 

57 

58 :returns: 

59 Dictionary containing the metadata. 

60 :rtype: 

61 dict 

62 

63 ''' 

64 

65 key_list = list(META_KEYS.keys()) 

66 val_list = list(META_KEYS.values()) 

67 

68 meta = dict() 

69 # the following is based on the PRODML format 

70 for field in ['Acquisition', 

71 'Acquisition/Custom/AdvancedUserSettings', 

72 'Acquisition/Custom/SystemSettings', 

73 'Acquisition/Custom/SystemInformation/GPS', 

74 'Acquisition/Custom/SystemInformation/OSVersion', 

75 'Acquisition/Custom/UserSettings', 

76 'Acquisition/Raw[0]']: 

77 try: 

78 field_keys = h5file[field].attrs.keys() 

79 for val in val_list: 

80 if val not in field_keys: 

81 continue 

82 meta[key_list[val_list.index(val)]] = h5file[field].attrs[val] 

83 except KeyError: 

84 raise KeyError(f"Key '{val}' not found in PRODML H5 file.") 

85 

86 # some type conversions 

87 for val in ( 

88 'p_coefficients', 

89 'receiver_gain', 

90 'source_mode', 

91 'unit', 

92 'idas_version'): 

93 meta[val] = meta[val].decode('utf-8') 

94 for val in ['decimation_filter', 'normalization']: 

95 meta[val] = bool(meta[val]) 

96 for val in ['receiver_gain']: 

97 meta[val] = tuple([float(item) for item in meta[val].split(';')]) 

98 

99 return meta 

100 

101 

102def iload(filename, load_data=True) -> Iterator[trace.Trace]: 

103 

104 # prevent hard dependency on h5py 

105 try: 

106 import h5py 

107 except ImportError as exc: 

108 raise ImportError( 

109 "Please install 'h5py' to proceed," 

110 "e.g. by running 'pip install h5py'") from exc 

111 

112 with h5py.File(filename, 'r') as file: 

113 # get the meta data 

114 meta = get_meta(file) 

115 # get the actual time series if load_data 

116 data = file['Acquisition/Raw[0]/RawData'][:] if load_data else None 

117 # get the sampling rate, starttime, number of samples in space and time 

118 deltat = 1.0 / file['Acquisition/Raw[0]'].attrs['OutputDataRate'] 

119 tmin = datetime.fromisoformat( 

120 file['Acquisition/Raw[0]/RawData'].attrs['PartStartTime'] 

121 .decode('ascii')).timestamp() 

122 nchannels = int(file['Acquisition/Raw[0]'].attrs['NumberOfLoci']) 

123 nsamples = int(file['Acquisition/Raw[0]/RawDataTime'].attrs['Count']) 

124 

125 for icha in range(nchannels): 

126 station = '%05i' % icha 

127 meta_icha = meta.copy() 

128 meta_icha['channel'] = icha 

129 

130 tr = trace.Trace( 

131 network='DA', 

132 station=station, 

133 ydata=None, 

134 deltat=deltat, 

135 tmin=tmin, 

136 tmax=tmin + (nsamples - 1) * deltat, 

137 meta=meta_icha) 

138 

139 if data: 

140 tr.set_ydata(data[:, icha]) 

141 

142 yield tr 

143 

144 

145def detect(first512) -> bool: 

146 return first512.startswith(b'\x89HDF')