Add datatype constraints to lastKnownValue

bricksrc/entity_properties.py

@@ -1,8 +1,9 @@
 """
 Entity property definitions
 """
+from typing_extensions import dataclass_transform
 from rdflib import Literal
-from .namespaces import BRICK, RDFS, SKOS, UNIT, XSD, SH, BSH, REF
+from .namespaces import BRICK, RDFS, SKOS, UNIT, XSD, SH, BSH, A
 
 # these are the "relationship"/predicates/OWL properties that
 # relate a Brick entity to a structured value.
@@ -555,13 +556,32 @@ def get_shapes(G):
 
 def generate_quantity_shapes(G):
     quantities = G.query(
-        "SELECT ?q WHERE { ?q a brick:Quantity . ?q qudt:applicableUnit ?unit }"
+        """SELECT ?q ?datatype WHERE {
+                ?q a brick:Quantity .
+                ?q qudt:applicableUnit ?unit .
+                OPTIONAL { ?q skos:broader*/bsh:preferredDatatype ?datatype } }"""
     )
     d = {}
-    for (quantity,) in quantities:
-        shape = BSH[f"{quantity.split('#')[-1]}Shape"]
+    for (quantity, datatype) in quantities:
+        quantity_name = quantity.split("#")[-1]
+        shape = BSH[f"{quantity_name}Shape"]
         d[shape] = {
             "unitsFromQuantity": quantity,
-            "datatype": BSH.NumericValue,
+            "datatype": BSH.NumericValue if datatype is None else datatype,
         }
     return d
+
+
+def generate_last_known_value_shapes(G):
+    generated = {}
+    for quantity in G.subjects(A, BRICK.Quantity):
+        quantity_name = quantity.split("#")[-1]
+        generated[BSH[f"LastKnown{quantity_name}ValueShape"]] = {
+            "properties": {
+                BRICK.timestamp: {"datatype": XSD.dateTime},
+            },
+            RDFS.subClassOf: BSH.LastKnownValueShape,
+            SH.node: BSH[f"{quantity_name}Shape"],
+        }
+    G.remove((None, BSH.preferredDatatype, None))
+    return generated

bricksrc/quantities.py

@@ -1,6 +1,6 @@
 from brickschema.graph import Graph
 from rdflib import Literal, URIRef
-from .namespaces import SKOS, OWL, RDFS, BRICK, QUDTQK, QUDTDV, QUDT, UNIT
+from .namespaces import SKOS, OWL, RDFS, BRICK, QUDTQK, QUDTDV, QUDT, UNIT, XSD
 
 
 g = Graph()
@@ -35,6 +35,7 @@ def get_units(qudt_quantity):
 """
 quantity_definitions = {
     "Air_Quality": {
+        "preferredDatatype": XSD.float,
         SKOS.narrower: {
             "Ammonia_Concentration": {
                 QUDT.applicableUnit: [UNIT.PPM, UNIT.PPB],
@@ -706,6 +707,7 @@ def get_units(qudt_quantity):
         },
     },
     "Temperature": {
+        "preferredDatatype": XSD.float,
         BRICK.hasQUDTReference: QUDTQK["Temperature"],
         SKOS.narrower: {
             "Differential_Temperature": {

bricksrc/rules.ttl

@@ -300,3 +300,20 @@ bsh:hasSubstance a sh:NodeShape ;
     sh:targetObjectsOf brick:hasSubstance ;
     sh:class brick:Substance ;
 .
+
+bsh:PropagateUnitslastKnownValue a sh:NodeShape ;
+    sh:targetSubjectsOf brick:lastKnownValue ;
+    sh:rule [
+        a sh:SPARQLRule ;
+        sh:construct """
+        CONSTRUCT {
+            ?val brick:hasUnit ?unit .
+        }
+        WHERE {
+            $this brick:hasUnit ?unit .
+            $this brick:lastKnownValue ?val .
+            FILTER NOT EXISTS { ?val brick:hasUnit ?unit }
+        }
+        """ ;
+    ] ;
+.

generate_brick.py

@@ -46,7 +46,12 @@
 from bricksrc.substances import substances
 from bricksrc.quantities import quantity_definitions, get_units
 from bricksrc.properties import properties
-from bricksrc.entity_properties import shape_properties, entity_properties, get_shapes
+from bricksrc.entity_properties import (
+    shape_properties,
+    entity_properties,
+    get_shapes,
+    generate_last_known_value_shapes,
+)
 from bricksrc.deprecations import deprecations
 
 logging.basicConfig(
@@ -235,6 +240,12 @@ def define_concept_hierarchy(definitions, typeclasses, broader=None, related=Non
         if not has_label(concept):
             G.add((concept, RDFS.label, Literal(label)))
 
+        # setup an annotation for the quantity-flavored lastknownvalue shapes.
+        # The BSH.preferredDatatype will be removed in a later stage of Brick compilation
+        if "preferredDatatype" in defn:
+            preferredDatatype = defn.pop("preferredDatatype")
+            G.add((concept, BSH.preferredDatatype, preferredDatatype))
+
         # define concept hierarchy
         # this is a nested dictionary
         narrower_defs = defn.get(SKOS.narrower, {})
@@ -289,6 +300,15 @@ def define_classes(definitions, parent, pun_classes=False):
         if pun_classes:
             G.add((classname, A, classname))
 
+        if BRICK.hasQuantity in defn:
+            quantity = defn[BRICK.hasQuantity].split("#")[-1]
+            lkv_shape = BSH[f"LastKnown{quantity}ValueShape"]
+            lkv_prop_shape = BNode()
+            G.add((classname, SH.property, lkv_prop_shape))
+            G.add((lkv_prop_shape, SH.path, BRICK.lastKnownValue))
+            G.add((lkv_prop_shape, SH.node, lkv_shape))
+            G.add((lkv_prop_shape, SH.maxCount, Literal(1)))
+
         # define mapping to tags if it exists
         # "tags" property is a list of URIs naming Tags
         taglist = defn.get("tags", [])
@@ -470,6 +490,10 @@ def define_shape_properties(definitions):
             G.add((brick_value_shape, SH.minCount, Literal(1)))
             G.add((brick_value_shape, SH.maxCount, Literal(1)))
 
+        # handle RDF annotations on the shape
+        other_props = {k: v for k, v in defn.items() if isinstance(k, URIRef)}
+        add_properties(shape_name, other_props)
+
         v = BNode()
         # prop:value PropertyShape
         if "values" in defn:
@@ -873,6 +897,7 @@ def handle_deprecations():
 G.add((BSH.ValueShape, A, OWL.Class))
 define_shape_properties(get_shapes(G))
 define_entity_properties(entity_properties)
+define_shape_properties(generate_last_known_value_shapes(G))
 
 handle_deprecations()