Coverage for pygeoapi/api/coverages.py: 22%

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 

40from copy import deepcopy 

41import logging 

42from http import HTTPStatus 

43from typing import Tuple 

44 

45from pygeoapi import l10n 

46from pygeoapi.formats import F_JSON 

47from pygeoapi.openapi import get_oas_30_parameters 

48from pygeoapi.plugin import load_plugin 

49from pygeoapi.provider.base import ProviderGenericError, ProviderTypeError 

50from pygeoapi.provider import get_provider_by_type 

51from pygeoapi.util import filter_dict_by_key_value, to_json 

52 

53from . import ( 

54 APIRequest, API, SYSTEM_LOCALE, validate_bbox, validate_datetime, 

55 validate_subset 

56) 

57 

58LOGGER = logging.getLogger(__name__) 

59 

60CONFORMANCE_CLASSES = [ 

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

62 'http://www.opengis.net/spec/ogcapi-coverages-1/1.0/conf/oas30', 

63 'http://www.opengis.net/spec/ogcapi-coverages-1/1.0/conf/html', 

64 'http://www.opengis.net/spec/ogcapi-coverages-1/1.0/conf/geodata-coverage', 

65 'http://www.opengis.net/spec/ogcapi-coverages-1/1.0/conf/coverage-subset', 

66 'http://www.opengis.net/spec/ogcapi-coverages-1/1.0/conf/coverage-rangesubset', # noqa 

67 'http://www.opengis.net/spec/ogcapi-coverages-1/1.0/conf/coverage-bbox', 

68 'http://www.opengis.net/spec/ogcapi-coverages-1/1.0/conf/coverage-datetime' 

69] 

70 

71 

72def get_collection_coverage( 

73 api: API, request: APIRequest, 

74 dataset: str) -> Tuple[dict, int, str]: 

75 """ 

76 Returns a subset of a collection coverage 

77 

78 :param request: A request object 

79 :param dataset: dataset name 

80 

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

82 """ 

83 

84 query_args = {} 

85 format_ = request.format or F_JSON 

86 

87 # Force response content type and language (en-US only) headers 

88 headers = request.get_response_headers(SYSTEM_LOCALE, **api.api_headers) 

89 

90 LOGGER.debug('Loading provider') 

91 try: 

92 collection_def = get_provider_by_type( 

93 api.config['resources'][dataset]['providers'], 'coverage') 

94 

95 p = load_plugin('provider', collection_def) 

96 except KeyError: 

97 msg = 'collection does not exist' 

98 return api.get_exception( 

99 HTTPStatus.NOT_FOUND, headers, format_, 

100 'InvalidParameterValue', msg) 

101 except ProviderGenericError as err: 

102 return api.get_exception( 

103 err.http_status_code, headers, request.format, 

104 err.ogc_exception_code, err.message) 

105 

106 LOGGER.debug('Processing bbox parameter') 

107 

108 bbox = request.params.get('bbox') 

109 

110 if bbox is None: 

111 bbox = [] 

112 else: 

113 try: 

114 bbox = validate_bbox(bbox) 

115 except ValueError as err: 

116 msg = str(err) 

117 return api.get_exception( 

118 HTTPStatus.INTERNAL_SERVER_ERROR, headers, format_, 

119 'InvalidParameterValue', msg) 

120 

121 query_args['bbox'] = bbox 

122 

123 LOGGER.debug('Processing bbox-crs parameter') 

124 

125 bbox_crs = request.params.get('bbox-crs') 

126 if bbox_crs is not None: 

127 query_args['bbox_crs'] = bbox_crs 

128 

129 LOGGER.debug('Processing datetime parameter') 

130 

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

132 

133 try: 

134 datetime_ = validate_datetime( 

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

136 except ValueError as err: 

137 msg = str(err) 

138 return api.get_exception( 

139 HTTPStatus.BAD_REQUEST, headers, format_, 

140 'InvalidParameterValue', msg) 

141 

142 query_args['datetime_'] = datetime_ 

143 query_args['format_'] = format_ 

144 

145 properties = request.params.get('properties') 

146 if properties: 

147 LOGGER.debug('Processing properties parameter') 

148 query_args['properties'] = [rs for 

149 rs in properties.split(',') if rs] 

150 LOGGER.debug(f"Fields: {query_args['properties']}") 

151 

152 for a in query_args['properties']: 

153 if a not in p.fields: 

154 msg = 'Invalid field specified' 

155 return api.get_exception( 

156 HTTPStatus.BAD_REQUEST, headers, format_, 

157 'InvalidParameterValue', msg) 

158 

159 if 'subset' in request.params: 

160 LOGGER.debug('Processing subset parameter') 

161 try: 

162 subsets = validate_subset(request.params['subset'] or '') 

163 except (AttributeError, ValueError) as err: 

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

165 return api.get_exception( 

166 HTTPStatus.BAD_REQUEST, headers, format_, 

167 'InvalidParameterValue', msg) 

168 

169 if not set(subsets.keys()).issubset(p.axes): 

170 msg = 'Invalid axis name' 

171 return api.get_exception( 

172 HTTPStatus.BAD_REQUEST, headers, format_, 

173 'InvalidParameterValue', msg) 

174 

175 query_args['subsets'] = subsets 

176 LOGGER.debug(f"Subsets: {query_args['subsets']}") 

177 

178 LOGGER.debug('Querying coverage') 

179 try: 

180 data = p.query(**query_args) 

181 except ProviderGenericError as err: 

182 return api.get_exception( 

183 err.http_status_code, headers, request.format, 

184 err.ogc_exception_code, err.message) 

185 

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

187 if format_ == mt: # native format 

188 if p.filename is not None: 

189 cd = f'attachment; filename="{p.filename}"' 

190 headers['Content-Disposition'] = cd 

191 

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

193 return headers, HTTPStatus.OK, data 

194 elif format_ == F_JSON: 

195 headers['Content-Type'] = 'application/prs.coverage+json' 

196 return headers, HTTPStatus.OK, to_json(data, api.pretty_print) 

197 else: 

198 return api.get_format_exception(request) 

199 

200 

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

202 """ 

203 Get OpenAPI fragments 

204 

205 :param cfg: `dict` of configuration 

206 :param locale: `str` of locale 

207 

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

209 """ 

210 

211 from pygeoapi.openapi import OPENAPI_YAML, get_visible_collections 

212 

213 paths = {} 

214 

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

216 'type', 'collection') 

217 

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

219 try: 

220 p = load_plugin('provider', get_provider_by_type( 

221 collections[k]['providers'], 'coverage')) 

222 except ProviderTypeError: 

223 LOGGER.debug('collection is not coverage based') 

224 continue 

225 

226 coverage_path = f'/collections/{k}/coverage' 

227 title = l10n.translate(v['title'], locale) 

228 description = l10n.translate(v['description'], locale) 

229 

230 parameters = get_oas_30_parameters(cfg, locale) 

231 

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

233 coll_properties['schema']['items']['enum'] = list(p.fields.keys()) 

234 

235 paths[coverage_path] = { 

236 'get': { 

237 'summary': f'Get {title} coverage', 

238 'description': description, 

239 'tags': [k], 

240 'operationId': f'get{k.capitalize()}Coverage', 

241 'parameters': [ 

242 {'$ref': '#/components/parameters/lang'}, 

243 {'$ref': '#/components/parameters/f'}, 

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

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

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

247 coll_properties 

248 ], 

249 'responses': { 

250 '200': {'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/responses/Features"}, # noqa 

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

252 '404': {'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/responses/NotFound"}, # noqa 

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

254 } 

255 } 

256 } 

257 

258 return [{'name': 'coverages'}], {'paths': paths}