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 2023-10-06 06:59 +0000
« prev ^ index » next coverage.py v6.5.0, created at 2023-10-06 06:59 +0000
1# https://pyrocko.org - GPLv3
2#
3# The Pyrocko Developers, 21st Century
4# ---|P------/S----------~Lg----------
6from pyrocko.gui.snuffler import snuffling as sn
7from pyrocko import obspy_compat as oc
10class ObsPyStreamSnuffling(sn.Snuffling):
11 '''
12 Snuffling to fiddle with an ObsPy stream.
13 '''
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()
20 def setup(self):
21 self.set_name('ObsPy Stream Fiddler')
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.
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))
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)
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)
54 elif self.lowpass_corner is not None:
55 obspy_stream.filter(
56 'lowpass',
57 freq=self.lowpass_corner)
59 elif self.highpass_corner is not None:
60 obspy_stream.filter(
61 'highpass',
62 freq=self.highpass_corner)
64 self.cleanup()
65 pyrocko_traces = oc.to_pyrocko_traces(obspy_stream)
66 self.add_traces(pyrocko_traces)
67 self.obspy_stream = obspy_stream
69 except Exception:
70 raise # logged by caller
72 def get_obspy_stream(self):
73 return self.obspy_stream