Add blob, capability

app/models/blob.py

@@ -22,17 +22,79 @@
 from enum import Enum
 from typing import Any, List, Optional, Union, Literal
 
+import rdflib
 from pydantic import BaseModel, Field, constr
+from rdflib import RDF
 
+from app.models.aas_namespace import AASNameSpace
 from app.models.data_element import DataElement
 from app.models.data_type_def_xsd import DataTypeDefXsd
 from app.models.lang_string_text_type import LangStringTextType
 from app.models.model_type import ModelType
 from app.models.reference import Reference
+from app.models.submodel_element import SubmodelElement
+import base64
 
 
 # TODO: check content type
 class Blob(DataElement):
     value: Optional[str] = None
     contentType: str
     modelType: Literal["Blob"] = ModelType.Blob.value
+
+    def to_rdf(
+        self,
+        graph: rdflib.Graph = None,
+        parent_node: rdflib.IdentifiedNode = None,
+        prefix_uri: str = "",
+        base_uri: str = "",
+    ) -> (rdflib.Graph, rdflib.IdentifiedNode):
+        created_graph, created_node = super().to_rdf(graph, parent_node, prefix_uri, base_uri)
+        created_graph.add((created_node, RDF.type, AASNameSpace.AAS["Blob"]))
+        if self.value != None:
+            created_graph.add(
+                (
+                    created_node,
+                    AASNameSpace.AAS["Blob/value"],
+                    rdflib.Literal(self.value, datatype=rdflib.XSD.base64Binary),
+                )
+            )
+        created_graph.add(
+            (
+                created_node,
+                AASNameSpace.AAS["Blob/contentType"],
+                rdflib.Literal(self.contentType),
+            )
+        )
+        return created_graph, created_node
+
+    @staticmethod
+    def from_rdf(graph: rdflib.Graph, subject: rdflib.IdentifiedNode) -> "Blob":
+        value_value = None
+        value_ref: rdflib.Literal = next(
+            graph.objects(subject=subject, predicate=AASNameSpace.AAS["Blob/value"]),
+            None,
+        )
+        if value_ref:
+            value_value = value_ref
+        content_type_value = None
+        content_type_ref: rdflib.Literal = next(
+            graph.objects(subject=subject, predicate=AASNameSpace.AAS["Blob/contentType"]),
+            None,
+        )
+        if content_type_ref:
+            content_type_value = content_type_ref.value
+        submodel_element = SubmodelElement.from_rdf(graph, subject)
+        return Blob(
+            value=value_value,
+            contentType=content_type_value,
+            qualifiers=submodel_element.qualifiers,
+            category=submodel_element.category,
+            idShort=submodel_element.idShort,
+            displayName=submodel_element.displayName,
+            description=submodel_element.description,
+            extensions=submodel_element.extensions,
+            semanticId=submodel_element.semanticId,
+            supplementalSemanticIds=submodel_element.supplementalSemanticIds,
+            embeddedDataSpecifications=submodel_element.embeddedDataSpecifications,
+        )

app/models/capability.py

@@ -20,9 +20,39 @@
 #  OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 from typing import Literal
 
+import rdflib
+from rdflib import RDF
+
+from app.models.aas_namespace import AASNameSpace
 from app.models.model_type import ModelType
 from app.models.submodel_element import SubmodelElement
 
 
 class Capability(SubmodelElement):
     modelType: Literal["Capability"] = ModelType.Capability.value
+
+    def to_rdf(
+        self,
+        graph: rdflib.Graph = None,
+        parent_node: rdflib.IdentifiedNode = None,
+        prefix_uri: str = "",
+        base_uri: str = "",
+    ) -> (rdflib.Graph, rdflib.IdentifiedNode):
+        created_graph, created_node = super().to_rdf(graph, parent_node, prefix_uri, base_uri)
+        created_graph.add((created_node, RDF.type, AASNameSpace.AAS["Capability"]))
+        return created_graph, created_node
+
+    @staticmethod
+    def from_rdf(graph: rdflib.Graph, subject: rdflib.IdentifiedNode) -> "Capability":
+        submodel_element = SubmodelElement.from_rdf(graph, subject)
+        return Capability(
+            qualifiers=submodel_element.qualifiers,
+            category=submodel_element.category,
+            idShort=submodel_element.idShort,
+            displayName=submodel_element.displayName,
+            description=submodel_element.description,
+            extensions=submodel_element.extensions,
+            semanticId=submodel_element.semanticId,
+            supplementalSemanticIds=submodel_element.supplementalSemanticIds,
+            embeddedDataSpecifications=submodel_element.embeddedDataSpecifications,
+        )

app/models/util.py

@@ -24,6 +24,8 @@
 
 from app.models.aas_namespace import AASNameSpace
 from app.models.annotated_relationship_element import AnnotatedRelationshipElement
+from app.models.blob import Blob
+from app.models.capability import Capability
 from app.models.file import File
 from app.models.multi_language_property import MultiLanguageProperty
 from app.models.property import Property
@@ -44,7 +46,7 @@ def from_unknown_rdf(graph: rdflib.Graph, subject: rdflib.IdentifiedNode) -> Sub
     if type_ref == AASNameSpace.AAS["BasicEventElement"]:
         raise NotImplementedError()
     if type_ref == AASNameSpace.AAS["Blob"]:
-        raise NotImplementedError()
+        return Blob.from_rdf(graph, subject)
     if type_ref == AASNameSpace.AAS["File"]:
         return File.from_rdf(graph, subject)
     if type_ref == AASNameSpace.AAS["MultiLanguageProperty"]:
@@ -62,6 +64,6 @@ def from_unknown_rdf(graph: rdflib.Graph, subject: rdflib.IdentifiedNode) -> Sub
     if type_ref == AASNameSpace.AAS["Entity"]:
         raise NotImplementedError()
     if type_ref == AASNameSpace.AAS["Capability"]:
-        raise NotImplementedError()
+        return Capability.from_rdf(graph, subject)
     if type_ref == AASNameSpace.AAS["Operation"]:
         raise NotImplementedError()

sonar-project.properties

@@ -2,4 +2,4 @@ sonar.projectKey=mhrimaz_aasbrain-concept-description-repo
 sonar.organization=mhrimaz
 
 sonar.python.coverage.reportPaths=coverage.xml
-sonar.coverage.exclusions=**/tests/**
+sonar.coverage.exclusions=tests/**

tests/serialization/shell_serialization_test.py

@@ -362,15 +362,15 @@ def test_any_submodel_element_minimal_to_rdf():
         "SubmodelElementCollection",
         "AnnotatedRelationshipElement",
         "File",
-        # "Capability",
-        # "SubmodelElementList",
+        "Capability",
+        "Blob",
         # "Range",
+        # "ReferenceElement",
         # "Operation",
         # "Entity",
-        # "Blob",
         # "BasicEventElement",
-        # "ReferenceElement",
         # "RelationshipElement",
+        # "SubmodelElementList",
     ]:
         payload_json = json.loads(get_testdata_json(model, "minimal"))["submodels"][0]
         payload = Submodel(**payload_json)
@@ -387,15 +387,15 @@ def test_any_submodel_element_maximal_to_rdf():
         "SubmodelElementCollection",
         "AnnotatedRelationshipElement",
         "File",
-        # "Capability",
-        # "SubmodelElementList",
+        "Capability",
+        "Blob",
         # "Range",
+        # "ReferenceElement",
         # "Operation",
         # "Entity",
-        # "Blob",
         # "BasicEventElement",
-        # "ReferenceElement",
         # "RelationshipElement",
+        # "SubmodelElementList",
     ]:
         payload_json = json.loads(get_testdata_json(model, "maximal"))["submodels"][0]
         payload = Submodel(**payload_json)