BGS vocabularies updates for Prez

compound.py

@@ -0,0 +1,17 @@
+from pathlib import Path
+
+from rdflib import Graph
+
+spatial_path = Path("spatial")
+vocabularies_path = Path("vocabularies")
+
+files = list(spatial_path.glob("**/*")) + list(vocabularies_path.glob("**/*"))
+graph = Graph()
+for file in files:
+    if file.is_file() and file.name != ".DS_Store" and file.name != "625KGeologyMap_all.nt" and file.name != "dataholdings.nt":
+        try:
+            graph.parse(file)
+        except Exception as err:
+            raise RuntimeError(f"Failed to parse file {file.name}. {err}") from err
+
+graph.serialize("compounded.ttl", format="longturtle")

download_reg_status.py

@@ -0,0 +1,22 @@
+import time
+
+from rdflib import Graph
+
+
+def main() -> None:
+    starttime = time.time()
+
+    try:
+        graph = Graph()
+        graph.parse("https://linked.data.gov.au/def/reg-statuses", format="text/turtle")
+        graph.serialize(
+            "vocabularies/reg-status.nt", format="ntriples", encoding="utf-8"
+        )
+
+    finally:
+        endtime = time.time() - starttime
+        print(f"Completed in {endtime:0.2f} seconds")
+
+
+if __name__ == "__main__":
+    main()

fix.py

@@ -0,0 +1,89 @@
+import time
+
+from rdflib import Graph, RDF, SKOS
+
+
+def fix_dataholdings():
+    """
+    Remove concept scheme declaration as it breaks Prez trying to list it in the vocabularies listing page.
+
+    Error is: ValueError: Can't split 'http://data.bgs.ac.uk/ref/dataHolding/'
+
+    Remove collection declaration as it doesn't make sense to list these as a SKOS collection.
+    Each item in the collection also don't load correctly as they are not SKOS concepts.
+    """
+    filepath = "vocabularies/dataholdings.nt"
+    fileformat = "ntriples"
+    graph = Graph()
+    graph.parse(filepath, format=fileformat)
+    graph.remove((None, RDF.type, SKOS.ConceptScheme))
+    graph.remove((None, RDF.type, SKOS.Collection))
+    graph.serialize(filepath, fileformat, encoding="utf-8")
+
+
+def fix_geology_map():
+    """
+    Remove concept scheme declarations.
+
+    We use normal file and line processing here because rdflib cannot parse this file containing data with XML.
+    """
+    filepath = "vocabularies/625KGeologyMap_all.nt"
+    lines = []
+    with open(filepath, "r", encoding="utf-8") as file:
+        for line in file.readlines():
+            if not "<http://www.w3.org/2004/02/skos/core#ConceptScheme>" in line:
+                lines.append(line)
+
+    with open(filepath, "w", encoding="utf-8") as file:
+        file.write("".join(lines))
+
+
+def fix_bedding_surface_structure():
+    """
+    Vocab is defined as both concept scheme and collection.
+
+    Remove the collection type declaration.
+    """
+    filepath = "vocabularies/simple-dictionaries/BeddingSurfaceStructure_scheme.nt"
+    fileformat = "ntriples"
+    graph = Graph()
+    graph.parse(filepath, format=fileformat)
+    graph.remove((None, RDF.type, SKOS.Collection))
+    graph.serialize(filepath, fileformat, encoding="utf-8")
+
+
+def fix_borehole_material_type():
+    filepath = "vocabularies/simple-dictionaries/BoreholeMaterialType_scheme.nt"
+    fileformat = "ntriples"
+    graph = Graph()
+    graph.parse(filepath, format=fileformat)
+    graph.remove((None, RDF.type, SKOS.Collection))
+    graph.serialize(filepath, fileformat, encoding="utf-8")
+
+
+def fix_rockunitrank():
+    filepath = "vocabularies/LexiconRockUnitName/RockUnitRank_scheme.nt"
+    fileformat = "ntriples"
+    graph = Graph()
+    graph.parse(filepath, format=fileformat)
+    graph.remove((None, RDF.type, SKOS.Collection))
+    graph.serialize(filepath, fileformat, encoding="utf-8")
+
+
+def main() -> None:
+    starttime = time.time()
+
+    try:
+        fix_dataholdings()
+        fix_geology_map()
+        fix_bedding_surface_structure()
+        fix_borehole_material_type()
+        fix_rockunitrank()
+
+    finally:
+        endtime = time.time() - starttime
+        print(f"Completed in {endtime:0.2f} seconds")
+
+
+if __name__ == "__main__":
+    main()

generate_geochronology_colour.py

@@ -0,0 +1,63 @@
+import time
+from textwrap import dedent
+
+from rdflib import Graph, URIRef, Literal, SDO, SKOS
+
+
+def main() -> None:
+    starttime = time.time()
+
+    try:
+        gts_graph = Graph()
+        gts_graph.parse(
+            "https://raw.githubusercontent.com/GeoscienceAustralia/cgi-vocabs/bc48b3b81727fb8681c5b6c833b41d46c13104aa/vocabularies/ics/gts-chart-colours.ttl",
+            format="text/turtle",
+        )
+
+        geochrono_graph = Graph()
+        geochrono_graph.parse(
+            "vocabularies/Geochronology/geochronology_alignments_cgi.nt",
+            format="ntriples",
+        )
+
+        graph = Graph()
+
+        def get_colour(iri: URIRef, graph: Graph) -> str | None:
+            query = dedent(
+                f"""
+                PREFIX : <http://resource.geosciml.org/classifier/ics/gts-chart-colours/>
+                PREFIX skos: <http://www.w3.org/2004/02/skos/core#>
+
+                SELECT ?colour
+                WHERE {{
+                    <{iri}> skos:notation ?colour .
+
+                    FILTER(datatype(?colour) = :RGBHex)
+                }}
+            """
+            )
+            result = graph.query(query)
+
+            for row in result:
+                return str(row.colour)
+
+        for s, _, o in geochrono_graph.triples((None, SKOS.exactMatch, None)):
+            colour = get_colour(o, gts_graph)
+            if colour is not None:
+                graph.add((s, SDO.color, Literal(colour)))
+            else:
+                print(f"No colour found for {s} to {o}")
+
+        graph.serialize(
+            "vocabularies/Geochronology/geochronology_colours.nt",
+            format="ntriples",
+            encoding="utf-8",
+        )
+
+    finally:
+        endtime = time.time() - starttime
+        print(f"Completed in {endtime:0.2f} seconds")
+
+
+if __name__ == "__main__":
+    main()

generate_geology_bedrock_fc.py

@@ -0,0 +1,96 @@
+import time
+
+import httpx
+from rdflib import (
+    Graph,
+    Namespace,
+    RDF,
+    DCAT,
+    SDO,
+    Literal,
+    BNode,
+    DCTERMS,
+    XSD,
+    RDFS,
+    URIRef,
+)
+from shapely.geometry import box
+
+
+dataset_json_url = "https://ogcapi.bgs.ac.uk/collections/bgsgeology625kbedrock?f=json"
+url = "https://ogcapi.bgs.ac.uk/collections/bgsgeology625kbedrock/items?f=json&offset=0"
+
+BGS = Namespace("http://data.bgs.ac.uk/id/")
+GEOLOGY_MAP_DATASET = BGS.geology625k
+FEATURE_COLLECTION_IRI = URIRef("http://data.bgs.ac.uk/id/geology625k/bedrock")
+GEO = Namespace("http://www.opengis.net/ont/geosparql#")
+
+
+def get_bbox(data):
+    crs = data["crs"]
+    bbox_coords = data["bbox"][0]
+    bbox_geometry = box(*bbox_coords)
+    wkt = bbox_geometry.wkt
+    return f"<{crs}> {wkt}"
+
+
+def generate():
+    graph = Graph()
+
+    response = httpx.get(dataset_json_url)
+    response.raise_for_status()
+
+    data = response.json()
+
+    iri = FEATURE_COLLECTION_IRI
+    graph.add((GEOLOGY_MAP_DATASET, RDFS.member, iri))
+    graph.add((iri, RDF.type, GEO.FeatureCollection))
+
+    name = Literal(data["title"])
+    graph.add((iri, SDO.name, name))
+
+    description = Literal(data["description"])
+    graph.add((iri, SDO.description, description))
+
+    keywords = [Literal(keyword) for keyword in data["keywords"]]
+    for keyword in keywords:
+        graph.add((iri, DCAT.keyword, keyword))
+
+    bbox = get_bbox(data["extent"]["spatial"])
+    geom = BNode()
+    graph.add((iri, GEO.hasBoundingBox, geom))
+    graph.add((geom, RDF.type, GEO.Geometry))
+    graph.add((geom, GEO.asWKT, Literal(bbox, datatype=GEO.wktLiteral)))
+
+    temporal_start = data["extent"]["temporal"]["interval"][0][0]
+    temporal_end = data["extent"]["temporal"]["interval"][0][1]
+    period_of_time = BNode()
+    if temporal_start is not None:
+        graph.add((iri, DCTERMS.temporal, period_of_time))
+        graph.add((period_of_time, RDF.type, DCTERMS.PeriodOfTime))
+        graph.add(
+            (period_of_time, DCAT.startDate, Literal(temporal_start, datatype=XSD.date))
+        )
+    if temporal_start is not None and temporal_end is not None:
+        graph.add(
+            (period_of_time, DCAT.endDate, Literal(temporal_end, datatype=XSD.date))
+        )
+
+    graph.serialize(
+        "spatial/geologymap/bedrock-feature-collection.ttl", format="longturtle"
+    )
+
+
+def main() -> None:
+    starttime = time.time()
+
+    try:
+        generate()
+
+    finally:
+        endtime = time.time() - starttime
+        print(f"Completed in {endtime:0.2f} seconds")
+
+
+if __name__ == "__main__":
+    main()