Coverage for /usr/local/lib/python3.11/dist-packages/pyrocko/squirrel/io/backends/textfiles.py: 85%

1# http://pyrocko.org - GPLv3 

2# 

3# The Pyrocko Developers, 21st Century 

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

5 

6''' 

7Squirrel IO adaptor to read classic Pyrocko station and event files. 

8''' 

9 

10import logging 

11 

12from pyrocko.io.io_common import get_stats, touch # noqa 

13from ... import model 

14 

15logger = logging.getLogger('psq.io.textfiles') 

16 

17 

18def provided_formats(): 

19 return ['pyrocko_stations', 'pyrocko_events'] 

20 

21 

22def detect_pyrocko_stations(first512): 

23 try: 

24 first512 = first512.decode('utf-8') 

25 except UnicodeDecodeError: 

26 return False 

27 

28 for line in first512.splitlines(): 

29 t = line.split(None, 5) 

30 if len(t) in (5, 6): 

31 if len(t[0].split('.')) != 3: 

32 return False 

33 

34 try: 

35 lat, lon, ele, dep = map(float, t[1:5]) 

36 if lat < -90. or 90 < lat: 

37 return False 

38 if lon < -180. or 180 < lon: 

39 return False 

40 

41 return True 

42 

43 except Exception: 

44 return False 

45 

46 return False 

47 

48 

49g_event_keys = set(''' 

50name region catalog magnitude_type latitude longitude magnitude depth duration 

51north_shift east_shift mnn mee mdd mne mnd med strike1 dip1 rake1 strike2 dip2 

52rake2 duration time tags moment 

53'''.split()) 

54 

55 

56def detect_pyrocko_events(first512): 

57 try: 

58 first512 = first512.decode('utf-8') 

59 except UnicodeDecodeError: 

60 return False 

61 

62 lines = first512.splitlines()[:-1] 

63 ok = 0 

64 for line in lines: 

65 line = line.strip() 

66 if not line or line.startswith('#'): 

67 continue 

68 

69 t = line.split(' = ', 1) 

70 if len(t) == 2: 

71 if t[0].strip() in g_event_keys: 

72 ok += 1 

73 continue 

74 else: 

75 return False 

76 

77 if line.startswith('---'): 

78 ok += 1 

79 continue 

80 

81 return False 

82 

83 return ok > 2 

84 

85 

86def detect(first512): 

87 if detect_pyrocko_stations(first512): 

88 return 'pyrocko_stations' 

89 

90 elif detect_pyrocko_events(first512): 

91 return 'pyrocko_events' 

92 

93 return None 

94 

95 

96def float_or_none(s): 

97 if s.lower() == 'nan': 

98 return None 

99 else: 

100 return float(s) 

101 

102 

103def iload(format, file_path, segment, content): 

104 if format == 'pyrocko_stations': 

105 return iload_pyrocko_stations(file_path, segment, content) 

106 

107 if format == 'pyrocko_events': 

108 return iload_pyrocko_events(file_path, segment, content) 

109 

110 

111def iload_pyrocko_stations(file_path, segment, content): 

112 

113 inut = 0 

114 tmin = None 

115 tmax = None 

116 with open(file_path, 'r') as f: 

117 

118 have_station = False 

119 for (iline, line) in enumerate(f): 

120 try: 

121 toks = line.split(None, 5) 

122 if len(toks) == 5 or len(toks) == 6: 

123 net, sta, loc = toks[0].split('.') 

124 lat, lon, elevation, depth = [float(x) for x in toks[1:5]] 

125 if len(toks) == 5: 

126 description = u'' 

127 else: 

128 description = str(toks[5]) 

129 

130 nut = model.make_station_nut( 

131 file_segment=0, 

132 file_element=inut, 

133 codes=model.CodesNSL(net, sta, loc), 

134 tmin=tmin, 

135 tmax=tmax) 

136 

137 if 'station' in content: 

138 nut.content = model.Station( 

139 lat=lat, 

140 lon=lon, 

141 elevation=elevation, 

142 depth=depth, 

143 description=description, 

144 **nut.station_kwargs) 

145 

146 yield nut 

147 inut += 1 

148 

149 have_station = True 

150 

151 elif len(toks) == 4 and have_station: 

152 cha = toks[0] 

153 azi = float_or_none(toks[1]) 

154 dip = float_or_none(toks[2]) 

155 gain = float(toks[3]) 

156 

157 if gain != 1.0: 

158 logger.warning( 

159 '%s.%s.%s.%s gain value from stations ' 

160 'file ignored - please check' % ( 

161 net, sta, loc, cha)) 

162 

163 nut = model.make_channel_nut( 

164 file_segment=0, 

165 file_element=inut, 

166 codes=model.CodesNSLCE(net, sta, loc, cha, ''), 

167 tmin=tmin, 

168 tmax=tmax) 

169 

170 if 'channel' in content: 

171 nut.content = model.Channel( 

172 lat=lat, 

173 lon=lon, 

174 elevation=elevation, 

175 depth=depth, 

176 azimuth=azi, 

177 dip=dip, 

178 **nut.channel_kwargs) 

179 

180 yield nut 

181 inut += 1 

182 

183 else: 

184 raise Exception('invalid syntax') 

185 

186 except Exception as e: 

187 logger.warning( 

188 'skipping invalid station/channel definition: %s ' 

189 '(line: %i, file: %s' % (str(e), iline, file_path)) 

190 

191 

192def iload_pyrocko_events(file_path, segment, content): 

193 from pyrocko import model as pmodel 

194 

195 for iev, ev in enumerate(pmodel.Event.load_catalog(file_path)): 

196 nut = model.make_event_nut( 

197 file_segment=0, 

198 file_element=iev, 

199 codes=model.CodesX(ev.catalog or ''), 

200 tmin=ev.time, 

201 tmax=ev.time) 

202 

203 if 'event' in content: 

204 nut.content = ev 

205 

206 yield nut