mapfile util script to convert existing mapfiles to PDOK WMS CR YAML

mapfile-util.py

#!/usr/bin/env python3
import os
import urllib.parse as urlparse
from urllib.parse import parse_qs
import sys
import re
import yaml
import mappyfile
import click


def get_dataset_md_identifier(metadata_url):
    """
    Arguments:
    - metadata_url: url to metadata record either a full CSW getrecordbyid request 
    or like xml.metadata.get?uuid=$UUID

    Return value: the metadata record identifier (string)
    """
    parsed = urlparse.urlparse(metadata_url)
    parsed_qs = parse_qs(parsed.query)
    ds_md_id = ""
    if "id" in parsed_qs:
        ds_md_id = parsed_qs["id"][0]
    elif "uuid" in parsed_qs:
        ds_md_id = parsed_qs["uuid"][0]
    return ds_md_id

def get_styles_for_layer(classes, metadata):
    """
    Arguments:
    - classes: list of mappyfile class objects of mappyfile layer x
    - metadata: mappyfile layer metadata object of mappyfile layer x

    Return value: a list of style objects like {"title": $title, "name": $name, "file":$file}
    """
    styles = {}
    for cl in classes:
        style_name = cl['name']
        group_name = cl['group']
        if not group_name in styles:
            for key in metadata:
                if key.startswith("wms_style_") and key.endswith("_title"):
                    m = re.match(r"wms_style_(.*?)_title", key)
                    if len(m.groups()) == 0:
                        continue
                    mf_group_name = m.group(1)
                    if group_name.lower() == mf_group_name.lower():
                        style_title = metadata[key]
                        styles[group_name] = {"title": style_title, "name": group_name, "file": get_style_filename(style_name, group_name)}
    return [styles[key] for key in styles]

def get_style_filename(style_name, group_name):
    """
     Arguments:
    - style_name: mapfile internal stylename (string)
    - group_name: external stylename, indicated in mapfile by "groupname" (string)

    Return value: filename of output style (string)
    """
    style_name = style_name.replace(" ", "_")
    group_name = group_name.replace(" ", "_").replace(":", "_")
    return f"{style_name}_{group_name}.style"

def get_web_metadata_elements(mapfile):
    """
     mapfile:
    - mapfile: path to mapfile (string)

    Return value: list of string containing the web/metadata elements (including web element),
    note function does not follow includes, so web/metadata element that are in INCLUDE are not
    considered
    """
    result = []
    with open(mapfile, 'r') as f:
        line = f.readline()
        in_web = False
        first_end = False
        second_end = False
        web_result = ""
        while line:
            if not in_web:
                match = re.match(r"\s*WEB", line)
                if match:
                    in_web = True
                    web_result += line
            else:
                web_result += line
                match = re.match(r"\s*END", line)
                if match:
                    if not first_end:
                        first_end = True
                    elif not second_end:
                        second_end = True
                        result.append(web_result)
                        in_web = False
                        first_end = False
                        second_end = False
                        web_result = ""
            line = f.readline()
    return result


def get_gpgk_tablename(lyr):
    """
    Arguments:
    - lyr: mappyfile layer object
    Return value: the gpkg table name used in the lyr/data field (string)
    """
    data = lyr["data"][0]
    match = re.match(r".*\sfrom\s(.*?)\)", data)
    table_name = ""
    if len(match.groups()) > 0:
        table_name = match.group(1)
    return table_name

@click.group()
def cli():
    """mapfile util script to convert existing mapfiles to PDOK WMS CR YAML\n
    Python packages required:\n
    - Click==7.0\n
    - PyYAML==5.3.1\n
    - mappyfile==0.8.4\n

    Python version: 3.6.9
    """
    return

@cli.command(name="convert-mapfile")
@click.argument('mapfile', type=click.Path(exists=True))
@click.option('--columns')
@click.option('--ds-id')
def convert_mapfile(mapfile, columns="", ds_id=""):
    """
    Convert mapfile into WMS Custom Resource Yaml
    Arguments:

    - mapfile: path to mapfile (string)\n
    - columns: [optional] comma seperated list of columns,
    used for getfeaturinfo template generator (string)\n
    - ds_id: [optional] dataset source identifier (string)

    Return value: none, outputs WMS Custom Resource Yaml to stdout
    """
    result = {}
    columns = columns.split(",")
    layers_result = []

    # mappyfile only allows for one web/metadata element, 
    # only the last web/metadata element will be read
    # therefore extract and parse the web/metadata elements seperately
    web_md_strings = get_web_metadata_elements(mapfile)
    for web_md_string in web_md_strings:
        md_el = mappyfile.loads(web_md_string)["metadata"]
        if "ows_title" in md_el.keys():
            result["title"] = md_el["ows_title"]
        if "ows_abstract" in md_el.keys():
            result["abstract"] = md_el["ows_abstract"]
        if "ows_keywordlist" in md_el.keys():
            result["keywords"] = md_el["ows_keywordlist"].split(",")
        if "wms_inspire_metadataurl_href" in md_el.keys():
            md_url = md_el["wms_inspire_metadataurl_href"]
            result["serviceMetadataId"] = get_dataset_md_identifier(md_url)

    mapfile = mappyfile.open(mapfile)
    result["extent"] = " ".join([str(item) for item in  mapfile["extent"]])
    first_layer = layers = mapfile["layers"][0]

    result["authority"] = first_layer["metadata"]["wms_authorityurl_name"]
    result["authorityUrl"] = first_layer["metadata"]["wms_authorityurl_href"]
    result["geometryEpsgCode"] = mapfile["projection"][0].replace("init=epsg:", "")

    layers = mapfile["layers"]
    for lyr in layers:
        lyr_result = {}
        metadata = lyr["metadata"]
        lyr_result["name"] = lyr["name"]
        lyr_result["title"] = metadata["wms_title"]
        lyr_result["gpkg"] = os.path.basename(lyr["connection"])
        lyr_result["gpkgTablename"] = get_gpgk_tablename(lyr)
        lyr_result["abstract"] = metadata["wms_abstract"]
        lyr_result["visible"] = True if lyr["status"] == "ON" else False
        lyr_result["geometryType"] = lyr["type"]
        lyr_result["datasetMetadataId"] = \
            get_dataset_md_identifier(metadata["ows_metadataurl_href"])
        lyr_result["datasetSourceId"] = ds_id
        lyr_result["keywords"] = metadata["wms_keywordlist"].split(",")
        lyr_result["extent"] = metadata["wms_extent"]
        lyr_result["styles"] = get_styles_for_layer(lyr["classes"], metadata)
        lyr_result["maxscale"] = str(lyr["maxscaledenom"])
        lyr_result["minscale"] = str(lyr["minscaledenom"])
        lyr_result["columns"] = columns
        layers_result.append(lyr_result)
    result["layers"] = layers_result
    yaml.Dumper.ignore_aliases = lambda *args : True
    yaml.dump(result, sys.stdout, default_flow_style=False, sort_keys=False)

@cli.command(name="export-styles")
@click.argument('mapfile', type=click.Path(exists=True))
@click.argument('destination-dir', type=click.Path())
def export_styles(mapfile, destination_dir):
    """
    Export styles from mapfile to seperate files for all layers in mapfile
    Arguments:\n
    - mapfile: path to mapfile (string)\n
    - destination_dir: destination directory style files will be saved to

    Return value: none, ouputs status message to stdout
    """
    mapfile = mappyfile.open(mapfile)
    layers = mapfile["layers"]
    if not os.path.exists(destination_dir):
        os.mkdir(destination_dir)
    for lyr in layers:
        styles = {}
        for cl in lyr["classes"]:
            style_name = cl['name']
            group_name = cl['group']
            if not group_name in styles:
                styles[group_name] = []
            styles[group_name].append(cl)
        for group_name in styles:
            file_path = os.path.join\
                (destination_dir, get_style_filename(style_name, group_name))
            with open(file_path, 'w') as style_file:
                for cl in styles[group_name]:
                    mappyfile.dump(cl, style_file)
                    style_file.write("\n")
    full_path = os.path.abspath(destination_dir)
    print(f"styles exported to {full_path}")

if __name__ == "__main__":
    cli()