Source code for GeoHealthCheck.plugins.probe.tms

from GeoHealthCheck.probe import Probe
from GeoHealthCheck.plugin import Plugin
from owslib.tms import TileMapService

[docs]class TmsCaps(Probe): """Probe for TMS main endpoint url""" NAME = 'TMS Capabilities' DESCRIPTION = 'Perform TMS Capabilities Operation and check validity' RESOURCE_TYPE = 'OSGeo:TMS' REQUEST_METHOD = 'GET' def __init__(self): Probe.__init__(self) CHECKS_AVAIL = { 'GeoHealthCheck.plugins.check.checks.XmlParse': { 'default': True }, 'GeoHealthCheck.plugins.check.checks.ContainsStrings': { 'default': True, 'set_params': { 'strings': { 'name': 'Must contain TileMap Element', 'value': ['<TileMap'] } } }, } """ Checks avail for all specific Caps checks. Optionally override Check.PARAM_DEFS using set_params e.g. with specific `value` or even `name`. """
[docs]class TmsGetTile(Probe): """Fetch TMS tile and check result""" NAME = 'TMS GetTile Single - get SINGLE Tile Image' DESCRIPTION = """Fetch SINGLE TMS-tile. NB extension should match last string of layer.""" RESOURCE_TYPE = 'OSGeo:TMS' REQUEST_METHOD = 'GET' # e.g. # brtachtergrondkaart/1/0/0.png REQUEST_TEMPLATE = '/{layer}/{zoom}/{x}/{y}.{extension}' PARAM_DEFS = { 'layer': { 'type': 'string', 'description': 'The TMS Layer within resource endpoint', 'default': None, 'required': True, 'range': None }, 'zoom': { 'type': 'string', 'description': 'The tile pyramid zoomlevel', 'default': '0', 'required': True, 'range': None }, 'x': { 'type': 'string', 'description': 'The tile x offset', 'default': '0', 'required': True, 'range': None }, 'y': { 'type': 'string', 'description': 'The tile y offset', 'default': '0', 'required': True, 'range': None }, 'extension': { 'type': 'string', 'description': 'The tile image extension', 'default': 'png', 'required': True, 'range': None } } """Param defs""" CHECKS_AVAIL = { 'GeoHealthCheck.plugins.check.checks.HttpHasImageContentType': { 'default': True } } """Check for TMS GetTile""" def __init__(self): Probe.__init__(self) self.layer_count = 0
[docs] def get_metadata(self, resource, version='1.0.0'): """ Get metadata, specific per Resource type. :param resource: :param version: :return: Metadata object """ return TileMapService(resource.url, version=version)
# Overridden: expand param-ranges from WMS metadata
[docs] def expand_params(self, resource): # Use WMS Capabilities doc to get metadata for # PARAM_DEFS ranges/defaults try: tms = self.get_metadata_cached(resource, version='1.0.0') layers = tms.contents self.layer_count = len(layers) # Determine Layers and Extensions layer_list = [layer_name.split('1.0.0/')[-1] for layer_name in layers] self.PARAM_DEFS['layer']['range'] = layer_list # Make a set of all extensions extensions = set([layer.extension for layer_name, layer in layers.items()]) self.PARAM_DEFS['extension']['range'] = list(extensions) except Exception as err: raise err
[docs]class TmsGetTileAll(TmsGetTile): """ Get TMS map image for each Layer using the TMS GetTile operation. """ NAME = 'TMS GetTile All - get Tile Image for ALL Layers' DESCRIPTION = """ Do TMS GetTile request for each Layer. """ PARAM_DEFS = Plugin.merge(TmsGetTile.PARAM_DEFS, {}) def __init__(self): TmsGetTile.__init__(self) self.tms = None self.layers = None # Overridden: expand param-ranges from TMS metadata # from single-layer GetTile parent Probe and set layers # fixed to *
[docs] def expand_params(self, resource): TmsGetTile.expand_params(self, resource) layer_val = 'all %d layers' % self.layer_count extension_val = 'all extensions' self.PARAM_DEFS['layer']['range'] = [layer_val] self.PARAM_DEFS['layer']['value'] = layer_val self.PARAM_DEFS['layer']['default'] = layer_val self.PARAM_DEFS['extension']['range'] = [extension_val] self.PARAM_DEFS['extension']['value'] = extension_val self.PARAM_DEFS['extension']['default'] = extension_val
[docs] def before_request(self): """ Before request to service, overridden from base class""" # Get capabilities doc to get all layers try: self.tms = self.get_metadata_cached(self._resource, version='1.0.0') self.layers = self.tms.contents except Exception as err: self.result.set(False, str(err))
[docs] def perform_request(self): """ Perform actual request to service, overridden from base class""" if not self.layers: self.result.set(False, 'Found no TMS Layers') return self.result.start() results_failed_total = [] for layer_name in self.layers.keys(): # Layer name is last part of full URL self._parameters['layer'] = layer_name.split('1.0.0/')[-1] self._parameters['extension'] = self.layers[layer_name].extension # Let the templated parent perform Probe.perform_request(self) self.run_checks() # Only keep failed Layer results # otherwise with 100s of Layers the report grows out of hand... results_failed = self.result.results_failed if len(results_failed) > 0: # We have a failed layer: add to result message for result in results_failed: result.message = 'layer %s: %s' % \ (layer_name, result.message) results_failed_total += results_failed self.result.results_failed = [] self.result.results = [] self.result.results_failed = results_failed_total self.result.results = results_failed_total