Coverage for /usr/local/lib/python3.11/dist-packages/pyrocko/config.py: 91%

119 statements  

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

1# http://pyrocko.org - GPLv3 

2# 

3# The Pyrocko Developers, 21st Century 

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

5 

6''' 

7User configuration file handling. 

8''' 

9 

10import os 

11import os.path as op 

12from copy import deepcopy 

13import logging 

14 

15from . import util 

16from .guts import Object, Float, String, load, dump, List, Dict, TBase, \ 

17 Tuple, StringChoice, Bool 

18 

19 

20logger = logging.getLogger('pyrocko.config') 

21 

22guts_prefix = 'pf' 

23 

24pyrocko_dir_tmpl = os.environ.get( 

25 'PYROCKO_DIR', 

26 os.path.join('~', '.pyrocko')) 

27 

28 

29def make_conf_path_tmpl(name='config'): 

30 return op.join(pyrocko_dir_tmpl, '%s.pf' % name) 

31 

32 

33default_phase_key_mapping = { 

34 'F1': 'P', 'F2': 'S', 'F3': 'R', 'F4': 'Q', 'F5': '?'} 

35 

36 

37class BadConfig(Exception): 

38 pass 

39 

40 

41class PathWithPlaceholders(String): 

42 ''' 

43 Path, possibly containing placeholders. 

44 ''' 

45 pass 

46 

47 

48class VisibleLengthSetting(Object): 

49 class __T(TBase): 

50 def regularize_extra(self, val): 

51 if isinstance(val, list): 

52 return self._cls(key=val[0], value=val[1]) 

53 

54 return val 

55 

56 def to_save(self, val): 

57 return (val.key, val.value) 

58 

59 def to_save_xml(self, val): 

60 raise NotImplementedError() 

61 

62 key = String.T() 

63 value = Float.T() 

64 

65 

66class ConfigBase(Object): 

67 @classmethod 

68 def default(cls): 

69 return cls() 

70 

71 

72class SnufflerConfig(ConfigBase): 

73 visible_length_setting = List.T( 

74 VisibleLengthSetting.T(), 

75 default=[VisibleLengthSetting(key='Short', value=20000.), 

76 VisibleLengthSetting(key='Medium', value=60000.), 

77 VisibleLengthSetting(key='Long', value=120000.), 

78 VisibleLengthSetting(key='Extra Long', value=600000.)]) 

79 phase_key_mapping = Dict.T( 

80 String.T(), String.T(), default=default_phase_key_mapping) 

81 demean = Bool.T(default=True) 

82 show_scale_ranges = Bool.T(default=False) 

83 show_scale_axes = Bool.T(default=False) 

84 trace_scale = String.T(default='individual_scale') 

85 show_boxes = Bool.T(default=True) 

86 clip_traces = Bool.T(default=True) 

87 first_start = Bool.T(default=True) 

88 

89 def get_phase_name(self, key): 

90 return self.phase_key_mapping.get('F%s' % key, 'Undefined') 

91 

92 

93class PyrockoConfig(ConfigBase): 

94 cache_dir = PathWithPlaceholders.T( 

95 default=os.path.join(pyrocko_dir_tmpl, 'cache')) 

96 earthradius = Float.T(default=6371.*1000.) 

97 fdsn_timeout = Float.T(default=None, optional=True) 

98 gf_store_dirs = List.T(PathWithPlaceholders.T()) 

99 gf_store_superdirs = List.T(PathWithPlaceholders.T()) 

100 topo_dir = PathWithPlaceholders.T( 

101 default=os.path.join(pyrocko_dir_tmpl, 'topo')) 

102 tectonics_dir = PathWithPlaceholders.T( 

103 default=os.path.join(pyrocko_dir_tmpl, 'tectonics')) 

104 geonames_dir = PathWithPlaceholders.T( 

105 default=os.path.join(pyrocko_dir_tmpl, 'geonames')) 

106 crustdb_dir = PathWithPlaceholders.T( 

107 default=os.path.join(pyrocko_dir_tmpl, 'crustdb')) 

108 gshhg_dir = PathWithPlaceholders.T( 

109 default=os.path.join(pyrocko_dir_tmpl, 'gshhg')) 

110 volcanoes_dir = PathWithPlaceholders.T( 

111 default=os.path.join(pyrocko_dir_tmpl, 'volcanoes')) 

112 fault_lines_dir = PathWithPlaceholders.T( 

113 default=os.path.join(pyrocko_dir_tmpl, 'fault_lines')) 

114 colortables_dir = PathWithPlaceholders.T( 

115 default=os.path.join(pyrocko_dir_tmpl, 'colortables')) 

116 leapseconds_path = PathWithPlaceholders.T( 

117 default=os.path.join(pyrocko_dir_tmpl, 'leap-seconds.list')) 

118 leapseconds_url = String.T( 

119 default='https://hpiers.obspm.fr/iers/bul/bulc/ntp/leap-seconds.list') 

120 earthdata_credentials = Tuple.T( 

121 2, String.T(), 

122 optional=True) 

123 gui_toolkit = StringChoice.T( 

124 choices=['auto', 'qt4', 'qt5'], 

125 default='auto') 

126 use_high_precision_time = Bool.T(default=False) 

127 

128 

129config_cls = { 

130 'config': PyrockoConfig, 

131 'snuffler': SnufflerConfig 

132} 

133 

134 

135def fill_template(tmpl, config_type): 

136 tmpl = tmpl .format( 

137 module=('.' + config_type) if config_type != 'pyrocko' else '') 

138 return tmpl 

139 

140 

141def expand(x): 

142 x = op.expanduser(op.expandvars(x)) 

143 return x 

144 

145 

146def rec_expand(x): 

147 for prop, val in x.T.ipropvals(x): 

148 if prop.multivalued: 

149 if val is not None: 

150 for i, ele in enumerate(val): 

151 if isinstance(prop.content_t, PathWithPlaceholders.T): 

152 newele = expand(ele) 

153 if newele != ele: 

154 val[i] = newele 

155 

156 elif isinstance(ele, Object): 

157 rec_expand(ele) 

158 else: 

159 if isinstance(prop, PathWithPlaceholders.T): 

160 newval = expand(val) 

161 if newval != val: 

162 setattr(x, prop.name, newval) 

163 

164 elif isinstance(val, Object): 

165 rec_expand(val) 

166 

167 

168def processed(config): 

169 config = deepcopy(config) 

170 rec_expand(config) 

171 return config 

172 

173 

174def mtime(p): 

175 return os.stat(p).st_mtime 

176 

177 

178g_conf_mtime = {} 

179g_conf = {} 

180 

181 

182def raw_config(config_name='config'): 

183 

184 conf_path = expand(make_conf_path_tmpl(config_name)) 

185 

186 if not op.exists(conf_path): 

187 g_conf[config_name] = config_cls[config_name].default() 

188 write_config(g_conf[config_name], config_name) 

189 

190 conf_mtime_now = mtime(conf_path) 

191 if conf_mtime_now != g_conf_mtime.get(config_name, None): 

192 g_conf[config_name] = load(filename=conf_path) 

193 if not isinstance(g_conf[config_name], config_cls[config_name]): 

194 with open(conf_path, 'r') as fconf: 

195 logger.warning('Config file content:') 

196 for line in fconf: 

197 logger.warning(' ' + line) 

198 

199 raise BadConfig('config file does not contain a ' 

200 'valid "%s" section. Found: %s' % ( 

201 config_cls[config_name].__name__, 

202 type(g_conf[config_name]))) 

203 

204 g_conf_mtime[config_name] = conf_mtime_now 

205 

206 return g_conf[config_name] 

207 

208 

209def config(config_name='config'): 

210 return processed(raw_config(config_name)) 

211 

212 

213def write_config(conf, config_name='config'): 

214 conf_path = expand(make_conf_path_tmpl(config_name)) 

215 util.ensuredirs(conf_path) 

216 dump(conf, filename=conf_path) 

217 

218 

219override_gui_toolkit = None 

220 

221 

222def effective_gui_toolkit(): 

223 return override_gui_toolkit or config().gui_toolkit 

224 

225 

226if __name__ == '__main__': 

227 print(config())