NERC Vocabulary Server Concept Collection to OWL Named Individuals

nvs_collection_to_owl_named_individuals.py

import urllib.request
from rdflib import Graph, Literal, OWL, RDF, RDFS, URIRef, SKOS
import rdflib.namespace


def nvs_fetcher(nvs_collection: str, others: list) -> Graph:
    """Fetches a NERC Vocabulary server collection and returns it
        as a set of OWL Named Individuals with optional additional OWL or
        RDF classes added to the graph.

    Args:
        nvs_collection (str): The NERC Vocabulary Server collection to be
            parsed, e.g. "L22"
        others (list): A list of strings for rdf:types to be applied to the
            parsed Concepts, other than skos:Concept, e.g.
            ["EMODnetHumanActivities"]

    Returns:
        An rdflib.Graph object of OWL Named Individuals parsed from the
            passed graph of SKOS Concepts
    """
    nvs_url_pattern = "http://vocab.nerc.ac.uk/collection/{}/" + \
        "current/?_profile=nvs&_mediatype=text/turtle"
    with urllib.request.urlopen(
            nvs_url_pattern.format(nvs_collection)) as resp:
        nvs_ttl_str: str = resp.read().decode("utf-8")
    g: Graph = nvs_to_rdflib_graph(nvs_ttl_str)
    return skos_graph_to_owl_graph(g, others)


def eu_voc_fetcher(eu_collection_name: str, others: list) -> Graph:
    """Fetches an INSPIRE vocabulary as RDF/XML and returns it
        as a set of OWL Named Individuals with optional additional OWL or
        RDF classes added to the graph.

    Args:
        eu_collection_name (str): The INSPIRE Vocabulary to be parsed, e.g.
            "HILUCSValue"
        others (list): A list of strings for rdf:types to be applied to the
            parsed Concepts, other than skos:Concept, e.g.
            ["HILUCSClassificationCode"]

    Returns:
        An rdflib.Graph object of OWL Named Individuals parsed from the
            passed graph of SKOS Concepts
    """
    nvs_url_pattern = "https://inspire.ec.europa.eu/codelist/{0}/{0}.en.rdf"
    with urllib.request.urlopen(
            nvs_url_pattern.format(eu_collection_name)) as resp:
        nvs_ttl_str: str = resp.read().decode("utf-8")
    g: Graph = eu_voc_to_rdflib_graph(nvs_ttl_str)
    return skos_graph_to_owl_graph(g, others)


def nvs_to_rdflib_graph(ttl: str) -> Graph:
    """Convert a TTL string from the NERC Vocabulary to a rdflib.Graph

    Args:
        ttl(str): An RDF graph represented in the Terse Triple Language (TTL)
            stored as a Python string

    Returns:
        An rdflib.Graph object containing the graph data passed to the function
    """
    g: Graph = Graph().parse(data=ttl)
    return g


def eu_voc_to_rdflib_graph(rdfxml: str) -> Graph:
    """Convert an RDF/XML string from the EU INSPIRE vocabularies to a
        rdflib.Graph

    Args:
        rdfxml(str): An RDF graph represented as RDF/XML stored as a Python
            string

    Returns:
        An rdflib.Graph object containing the graph data passed to the function
    """
    g: Graph = Graph().parse(data=rdfxml, format="xml")
    return g


def skos_graph_to_owl_graph(g: Graph, others: list = []) -> Graph:
    """Converts a graph of mutliple SKOS Concept triples to graph of OWL
        Named Individuals

    Args:
        g (rdflib.Graph): The RDF graph of SKOS Concepts to be converted

    Returns:
        An rdflib.Graph object of OWL Named Individuals parsed from the
            passed graph of SKOS Concepts
    """
    dummy_uri: str = "https://foo/bar#"
    return_graph: Graph = Graph()
    concept: URIRef
    pred: URIRef
    object_literal: Literal
    object_ref: URIRef
    for s, p, o in g.triples((None, RDF.type, SKOS.Concept)):
        for ss, pp, oo in g.triples((s, SKOS.prefLabel, None)):
            concept = URIRef("{}{}".format(
                dummy_uri, oo.title().replace(
                    " ", "").replace(",", "").replace("-", "").replace(
                        "(", "").replace(")", "")))
            pred = URIRef(RDFS.label)
            object_literal = Literal("{}".format(oo.title().replace(
                " ", "").replace(",", "").replace("-", "").replace(
                    "(", "").replace(")", "")))
            return_graph.add((concept, pred, object_literal))
            pred = URIRef(SKOS.prefLabel)
            object_literal = Literal(oo)
            return_graph.add((concept, pred, object_literal))
            pred = URIRef(RDF.type)
            object_ref = URIRef(OWL.NamedIndividual)
            return_graph.add((concept, pred, object_ref))
            object_ref = URIRef(SKOS.Concept)
            return_graph.add((concept, pred, object_ref))
            # print(":{} a owl:NamedIndividual, skos:Concept{};".format(
            #    oo.title().replace(" ", "").replace(",", "").replace(
            #        "-", "").replace("(", "").replace(")", ""), other_types))
        for ss, pp, oo in g.triples((s, SKOS.definition, None)):
            pred = URIRef(RDFS.comment)
            object_literal = Literal(oo)
            return_graph.add((concept, pred, object_literal))
        for other_type in others:
            pred = URIRef(SKOS.inScheme)
            object_ref = URIRef("{}{}".format(
                dummy_uri, other_type))
            return_graph.add((concept, pred, object_ref))
            pred = URIRef(RDF.type)
            return_graph.add((concept, pred, object_ref))
        pred = URIRef(OWL.sameAs)
    return_graph.add((concept, pred, concept))
    return_graph.bind("owl", rdflib.namespace.OWL)
    return_graph.bind("skos", rdflib.namespace.SKOS)
    return_graph.bind("", dummy_uri)
    return return_graph


print(nvs_fetcher('C45', ['MSFDDescriptor']).serialize(format="turtle"))
# eu_voc_fetcher('HILUCSValue', ['HILUCSClassificationCode'])

requirements.txt

rdflib