Python script to parse WCS 1.1.0 capabilities document with OWSLib #wcs #monkeypatch #owslib #python

script.py

from owslib.wcs import WebCoverageService
from owslib.wcs import wcs110

from types import MethodType


import logging
import requests
import json

LOG_LEVEL = "DEBUG"

logging.basicConfig(
    level=LOG_LEVEL,
    format="%(asctime)s - %(levelname)s: %(message)s",
)


def OWS(cls, tag):
    print("Namespaces_1_1_0_fix", "{http://www.opengis.net/ows/1.1}" + tag)
    return "{http://www.opengis.net/ows/1.1}" + tag


def get_wcs_cap(url, version):
    def convert_layer(lyr):
        return {
            "title": wcs[lyr].title,
            "abstract": wcs[lyr].abstract,
            "name": wcs[lyr].id,
            "dataset_md_id": "",  # pdok wcs services do not advertise dataset md link for now, so left empty since unsure how to access dataset md link with owslib for wcs
        }

    try:
        logging.info(f"{version} - {url}")
        # monkeypatch OWS method to fix namespace issue
        wcs110.Namespaces_1_1_0.OWS = MethodType(OWS, wcs110.Namespaces_1_1_0)
        
        wcs = WebCoverageService(url, version=version)
        keywords = wcs.identification.keywords
        getcoverage_op = next(
            (x for x in wcs.operations if x.name == "GetCoverage"), None
        )
        result = {}
        result["formats"] = ",".join(getcoverage_op.formatOptions)
        layers = list(wcs.contents)
        result["title"] = wcs.identification.title
        result["abstract"] = wcs.identification.abstract
        result["layers"] = list(map(convert_layer, layers))
        result["keywords"] = keywords
        return result

    except requests.exceptions.HTTPError as e:
        logging.error(f"url: {url} - {e}")
    except Exception:
        message = f"exception while retrieving WCS cap for service-url: {url}"
        logging.exception(message)


if __name__ == "__main__":
    # url = "http://localhost:8000"
    # url = "https://service.pdok.nl/rws/ahn3/wcs/v1_0"
    url = "https://geodata.nationaalgeoregister.nl/ahn3/wcs"

    print(json.dumps(get_wcs_cap(url, "1.1.0"), indent=4))