Coverage for /usr/local/lib/python3.11/dist-packages/pyrocko/obspy_compat/snuffling.py: 58%

36 statements  

« prev     ^ index     » next       coverage.py v6.5.0, created at 2024-03-07 11:54 +0000

1# https://pyrocko.org - GPLv3 

2# 

3# The Pyrocko Developers, 21st Century 

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

5 

6from pyrocko.gui.snuffler import snuffling as sn 

7from pyrocko import obspy_compat as oc 

8 

9 

10class ObsPyStreamSnuffling(sn.Snuffling): 

11 ''' 

12 Snuffling to fiddle with an ObsPy stream. 

13 ''' 

14 

15 def __init__(self, obspy_stream=None, *args, **kwargs): 

16 sn.Snuffling.__init__(self, *args, **kwargs) 

17 self.obspy_stream_orig = obspy_stream 

18 self.obspy_stream = obspy_stream.copy() 

19 

20 def setup(self): 

21 self.set_name('ObsPy Stream Fiddler') 

22 

23 if len(self.obspy_stream_orig) != 0: 

24 fmax = 0.5/min( 

25 tr.stats.delta for tr in self.obspy_stream_orig) 

26 fmin = fmax / 1000. 

27 else: 

28 fmin = 0.001 

29 fmax = 1000. 

30 

31 self.add_parameter( 

32 sn.Param( 

33 'Highpass', 'highpass_corner', None, fmin, fmax, 

34 low_is_none=True)) 

35 self.add_parameter( 

36 sn.Param( 

37 'Lowpass', 'lowpass_corner', None, fmin, fmax, 

38 high_is_none=True)) 

39 

40 def init_gui(self, *args, **kwargs): 

41 sn.Snuffling.init_gui(self, *args, **kwargs) 

42 pyrocko_traces = oc.to_pyrocko_traces(self.obspy_stream_orig) 

43 self.add_traces(pyrocko_traces) 

44 

45 def call(self): 

46 try: 

47 obspy_stream = self.obspy_stream_orig.copy() 

48 if None not in (self.highpass_corner, self.lowpass_corner): 

49 obspy_stream.filter( 

50 'bandpass', 

51 freqmin=self.highpass_corner, 

52 freqmax=self.lowpass_corner) 

53 

54 elif self.lowpass_corner is not None: 

55 obspy_stream.filter( 

56 'lowpass', 

57 freq=self.lowpass_corner) 

58 

59 elif self.highpass_corner is not None: 

60 obspy_stream.filter( 

61 'highpass', 

62 freq=self.highpass_corner) 

63 

64 self.cleanup() 

65 pyrocko_traces = oc.to_pyrocko_traces(obspy_stream) 

66 self.add_traces(pyrocko_traces) 

67 self.obspy_stream = obspy_stream 

68 

69 except Exception: 

70 raise # logged by caller 

71 

72 def get_obspy_stream(self): 

73 return self.obspy_stream