Coverage for gws-app/gws/gis/mpx/config.py: 57%

1"""MapProxy configuration module for GWS. 

2 

3This module provides functions and classes to create and manage MapProxy configurations. 

4""" 

5 

6from typing import Any, Dict, Generator, List, Optional, Tuple, cast 

7 

8import yaml 

9 

10from mapproxy.wsgiapp import make_wsgi_app 

11 

12import gws 

13import gws.lib.osx 

14 

15CONFIG_PATH = gws.c.CONFIG_DIR + '/mapproxy.yaml' 

16 

17DEFAULT_CONFIG = { 

18 "services": { 

19 "wmts": { 

20 } 

21 }, 

22 "sources": { 

23 "test": { 

24 "type": "tile", 

25 "url": "https://osmtiles.gbd-consult.de/ows/%(z)s/%(x)s/%(y)s.png", 

26 } 

27 }, 

28 "layers": [ 

29 { 

30 "name": "test", 

31 "title": "test", 

32 "sources": [ 

33 "test" 

34 ] 

35 } 

36 ] 

37} 

38 

39 

40class _Config: 

41 """Internal configuration builder for MapProxy. 

42  

43 This class helps build a MapProxy configuration by collecting and organizing 

44 configuration elements from GWS layers. 

45 """ 

46 

47 def __init__(self) -> None: 

48 """Initialize a new MapProxy configuration builder.""" 

49 self.c = 0 

50 

51 self.services = { 

52 'wms': { 

53 'image_formats': ['image/png'], 

54 'max_output_pixels': [9000, 9000] 

55 }, 

56 'wmts': { 

57 'kvp': True, 

58 'restful': False 

59 } 

60 } 

61 

62 self.globals = { 

63 # https://mapproxy.org/docs/1.11.0/configuration.html#id14 

64 # "By default MapProxy assumes lat/long (north/east) order for all geographic and x/y (east/north) order for all projected SRS." 

65 # we need to change that because our extents are always x/y (lon/lat) even if a CRS says otherwise 

66 'srs': { 

67 'axis_order_en': ['EPSG:4326'] 

68 }, 

69 'cache': { 

70 'base_dir': gws.c.MAPPROXY_CACHE_DIR, 

71 'lock_dir': gws.u.ensure_dir(gws.c.TRANSIENT_DIR + '/mpx_locks_' + gws.u.random_string(16)), 

72 'tile_lock_dir': gws.u.ensure_dir(gws.c.TRANSIENT_DIR + '/mpx_tile_locks_' + gws.u.random_string(16)), 

73 'concurrent_tile_creators': 1, 

74 'max_tile_limit': 5000, 

75 

76 }, 

77 'image': { 

78 'resampling_method': 'bicubic', 

79 'stretch_factor': 1.15, 

80 'max_shrink_factor': 4.0, 

81 

82 'formats': { 

83 'png8': { 

84 'format': 'image/png', 

85 'mode': 'P', 

86 'colors': 256, 

87 'transparent': True, 

88 'resampling_method': 'bicubic', 

89 }, 

90 'png24': { 

91 'format': 'image/png', 

92 'mode': 'RGBA', 

93 'colors': 0, 

94 'transparent': True, 

95 'resampling_method': 'bicubic', 

96 } 

97 

98 } 

99 }, 

100 'http': { 

101 'hide_error_details': False, 

102 } 

103 } 

104 

105 self.cfg = {} 

106 

107 def _add(self, kind: str, c: Dict[str, Any]) -> str: 

108 """Add a configuration element to the internal registry. 

109  

110 Args: 

111 kind: The type of configuration element ('source', 'grid', etc.). 

112 c: The configuration dictionary. 

113  

114 Returns: 

115 A unique identifier for the added configuration element. 

116 """ 

117 # mpx doesn't like tuples 

118 for k, v in c.items(): 

119 if isinstance(v, tuple): 

120 c[k] = list(v) 

121 

122 uid = kind + '_' + gws.u.sha256(c) 

123 

124 # clients might add their hash params starting with '$' 

125 c = { 

126 k: v 

127 for k, v in c.items() 

128 if not k.startswith('$') 

129 } 

130 

131 self.cfg[uid] = {'kind': kind, 'c': c} 

132 return uid 

133 

134 def _items(self, kind: str) -> Generator[Tuple[str, Dict[str, Any]], None, None]: 

135 """Get all configuration elements of a specific kind. 

136  

137 Args: 

138 kind: The type of configuration element to retrieve. 

139  

140 Yields: 

141 Tuples of (uid, configuration) for each matching element. 

142 """ 

143 for k, v in self.cfg.items(): 

144 if v['kind'] == kind: 

145 yield k, v['c'] 

146 

147 def cache(self, c: Dict[str, Any]) -> str: 

148 """Add a cache configuration. 

149  

150 Args: 

151 c: The cache configuration dictionary. 

152  

153 Returns: 

154 A unique identifier for the added cache configuration. 

155 """ 

156 return self._add('cache', c) 

157 

158 def source(self, c: Dict[str, Any]) -> str: 

159 """Add a source configuration. 

160  

161 Args: 

162 c: The source configuration dictionary. 

163  

164 Returns: 

165 A unique identifier for the added source configuration. 

166 """ 

167 return self._add('source', c) 

168 

169 def grid(self, c: Dict[str, Any]) -> str: 

170 """Add a grid configuration. 

171  

172 Args: 

173 c: The grid configuration dictionary. 

174  

175 Returns: 

176 A unique identifier for the added grid configuration. 

177 """ 

178 # self._transform_extent(c) 

179 return self._add('grid', c) 

180 

181 def layer(self, c: Dict[str, Any]) -> str: 

182 """Add a layer configuration. 

183  

184 Args: 

185 c: The layer configuration dictionary. 

186  

187 Returns: 

188 A unique identifier for the added layer configuration. 

189 """ 

190 c['title'] = '' 

191 return self._add('layer', c) 

192 

193 def to_dict(self) -> Dict[str, Any]: 

194 """Convert the configuration to a dictionary suitable for MapProxy. 

195  

196 Returns: 

197 A dictionary containing the complete MapProxy configuration. 

198 """ 

199 d = { 

200 'services': self.services, 

201 'globals': self.globals, 

202 } 

203 

204 kinds = ['source', 'grid', 'cache', 'layer'] 

205 for kind in kinds: 

206 d[kind + 's'] = { 

207 key: c 

208 for key, c in self._items(kind) 

209 } 

210 

211 d['layers'] = sorted(d['layers'].values(), key=lambda x: x['name']) 

212 

213 return d 

214 

215 

216def create(root: gws.Root) -> Optional[Dict[str, Any]]: 

217 """Create a MapProxy configuration from the GWS root object. 

218  

219 This function collects configuration from all layers that provide 

220 MapProxy configuration and builds a complete MapProxy configuration. 

221  

222 Args: 

223 root: The GWS root object. 

224  

225 Returns: 

226 A dictionary containing the complete MapProxy configuration, 

227 or None if no layers provide MapProxy configuration. 

228 """ 

229 mc = _Config() 

230 

231 for layer in root.find_all(gws.ext.object.layer): 

232 m = getattr(layer, 'mapproxy_config', None) 

233 if m: 

234 m(mc) 

235 

236 cfg = mc.to_dict() 

237 if not cfg.get('layers'): 

238 return None 

239 

240 crs: list[gws.Crs] = [] 

241 for p in root.find_all(gws.ext.object.map): 

242 crs.append(cast(gws.Map, p).bounds.crs) 

243 for p in root.find_all(gws.ext.object.owsService): 

244 crs.extend(gws.u.get(p, 'supported_crs', default=[])) 

245 cfg['services']['wms']['srs'] = sorted(set(c.epsg for c in crs)) 

246 

247 return cfg 

248 

249 

250def create_and_save(root: gws.Root) -> Optional[Dict[str, Any]]: 

251 """Create a MapProxy configuration and save it to disk. 

252  

253 This function creates a MapProxy configuration and saves it to the 

254 configured path. It also validates the configuration by attempting 

255 to load it with MapProxy. 

256  

257 Args: 

258 root: The GWS root object. 

259  

260 Returns: 

261 The created configuration dictionary, or None if no configuration 

262 was created and force start is not enabled. 

263  

264 Raises: 

265 gws.Error: If the configuration is invalid. 

266 """ 

267 cfg = create(root) 

268 

269 if not cfg: 

270 force = root.app.cfg('server.mapproxy.forceStart') 

271 if force: 

272 gws.log.warning('mapproxy: no configuration, using default') 

273 cfg = DEFAULT_CONFIG 

274 else: 

275 gws.log.warning('mapproxy: no configuration, not starting') 

276 gws.lib.osx.unlink(CONFIG_PATH) 

277 return None 

278 

279 cfg_str = yaml.dump(cfg) 

280 

281 # make sure the config is ok before starting the server! 

282 test_path = CONFIG_PATH + '.test.yaml' 

283 gws.u.write_file(test_path, cfg_str) 

284 

285 try: 

286 make_wsgi_app(test_path) 

287 except Exception as e: 

288 raise gws.Error(f'MAPPROXY ERROR: {e!r}') from e 

289 

290 gws.lib.osx.unlink(test_path) 

291 

292 # write into the real config path 

293 gws.u.write_file(CONFIG_PATH, cfg_str) 

294 

295 return cfg