diff --git a/basyx/aas/adapter/rdf/rdf_serialization.py b/basyx/aas/adapter/rdf/rdf_serialization.py index 863a2ea5..678264f8 100644 --- a/basyx/aas/adapter/rdf/rdf_serialization.py +++ b/basyx/aas/adapter/rdf/rdf_serialization.py @@ -5,7 +5,7 @@ # # SPDX-License-Identifier: MIT """ -.. _adapter.json.rdf_serialization: +.. _adapter.rdf.rdf_serialization: Module for serializing Asset Administration Shell objects to the official RDF format @@ -19,7 +19,6 @@ part of the shacl validation. """ - from typing import Dict, Type, Union import base64 @@ -46,8 +45,8 @@ def object_store_to_rdflib_graph(self, data: model.AbstractObjectStore) -> Graph This class function is used internally by :meth:`write_aas_rdf_file` and shouldn't be called directly for most use-cases. - :param data: :class:`ObjectStore ` which contains different objects of - the AAS meta model which should be serialized to an RDF file + :param data: :class:`ObjectStore ` which contains different + objects of the AAS meta model which should be serialized to an RDF file """ for obj in data: if isinstance(obj, model.AssetAdministrationShell): @@ -74,8 +73,8 @@ def _abstract_classes_to_rdf(self, obj: object, parent: Union[URIRef, BNode]) -> """ Adds attributes of abstract base classes of ``obj``. - If the object obj is inheriting from any abstract AAS class, this function adds all the serialized information of - those abstract classes to the generated element. + If the object obj is inheriting from any abstract AAS class, this function adds all the serialized information + of those abstract classes to the generated element. :param obj: An object of the AAS :param parent: The parent node. Can either be :class:`~rdflib.term.URIRef` or :class:`~rdflib.term.BNode` @@ -88,7 +87,8 @@ def _abstract_classes_to_rdf(self, obj: object, parent: Union[URIRef, BNode]) -> self._extension_to_rdf(extension, parent) if isinstance(obj, model.Referable): if obj.category: - self.graph.add((parent, self.aas["Referable/category"], Literal(str(obj.category), datatype=XSD.string))) + self.graph.add( + (parent, self.aas["Referable/category"], Literal(str(obj.category), datatype=XSD.string))) if obj.id_short and not isinstance(obj.parent, model.SubmodelElementList): self.graph.add((parent, self.aas["Referable/idShort"], Literal(str(obj.id_short), datatype=XSD.string))) if obj.display_name: @@ -109,7 +109,8 @@ def _abstract_classes_to_rdf(self, obj: object, parent: Union[URIRef, BNode]) -> self._reference_to_rdf(obj.semantic_id, parent, self.aas["HasSemantics/semanticId"]) if obj.supplemental_semantic_id: for supplemental_semantic_id in obj.supplemental_semantic_id: - self._reference_to_rdf(supplemental_semantic_id, parent, self.aas["HasSemantics/supplementalSemanticIds"]) + self._reference_to_rdf(supplemental_semantic_id, parent, + self.aas["HasSemantics/supplementalSemanticIds"]) if isinstance(obj, model.Qualifiable): if obj.qualifier: for qualifier in obj.qualifier: @@ -123,26 +124,30 @@ def _abstract_classes_to_rdf(self, obj: object, parent: Union[URIRef, BNode]) -> # transformation functions to serialize classes from model.base # ############################################################## - def _value_to_rdf(self, value: model.ValueDataType, value_type: model.DataTypeDefXsd, parent: Union[URIRef, BNode], objectProperty: URIRef) -> None: + def _value_to_rdf(self, value: model.ValueDataType, value_type: model.DataTypeDefXsd, parent: Union[URIRef, BNode], + objectProperty: URIRef) -> None: """ Serialization of objects of :class:`~basyx.aas.model.base.ValueDataType` to a joint rdflib Graph object :param value: :class:`~basyx.aas.model.base.ValueDataType` object :param value_type: Corresponding :class:`~basyx.aas.model.base.DataTypeDefXsd` :param parent: The parent node. Can either be :class:`~rdflib.term.URIRef` or :class:`~rdflib.term.BNode` - :param objectProperty: The object property that is used to connect to from the parent, :class:`~rdflib.term.URIRef` + :param objectProperty: The object property that is used to connect to from the + parent, :class:`~rdflib.term.URIRef` :return: None """ self.graph.add((parent, objectProperty, Literal(model.datatypes.xsd_repr(value), datatype=XSD.string))) - def _lang_string_set_to_rdf(self, obj: model.LangStringSet, parent: Union[URIRef, BNode], objectProperty: URIRef) -> None: + def _lang_string_set_to_rdf(self, obj: model.LangStringSet, parent: Union[URIRef, BNode], + objectProperty: URIRef) -> None: """ Serialization of objects of class :class:`~basyx.aas.model.base.LangStringSet` to a joint rdflib Graph object :param obj: Object of class :class:`~basyx.aas.model.base.LangStringSet` :param parent: The parent node. Can either be :class:`~rdflib.term.URIRef` or :class:`~rdflib.term.BNode` - :param objectProperty: The object property that is used to connect to from the parent, :class:`~rdflib.term.URIRef` + :param objectProperty: The object property that is used to connect to from the + parent, :class:`~rdflib.term.URIRef` :return: None """ # TODO: There is an ongoing bugfix of the validation shacl scheme for @@ -158,7 +163,8 @@ def _lang_string_set_to_rdf(self, obj: model.LangStringSet, parent: Union[URIRef lang_string = BNode() self.graph.add((parent, objectProperty, lang_string)) self.graph.add((lang_string, RDF.type, LANG_STRING_SET_TAGS[type(obj)])) - self.graph.add((lang_string, self.aas["AbstractLangString/language"], Literal(str(language), datatype=XSD.string))) + self.graph.add( + (lang_string, self.aas["AbstractLangString/language"], Literal(str(language), datatype=XSD.string))) self.graph.add((lang_string, self.aas["AbstractLangString/text"], Literal(str(text), datatype=XSD.string))) def _key_to_rdf(self, obj: model.Key, parent: Union[URIRef, BNode]) -> None: @@ -176,7 +182,8 @@ def _key_to_rdf(self, obj: model.Key, parent: Union[URIRef, BNode]) -> None: self.graph.add((key, self.aas["Key/type"], self.aas[f"KeyTypes/{_generic.KEY_TYPES[obj.type]}"])) self.graph.add((key, self.aas["Key/value"], Literal(str(obj.value), datatype=XSD.string))) - def _administrative_information_to_rdf(self, obj: model.AdministrativeInformation, parent: Union[URIRef, BNode]) -> None: + def _administrative_information_to_rdf(self, obj: model.AdministrativeInformation, + parent: Union[URIRef, BNode]) -> None: """ Serialization of objects of class :class:`~basyx.aas.model.base.AdministrativeInformation` to a joint rdflib Graph object @@ -189,13 +196,17 @@ def _administrative_information_to_rdf(self, obj: model.AdministrativeInformatio self.graph.add((parent, self.aas["Identifiable/administration"], administrative_information)) self._abstract_classes_to_rdf(obj, administrative_information) if obj.version: - self.graph.add((administrative_information, self.aas["AdministrativeInformation/version"], Literal(str(obj.version), datatype=XSD.string))) + self.graph.add((administrative_information, self.aas["AdministrativeInformation/version"], + Literal(str(obj.version), datatype=XSD.string))) if obj.revision: - self.graph.add((administrative_information, self.aas["AdministrativeInformation/revision"], Literal(str(obj.version), datatype=XSD.string))) + self.graph.add((administrative_information, self.aas["AdministrativeInformation/revision"], + Literal(str(obj.version), datatype=XSD.string))) if obj.creator: - self._reference_to_rdf(obj.creator, administrative_information, self.aas["AdministrativeInformation/creator"]) + self._reference_to_rdf(obj.creator, administrative_information, + self.aas["AdministrativeInformation/creator"]) if obj.template_id: - self.graph.add((administrative_information, self.aas["AdministrativeInformation/templateId"], Literal(str(obj.version), datatype=XSD.string))) + self.graph.add((administrative_information, self.aas["AdministrativeInformation/templateId"], + Literal(str(obj.version), datatype=XSD.string))) def _reference_to_rdf(self, obj: model.Reference, parent: Union[URIRef, BNode], objectProperty: URIRef) -> None: """ @@ -204,13 +215,15 @@ def _reference_to_rdf(self, obj: model.Reference, parent: Union[URIRef, BNode], :param obj: Object of class :class:`~basyx.aas.model.base.Reference` :param parent: The parent node. Can either be :class:`~rdflib.term.URIRef` or :class:`~rdflib.term.BNode` - :param objectProperty: The object property that is used to connect to from the parent, :class:`~rdflib.term.URIRef` + :param objectProperty: The object property that is used to connect to from the + parent, :class:`~rdflib.term.URIRef` :return: None """ reference = BNode() self.graph.add((parent, objectProperty, reference)) self.graph.add((reference, RDF.type, self.aas["Reference"])) - self.graph.add((reference, self.aas["Reference/type"], self.aas[f"ReferenceTypes/{_generic.REFERENCE_TYPES[obj.__class__]}"])) + self.graph.add((reference, self.aas["Reference/type"], + self.aas[f"ReferenceTypes/{_generic.REFERENCE_TYPES[obj.__class__]}"])) if obj.referred_semantic_id is not None: self._reference_to_rdf(obj.referred_semantic_id, reference, self.aas["Reference/referredSemanticId"]) for aas_key in obj.key: @@ -229,9 +242,11 @@ def _qualifier_to_rdf(self, obj: model.Qualifier, parent: Union[URIRef, BNode]) self.graph.add((parent, self.aas["Qualifiable/qualifiers"], qualifier)) self.graph.add((qualifier, RDF.type, self.aas["Qualifier"])) self._abstract_classes_to_rdf(obj, qualifier) - self.graph.add((qualifier, self.aas["Qualifier/kind"], self.aas[f"QualifierKind/{_generic.QUALIFIER_KIND[obj.kind]}"])) + self.graph.add( + (qualifier, self.aas["Qualifier/kind"], self.aas[f"QualifierKind/{_generic.QUALIFIER_KIND[obj.kind]}"])) self.graph.add((qualifier, self.aas["Qualifier/type"], Literal(str(obj.type), datatype=XSD.string))) - self.graph.add((qualifier, self.aas["Qualifier/valueType"], self.aas[f"DataTypeDefXsd/{model.datatypes.XSD_TYPE_NAMES[obj.value_type]}"])) + self.graph.add((qualifier, self.aas["Qualifier/valueType"], + self.aas[f"DataTypeDefXsd/{model.datatypes.XSD_TYPE_NAMES[obj.value_type]}"])) if obj.value: self._value_to_rdf(obj.value, obj.value_type, qualifier, self.aas["Qualifier/value"]) if obj.value_id: @@ -252,7 +267,8 @@ def _extension_to_rdf(self, obj: model.Extension, parent: Union[URIRef, BNode]) self._abstract_classes_to_rdf(obj, extension) self.graph.add((extension, self.aas["Extension/name"], Literal(str(obj.name), datatype=XSD.string))) if obj.value_type: - self.graph.add((extension, self.aas["Extension/valueType"], self.aas[f"DataTypeDefXsd/{model.datatypes.XSD_TYPE_NAMES[obj.value_type]}"])) + self.graph.add((extension, self.aas["Extension/valueType"], + self.aas[f"DataTypeDefXsd/{model.datatypes.XSD_TYPE_NAMES[obj.value_type]}"])) if obj.value: # Todo: Figure out why mypy complains about this function call and not about others self._value_to_rdf(obj.value, obj.value_type, extension, self.aas["Extension/value"]) # type: ignore @@ -260,19 +276,23 @@ def _extension_to_rdf(self, obj: model.Extension, parent: Union[URIRef, BNode]) for reference in obj.refers_to: self._reference_to_rdf(reference, extension, self.aas["Extension/refersTo"]) - def _value_reference_pair_to_rdf(self, obj: model.ValueReferencePair, parent: Union[URIRef, BNode], objectProperty: URIRef) -> None: + def _value_reference_pair_to_rdf(self, obj: model.ValueReferencePair, parent: Union[URIRef, BNode], + objectProperty: URIRef) -> None: """ - Serialization of objects of class :class:`~basyx.aas.model.base.ValueReferencePair` to a joint rdflib Graph object + Serialization of objects of class :class:`~basyx.aas.model.base.ValueReferencePair` to a joint rdflib Graph + object :param obj: Object of class :class:`~basyx.aas.model.base.ValueReferencePair` :param parent: The parent node. Can either be :class:`~rdflib.term.URIRef` or :class:`~rdflib.term.BNode` - :param objectProperty: The object property that is used to connect to from the parent, :class:`~rdflib.term.URIRef` + :param objectProperty: The object property that is used to connect to from the + parent, :class:`~rdflib.term.URIRef` :return: None """ value_reference_pair = BNode() self.graph.add((parent, objectProperty, value_reference_pair)) self.graph.add((value_reference_pair, RDF.type, self.aas["ValueReferencePair"])) - self.graph.add((value_reference_pair, self.aas["ValueReferencePair/value"], Literal(str(obj.value), datatype=XSD.string))) + self.graph.add( + (value_reference_pair, self.aas["ValueReferencePair/value"], Literal(str(obj.value), datatype=XSD.string))) self._reference_to_rdf(obj.value_id, value_reference_pair, self.aas["ValueReferencePair/valueId"]) def _value_list_to_rdf(self, obj: model.ValueList, parent: Union[URIRef, BNode], objectProperty: URIRef) -> None: @@ -281,7 +301,8 @@ def _value_list_to_rdf(self, obj: model.ValueList, parent: Union[URIRef, BNode], :param obj: Object of class :class:`~basyx.aas.model.base.ValueList` :param parent: The parent node. Can either be :class:`~rdflib.term.URIRef` or :class:`~rdflib.term.BNode` - :param objectProperty: The object property that is used to connect to from the parent, :class:`~rdflib.term.URIRef` + :param objectProperty: The object property that is used to connect to from the + parent, :class:`~rdflib.term.URIRef` :return: None """ value_list = BNode() @@ -295,23 +316,28 @@ def _value_list_to_rdf(self, obj: model.ValueList, parent: Union[URIRef, BNode], # transformation functions to serialize classes from model.aas # ############################################################ - def _specific_asset_id_to_rdf(self, obj: model.SpecificAssetId, parent: Union[URIRef, BNode], objectProperty: URIRef) -> None: + def _specific_asset_id_to_rdf(self, obj: model.SpecificAssetId, parent: Union[URIRef, BNode], + objectProperty: URIRef) -> None: """ Serialization of objects of class :class:`~basyx.aas.model.base.SpecificAssetId` to a joint rdflib Graph object :param obj: Object of class :class:`~basyx.aas.model.base.SpecificAssetId` :param parent: The parent node. Can either be :class:`~rdflib.term.URIRef` or :class:`~rdflib.term.BNode` - :param objectProperty: The object property that is used to connect to from the parent, :class:`~rdflib.term.URIRef` + :param objectProperty: The object property that is used to connect to from the + parent, :class:`~rdflib.term.URIRef` :return: None """ specific_asset_id = BNode() self.graph.add((parent, objectProperty, specific_asset_id)) self.graph.add((specific_asset_id, RDF.type, self.aas["SpecificAssetId"])) self._abstract_classes_to_rdf(obj, specific_asset_id) - self.graph.add((specific_asset_id, self.aas["SpecificAssetId/name"], Literal(str(obj.name), datatype=XSD.string))) - self.graph.add((specific_asset_id, self.aas["SpecificAssetId/value"], Literal(str(obj.value), datatype=XSD.string))) + self.graph.add( + (specific_asset_id, self.aas["SpecificAssetId/name"], Literal(str(obj.name), datatype=XSD.string))) + self.graph.add( + (specific_asset_id, self.aas["SpecificAssetId/value"], Literal(str(obj.value), datatype=XSD.string))) if obj.external_subject_id: - self._reference_to_rdf(obj.external_subject_id, specific_asset_id, self.aas["SpecificAssetId/externalSubjectId"]) + self._reference_to_rdf(obj.external_subject_id, specific_asset_id, + self.aas["SpecificAssetId/externalSubjectId"]) def _asset_information_to_rdf(self, obj: model.AssetInformation, parent: URIRef) -> None: """ @@ -320,21 +346,26 @@ def _asset_information_to_rdf(self, obj: model.AssetInformation, parent: URIRef) :param obj: Object of class :class:`~basyx.aas.model.aas.AssetInformation` :param parent: The parent node. Can either be :class:`~rdflib.term.URIRef` or :class:`~rdflib.term.BNode` - :param objectProperty: The object property that is used to connect to from the parent, :class:`~rdflib.term.URIRef` + :param objectProperty: The object property that is used to connect to from the + parent, :class:`~rdflib.term.URIRef` :return: None """ asset_info = BNode() self.graph.add((parent, self.aas["AssetAdministrationShell/assetInformation"], asset_info)) self.graph.add((asset_info, RDF.type, self.aas["AssetInformation"])) self._abstract_classes_to_rdf(obj, asset_info) - self.graph.add((asset_info, self.aas["AssetInformation/assetKind"], self.aas[f"AssetKind/{_generic.ASSET_KIND[obj.asset_kind]}"])) + self.graph.add((asset_info, self.aas["AssetInformation/assetKind"], + self.aas[f"AssetKind/{_generic.ASSET_KIND[obj.asset_kind]}"])) if obj.global_asset_id: - self.graph.add((asset_info, self.aas["AssetInformation/globalAssetId"], Literal(str(obj.global_asset_id), datatype=XSD.string))) + self.graph.add((asset_info, self.aas["AssetInformation/globalAssetId"], + Literal(str(obj.global_asset_id), datatype=XSD.string))) if obj.specific_asset_id: for specific_asset_id in obj.specific_asset_id: - self._specific_asset_id_to_rdf(specific_asset_id, asset_info, self.aas["AssetInformation/specificAssetIds"]) + self._specific_asset_id_to_rdf(specific_asset_id, asset_info, + self.aas["AssetInformation/specificAssetIds"]) if obj.asset_type: - self.graph.add((asset_info, self.aas["AssetInformation/assetType"], Literal(str(obj.asset_type), datatype=XSD.string))) + self.graph.add( + (asset_info, self.aas["AssetInformation/assetType"], Literal(str(obj.asset_type), datatype=XSD.string))) if obj.default_thumbnail: self._resource_to_rdf(obj.default_thumbnail, asset_info, self.aas["AssetInformation/defaultThumbnail"]) @@ -353,7 +384,8 @@ def _concept_description_to_rdf(self, obj: model.ConceptDescription) -> None: for reference in obj.is_case_of: self._reference_to_rdf(reference, subject, self.aas["ConceptDescription/isCaseOf"]) - def _embedded_data_specification_to_rdf(self, obj: model.EmbeddedDataSpecification, parent: Union[URIRef, BNode]) -> None: + def _embedded_data_specification_to_rdf(self, obj: model.EmbeddedDataSpecification, + parent: Union[URIRef, BNode]) -> None: """ Serialization of objects of class :class:`~basyx.aas.model.base.EmbeddedDataSpecification` to to a joint rdflib Graph object. @@ -363,13 +395,16 @@ def _embedded_data_specification_to_rdf(self, obj: model.EmbeddedDataSpecificati :return: None """ embedded_data_specification = BNode() - self.graph.add((parent, self.aas["HasDataSpecification/embeddedDataSpecifications"], embedded_data_specification)) + self.graph.add( + (parent, self.aas["HasDataSpecification/embeddedDataSpecifications"], embedded_data_specification)) self.graph.add((embedded_data_specification, RDF.type, self.aas["EmbeddedDataSpecification"])) self._abstract_classes_to_rdf(obj, embedded_data_specification) - self._reference_to_rdf(obj.data_specification, embedded_data_specification, self.aas["EmbeddedDataSpecification/dataSpecification"]) + self._reference_to_rdf(obj.data_specification, embedded_data_specification, + self.aas["EmbeddedDataSpecification/dataSpecification"]) self._data_specification_content_to_rdf(obj.data_specification_content, embedded_data_specification) - def _data_specification_content_to_rdf(self, obj: model.DataSpecificationContent, parent: Union[URIRef, BNode]) -> None: + def _data_specification_content_to_rdf(self, obj: model.DataSpecificationContent, + parent: Union[URIRef, BNode]) -> None: """ Serialization of objects of class :class:`~basyx.aas.model.base.DataSpecificationContent` to a joint rdflib Graph object @@ -379,13 +414,15 @@ def _data_specification_content_to_rdf(self, obj: model.DataSpecificationContent :return: None """ data_specification_content = BNode() - self.graph.add((parent, self.aas["EmbeddedDataSpecification/dataSpecificationContent"], data_specification_content)) + self.graph.add( + (parent, self.aas["EmbeddedDataSpecification/dataSpecificationContent"], data_specification_content)) if isinstance(obj, model.DataSpecificationIEC61360): self._data_specification_iec61360_to_rdf(obj, data_specification_content) else: raise TypeError(f"Serialization of {obj.__class__} to a joint rdflib Graph object is not supported!") - def _data_specification_iec61360_to_rdf(self, obj: model.DataSpecificationIEC61360, parent: Union[URIRef, BNode]) -> None: + def _data_specification_iec61360_to_rdf(self, obj: model.DataSpecificationIEC61360, + parent: Union[URIRef, BNode]) -> None: """ Serialization of objects of class :class:`~basyx.aas.model.base.DataSpecificationIEC61360` to a joint rdflib Graph object @@ -404,27 +441,33 @@ def _data_specification_iec61360_to_rdf(self, obj: model.DataSpecificationIEC613 if obj.unit_id is not None: self._reference_to_rdf(obj.unit_id, parent, self.aas["DataSpecificationIec61360/unitId"]) if obj.source_of_definition is not None: - self.graph.add((parent, self.aas["DataSpecificationIec61360/sourceOfDefinition"], Literal(obj.source_of_definition, datatype=XSD.string))) + self.graph.add((parent, self.aas["DataSpecificationIec61360/sourceOfDefinition"], + Literal(obj.source_of_definition, datatype=XSD.string))) if obj.symbol is not None: - self.graph.add((parent, self.aas["DataSpecificationIec61360/symbol"], Literal(obj.symbol, datatype=XSD.string))) + self.graph.add( + (parent, self.aas["DataSpecificationIec61360/symbol"], Literal(obj.symbol, datatype=XSD.string))) if obj.data_type is not None: - self.graph.add((parent, self.aas["DataSpecificationIec61360/dataType"], self.aas[f"DataSpecificationIec61360/{_generic.IEC61360_DATA_TYPES[obj.data_type]}"])) + self.graph.add((parent, self.aas["DataSpecificationIec61360/dataType"], + self.aas[f"DataSpecificationIec61360/{_generic.IEC61360_DATA_TYPES[obj.data_type]}"])) if obj.definition is not None: self._lang_string_set_to_rdf(obj.definition, parent, self.aas["DataSpecificationIec61360/definition"]) if obj.value_format is not None: - self.graph.add((parent, self.aas["DataSpecificationIec61360/valueFormat"], Literal(obj.value_format, datatype=XSD.string))) + self.graph.add((parent, self.aas["DataSpecificationIec61360/valueFormat"], + Literal(obj.value_format, datatype=XSD.string))) # # this can be either None or an empty set, both of which are equivalent to the bool false # # thus we don't check 'is not None' for this property if obj.value_list: self._value_list_to_rdf(obj.value_list, parent, self.aas["HasDataSpecification/valueList"]) if obj.value is not None: - self.graph.add((parent, self.aas["DataSpecificationIec61360/value"], Literal(obj.value, datatype=XSD.string))) + self.graph.add( + (parent, self.aas["DataSpecificationIec61360/value"], Literal(obj.value, datatype=XSD.string))) if obj.level_types: level_type_node = BNode() self.graph.add((parent, self.aas["DataSpecificationIec61360/levelType"], level_type_node)) self.graph.add((level_type_node, RDF.type, self.aas["LevelType"])) for k, v in _generic.IEC61360_LEVEL_TYPES.items(): - self.graph.add((level_type_node, self.aas[f"LevelType/{v}"], Literal(self._boolean_to_rdf(k in obj.level_types), datatype=XSD.boolean))) + self.graph.add((level_type_node, self.aas[f"LevelType/{v}"], + Literal(self._boolean_to_rdf(k in obj.level_types), datatype=XSD.boolean))) def _asset_administration_shell_to_rdf(self, obj: model.AssetAdministrationShell) -> None: """ @@ -449,14 +492,16 @@ def _asset_administration_shell_to_rdf(self, obj: model.AssetAdministrationShell # transformation functions to serialize classes from model.submodel # ################################################################# - def _submodel_element_to_rdf(self, obj: model.SubmodelElement, parent: Union[URIRef, BNode], objectProperty: URIRef) -> None: + def _submodel_element_to_rdf(self, obj: model.SubmodelElement, parent: Union[URIRef, BNode], + objectProperty: URIRef) -> None: """ Serialization of objects of class :class:`~basyx.aas.model.submodel.SubmodelElement` to a joint rdflib Graph object :param obj: Object of class :class:`~basyx.aas.model.submodel.SubmodelElement` :param parent: The parent node. Can either be :class:`~rdflib.term.URIRef` or :class:`~rdflib.term.BNode` - :param objectProperty: The object property that is used to connect to from the parent, :class:`~rdflib.term.URIRef` + :param objectProperty: The object property that is used to connect to from the + parent, :class:`~rdflib.term.URIRef` :return: None """ submodel_element = BNode() @@ -532,7 +577,8 @@ def _property_to_rdf(self, obj: model.Property, parent: Union[URIRef, BNode]) -> """ self.graph.add((parent, RDF.type, self.aas["Property"])) self._abstract_classes_to_rdf(obj, parent) - self.graph.add((parent, self.aas["Property/valueType"], self.aas[f"DataTypeDefXsd/{model.datatypes.XSD_TYPE_NAMES[obj.value_type]}"])) + self.graph.add((parent, self.aas["Property/valueType"], + self.aas[f"DataTypeDefXsd/{model.datatypes.XSD_TYPE_NAMES[obj.value_type]}"])) if obj.value is not None: self._value_to_rdf(obj.value, obj.value_type, parent, self.aas["Property/value"]) if obj.value_id: @@ -565,7 +611,8 @@ def _range_to_rdf(self, obj: model.Range, parent: Union[URIRef, BNode]) -> None: """ self.graph.add((parent, RDF.type, self.aas["Range"])) self._abstract_classes_to_rdf(obj, parent) - self.graph.add((parent, self.aas["Range/valueType"], self.aas[f"DataTypeDefXsd/{model.datatypes.XSD_TYPE_NAMES[obj.value_type]}"])) + self.graph.add((parent, self.aas["Range/valueType"], + self.aas[f"DataTypeDefXsd/{model.datatypes.XSD_TYPE_NAMES[obj.value_type]}"])) if obj.min is not None: self._value_to_rdf(obj.min, obj.value_type, parent, self.aas["Range/min"]) if obj.max is not None: @@ -583,7 +630,8 @@ def _blob_to_rdf(self, obj: model.Blob, parent: Union[URIRef, BNode]) -> None: self.graph.add((parent, RDF.type, self.aas["Blob"])) self._abstract_classes_to_rdf(obj, parent) if obj.value: - self.graph.add((parent, self.aas["Blob/value"], Literal(base64.b64encode(obj.value).decode(), datatype=XSD.base64Binary))) + self.graph.add((parent, self.aas["Blob/value"], + Literal(base64.b64encode(obj.value).decode(), datatype=XSD.base64Binary))) self.graph.add((parent, self.aas["Blob/contentType"], Literal(str(obj.content_type), datatype=XSD.string))) def _file_to_rdf(self, obj: model.File, parent: Union[URIRef, BNode]) -> None: @@ -593,7 +641,8 @@ def _file_to_rdf(self, obj: model.File, parent: Union[URIRef, BNode]) -> None: :param obj: Object of class :class:`~basyx.aas.model.submodel.File` :param parent: The parent node. Can either be :class:`~rdflib.term.URIRef` or :class:`~rdflib.term.BNode` - :param objectProperty: The object property that is used to connect to from the parent, :class:`~rdflib.term.URIRef` + :param objectProperty: The object property that is used to connect to from the + parent, :class:`~rdflib.term.URIRef` :return: None """ self.graph.add((parent, RDF.type, self.aas["File"])) @@ -609,7 +658,8 @@ def _resource_to_rdf(self, obj: model.Resource, parent: Union[URIRef, BNode], ob :param obj: Object of class :class:`~basyx.aas.model.base.Resource` :param parent: The parent node. Can either be :class:`~rdflib.term.URIRef` or :class:`~rdflib.term.BNode` - :param objectProperty: The object property that is used to connect to from the parent, :class:`~rdflib.term.URIRef` + :param objectProperty: The object property that is used to connect to from the + parent, :class:`~rdflib.term.URIRef` :return: None """ ressource = BNode() @@ -618,7 +668,8 @@ def _resource_to_rdf(self, obj: model.Resource, parent: Union[URIRef, BNode], ob self._abstract_classes_to_rdf(obj, ressource) self.graph.add((ressource, self.aas["Resource/path"], Literal(str(obj.path), datatype=XSD.string))) if obj.content_type: - self.graph.add((ressource, self.aas["Ressource/contentType"], Literal(str(obj.content_type), datatype=XSD.string))) + self.graph.add( + (ressource, self.aas["Ressource/contentType"], Literal(str(obj.content_type), datatype=XSD.string))) def _reference_element_to_rdf(self, obj: model.ReferenceElement, parent: Union[URIRef, BNode]) -> None: """ @@ -634,7 +685,8 @@ def _reference_element_to_rdf(self, obj: model.ReferenceElement, parent: Union[U if obj.value: self._reference_to_rdf(obj.value, parent, self.aas["ReferenceElement/value"]) - def _submodel_element_collection_to_rdf(self, obj: model.SubmodelElementCollection, parent: Union[URIRef, BNode]) -> None: + def _submodel_element_collection_to_rdf(self, obj: model.SubmodelElementCollection, + parent: Union[URIRef, BNode]) -> None: """ Serialization of objects of class :class:`~basyx.aas.model.submodel.SubmodelElementCollection` to a joint rdflib Graph object @@ -660,11 +712,16 @@ def _submodel_element_list_to_rdf(self, obj: model.SubmodelElementList, parent: """ self.graph.add((parent, RDF.type, self.aas["SubmodelElementList"])) self._abstract_classes_to_rdf(obj, parent) - self.graph.add((parent, self.aas["SubmodelElementList/orderRelevant"], Literal(self._boolean_to_rdf(obj.order_relevant), datatype=XSD.boolean))) + self.graph.add((parent, self.aas["SubmodelElementList/orderRelevant"], + Literal(self._boolean_to_rdf(obj.order_relevant), datatype=XSD.boolean))) if obj.semantic_id_list_element is not None: - self._reference_to_rdf(obj.semantic_id_list_element, parent, self.aas["SubmodelElementList/semanticIdListElement"]) - self.graph.add((parent, self.aas["SubmodelElementList/typeValueListElement"], - self.aas[f"AasSubmodelElements/{_generic.KEY_TYPES[model.KEY_TYPES_CLASSES[obj.type_value_list_element]]}"])) + self._reference_to_rdf(obj.semantic_id_list_element, parent, + self.aas["SubmodelElementList/semanticIdListElement"]) + self.graph.add(( + parent, + self.aas["SubmodelElementList/typeValueListElement"], + self.aas[f"AasSubmodelElements/{_generic.KEY_TYPES[model.KEY_TYPES_CLASSES[obj.type_value_list_element]]}"] + )) if obj.value_type_list_element is not None: self.graph.add((parent, self.aas["SubmodelElementList/valueTypeListElement"], self.aas[f"DataTypeDefXsd/{model.datatypes.XSD_TYPE_NAMES[obj.value_type_list_element]}"])) @@ -686,7 +743,8 @@ def _relationship_element_to_rdf(self, obj: model.RelationshipElement, parent: U self._reference_to_rdf(obj.first, parent, self.aas["RelationshipElement/first"]) self._reference_to_rdf(obj.second, parent, self.aas["RelationshipElement/second"]) - def _annotated_relationship_element_to_rdf(self, obj: model.AnnotatedRelationshipElement, parent: Union[URIRef, BNode]) -> None: + def _annotated_relationship_element_to_rdf(self, obj: model.AnnotatedRelationshipElement, + parent: Union[URIRef, BNode]) -> None: """ Serialization of objects of class :class:`~basyx.aas.model.submodel.AnnotatedRelationshipElement` to a joint rdflib Graph object @@ -701,9 +759,11 @@ def _annotated_relationship_element_to_rdf(self, obj: model.AnnotatedRelationshi self._reference_to_rdf(obj.second, parent, self.aas["RelationshipElement/second"]) if obj.annotation: for data_element in obj.annotation: - self._submodel_element_to_rdf(data_element, parent, self.aas["AnnotatedRelationshipElement/annotations"]) + self._submodel_element_to_rdf(data_element, parent, + self.aas["AnnotatedRelationshipElement/annotations"]) - def _operation_variable_to_rdf(self, obj: model.SubmodelElement, operation: Union[URIRef, BNode], objectProperty: URIRef) -> None: + def _operation_variable_to_rdf(self, obj: model.SubmodelElement, operation: Union[URIRef, BNode], + objectProperty: URIRef) -> None: """ Serialization of :class:`~basyx.aas.model.submodel.SubmodelElement` to a joint rdflib Graph object. Since we don't implement the ``OperationVariable`` class, which is just a wrapper for a single @@ -712,7 +772,8 @@ def _operation_variable_to_rdf(self, obj: model.SubmodelElement, operation: Unio :param obj: Object of class :class:`~basyx.aas.model.submodel.SubmodelElement` :param parent: The parent node. Can either be :class:`~rdflib.term.URIRef` or :class:`~rdflib.term.BNode` - :param objectProperty: The object property that is used to connect to from the parent, :class:`~rdflib.term.URIRef` + :param objectProperty: The object property that is used to connect to from the + parent, :class:`~rdflib.term.URIRef` :return: None """ operation_variable = BNode() @@ -765,9 +826,11 @@ def _entity_to_rdf(self, obj: model.Entity, parent: Union[URIRef, BNode]) -> Non if obj.statement: for statement in obj.statement: self._submodel_element_to_rdf(statement, parent, self.aas["Entity/statements"]) - self.graph.add((parent, self.aas["Entity/entityType"], self.aas[f"EntityType/{_generic.ENTITY_TYPES[obj.entity_type]}"])) + self.graph.add( + (parent, self.aas["Entity/entityType"], self.aas[f"EntityType/{_generic.ENTITY_TYPES[obj.entity_type]}"])) if obj.global_asset_id: - self.graph.add((parent, self.aas["Entity/globalAssetId"], Literal(str(obj.global_asset_id), datatype=XSD.string))) + self.graph.add( + (parent, self.aas["Entity/globalAssetId"], Literal(str(obj.global_asset_id), datatype=XSD.string))) if obj.specific_asset_id: for specific_asset_id in obj.specific_asset_id: self._specific_asset_id_to_rdf(specific_asset_id, parent, self.aas["Entity/specificAssetIds"]) @@ -784,18 +847,24 @@ def _basic_event_element_to_rdf(self, obj: model.BasicEventElement, parent: Unio self.graph.add((parent, RDF.type, self.aas["BasicEventElement"])) self._abstract_classes_to_rdf(obj, parent) self._reference_to_rdf(obj.observed, parent, self.aas["BasicEventElement/observed"]) - self.graph.add((parent, self.aas["BasicEventElement/direction"], self.aas[f"Direction/{_generic.DIRECTION[obj.direction]}"])) - self.graph.add((parent, self.aas["BasicEventElement/state"], self.aas[f"StateOfEvent/{_generic.STATE_OF_EVENT[obj.state]}"])) + self.graph.add((parent, self.aas["BasicEventElement/direction"], + self.aas[f"Direction/{_generic.DIRECTION[obj.direction]}"])) + self.graph.add((parent, self.aas["BasicEventElement/state"], + self.aas[f"StateOfEvent/{_generic.STATE_OF_EVENT[obj.state]}"])) if obj.message_topic: - self.graph.add((parent, self.aas["BasicEventElement/messageTopic"], Literal(str(obj.message_topic), datatype=XSD.string))) + self.graph.add((parent, self.aas["BasicEventElement/messageTopic"], + Literal(str(obj.message_topic), datatype=XSD.string))) if obj.message_broker: self._reference_to_rdf(obj.message_broker, parent, self.aas["BasicEventElement/messageBroker"]) if obj.last_update is not None: - self.graph.add((parent, self.aas["BasicEventElement/lastUpdate"], Literal(model.datatypes.xsd_repr(obj.last_update), datatype=XSD.string))) + self.graph.add((parent, self.aas["BasicEventElement/lastUpdate"], + Literal(model.datatypes.xsd_repr(obj.last_update), datatype=XSD.string))) if obj.min_interval is not None: - self.graph.add((parent, self.aas["BasicEventElement/minInterval"], Literal(model.datatypes.xsd_repr(obj.min_interval), datatype=XSD.string))) + self.graph.add((parent, self.aas["BasicEventElement/minInterval"], + Literal(model.datatypes.xsd_repr(obj.min_interval), datatype=XSD.string))) if obj.max_interval is not None: - self.graph.add((parent, self.aas["BasicEventElement/maxInterval"], Literal(model.datatypes.xsd_repr(obj.max_interval), datatype=XSD.string))) + self.graph.add((parent, self.aas["BasicEventElement/maxInterval"], + Literal(model.datatypes.xsd_repr(obj.max_interval), datatype=XSD.string))) # ############################################################## diff --git a/pyproject.toml b/pyproject.toml index 436d6a4b..5b383008 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -46,7 +46,7 @@ dependencies = [ "hypothesis~=6.13", "lxml-stubs~=0.5.1", "rdflib~=7.0.0", - "pyshacl~=0.26.0" + "pyshacl~=0.27.0" ] [project.optional-dependencies] diff --git a/test/adapter/rdf/test_rdf_serialization.py b/test/adapter/rdf/test_rdf_serialization.py index f3c672e0..11ed1db8 100644 --- a/test/adapter/rdf/test_rdf_serialization.py +++ b/test/adapter/rdf/test_rdf_serialization.py @@ -14,7 +14,8 @@ from basyx.aas import model from basyx.aas.adapter.rdf import write_aas_rdf_file -from basyx.aas.examples.data import example_submodel_template, example_aas_mandatory_attributes, example_aas_missing_attributes, example_aas +from basyx.aas.examples.data import example_submodel_template, example_aas_mandatory_attributes, \ + example_aas_missing_attributes, example_aas RDF_ONTOLOGY_FILE = os.path.join(os.path.dirname(__file__), '../schemas/aasRDFOntology.ttl') RDF_SHACL_SCHEMA_FILE = os.path.join(os.path.dirname(__file__), '../schemas/aasRDFShaclSchema.ttl') @@ -56,15 +57,15 @@ def validate_graph(data_graph: io.BytesIO): # validate serialization against schema conforms, results_graph, results_text = validate( - data_graph=data_graph, # Passing the BytesIO object here - shacl_graph=shacl_graph, # The SHACL graph - ont_graph=aas_graph, - data_graph_format="turtle", # Specify the format for the data graph (since it's serialized) - inference='both', # Optional: perform RDFS inference - abort_on_first=True, # Don't continue validation after finding an error - allow_infos=True, # Allow informational messages - allow_warnings=True, # Allow warnings - advanced=True) + data_graph=data_graph, # Passing the BytesIO object here + shacl_graph=shacl_graph, # The SHACL graph + ont_graph=aas_graph, + data_graph_format="turtle", # Specify the format for the data graph (since it's serialized) + inference='both', # Optional: perform RDFS inference + abort_on_first=True, # Don't continue validation after finding an error + allow_infos=True, # Allow informational messages + allow_warnings=True, # Allow warnings + advanced=True) # print("Conforms:", conforms) # print("Validation Results:\n", results_text) assert conforms is True @@ -84,7 +85,7 @@ def test_random_object_serialization(self) -> None: submodel_reference = model.ModelReference(submodel_key, model.Submodel) submodel = model.Submodel(submodel_identifier, semantic_id=model.ExternalReference((model.Key(model.KeyTypes.GLOBAL_REFERENCE, - "http://acplt.org/TestSemanticId"),))) + "http://acplt.org/TestSemanticId"),))) test_aas = model.AssetAdministrationShell(model.AssetInformation(global_asset_id="test"), aas_identifier, submodel={submodel_reference})