Coverage for pygeoapi/api/collection.py: 43%
169 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 08:15 +0000
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#
10# Copyright (c) 2026 Tom Kralidis
11# Copyright (c) 2026 Francesco Bartoli
12# Copyright (c) 2022 John A Stevenson and Colin Blackburn
13# Copyright (c) 2023 Ricardo Garcia Silva
14#
15# Permission is hereby granted, free of charge, to any person
16# obtaining a copy of this software and associated documentation
17# files (the "Software"), to deal in the Software without
18# restriction, including without limitation the rights to use,
19# copy, modify, merge, publish, distribute, sublicense, and/or sell
20# copies of the Software, and to permit persons to whom the
21# Software is furnished to do so, subject to the following
22# conditions:
23#
24# The above copyright notice and this permission notice shall be
25# included in all copies or substantial portions of the Software.
26#
27# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
28# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
29# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
30# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
31# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
32# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
33# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
34# OTHER DEALINGS IN THE SOFTWARE.
35#
36# =================================================================
38from copy import deepcopy
39import logging
41from pygeoapi import l10n
42from pygeoapi.formats import (F_JSON, F_JSONLD, F_HTML, F_JPEG,
43 F_PNG, FORMAT_TYPES)
44from pygeoapi.crs import DEFAULT_STORAGE_CRS, get_supported_crs_list
45from pygeoapi.plugin import load_plugin
46from pygeoapi.provider import get_provider_by_type, get_provider_default
47from pygeoapi.provider.base import ProviderConnectionError, ProviderTypeError
48from pygeoapi.util import dategetter, get_dataset_formatters
50LOGGER = logging.getLogger(__name__)
52OGC_RELTYPES_BASE = 'http://www.opengis.net/def/rel/ogc/1.0'
55def gen_collection(api, request, dataset: str,
56 locale_: str) -> dict:
57 """
58 Generate OGC API Collection description
60 :param api: `APIRequest` object
61 :param dataset: `str` of dataset name
62 :param locale_: `str` of requested locale
64 :returns: `dict` of OGC API Collection description
65 """
67 config = api.config['resources'][dataset]
69 data = {
70 'id': dataset,
71 'links': []
72 }
74 collection_data = get_provider_default(config['providers'])
75 collection_data_type = collection_data['type']
77 collection_data_format = None
79 if 'format' in collection_data: 79 ↛ 80line 79 didn't jump to line 80 because the condition on line 79 was never true
80 collection_data_format = collection_data['format']
82 is_vector_tile = (collection_data_type == 'tile' and
83 collection_data_format['name'] not
84 in [F_PNG, F_JPEG])
86 data.update({
87 'title': l10n.translate(config['title'], locale_),
88 'description': l10n.translate(config['description'], locale_),
89 'keywords': l10n.translate(config['keywords'], locale_),
90 })
92 extents = deepcopy(config['extents'])
94 bbox = extents['spatial']['bbox']
95 LOGGER.debug('Setting spatial extents from configuration')
96 # The output should be an array of bbox, so if the user only
97 # provided a single bbox, wrap it in a array.
98 if not isinstance(bbox[0], list): 98 ↛ 101line 98 didn't jump to line 101 because the condition on line 98 was always true
99 bbox = [bbox]
101 data['extent'] = {
102 'spatial': {
103 'bbox': bbox
104 }
105 }
107 if 'crs' in extents['spatial']: 107 ↛ 110line 107 didn't jump to line 110 because the condition on line 107 was always true
108 data['extent']['spatial']['crs'] = extents['spatial']['crs']
110 t_ext = extents.get('temporal', {})
111 if t_ext: 111 ↛ 127line 111 didn't jump to line 127 because the condition on line 111 was always true
112 LOGGER.debug('Setting temporal extents from configuration')
113 begins = dategetter('begin', t_ext)
114 ends = dategetter('end', t_ext)
115 data['extent']['temporal'] = {
116 'interval': [[begins, ends]]
117 }
118 if 'trs' in t_ext: 118 ↛ 119line 118 didn't jump to line 119 because the condition on line 118 was never true
119 data['extent']['temporal']['trs'] = t_ext['trs']
120 if 'resolution' in t_ext: 120 ↛ 121line 120 didn't jump to line 121 because the condition on line 120 was never true
121 data['extent']['temporal']['grid'] = {
122 'resolution': t_ext['resolution']
123 }
124 if 'default' in t_ext: 124 ↛ 125line 124 didn't jump to line 125 because the condition on line 124 was never true
125 data['extent']['temporal']['default'] = t_ext['default']
127 _ = extents.pop('spatial', None)
128 _ = extents.pop('temporal', None)
130 for ek, ev in extents.items(): 130 ↛ 131line 130 didn't jump to line 131 because the loop on line 130 never started
131 LOGGER.debug(f'Adding extent {ek}')
132 data['extent'][ek] = {
133 'definition': ev['url'],
134 'interval': [ev['range']]
135 }
136 if 'units' in ev:
137 data['extent'][ek]['unit'] = ev['units']
139 if 'values' in ev:
140 data['extent'][ek]['grid'] = {
141 'cellsCount': len(ev['values']),
142 'coordinates': ev['values']
143 }
145 LOGGER.debug('Processing configured collection links')
146 for link in l10n.translate(config.get('links', []), locale_):
147 lnk = {
148 'type': link['type'],
149 'rel': link['rel'],
150 'title': l10n.translate(link['title'], locale_),
151 'href': l10n.translate(link['href'], locale_),
152 }
153 if 'hreflang' in link: 153 ↛ 156line 153 didn't jump to line 156 because the condition on line 153 was always true
154 lnk['hreflang'] = l10n.translate(
155 link['hreflang'], locale_)
156 content_length = link.get('length', 0)
158 if lnk['rel'] == 'enclosure' and content_length == 0: 158 ↛ 160line 158 didn't jump to line 160 because the condition on line 158 was never true
159 # Issue HEAD request for enclosure links without length
160 lnk_headers = api.prefetcher.get_headers(lnk['href'])
161 content_length = int(lnk_headers.get('content-length', 0))
162 content_type = lnk_headers.get('content-type', lnk['type'])
163 if content_length == 0:
164 # Skip this (broken) link
165 LOGGER.debug(f"Enclosure {lnk['href']} is invalid")
166 continue
167 if content_type != lnk['type']:
168 # Update content type if different from specified
169 lnk['type'] = content_type
170 LOGGER.debug(
171 f"Fixed media type for enclosure {lnk['href']}")
173 if content_length > 0: 173 ↛ 174line 173 didn't jump to line 174 because the condition on line 173 was never true
174 lnk['length'] = content_length
176 data['links'].append(lnk)
178 # TODO: provide translations
179 LOGGER.debug('Adding JSON and HTML link relations')
180 data['links'].extend([{
181 'type': FORMAT_TYPES[F_JSON],
182 'rel': 'root',
183 'title': l10n.translate('The landing page of this server as JSON', locale_), # noqa
184 'href': f"{api.base_url}?f={F_JSON}"
185 }, {
186 'type': FORMAT_TYPES[F_HTML],
187 'rel': 'root',
188 'title': l10n.translate('The landing page of this server as HTML', locale_), # noqa
189 'href': f"{api.base_url}?f={F_HTML}"
190 }, {
191 'type': FORMAT_TYPES[F_JSON],
192 'rel': request.get_linkrel(F_JSON),
193 'title': l10n.translate('This document as JSON', locale_),
194 'href': f'{api.get_collections_url()}/{dataset}?f={F_JSON}'
195 }, {
196 'type': FORMAT_TYPES[F_JSONLD],
197 'rel': request.get_linkrel(F_JSONLD),
198 'title': l10n.translate('This document as RDF (JSON-LD)', locale_),
199 'href': f'{api.get_collections_url()}/{dataset}?f={F_JSONLD}'
200 }, {
201 'type': FORMAT_TYPES[F_HTML],
202 'rel': request.get_linkrel(F_HTML),
203 'title': l10n.translate('This document as HTML', locale_),
204 'href': f'{api.get_collections_url()}/{dataset}?f={F_HTML}'
205 }])
207 if collection_data_type == 'record': 207 ↛ 208line 207 didn't jump to line 208 because the condition on line 207 was never true
208 data['links'].extend([{
209 'type': FORMAT_TYPES[F_JSON],
210 'rel': f'{OGC_RELTYPES_BASE}/ogc-catalog',
211 'title': l10n.translate('Record catalogue as JSON', locale_),
212 'href': f'{api.get_collections_url()}/{dataset}?f={F_JSON}'
213 }, {
214 'type': FORMAT_TYPES[F_HTML],
215 'rel': f'{OGC_RELTYPES_BASE}/ogc-catalog',
216 'title': l10n.translate('Record catalogue as HTML', locale_),
217 'href': f'{api.get_collections_url()}/{dataset}?f={F_HTML}'
218 }])
220 if collection_data_type in ['feature', 'coverage', 'record']: 220 ↛ 233line 220 didn't jump to line 233 because the condition on line 220 was always true
221 data['links'].extend([{
222 'type': 'application/schema+json',
223 'rel': f'{OGC_RELTYPES_BASE}/schema',
224 'title': l10n.translate('Schema of collection in JSON', locale_),
225 'href': f'{api.get_collections_url()}/{dataset}/schema?f={F_JSON}'
226 }, {
227 'type': FORMAT_TYPES[F_HTML],
228 'rel': f'{OGC_RELTYPES_BASE}/schema',
229 'title': l10n.translate('Schema of collection in HTML', locale_),
230 'href': f'{api.get_collections_url()}/{dataset}/schema?f={F_HTML}'
231 }])
233 if is_vector_tile or collection_data_type in ['feature', 'record']: 233 ↛ 273line 233 didn't jump to line 273 because the condition on line 233 was always true
234 # TODO: translate
235 data['itemType'] = collection_data_type
236 LOGGER.debug('Adding feature/record based links')
237 data['links'].extend([{
238 'type': 'application/schema+json',
239 'rel': f'{OGC_RELTYPES_BASE}/queryables',
240 'title': l10n.translate('Queryables for this collection as JSON', locale_), # noqa
241 'href': f'{api.get_collections_url()}/{dataset}/queryables?f={F_JSON}' # noqa
242 }, {
243 'type': FORMAT_TYPES[F_HTML],
244 'rel': f'{OGC_RELTYPES_BASE}/queryables',
245 'title': l10n.translate('Queryables for this collection as HTML', locale_), # noqa
246 'href': f'{api.get_collections_url()}/{dataset}/queryables?f={F_HTML}' # noqa
247 }, {
248 'type': 'application/geo+json',
249 'rel': 'items',
250 'title': l10n.translate('Items as GeoJSON', locale_),
251 'href': f'{api.get_collections_url()}/{dataset}/items?f={F_JSON}'
252 }, {
253 'type': FORMAT_TYPES[F_JSONLD],
254 'rel': 'items',
255 'title': l10n.translate('Items as RDF (GeoJSON-LD)', locale_),
256 'href': f'{api.get_collections_url()}/{dataset}/items?f={F_JSONLD}'
257 }, {
258 'type': FORMAT_TYPES[F_HTML],
259 'rel': 'items',
260 'title': l10n.translate('Items as HTML', locale_), # noqa
261 'href': f'{api.get_collections_url()}/{dataset}/items?f={F_HTML}'
262 }])
264 for key, value in get_dataset_formatters(config).items():
265 data['links'].append({
266 'type': value.mimetype,
267 'rel': 'items',
268 'title': l10n.translate(f'Items as {key}', locale_), # noqa
269 'href': f'{api.get_collections_url()}/{dataset}/items?f={value.f}' # noqa
270 })
272 # OAPIF Part 2 - list supported CRSs and StorageCRS
273 if collection_data_type in ['edr', 'feature']: 273 ↛ 279line 273 didn't jump to line 279 because the condition on line 273 was always true
274 data['crs'] = get_supported_crs_list(collection_data)
275 data['storageCrs'] = collection_data.get('storage_crs', DEFAULT_STORAGE_CRS) # noqa
276 if 'storage_crs_coordinate_epoch' in collection_data: 276 ↛ 277line 276 didn't jump to line 277 because the condition on line 276 was never true
277 data['storageCrsCoordinateEpoch'] = collection_data.get('storage_crs_coordinate_epoch') # noqa
279 elif collection_data_type == 'coverage':
280 LOGGER.debug('Adding coverage based links')
281 data['links'].append({
282 'type': 'application/prs.coverage+json',
283 'rel': f'{OGC_RELTYPES_BASE}/coverage',
284 'title': l10n.translate('Coverage data', locale_),
285 'href': f'{api.get_collections_url()}/{dataset}/coverage?f={F_JSON}' # noqa
286 })
287 if collection_data_format is not None:
288 title_ = l10n.translate('Coverage data as', locale_)
289 title_ = f"{title_} {collection_data_format['name']}"
290 data['links'].append({
291 'type': collection_data_format['mimetype'],
292 'rel': f'{OGC_RELTYPES_BASE}/coverage',
293 'title': title_,
294 'href': f"{api.get_collections_url()}/{dataset}/coverage?f={collection_data_format['name']}" # noqa
295 })
296 if dataset is not None:
297 LOGGER.debug('Creating extended coverage metadata')
298 try:
299 provider_def = get_provider_by_type(
300 api.config['resources'][dataset]['providers'],
301 'coverage')
302 p = load_plugin('provider', provider_def)
303 except ProviderConnectionError:
304 raise
305 except ProviderTypeError:
306 pass
307 else:
308 data['extent']['spatial']['grid'] = [{
309 'cellsCount': p._coverage_properties['width'],
310 'resolution': p._coverage_properties['resx']
311 }, {
312 'cellsCount': p._coverage_properties['height'],
313 'resolution': p._coverage_properties['resy']
314 }]
315 if 'time_range' in p._coverage_properties:
316 data['extent']['temporal'] = {
317 'interval': [p._coverage_properties['time_range']]
318 }
319 if 'restime' in p._coverage_properties:
320 data['extent']['temporal']['grid'] = {
321 'resolution': p._coverage_properties['restime']
322 }
323 if 'uad' in p._coverage_properties:
324 data['extent'].update(p._coverage_properties['uad'])
326 try:
327 tile = get_provider_by_type(config['providers'], 'tile')
328 p = load_plugin('provider', tile)
329 except ProviderConnectionError:
330 raise
331 except ProviderTypeError:
332 tile = None
334 if tile: 334 ↛ 335line 334 didn't jump to line 335 because the condition on line 334 was never true
335 LOGGER.debug('Adding tile links')
336 data['links'].extend([{
337 'type': FORMAT_TYPES[F_JSON],
338 'rel': f'{OGC_RELTYPES_BASE}/tilesets-{p.tile_type}',
339 'title': l10n.translate('Tiles as JSON', locale_),
340 'href': f'{api.get_collections_url()}/{dataset}/tiles?f={F_JSON}'
341 }, {
342 'type': FORMAT_TYPES[F_HTML],
343 'rel': f'{OGC_RELTYPES_BASE}/tilesets-{p.tile_type}',
344 'title': l10n.translate('Tiles as HTML', locale_),
345 'href': f'{api.get_collections_url()}/{dataset}/tiles?f={F_HTML}'
346 }])
348 try:
349 map_ = get_provider_by_type(config['providers'], 'map')
350 p = load_plugin('provider', map_)
351 except ProviderTypeError:
352 map_ = None
354 if map_: 354 ↛ 355line 354 didn't jump to line 355 because the condition on line 354 was never true
355 LOGGER.debug('Adding map links')
357 map_mimetype = map_['format']['mimetype']
358 map_format = map_['format']['name']
360 title_ = l10n.translate('Map as', locale_)
361 title_ = f'{title_} {map_format}'
363 data['links'].append({
364 'type': map_mimetype,
365 'rel': f'{OGC_RELTYPES_BASE}/map',
366 'title': title_,
367 'href': f'{api.get_collections_url()}/{dataset}/map?f={map_format}'
368 })
370 if p._fields:
371 schema_reltype = f'{OGC_RELTYPES_BASE}/schema',
372 schema_links = [s for s in data['links'] if
373 schema_reltype in s]
375 if not schema_links:
376 title_ = l10n.translate('Schema of collection in JSON', locale_) # noqa
377 data['links'].append({
378 'type': 'application/schema+json',
379 'rel': f'{OGC_RELTYPES_BASE}/schema',
380 'title': title_,
381 'href': f'{api.get_collections_url()}/{dataset}/schema?f=json' # noqa
382 })
383 title_ = l10n.translate('Schema of collection in HTML', locale_) # noqa
384 data['links'].append({
385 'type': 'text/html',
386 'rel': f'{OGC_RELTYPES_BASE}/schema',
387 'title': title_,
388 'href': f'{api.get_collections_url()}/{dataset}/schema?f=html' # noqa
389 })
391 try:
392 edr = get_provider_by_type(config['providers'], 'edr')
393 p = load_plugin('provider', edr)
394 except ProviderConnectionError:
395 raise
396 except ProviderTypeError:
397 edr = None
399 if edr: 399 ↛ 401line 399 didn't jump to line 401 because the condition on line 399 was never true
400 # TODO: translate
401 LOGGER.debug('Adding EDR links')
402 data['data_queries'] = {}
403 parameters = p.get_fields()
404 if parameters:
405 data['parameter_names'] = {}
406 for key, value in parameters.items():
407 data['parameter_names'][key] = {
408 'id': key,
409 'type': 'Parameter',
410 'name': value['title'],
411 'observedProperty': {
412 'label': {
413 'id': key,
414 'en': value['title']
415 },
416 },
417 'unit': {
418 'label': {
419 'en': value['title']
420 },
421 'symbol': {
422 'value': value['x-ogc-unit'],
423 'type': 'http://www.opengis.net/def/uom/UCUM/'
424 }
425 }
426 }
428 data['parameter_names'][key].update({
429 'description': value['description']}
430 if 'description' in value else {}
431 )
433 for qt in p.get_query_types():
434 data_query = {
435 'link': {
436 'href': f'{api.get_collections_url()}/{dataset}/{qt}',
437 'rel': 'data',
438 'variables': {
439 'query_type': qt
440 }
441 }
442 }
444 if request.format is not None and request.format == 'json':
445 data_query['link']['type'] = 'application/vnd.cov+json'
447 data['data_queries'][qt] = data_query
449 title1 = l10n.translate('query for this collection as JSON', locale_) # noqa
450 title1 = f'{qt} {title1}'
451 title2 = l10n.translate('query for this collection as HTML', locale_) # noqa
452 title2 = f'{qt} {title2}'
454 data['links'].extend([{
455 'type': 'application/json',
456 'rel': 'data',
457 'title': title1,
458 'href': f'{api.get_collections_url()}/{dataset}/{qt}?f={F_JSON}' # noqa
459 }, {
460 'type': FORMAT_TYPES[F_HTML],
461 'rel': 'data',
462 'title': title2,
463 'href': f'{api.get_collections_url()}/{dataset}/{qt}?f={F_HTML}' # noqa
464 }])
466 for key, value in get_dataset_formatters(config).items():
467 title3 = f'{qt} query for this collection as {key}'
468 data['links'].append({
469 'type': value.mimetype,
470 'rel': 'data',
471 'title': title3,
472 'href': f'{api.get_collections_url()}/{dataset}/{qt}?f={value.f}' # noqa
473 })
475 return data