Coverage for C: \ Users \ peaco \ OneDrive \ Documents \ GitHub \ mth5 \ mth5 \ tables \ tf_table.py: 96%

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

2""" 

3Transfer function summary table utilities. 

4 

5Summarize `TransferFunction` groups stored in an MTH5 file into a structured 

6table and provide a convenient `pandas.DataFrame` view for querying. 

7 

8Notes 

9----- 

10- Traversal searches for groups with attribute ``mth5_type='transferfunction'`` 

11 and collects basic availability flags (impedance, tipper, covariance) along 

12 with period range and references. 

13 

14""" 

15 

16from __future__ import annotations 

17 

18import h5py 

19import numpy as np 

20 

21# ============================================================================= 

22# Imports 

23# ============================================================================= 

24import pandas as pd 

25 

26from mth5 import TF_DTYPE 

27from mth5.helpers import validate_name 

28from mth5.tables import MTH5Table 

29 

30 

31# ============================================================================= 

32 

33 

34class TFSummaryTable(MTH5Table): 

35 """ 

36 Summary table for `TransferFunction` groups. 

37 

38 Provides convenience functions to populate the table (`summarize`) and 

39 export to `pandas.DataFrame` (`to_dataframe`). 

40 

41 Examples 

42 -------- 

43 Build and export a TF summary:: 

44 

45 >>> import h5py 

46 >>> from mth5.tables.tf_table import TFSummaryTable 

47 >>> f = h5py.File('example.mth5', 'r') 

48 >>> tf_summary_ds = f['Exchange']['TF_Summary'] 

49 >>> tf_table = TFSummaryTable(tf_summary_ds) 

50 >>> tf_table.summarize() 

51 >>> df = tf_table.to_dataframe() 

52 >>> df.head() 

53 """ 

54 

55 def __init__(self, hdf5_dataset: h5py.Dataset) -> None: 

56 super().__init__(hdf5_dataset, TF_DTYPE) 

57 

58 def to_dataframe(self) -> pd.DataFrame: 

59 """ 

60 Convert the table to a `pandas.DataFrame` for easier querying. 

61 

62 Returns 

63 ------- 

64 pandas.DataFrame 

65 A dataframe with decoded string columns. 

66 

67 Examples 

68 -------- 

69 Filter transfer functions that include tipper:: 

70 

71 >>> df = tf_table.to_dataframe() 

72 >>> df[df.has_tipper] 

73 """ 

74 

75 df = pd.DataFrame(self.array[()]) 

76 for key in [ 

77 "station", 

78 "survey", 

79 "tf_id", 

80 "units", 

81 ]: 

82 setattr(df, key, getattr(df, key).str.decode("utf-8")) 

83 return df 

84 

85 def summarize(self) -> None: 

86 """ 

87 Populate the summary table by traversing the HDF5 hierarchy. 

88 

89 Searches for groups where ``mth5_type`` equals ``'transferfunction'`` 

90 and adds a row indicating available datasets (impedance, tipper, 

91 covariance), period min/max, and relevant references. 

92 

93 Returns 

94 ------- 

95 None 

96 

97 Examples 

98 -------- 

99 Refresh the TF summary:: 

100 

101 >>> tf_table.clear_table() 

102 >>> tf_table.summarize() 

103 """ 

104 self.clear_table() 

105 

106 def recursive_get_tf_entry( 

107 group: h5py.Group | h5py.File | h5py.Dataset, 

108 ) -> None: 

109 """Recursively collect TF summary entries from the hierarchy.""" 

110 if isinstance(group, (h5py.Group, h5py.File)): 

111 for key, node in group.items(): 

112 try: 

113 group_type = node.attrs["mth5_type"].lower() 

114 if group_type == "transferfunction": 

115 has_impedance = False 

116 has_tipper = False 

117 has_covariance = False 

118 if "transfer_function" in node.keys(): 

119 tf_dataset = node["transfer_function"] 

120 if tf_dataset != (1, 1, 1): 

121 nz = np.nonzero(tf_dataset) 

122 unique_values = np.unique(nz[1]) 

123 if 0 in unique_values or 1 in unique_values: 

124 has_impedance = True 

125 if 2 in unique_values: 

126 has_tipper = True 

127 if ( 

128 "residual_covariance" in node.keys() 

129 and "inverse_signal_power" in node.keys() 

130 ): 

131 res = node["residual_covariance"] 

132 isp = node["inverse_signal_power"] 

133 

134 if res.shape != (1, 1, 1) and isp.shape != ( 

135 1, 

136 1, 

137 1, 

138 ): 

139 has_covariance = True 

140 if "period" in node.keys(): 

141 period = node["period"][()] 

142 else: 

143 period = np.zeros(2) 

144 tf_entry = np.array( 

145 [ 

146 ( 

147 validate_name( 

148 node.parent.parent.attrs["id"] 

149 ).encode("utf-8"), 

150 validate_name( 

151 node.parent.parent.parent.parent.attrs["id"] 

152 ).encode("utf-8"), 

153 node.parent.parent.attrs["location.latitude"], 

154 node.parent.parent.attrs["location.longitude"], 

155 node.parent.parent.attrs["location.elevation"], 

156 validate_name(node.attrs["id"]).encode("utf-8"), 

157 node.attrs["units"], 

158 has_impedance, 

159 has_tipper, 

160 has_covariance, 

161 period.min(), 

162 period.max(), 

163 node.ref, 

164 node.parent.parent.ref, 

165 ) 

166 ], 

167 dtype=TF_DTYPE, 

168 ) 

169 self.add_row(tf_entry) 

170 else: 

171 recursive_get_tf_entry(node) 

172 except KeyError: 

173 recursive_get_tf_entry(node) 

174 elif isinstance(group, h5py.Dataset): 

175 pass 

176 

177 parent = self.array.parent 

178 # Allow Mock objects and dictionaries for testing, in addition to h5py types 

179 if not ( 

180 isinstance(parent, (h5py.Group, h5py.File)) 

181 or hasattr(parent, "items") 

182 or isinstance(parent, dict) 

183 ): 

184 raise TypeError("Unexpected parent type for summary dataset.") 

185 recursive_get_tf_entry(parent)