Coverage for /usr/local/lib/python3.11/dist-packages/grond/synthetic_tests.py: 36%

1# https://pyrocko.org/grond - GPLv3 

2# 

3# The Grond Developers, 21st Century 

4import numpy as num 

5import logging 

6 

7from pyrocko import trace 

8from pyrocko.guts import (Object, Dict, String, Float, Bool, Int) 

9 

10logger = logging.getLogger('grond.synthetic_tests') 

11 

12guts_prefix = 'grond' 

13 

14 

15class SyntheticWaveformNotAvailable(Exception): 

16 pass 

17 

18 

19class SyntheticTest(Object): 

20 inject_solution = Bool.T(default=False) 

21 respect_data_availability = Bool.T(default=False) 

22 real_noise_scale = Float.T(default=0.0) 

23 white_noise_scale = Float.T(default=0.0) 

24 relative_white_noise_scale = Float.T(default=0.0) 

25 random_response_scale = Float.T(default=0.0) 

26 real_noise_toffset = Float.T(default=-3600.) 

27 random_seed = Int.T(optional=True) 

28 x = Dict.T(String.T(), Float.T()) 

29 

30 def __init__(self, **kwargs): 

31 Object.__init__(self, **kwargs) 

32 self._problem = None 

33 self._synthetics = None 

34 

35 def set_problem(self, problem): 

36 self._problem = problem 

37 self._synthetics = None 

38 

39 def get_problem(self): 

40 if self._problem is None: 

41 raise SyntheticWaveformNotAvailable( 

42 'SyntheticTest.set_problem() has not been called yet') 

43 

44 return self._problem 

45 

46 def get_x(self): 

47 problem = self.get_problem() 

48 if self.x: 

49 x = problem.preconstrain( 

50 problem.get_parameter_array(self.x)) 

51 

52 else: 

53 x = problem.preconstrain( 

54 problem.pack( 

55 problem.base_source)) 

56 

57 return x 

58 

59 def get_synthetics(self): 

60 problem = self.get_problem() 

61 if self._synthetics is None: 

62 x = self.get_x() 

63 results = problem.forward(x) 

64 synthetics = {} 

65 for iresult, result in enumerate(results): 

66 tr = result.trace.pyrocko_trace() 

67 tfade = tr.tmax - tr.tmin 

68 tr_orig = tr.copy() 

69 tr.extend(tr.tmin - tfade, tr.tmax + tfade) 

70 rstate = num.random.RandomState( 

71 (self.random_seed or 0) + iresult) 

72 

73 if self.random_response_scale != 0: 

74 tf = RandomResponse(scale=self.random_response_scale) 

75 tf.set_random_state(rstate) 

76 tr = tr.transfer( 

77 tfade=tfade, 

78 transfer_function=tf) 

79 

80 if self.white_noise_scale != 0.0: 

81 u = rstate.normal( 

82 scale=self.white_noise_scale, 

83 size=tr.data_len()) 

84 

85 tr.ydata += u 

86 

87 if self.relative_white_noise_scale != 0.0: 

88 u = rstate.normal( 

89 scale=self.relative_white_noise_scale * num.std( 

90 tr_orig.ydata), 

91 size=tr.data_len()) 

92 

93 tr.ydata += u 

94 

95 synthetics[result.trace.codes] = tr 

96 

97 self._synthetics = synthetics 

98 

99 return self._synthetics 

100 

101 def get_waveform(self, nslc, tmin, tmax, tfade=0., freqlimits=None): 

102 synthetics = self.get_synthetics() 

103 if nslc not in synthetics: 

104 s = 'no synthetics for %s available' % '.'.join(nslc) 

105 logger.warning(s) 

106 from grond.dataset import NotFound 

107 raise NotFound(s) 

108 

109 tr = synthetics[nslc] 

110 tr.extend(tmin - tfade * 2.0, tmax + tfade * 2.0) 

111 

112 tr = tr.transfer( 

113 tfade=tfade, 

114 freqlimits=freqlimits) 

115 

116 tr.chop(tmin, tmax) 

117 return tr 

118 

119 

120class RandomResponse(trace.FrequencyResponse): 

121 

122 scale = Float.T(default=0.0) 

123 

124 def set_random_state(self, rstate): 

125 self._rstate = rstate 

126 

127 def evaluate(self, freqs): 

128 n = freqs.size 

129 return 1.0 + ( 

130 self._rstate.normal(scale=self.scale, size=n) + 

131 0.0J * self._rstate.normal(scale=self.scale, size=n)) 

132 

133 

134__all__ = ''' 

135SyntheticTest 

136SyntheticWaveformNotAvailable 

137'''.split()