Coverage for /usr/local/lib/python3.11/dist-packages/pyrocko/gui/snuffler/snufflings/rms.py: 26%

1# https://pyrocko.org - GPLv3 

2# 

3# The Pyrocko Developers, 21st Century 

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

5 

6import time 

7import logging 

8from collections import defaultdict 

9 

10import numpy as num 

11 

12from ..snuffling import Snuffling, Param, Switch 

13from pyrocko.trace import Trace, NoData 

14from pyrocko import util 

15 

16logger = logging.getLogger('pyrocko.gui.snuffling.rms') 

17 

18 

19class RootMeanSquareSnuffling(Snuffling): 

20 

21 ''' 

22 Create traces with blockwise root mean square values. 

23 ''' 

24 

25 def setup(self): 

26 ''' 

27 Customization of the snuffling. 

28 ''' 

29 

30 self.set_name('RMS') 

31 

32 self.add_parameter(Param( 

33 'Highpass [Hz]', 'highpass', None, 0.001, 1000., 

34 low_is_none=True)) 

35 

36 self.add_parameter(Param( 

37 'Lowpass [Hz]', 'lowpass', None, 0.001, 1000., 

38 high_is_none=True)) 

39 

40 self.add_parameter(Param( 

41 'Block Length [s]', 'block_length', 100., 0.1, 3600.)) 

42 

43 self.add_parameter(Switch( 

44 'Group channels', 'group_channels', True)) 

45 

46 self.add_parameter(Switch( 

47 'Log RMS', 'log', False)) 

48 

49 self.add_trigger( 

50 'Copy passband from Main', self.copy_passband) 

51 

52 self.set_live_update(False) 

53 

54 def panel_visibility_changed(self, visible): 

55 viewer = self.get_viewer() 

56 if visible: 

57 viewer.pile_has_changed_signal.connect(self.adjust_controls) 

58 self.adjust_controls() 

59 

60 else: 

61 viewer.pile_has_changed_signal.disconnect(self.adjust_controls) 

62 

63 def adjust_controls(self): 

64 viewer = self.get_viewer() 

65 dtmin, dtmax = viewer.content_deltat_range() 

66 maxfreq = 0.5/dtmin 

67 minfreq = (0.5/dtmax)*0.001 

68 self.set_parameter_range('lowpass', minfreq, maxfreq) 

69 self.set_parameter_range('highpass', minfreq, maxfreq) 

70 

71 def copy_passband(self): 

72 viewer = self.get_viewer() 

73 self.set_parameter('lowpass', viewer.lowpass) 

74 self.set_parameter('highpass', viewer.highpass) 

75 

76 def call(self): 

77 ''' 

78 Main work routine of the snuffling. 

79 ''' 

80 

81 self.cleanup() 

82 

83 tinc = self.block_length 

84 

85 tpad = 0.0 

86 for freq in (self.highpass, self.lowpass): 

87 if freq is not None: 

88 tpad = max(tpad, 1./freq) 

89 

90 targets = {} 

91 tlast = time.time() 

92 for batch in self.chopper_selected_traces( 

93 tinc=tinc, 

94 tpad=tpad, 

95 want_incomplete=False, 

96 style='batch', 

97 mode='visible', 

98 progress='Calculating RMS', 

99 responsive=True, 

100 fallback=True): 

101 

102 tcur = batch.tmin + 0.5 * tinc 

103 

104 if self.group_channels: 

105 def grouping(nslc): 

106 return nslc[:3] + (nslc[3][:-1],) 

107 

108 else: 

109 def grouping(nslc): 

110 return nslc 

111 

112 rms = defaultdict(list) 

113 for tr in batch.traces: 

114 

115 # don't work traces produced by this (and other) snuffling 

116 if tr.meta and 'tabu' in tr.meta and tr.meta['tabu']: 

117 continue 

118 

119 if self.lowpass is not None: 

120 tr.lowpass(4, self.lowpass, nyquist_exception=True) 

121 

122 if self.highpass is not None: 

123 tr.highpass(4, self.highpass, nyquist_exception=True) 

124 

125 try: 

126 tr.chop(batch.tmin, batch.tmax) 

127 val = 0.0 

128 if tr.ydata.size != 0: 

129 y = tr.ydata.astype(float) 

130 if num.any(num.isnan(y)): 

131 logger.error( 

132 'NaN value in trace %s (%s - %s)' % ( 

133 '.'.join(tr.nslc_id), 

134 util.time_to_str(tr.tmin), 

135 util.time_to_str(tr.tmax))) 

136 

137 val = num.sqrt(num.sum(y**2)/tr.ydata.size) 

138 

139 rms[grouping(tr.nslc_id)].append(val) 

140 

141 except NoData: 

142 continue 

143 

144 tnow = time.time() 

145 

146 insert_now = False 

147 if tnow > tlast + 0.8: 

148 insert_now = True 

149 tlast = tnow 

150 

151 to_add = [] 

152 for key, values in rms.items(): 

153 target = targets.get(key, None) 

154 

155 if not target \ 

156 or abs((target.tmax + tinc) - tcur) > 0.01 * tinc \ 

157 or insert_now: 

158 

159 if target: 

160 to_add.append(target) 

161 

162 target = targets[key] = Trace( 

163 network=key[0], 

164 station=key[1], 

165 location=key[2], 

166 channel=key[3] + '-RMS', 

167 tmin=tcur, 

168 deltat=tinc, 

169 ydata=num.zeros(0, dtype=float), 

170 meta={'tabu': True}) 

171 

172 value = num.atleast_1d( 

173 num.sqrt(num.sum(num.array(values, dtype=float)**2))) 

174 

175 if self.log and value != 0.0: 

176 value = num.log(value) 

177 

178 targets[key].append(value) 

179 

180 if to_add: 

181 self.add_traces(to_add) 

182 

183 # add the newly created traces to the viewer 

184 if targets.values(): 

185 self.add_traces(list(targets.values())) 

186 

187 

188def __snufflings__(): 

189 ''' 

190 Returns a list of snufflings to be exported by this module. 

191 ''' 

192 

193 return [RootMeanSquareSnuffling()]