Coverage for pygeoapi/api/maps.py: 15%

1# ================================================================= 

2 

3# Authors: Tom Kralidis <tomkralidis@gmail.com> 

4# Francesco Bartoli <xbartolone@gmail.com> 

5# Sander Schaminee <sander.schaminee@geocat.net> 

6# John A Stevenson <jostev@bgs.ac.uk> 

7# Colin Blackburn <colb@bgs.ac.uk> 

8# Ricardo Garcia Silva <ricardo.garcia.silva@geobeyond.it> 

9# Bernhard Mallinger <bernhard.mallinger@eox.at> 

10# 

11# Copyright (c) 2026 Tom Kralidis 

12# Copyright (c) 2025 Francesco Bartoli 

13# Copyright (c) 2022 John A Stevenson and Colin Blackburn 

14# Copyright (c) 2023 Ricardo Garcia Silva 

15# Copyright (c) 2024 Bernhard Mallinger 

16# 

17# Permission is hereby granted, free of charge, to any person 

18# obtaining a copy of this software and associated documentation 

19# files (the "Software"), to deal in the Software without 

20# restriction, including without limitation the rights to use, 

21# copy, modify, merge, publish, distribute, sublicense, and/or sell 

22# copies of the Software, and to permit persons to whom the 

23# Software is furnished to do so, subject to the following 

24# conditions: 

25# 

26# The above copyright notice and this permission notice shall be 

27# included in all copies or substantial portions of the Software. 

28# 

29# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, 

30# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES 

31# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND 

32# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT 

33# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, 

34# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING 

35# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR 

36# OTHER DEALINGS IN THE SOFTWARE. 

37# 

38# ================================================================= 

39 

40 

41from copy import deepcopy 

42from http import HTTPStatus 

43import logging 

44from typing import Tuple 

45 

46from pygeoapi.crs import transform_bbox 

47from pygeoapi.formats import F_JSON, FORMAT_TYPES 

48from pygeoapi.openapi import get_oas_30_parameters 

49from pygeoapi.plugin import load_plugin 

50from pygeoapi.provider import filter_providers_by_type, get_provider_by_type 

51from pygeoapi.provider.base import ( 

52 ProviderGenericError, ProviderInvalidDataError 

53) 

54from pygeoapi.util import to_json, filter_dict_by_key_value 

55 

56from . import APIRequest, API, validate_datetime, validate_subset 

57 

58LOGGER = logging.getLogger(__name__) 

59 

60CONFORMANCE_CLASSES = [ 

61 'http://www.opengis.net/spec/ogcapi-maps-1/1.0/conf/core' 

62] 

63 

64 

65DEFAULT_CRS = 'http://www.opengis.net/def/crs/EPSG/0/4326' 

66 

67 

68def get_collection_map(api: API, request: APIRequest, 

69 dataset: str, style: str | None = None 

70 ) -> Tuple[dict, int, str]: 

71 """ 

72 Returns an image of a collection map 

73 

74 :param request: A request object 

75 :param dataset: dataset name 

76 :param style: style name 

77 

78 :returns: tuple of headers, status code, content 

79 """ 

80 

81 query_args = {} 

82 

83 format_ = request.format or 'png' 

84 headers = request.get_response_headers(**api.api_headers) 

85 LOGGER.debug('Processing query parameters') 

86 

87 LOGGER.debug('Loading provider') 

88 try: 

89 collection_def = get_provider_by_type( 

90 api.config['resources'][dataset]['providers'], 'map') 

91 

92 p = load_plugin('provider', collection_def) 

93 except KeyError: 

94 exception = { 

95 'code': 'InvalidParameterValue', 

96 'description': 'collection does not exist' 

97 } 

98 headers['Content-type'] = 'application/json' 

99 LOGGER.error(exception) 

100 return headers, HTTPStatus.NOT_FOUND, to_json( 

101 exception, api.pretty_print) 

102 except ProviderGenericError as err: 

103 return api.get_exception( 

104 err.http_status_code, headers, request.format, 

105 err.ogc_exception_code, err.message) 

106 

107 query_args['format_'] = request.params.get('f', 'png') 

108 query_args['style'] = style 

109 query_args['crs'] = collection_def.get('crs', DEFAULT_CRS) 

110 query_args['bbox_crs'] = request.params.get( 

111 'bbox-crs', DEFAULT_CRS 

112 ) 

113 query_args['transparent'] = request.params.get('transparent', True) 

114 

115 try: 

116 query_args['width'] = int(request.params.get('width', 500)) 

117 query_args['height'] = int(request.params.get('height', 300)) 

118 except ValueError: 

119 exception = { 

120 'code': 'InvalidParameterValue', 

121 'description': 'invalid width/height' 

122 } 

123 headers['Content-type'] = 'application/json' 

124 LOGGER.error(exception) 

125 return headers, HTTPStatus.BAD_REQUEST, to_json( 

126 exception, api.pretty_print) 

127 

128 LOGGER.debug('Processing bbox parameter') 

129 try: 

130 bbox = request.params.get('bbox').split(',') 

131 if len(bbox) != 4: 

132 exception = { 

133 'code': 'InvalidParameterValue', 

134 'description': 'bbox values should be minx,miny,maxx,maxy' 

135 } 

136 headers['Content-type'] = 'application/json' 

137 LOGGER.error(exception) 

138 return headers, HTTPStatus.BAD_REQUEST, to_json( 

139 exception, api.pretty_print) 

140 except AttributeError: 

141 bbox = api.config['resources'][dataset]['extents']['spatial']['bbox'] # noqa 

142 try: 

143 bbox = [float(c) for c in bbox] 

144 except ValueError: 

145 exception = { 

146 'code': 'InvalidParameterValue', 

147 'description': 'bbox values must be numbers' 

148 } 

149 headers['Content-type'] = 'application/json' 

150 LOGGER.error(exception) 

151 return headers, HTTPStatus.BAD_REQUEST, to_json( 

152 exception, api.pretty_print) 

153 

154 if query_args['bbox_crs'] != query_args['crs']: 

155 LOGGER.debug(f'Reprojecting bbox CRS: {query_args["crs"]}') 

156 bbox = transform_bbox(bbox, query_args['bbox_crs'], query_args['crs']) 

157 

158 query_args['bbox'] = bbox 

159 

160 LOGGER.debug('Processing datetime parameter') 

161 datetime_ = request.params.get('datetime') 

162 try: 

163 query_args['datetime_'] = validate_datetime( 

164 api.config['resources'][dataset]['extents'], datetime_) 

165 except ValueError as err: 

166 msg = str(err) 

167 return api.get_exception( 

168 HTTPStatus.BAD_REQUEST, headers, request.format, 

169 'InvalidParameterValue', msg) 

170 

171 if 'subset' in request.params: 

172 # TODO get subsets from provider 

173 subsets = deepcopy(api.config['resources'][dataset]['extents']) 

174 subsets.pop('spatial', None) # bbox 

175 subsets.pop('temporal', None) # datetime 

176 LOGGER.debug('Processing subset parameter') 

177 try: 

178 query_args['subsets'] = validate_subset( 

179 request.params['subset'] or '') 

180 except (AttributeError, ValueError) as err: 

181 msg = f'Invalid subset: {err}' 

182 return api.get_exception( 

183 HTTPStatus.BAD_REQUEST, headers, format_, 

184 'InvalidParameterValue', msg) 

185 

186 for sk in query_args['subsets'].keys(): 

187 if sk not in subsets.keys(): 

188 msg = f'Subset not found; valid values are {subsets}' 

189 return api.get_exception( 

190 HTTPStatus.BAD_REQUEST, headers, format_, 

191 'InvalidParameterValue', msg) 

192 

193 if request.params.get('properties'): 

194 try: 

195 fields = p.get_fields() or {} 

196 except NotImplementedError: 

197 msg = 'No properties implemented' 

198 headers['Content-Type'] = FORMAT_TYPES[F_JSON] 

199 return api.get_exception( 

200 HTTPStatus.NOT_IMPLEMENTED, headers, format_, 

201 'InvalidParameterValue', msg) 

202 

203 LOGGER.debug('Processing properties parameter') 

204 properties = request.params.get('properties') or [] 

205 if isinstance(properties, str): 

206 properties = properties.split(',') 

207 

208 if properties and not any((fld in properties) 

209 for fld in fields.keys()): 

210 msg = f'Invalid property; valid property names are {list(fields.keys())}' # noqa 

211 headers['Content-Type'] = FORMAT_TYPES[F_JSON] 

212 return api.get_exception( 

213 HTTPStatus.BAD_REQUEST, headers, request.format, 

214 'InvalidParameterValue', msg) 

215 

216 query_args['select_properties'] = properties 

217 

218 LOGGER.debug('Generating map') 

219 try: 

220 data = p.query(**query_args) 

221 except (ProviderGenericError, ProviderInvalidDataError) as err: 

222 headers['Content-Type'] = FORMAT_TYPES[F_JSON] 

223 return api.get_exception( 

224 err.http_status_code, headers, request.format, 

225 err.ogc_exception_code, err.message) 

226 

227 mt = collection_def['format']['name'] 

228 

229 if format_ == mt: 

230 headers['Content-Type'] = collection_def['format']['mimetype'] 

231 return headers, HTTPStatus.OK, data 

232 elif format_ in [None, 'html']: 

233 headers['Content-Type'] = collection_def['format']['mimetype'] 

234 return headers, HTTPStatus.OK, data 

235 else: 

236 exception = { 

237 'code': 'InvalidParameterValue', 

238 'description': 'invalid format parameter' 

239 } 

240 LOGGER.error(exception) 

241 return headers, HTTPStatus.BAD_REQUEST, to_json( 

242 data, api.pretty_print) 

243 

244 

245def get_collection_map_legend(api: API, request: APIRequest, 

246 dataset: str, style: str | None = None 

247 ) -> Tuple[dict, int, str]: 

248 """ 

249 Returns an image of a collection map legend 

250 

251 :param request: A request object 

252 :param dataset: dataset name 

253 :param style: style name 

254 

255 :returns: tuple of headers, status code, content 

256 """ 

257 

258 format_ = 'png' 

259 headers = request.get_response_headers(**api.api_headers) 

260 LOGGER.debug('Processing query parameters') 

261 

262 LOGGER.debug('Loading provider') 

263 try: 

264 collection_def = get_provider_by_type( 

265 api.config['resources'][dataset]['providers'], 'map') 

266 

267 p = load_plugin('provider', collection_def) 

268 except KeyError: 

269 exception = { 

270 'code': 'InvalidParameterValue', 

271 'description': 'collection does not exist' 

272 } 

273 LOGGER.error(exception) 

274 return headers, HTTPStatus.NOT_FOUND, to_json( 

275 exception, api.pretty_print) 

276 except ProviderGenericError as err: 

277 return api.get_exception( 

278 err.http_status_code, headers, request.format, 

279 err.ogc_exception_code, err.message) 

280 

281 LOGGER.debug('Generating legend') 

282 try: 

283 data = p.get_legend(style, request.params.get('f', 'png')) 

284 except ProviderGenericError as err: 

285 return api.get_exception( 

286 err.http_status_code, headers, request.format, 

287 err.ogc_exception_code, err.message) 

288 

289 mt = collection_def['format']['name'] 

290 

291 if format_ == mt: 

292 headers['Content-Type'] = collection_def['format']['mimetype'] 

293 return headers, HTTPStatus.OK, data 

294 else: 

295 exception = { 

296 'code': 'InvalidParameterValue', 

297 'description': 'invalid format parameter' 

298 } 

299 LOGGER.error(exception) 

300 return headers, HTTPStatus.BAD_REQUEST, to_json( 

301 data, api.pretty_print) 

302 

303 

304def get_oas_30(cfg: dict, locale: str) -> tuple[list[dict[str, str]], dict[str, dict]]: # noqa 

305 """ 

306 Get OpenAPI fragments 

307 

308 :param cfg: `dict` of configuration 

309 :param locale: `str` of locale 

310 

311 :returns: `tuple` of `list` of tag objects, and `dict` of path objects 

312 """ 

313 

314 from pygeoapi.openapi import OPENAPI_YAML, get_visible_collections 

315 

316 LOGGER.debug('setting up maps endpoints') 

317 

318 paths = {} 

319 

320 collections = filter_dict_by_key_value(cfg['resources'], 

321 'type', 'collection') 

322 

323 parameters = get_oas_30_parameters(cfg, locale) 

324 for k, v in get_visible_collections(cfg).items(): 

325 map_extension = filter_providers_by_type( 

326 collections[k]['providers'], 'map') 

327 

328 if map_extension: 328 ↛ 329line 328 didn't jump to line 329 because the condition on line 328 was never true

329 mp = load_plugin('provider', map_extension) 

330 

331 coll_properties = deepcopy(parameters)['properties'] 

332 coll_properties['schema']['items']['enum'] = list(mp.fields.keys()) 

333 

334 map_f = deepcopy(parameters['f']) 

335 map_f['schema']['enum'] = [map_extension['format']['name']] 

336 map_f['schema']['default'] = map_extension['format']['name'] 

337 

338 pth = f'/collections/{k}/map' 

339 paths[pth] = { 

340 'get': { 

341 'summary': 'Get map', 

342 'description': f"{v['description']} map", 

343 'tags': [k], 

344 'operationId': 'getMap', 

345 'parameters': [ 

346 {'$ref': '#/components/parameters/bbox'}, 

347 {'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/parameters/datetime"}, # noqa 

348 {'$ref': f"{OPENAPI_YAML['oamaps']}#/components/parameters/subset"}, # noqa 

349 { 

350 'name': 'width', 

351 'in': 'query', 

352 'description': 'Response image width', 

353 'required': False, 

354 'schema': { 

355 'type': 'integer', 

356 }, 

357 'style': 'form', 

358 'explode': False 

359 }, 

360 { 

361 'name': 'height', 

362 'in': 'query', 

363 'description': 'Response image height', 

364 'required': False, 

365 'schema': { 

366 'type': 'integer', 

367 }, 

368 'style': 'form', 

369 'explode': False 

370 }, 

371 { 

372 'name': 'transparent', 

373 'in': 'query', 

374 'description': 'Background transparency of map (default=true).', # noqa 

375 'required': False, 

376 'schema': { 

377 'type': 'boolean', 

378 'default': True, 

379 }, 

380 'style': 'form', 

381 'explode': False 

382 }, 

383 {'$ref': '#/components/parameters/bbox-crs-epsg'}, 

384 map_f 

385 ], 

386 'responses': { 

387 '200': { 

388 'description': 'Response', 

389 'content': { 

390 'application/json': {} 

391 } 

392 }, 

393 '400': {'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/responses/InvalidParameter"}, # noqa 

394 '500': {'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/responses/ServerError"}, # noqa 

395 } 

396 } 

397 } 

398 if coll_properties['schema']['items']['enum']: 

399 paths[pth]['get']['parameters'].append(coll_properties) 

400 

401 if mp.time_field is not None: 

402 paths[pth]['get']['parameters'].append( 

403 {'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/parameters/datetime"}) # noqa 

404 

405 return [{'name': 'maps'}], {'paths': paths}