diff --git a/basyx/aas/adapter/json/json_deserialization.py b/basyx/aas/adapter/json/json_deserialization.py index 6524f4b4d..960523e0f 100644 --- a/basyx/aas/adapter/json/json_deserialization.py +++ b/basyx/aas/adapter/json/json_deserialization.py @@ -70,7 +70,7 @@ def _get_ts(dct: Dict[str, object], key: str, type_: Type[T]) -> T: return val -def _expect_type(object_: object, type_: Type, context: str, failsafe: bool) -> bool: +def _validate_type(object_: object, type_: Type, context: str, failsafe: bool) -> bool: """ Helper function to check type of an embedded object. @@ -232,48 +232,72 @@ def _amend_abstract_attributes(cls, obj: object, dct: Dict[str, object]) -> None :param dct: The object's dict representation from JSON """ if isinstance(obj, model.Referable): - if 'idShort' in dct: - obj.id_short = _get_ts(dct, 'idShort', str) - if 'category' in dct: - obj.category = _get_ts(dct, 'category', str) - if 'displayName' in dct: - obj.display_name = cls._construct_lang_string_set(_get_ts(dct, 'displayName', list), - model.MultiLanguageNameType) - if 'description' in dct: - obj.description = cls._construct_lang_string_set(_get_ts(dct, 'description', list), - model.MultiLanguageTextType) + cls._amend_referable_attrs(obj, dct) if isinstance(obj, model.Identifiable): - if 'administration' in dct: - obj.administration = cls._construct_administrative_information(_get_ts(dct, 'administration', dict)) + cls._amend_identifiable_attrs(obj, dct) if isinstance(obj, model.HasSemantics): - if 'semanticId' in dct: - obj.semantic_id = cls._construct_reference(_get_ts(dct, 'semanticId', dict)) - if 'supplementalSemanticIds' in dct: - for ref in _get_ts(dct, 'supplementalSemanticIds', list): - obj.supplemental_semantic_id.append(cls._construct_reference(ref)) + cls._amend_has_semantics_attrs(obj, dct) # `HasKind` provides only mandatory, immutable attributes; so we cannot do anything here, after object creation. # However, the `cls._get_kind()` function may assist by retrieving them from the JSON object if isinstance(obj, model.Qualifiable) and not cls.stripped: - if 'qualifiers' in dct: - for constraint_dct in _get_ts(dct, 'qualifiers', list): - constraint = cls._construct_qualifier(constraint_dct) - obj.qualifier.add(constraint) + cls._amend_qualifiable_attrs(obj, dct) if isinstance(obj, model.HasDataSpecification) and not cls.stripped: - if 'embeddedDataSpecifications' in dct: - for dspec in _get_ts(dct, 'embeddedDataSpecifications', list): - obj.embedded_data_specifications.append( - # TODO: remove the following type: ignore comment when mypy supports abstract types for Type[T] - # see https://github.com/python/mypy/issues/5374 - model.EmbeddedDataSpecification( - data_specification=cls._construct_reference(_get_ts(dspec, 'dataSpecification', dict)), - data_specification_content=_get_ts(dspec, 'dataSpecificationContent', - model.DataSpecificationContent) # type: ignore - ) - ) + cls._amend_has_data_specification_attrs(obj, dct) if isinstance(obj, model.HasExtension) and not cls.stripped: - if 'extensions' in dct: - for extension in _get_ts(dct, 'extensions', list): - obj.extension.add(cls._construct_extension(extension)) + cls._amend_has_extension_attrs(obj, dct) + + @classmethod + def _amend_referable_attrs(cls, obj: model.Referable, dct: Dict[str, object]): + if 'idShort' in dct: + obj.id_short = _get_ts(dct, 'idShort', str) + if 'category' in dct: + obj.category = _get_ts(dct, 'category', str) + if 'displayName' in dct: + obj.display_name = cls._construct_lang_string_set(_get_ts(dct, 'displayName', list), + model.MultiLanguageNameType) + if 'description' in dct: + obj.description = cls._construct_lang_string_set(_get_ts(dct, 'description', list), + model.MultiLanguageTextType) + + @classmethod + def _amend_identifiable_attrs(cls, obj: model.Identifiable, dct: Dict[str, object]): + if 'administration' in dct: + obj.administration = cls._construct_administrative_information(_get_ts(dct, 'administration', dict)) + + @classmethod + def _amend_has_semantics_attrs(cls, obj: model.HasSemantics, dct: Dict[str, object]): + if 'semanticId' in dct: + obj.semantic_id = cls._construct_reference(_get_ts(dct, 'semanticId', dict)) + if 'supplementalSemanticIds' in dct: + for ref in _get_ts(dct, 'supplementalSemanticIds', list): + obj.supplemental_semantic_id.append(cls._construct_reference(ref)) + + @classmethod + def _amend_qualifiable_attrs(cls, obj: model.Qualifiable, dct: Dict[str, object]): + if 'qualifiers' in dct: + for constraint_dct in _get_ts(dct, 'qualifiers', list): + constraint = cls._construct_qualifier(constraint_dct) + obj.qualifier.add(constraint) + + @classmethod + def _amend_has_data_specification_attrs(cls, obj: model.HasDataSpecification, dct: Dict[str, object]): + if 'embeddedDataSpecifications' in dct: + for dspec in _get_ts(dct, 'embeddedDataSpecifications', list): + obj.embedded_data_specifications.append( + # TODO: remove the following type: ignore comment when mypy supports abstract types for Type[T] + # see https://github.com/python/mypy/issues/5374 + model.EmbeddedDataSpecification( + data_specification=cls._construct_reference(_get_ts(dspec, 'dataSpecification', dict)), + data_specification_content=_get_ts(dspec, 'dataSpecificationContent', + model.DataSpecificationContent) # type: ignore + ) + ) + + @classmethod + def _amend_has_extension_attrs(cls, obj: model.HasExtension, dct: Dict[str, object]): + if 'extensions' in dct: + for extension in _get_ts(dct, 'extensions', list): + obj.extension.add(cls._construct_extension(extension)) @classmethod def _get_kind(cls, dct: Dict[str, object]) -> model.ModellingKind: @@ -517,7 +541,7 @@ def _construct_entity(cls, dct: Dict[str, object], object_class=model.Entity) -> cls._amend_abstract_attributes(ret, dct) if not cls.stripped and 'statements' in dct: for element in _get_ts(dct, "statements", list): - if _expect_type(element, model.SubmodelElement, str(ret), cls.failsafe): + if _validate_type(element, model.SubmodelElement, str(ret), cls.failsafe): ret.statement.add(element) return ret @@ -554,7 +578,7 @@ def _construct_submodel(cls, dct: Dict[str, object], object_class=model.Submodel cls._amend_abstract_attributes(ret, dct) if not cls.stripped and 'submodelElements' in dct: for element in _get_ts(dct, "submodelElements", list): - if _expect_type(element, model.SubmodelElement, str(ret), cls.failsafe): + if _validate_type(element, model.SubmodelElement, str(ret), cls.failsafe): ret.submodel_element.add(element) return ret @@ -633,7 +657,7 @@ def _construct_annotated_relationship_element( cls._amend_abstract_attributes(ret, dct) if not cls.stripped and 'annotations' in dct: for element in _get_ts(dct, 'annotations', list): - if _expect_type(element, model.DataElement, str(ret), cls.failsafe): + if _validate_type(element, model.DataElement, str(ret), cls.failsafe): ret.annotation.add(element) return ret @@ -645,7 +669,7 @@ def _construct_submodel_element_collection(cls, dct: Dict[str, object], cls._amend_abstract_attributes(ret, dct) if not cls.stripped and 'value' in dct: for element in _get_ts(dct, "value", list): - if _expect_type(element, model.SubmodelElement, str(ret), cls.failsafe): + if _validate_type(element, model.SubmodelElement, str(ret), cls.failsafe): ret.value.add(element) return ret @@ -670,7 +694,7 @@ def _construct_submodel_element_list(cls, dct: Dict[str, object], object_class=m cls._amend_abstract_attributes(ret, dct) if not cls.stripped and 'value' in dct: for element in _get_ts(dct, 'value', list): - if _expect_type(element, type_value_list_element, str(ret), cls.failsafe): + if _validate_type(element, type_value_list_element, str(ret), cls.failsafe): ret.value.add(element) return ret diff --git a/basyx/aas/adapter/json/json_serialization.py b/basyx/aas/adapter/json/json_serialization.py index 17fff2680..a34f9fac2 100644 --- a/basyx/aas/adapter/json/json_serialization.py +++ b/basyx/aas/adapter/json/json_serialization.py @@ -62,7 +62,7 @@ def default(self, obj: object) -> object: :param obj: The object to serialize to json :return: The serialized object """ - mapping: Dict[Type, Callable] = { + serialization_methods: Dict[Type, Callable] = { model.AdministrativeInformation: self._administrative_information_to_json, model.AnnotatedRelationshipElement: self._annotated_relationship_element_to_json, model.AssetAdministrationShell: self._asset_administration_shell_to_json, @@ -92,10 +92,10 @@ def default(self, obj: object) -> object: model.SubmodelElementList: self._submodel_element_list_to_json, model.ValueReferencePair: self._value_reference_pair_to_json, } - for typ in mapping: + for typ in serialization_methods: if isinstance(obj, typ): - mapping_method = mapping[typ] - return mapping_method(obj) + serialization_method = serialization_methods[typ] + return serialization_method(obj) return super().default(obj) @classmethod @@ -108,48 +108,76 @@ def _abstract_classes_to_json(cls, obj: object) -> Dict[str, object]: """ data: Dict[str, object] = {} if isinstance(obj, model.HasExtension) and not cls.stripped: - if obj.extension: - data['extensions'] = list(obj.extension) + cls._extend_with_has_extension_attrs(data, obj) if isinstance(obj, model.HasDataSpecification) and not cls.stripped: - if obj.embedded_data_specifications: - data['embeddedDataSpecifications'] = [ - {'dataSpecification': spec.data_specification, - 'dataSpecificationContent': spec.data_specification_content} - for spec in obj.embedded_data_specifications - ] - + cls._extend_with_has_data_specification_specific_attrs(data, obj) if isinstance(obj, model.Referable): - if obj.id_short and not isinstance(obj.parent, model.SubmodelElementList): - data['idShort'] = obj.id_short - if obj.display_name: - data['displayName'] = obj.display_name - if obj.category: - data['category'] = obj.category - if obj.description: - data['description'] = obj.description - try: - ref_type = next(iter(t for t in inspect.getmro(type(obj)) if t in model.KEY_TYPES_CLASSES)) - except StopIteration as e: - raise TypeError("Object of type {} is Referable but does not inherit from a known AAS type" - .format(obj.__class__.__name__)) from e - data['modelType'] = ref_type.__name__ + cls._extend_with_referable_attrs(data, obj) if isinstance(obj, model.Identifiable): - data['id'] = obj.id - if obj.administration: - data['administration'] = obj.administration + cls._extend_with_identifiable_attrs(data, obj) if isinstance(obj, model.HasSemantics): - if obj.semantic_id: - data['semanticId'] = obj.semantic_id - if obj.supplemental_semantic_id: - data['supplementalSemanticIds'] = list(obj.supplemental_semantic_id) + cls._extend_with_has_semantics_attrs(data, obj) if isinstance(obj, model.HasKind): - if obj.kind is model.ModellingKind.TEMPLATE: - data['kind'] = _generic.MODELLING_KIND[obj.kind] + cls._extend_with_has_kind_attrs(data, obj) if isinstance(obj, model.Qualifiable) and not cls.stripped: - if obj.qualifier: - data['qualifiers'] = list(obj.qualifier) + cls._extend_with_qualifiable_attrs(data, obj) return data + @classmethod + def _extend_with_has_extension_attrs(cls, data: Dict[str, object], obj: model.HasExtension): + if obj.extension: + data['extensions'] = list(obj.extension) + + @classmethod + def _extend_with_has_data_specification_specific_attrs(cls, data: Dict[str, object], + obj: model.HasDataSpecification): + if obj.embedded_data_specifications: + data['embeddedDataSpecifications'] = [ + {'dataSpecification': spec.data_specification, + 'dataSpecificationContent': spec.data_specification_content} + for spec in obj.embedded_data_specifications + ] + + @classmethod + def _extend_with_referable_attrs(cls, data: Dict[str, object], obj: model.Referable): + if obj.id_short and not isinstance(obj.parent, model.SubmodelElementList): + data['idShort'] = obj.id_short + if obj.display_name: + data['displayName'] = obj.display_name + if obj.category: + data['category'] = obj.category + if obj.description: + data['description'] = obj.description + try: + ref_type = next(iter(t for t in inspect.getmro(type(obj)) if t in model.KEY_TYPES_CLASSES)) + except StopIteration as e: + raise TypeError("Object of type {} is Referable but does not inherit from a known AAS type" + .format(obj.__class__.__name__)) from e + data['modelType'] = ref_type.__name__ + + @classmethod + def _extend_with_identifiable_attrs(cls, data: Dict[str, object], obj: model.Identifiable): + data['id'] = obj.id + if obj.administration: + data['administration'] = obj.administration + + @classmethod + def _extend_with_has_semantics_attrs(cls, data: Dict[str, object], obj: model.HasSemantics): + if obj.semantic_id: + data['semanticId'] = obj.semantic_id + if obj.supplemental_semantic_id: + data['supplementalSemanticIds'] = list(obj.supplemental_semantic_id) + + @classmethod + def _extend_with_has_kind_attrs(cls, data: Dict[str, object], obj: model.HasKind): + if obj.kind is model.ModellingKind.TEMPLATE: + data['kind'] = _generic.MODELLING_KIND[obj.kind] + + @classmethod + def _extend_with_qualifiable_attrs(cls, data: Dict[str, object], obj: model.Qualifiable): + if obj.qualifier: + data['qualifiers'] = list(obj.qualifier) + # ############################################################# # transformation functions to serialize classes from model.base # ############################################################# diff --git a/basyx/aas/adapter/xml/xml_deserialization.py b/basyx/aas/adapter/xml/xml_deserialization.py index e42a9a6f6..21cf3f5cf 100644 --- a/basyx/aas/adapter/xml/xml_deserialization.py +++ b/basyx/aas/adapter/xml/xml_deserialization.py @@ -436,53 +436,77 @@ def _amend_abstract_attributes(cls, obj: object, element: etree.Element) -> None :return: None """ if isinstance(obj, model.Referable): - id_short = _get_text_or_none(element.find(NS_AAS + "idShort")) - if id_short is not None: - obj.id_short = id_short - category = _get_text_or_none(element.find(NS_AAS + "category")) - display_name = _failsafe_construct(element.find(NS_AAS + "displayName"), - cls.construct_multi_language_name_type, cls.failsafe) - if display_name is not None: - obj.display_name = display_name - if category is not None: - obj.category = category - description = _failsafe_construct(element.find(NS_AAS + "description"), - cls.construct_multi_language_text_type, cls.failsafe) - if description is not None: - obj.description = description + cls._amend_referable_attrs(element, obj) if isinstance(obj, model.Identifiable): - administration = _failsafe_construct(element.find(NS_AAS + "administration"), - cls.construct_administrative_information, cls.failsafe) - if administration: - obj.administration = administration + cls._amend_identifiable_attrs(element, obj) if isinstance(obj, model.HasSemantics): - semantic_id = _failsafe_construct(element.find(NS_AAS + "semanticId"), cls.construct_reference, - cls.failsafe) - if semantic_id is not None: - obj.semantic_id = semantic_id - supplemental_semantic_ids = element.find(NS_AAS + "supplementalSemanticIds") - if supplemental_semantic_ids is not None: - for supplemental_semantic_id in _child_construct_multiple(supplemental_semantic_ids, - NS_AAS + "reference", cls.construct_reference, - cls.failsafe): - obj.supplemental_semantic_id.append(supplemental_semantic_id) + cls._amend_has_semantics_attrs(element, obj) if isinstance(obj, model.Qualifiable) and not cls.stripped: - qualifiers_elem = element.find(NS_AAS + "qualifiers") - if qualifiers_elem is not None and len(qualifiers_elem) > 0: - for qualifier in _failsafe_construct_multiple(qualifiers_elem, cls.construct_qualifier, cls.failsafe): - obj.qualifier.add(qualifier) + cls._amend_qualifiable_attrs(element, obj) if isinstance(obj, model.HasDataSpecification) and not cls.stripped: - embedded_data_specifications_elem = element.find(NS_AAS + "embeddedDataSpecifications") - if embedded_data_specifications_elem is not None: - for eds in _failsafe_construct_multiple(embedded_data_specifications_elem, - cls.construct_embedded_data_specification, cls.failsafe): - obj.embedded_data_specifications.append(eds) + cls.amend_has_data_specification_attrs(element, obj) if isinstance(obj, model.HasExtension) and not cls.stripped: - extension_elem = element.find(NS_AAS + "extensions") - if extension_elem is not None: - for extension in _child_construct_multiple(extension_elem, NS_AAS + "extension", - cls.construct_extension, cls.failsafe): - obj.extension.add(extension) + cls._amend_extension_attrs(element, obj) + + @classmethod + def _amend_referable_attrs(cls, element: etree.Element, obj: model.Referable): + id_short = _get_text_or_none(element.find(NS_AAS + "idShort")) + if id_short is not None: + obj.id_short = id_short + display_name = _failsafe_construct(element.find(NS_AAS + "displayName"), + cls.construct_multi_language_name_type, cls.failsafe) + if display_name is not None: + obj.display_name = display_name + category = _get_text_or_none(element.find(NS_AAS + "category")) + if category is not None: + obj.category = category + description = _failsafe_construct(element.find(NS_AAS + "description"), + cls.construct_multi_language_text_type, cls.failsafe) + if description is not None: + obj.description = description + + @classmethod + def _amend_identifiable_attrs(cls, element: etree.Element, obj: model.Identifiable): + administration = _failsafe_construct(element.find(NS_AAS + "administration"), + cls.construct_administrative_information, cls.failsafe) + if administration: + obj.administration = administration + + @classmethod + def _amend_has_semantics_attrs(cls, element: etree.Element, obj: model.HasSemantics): + semantic_id = _failsafe_construct(element.find(NS_AAS + "semanticId"), cls.construct_reference, + cls.failsafe) + if semantic_id is not None: + obj.semantic_id = semantic_id + supplemental_semantic_ids = element.find(NS_AAS + "supplementalSemanticIds") + if supplemental_semantic_ids is not None: + for supplemental_semantic_id in _child_construct_multiple(supplemental_semantic_ids, + NS_AAS + "reference", cls.construct_reference, + cls.failsafe): + obj.supplemental_semantic_id.append(supplemental_semantic_id) + + @classmethod + def _amend_qualifiable_attrs(cls, element: etree.Element, obj: model.Qualifiable): + qualifiers_elem = element.find(NS_AAS + "qualifiers") + if qualifiers_elem is not None and len(qualifiers_elem) > 0: + for qualifier in _failsafe_construct_multiple(qualifiers_elem, cls.construct_qualifier, cls.failsafe): + obj.qualifier.add(qualifier) + + @classmethod + def amend_has_data_specification_attrs(cls, element: etree.Element, obj: model.HasDataSpecification): + embedded_data_specifications_elem = element.find(NS_AAS + "embeddedDataSpecifications") + if embedded_data_specifications_elem is not None: + for eds in _failsafe_construct_multiple(embedded_data_specifications_elem, + cls.construct_embedded_data_specification, cls.failsafe): + obj.embedded_data_specifications.append(eds) + + @classmethod + def _amend_extension_attrs(cls, element: etree.Element, obj: model.HasExtension): + extension_elem = element.find(NS_AAS + "extensions") + if extension_elem is not None: + for extension in _child_construct_multiple(extension_elem, NS_AAS + "extension", + cls.construct_extension, cls.failsafe): + obj.extension.add(extension) @classmethod def _construct_relationship_element_internal(cls, element: etree.Element, object_class: Type[RE], **_kwargs: Any) \ @@ -707,16 +731,16 @@ def construct_submodel_element(cls, element: etree.Element, **kwargs: Any) -> mo This function doesn't support the object_class parameter. Overwrite each individual SubmodelElement/DataElement constructor function instead. """ - submodel_elements: Dict[str, Callable[..., model.SubmodelElement]] = {NS_AAS + k: v for k, v in { - "annotatedRelationshipElement": cls.construct_annotated_relationship_element, - "basicEventElement": cls.construct_basic_event_element, - "capability": cls.construct_capability, - "entity": cls.construct_entity, - "operation": cls.construct_operation, - "relationshipElement": cls.construct_relationship_element, - "submodelElementCollection": cls.construct_submodel_element_collection, - "submodelElementList": cls.construct_submodel_element_list - }.items()} + submodel_elements: Dict[str, Callable[..., model.SubmodelElement]] = { + f"{NS_AAS}annotatedRelationshipElement": cls.construct_annotated_relationship_element, + f"{NS_AAS}basicEventElement": cls.construct_basic_event_element, + f"{NS_AAS}capability": cls.construct_capability, + f"{NS_AAS}entity": cls.construct_entity, + f"{NS_AAS}operation": cls.construct_operation, + f"{NS_AAS}relationshipElement": cls.construct_relationship_element, + f"{NS_AAS}submodelElementCollection": cls.construct_submodel_element_collection, + f"{NS_AAS}submodelElementList": cls.construct_submodel_element_list + } if element.tag not in submodel_elements: return cls.construct_data_element(element, abstract_class_name="SubmodelElement", **kwargs) return submodel_elements[element.tag](element, **kwargs) @@ -728,14 +752,14 @@ def construct_data_element(cls, element: etree.Element, abstract_class_name: str This function does not support the object_class parameter. Overwrite each individual DataElement constructor function instead. """ - data_elements: Dict[str, Callable[..., model.DataElement]] = {NS_AAS + k: v for k, v in { - "blob": cls.construct_blob, - "file": cls.construct_file, - "multiLanguageProperty": cls.construct_multi_language_property, - "property": cls.construct_property, - "range": cls.construct_range, - "referenceElement": cls.construct_reference_element, - }.items()} + data_elements: Dict[str, Callable[..., model.DataElement]] = { + f"{NS_AAS}blob": cls.construct_blob, + f"{NS_AAS}file": cls.construct_file, + f"{NS_AAS}multiLanguageProperty": cls.construct_multi_language_property, + f"{NS_AAS}property": cls.construct_property, + f"{NS_AAS}range": cls.construct_range, + f"{NS_AAS}referenceElement": cls.construct_reference_element, + } if element.tag not in data_elements: raise KeyError(_element_pretty_identifier(element) + f" is not a valid {abstract_class_name}!") return data_elements[element.tag](element, **kwargs) @@ -1090,9 +1114,9 @@ def construct_data_specification_content(cls, element: etree.Element, **kwargs: Overwrite each individual DataSpecificationContent constructor function instead. """ data_specification_contents: Dict[str, Callable[..., model.DataSpecificationContent]] = \ - {NS_AAS + k: v for k, v in { - "dataSpecificationIec61360": cls.construct_data_specification_iec61360, - }.items()} + { + f"{NS_AAS}dataSpecificationIec61360": cls.construct_data_specification_iec61360, + } if element.tag not in data_specification_contents: raise KeyError(f"{_element_pretty_identifier(element)} is not a valid DataSpecificationContent!") return data_specification_contents[element.tag](element, **kwargs) @@ -1295,89 +1319,53 @@ def read_aas_xml_element(file: IO, construct: XMLConstructables, failsafe: bool :return: The constructed object or None, if an error occurred in failsafe mode. """ decoder_ = _select_decoder(failsafe, stripped, decoder) - constructor: Callable[..., object] - - if construct == XMLConstructables.KEY: - constructor = decoder_.construct_key - elif construct == XMLConstructables.REFERENCE: - constructor = decoder_.construct_reference - elif construct == XMLConstructables.MODEL_REFERENCE: - constructor = decoder_.construct_model_reference - elif construct == XMLConstructables.GLOBAL_REFERENCE: - constructor = decoder_.construct_external_reference - elif construct == XMLConstructables.ADMINISTRATIVE_INFORMATION: - constructor = decoder_.construct_administrative_information - elif construct == XMLConstructables.QUALIFIER: - constructor = decoder_.construct_qualifier - elif construct == XMLConstructables.ANNOTATED_RELATIONSHIP_ELEMENT: - constructor = decoder_.construct_annotated_relationship_element - elif construct == XMLConstructables.BASIC_EVENT_ELEMENT: - constructor = decoder_.construct_basic_event_element - elif construct == XMLConstructables.BLOB: - constructor = decoder_.construct_blob - elif construct == XMLConstructables.CAPABILITY: - constructor = decoder_.construct_capability - elif construct == XMLConstructables.ENTITY: - constructor = decoder_.construct_entity - elif construct == XMLConstructables.EXTENSION: - constructor = decoder_.construct_extension - elif construct == XMLConstructables.FILE: - constructor = decoder_.construct_file - elif construct == XMLConstructables.RESOURCE: - constructor = decoder_.construct_resource - elif construct == XMLConstructables.MULTI_LANGUAGE_PROPERTY: - constructor = decoder_.construct_multi_language_property - elif construct == XMLConstructables.OPERATION: - constructor = decoder_.construct_operation - elif construct == XMLConstructables.PROPERTY: - constructor = decoder_.construct_property - elif construct == XMLConstructables.RANGE: - constructor = decoder_.construct_range - elif construct == XMLConstructables.REFERENCE_ELEMENT: - constructor = decoder_.construct_reference_element - elif construct == XMLConstructables.RELATIONSHIP_ELEMENT: - constructor = decoder_.construct_relationship_element - elif construct == XMLConstructables.SUBMODEL_ELEMENT_COLLECTION: - constructor = decoder_.construct_submodel_element_collection - elif construct == XMLConstructables.SUBMODEL_ELEMENT_LIST: - constructor = decoder_.construct_submodel_element_list - elif construct == XMLConstructables.ASSET_ADMINISTRATION_SHELL: - constructor = decoder_.construct_asset_administration_shell - elif construct == XMLConstructables.ASSET_INFORMATION: - constructor = decoder_.construct_asset_information - elif construct == XMLConstructables.SPECIFIC_ASSET_ID: - constructor = decoder_.construct_specific_asset_id - elif construct == XMLConstructables.SUBMODEL: - constructor = decoder_.construct_submodel - elif construct == XMLConstructables.VALUE_REFERENCE_PAIR: - constructor = decoder_.construct_value_reference_pair - elif construct == XMLConstructables.CONCEPT_DESCRIPTION: - constructor = decoder_.construct_concept_description - elif construct == XMLConstructables.MULTI_LANGUAGE_NAME_TYPE: - constructor = decoder_.construct_multi_language_name_type - elif construct == XMLConstructables.MULTI_LANGUAGE_TEXT_TYPE: - constructor = decoder_.construct_multi_language_text_type - elif construct == XMLConstructables.DEFINITION_TYPE_IEC61360: - constructor = decoder_.construct_definition_type_iec61360 - elif construct == XMLConstructables.PREFERRED_NAME_TYPE_IEC61360: - constructor = decoder_.construct_preferred_name_type_iec61360 - elif construct == XMLConstructables.SHORT_NAME_TYPE_IEC61360: - constructor = decoder_.construct_short_name_type_iec61360 - elif construct == XMLConstructables.EMBEDDED_DATA_SPECIFICATION: - constructor = decoder_.construct_embedded_data_specification - elif construct == XMLConstructables.DATA_SPECIFICATION_IEC61360: - constructor = decoder_.construct_data_specification_iec61360 - # the following constructors decide which constructor to call based on the elements tag - elif construct == XMLConstructables.DATA_ELEMENT: - constructor = decoder_.construct_data_element - elif construct == XMLConstructables.SUBMODEL_ELEMENT: - constructor = decoder_.construct_submodel_element - elif construct == XMLConstructables.DATA_SPECIFICATION_CONTENT: - constructor = decoder_.construct_data_specification_content - # type aliases - elif construct == XMLConstructables.VALUE_LIST: - constructor = decoder_.construct_value_list - else: + + type_constructors: Dict[XMLConstructables, Callable[..., object]] = { + XMLConstructables.KEY: decoder_.construct_key, + XMLConstructables.REFERENCE: decoder_.construct_reference, + XMLConstructables.MODEL_REFERENCE: decoder_.construct_model_reference, + XMLConstructables.GLOBAL_REFERENCE: decoder_.construct_external_reference, + XMLConstructables.ADMINISTRATIVE_INFORMATION: decoder_.construct_administrative_information, + XMLConstructables.QUALIFIER: decoder_.construct_qualifier, + XMLConstructables.ANNOTATED_RELATIONSHIP_ELEMENT: decoder_.construct_annotated_relationship_element, + XMLConstructables.BASIC_EVENT_ELEMENT: decoder_.construct_basic_event_element, + XMLConstructables.BLOB: decoder_.construct_blob, + XMLConstructables.CAPABILITY: decoder_.construct_capability, + XMLConstructables.ENTITY: decoder_.construct_entity, + XMLConstructables.EXTENSION: decoder_.construct_extension, + XMLConstructables.FILE: decoder_.construct_file, + XMLConstructables.RESOURCE: decoder_.construct_resource, + XMLConstructables.MULTI_LANGUAGE_PROPERTY: decoder_.construct_multi_language_property, + XMLConstructables.OPERATION: decoder_.construct_operation, + XMLConstructables.PROPERTY: decoder_.construct_property, + XMLConstructables.RANGE: decoder_.construct_range, + XMLConstructables.REFERENCE_ELEMENT: decoder_.construct_reference_element, + XMLConstructables.RELATIONSHIP_ELEMENT: decoder_.construct_relationship_element, + XMLConstructables.SUBMODEL_ELEMENT_COLLECTION: decoder_.construct_submodel_element_collection, + XMLConstructables.SUBMODEL_ELEMENT_LIST: decoder_.construct_submodel_element_list, + XMLConstructables.ASSET_ADMINISTRATION_SHELL: decoder_.construct_asset_administration_shell, + XMLConstructables.ASSET_INFORMATION: decoder_.construct_asset_information, + XMLConstructables.SPECIFIC_ASSET_ID: decoder_.construct_specific_asset_id, + XMLConstructables.SUBMODEL: decoder_.construct_submodel, + XMLConstructables.VALUE_REFERENCE_PAIR: decoder_.construct_value_reference_pair, + XMLConstructables.CONCEPT_DESCRIPTION: decoder_.construct_concept_description, + XMLConstructables.MULTI_LANGUAGE_NAME_TYPE: decoder_.construct_multi_language_name_type, + XMLConstructables.MULTI_LANGUAGE_TEXT_TYPE: decoder_.construct_multi_language_text_type, + XMLConstructables.DEFINITION_TYPE_IEC61360: decoder_.construct_definition_type_iec61360, + XMLConstructables.PREFERRED_NAME_TYPE_IEC61360: decoder_.construct_preferred_name_type_iec61360, + XMLConstructables.SHORT_NAME_TYPE_IEC61360: decoder_.construct_short_name_type_iec61360, + XMLConstructables.EMBEDDED_DATA_SPECIFICATION: decoder_.construct_embedded_data_specification, + XMLConstructables.DATA_SPECIFICATION_IEC61360: decoder_.construct_data_specification_iec61360, + # the following constructors decide which constructor to call based on the elements tag + XMLConstructables.DATA_ELEMENT: decoder_.construct_data_element, + XMLConstructables.SUBMODEL_ELEMENT: decoder_.construct_submodel_element, + XMLConstructables.DATA_SPECIFICATION_CONTENT: decoder_.construct_data_specification_content, + # type aliases + XMLConstructables.VALUE_LIST: decoder_.construct_value_list, + } + + constructor: Optional[Callable[..., object]] = type_constructors.get(construct) + if constructor is None: raise ValueError(f"{construct.name} cannot be constructed!") element = _parse_xml_document(file, failsafe=decoder_.failsafe) @@ -1413,13 +1401,11 @@ def read_aas_xml_file_into(object_store: model.AbstractObjectStore[model.Identif decoder_ = _select_decoder(failsafe, stripped, decoder) element_constructors: Dict[str, Callable[..., model.Identifiable]] = { - "assetAdministrationShell": decoder_.construct_asset_administration_shell, - "conceptDescription": decoder_.construct_concept_description, - "submodel": decoder_.construct_submodel + f"{NS_AAS}assetAdministrationShell": decoder_.construct_asset_administration_shell, + f"{NS_AAS}conceptDescription": decoder_.construct_concept_description, + f"{NS_AAS}submodel": decoder_.construct_submodel } - element_constructors = {NS_AAS + k: v for k, v in element_constructors.items()} - root = _parse_xml_document(file, failsafe=decoder_.failsafe, **parser_kwargs) if root is None: diff --git a/basyx/aas/adapter/xml/xml_serialization.py b/basyx/aas/adapter/xml/xml_serialization.py index 106535ae1..a621d1d06 100644 --- a/basyx/aas/adapter/xml/xml_serialization.py +++ b/basyx/aas/adapter/xml/xml_serialization.py @@ -83,54 +83,82 @@ def abstract_classes_to_xml(tag: str, obj: object) -> etree.Element: """ elm = _generate_element(tag) if isinstance(obj, model.HasExtension): - if obj.extension: - et_extension = _generate_element(NS_AAS + "extensions") - for extension in obj.extension: - if isinstance(extension, model.Extension): - et_extension.append(extension_to_xml(extension, tag=NS_AAS + "extension")) - elm.append(et_extension) + _extend_with_has_extension_attrs(elm, obj) if isinstance(obj, model.Referable): - if obj.category: - elm.append(_generate_element(name=NS_AAS + "category", text=obj.category)) - if obj.id_short and not isinstance(obj.parent, model.SubmodelElementList): - elm.append(_generate_element(name=NS_AAS + "idShort", text=obj.id_short)) - if obj.display_name: - elm.append(lang_string_set_to_xml(obj.display_name, tag=NS_AAS + "displayName")) - if obj.description: - elm.append(lang_string_set_to_xml(obj.description, tag=NS_AAS + "description")) + _extend_with_referable_attrs(elm, obj) if isinstance(obj, model.Identifiable): - if obj.administration: - elm.append(administrative_information_to_xml(obj.administration)) - elm.append(_generate_element(name=NS_AAS + "id", text=obj.id)) + _extend_with_identifiable_attrs(elm, obj) if isinstance(obj, model.HasKind): - if obj.kind is model.ModellingKind.TEMPLATE: - elm.append(_generate_element(name=NS_AAS + "kind", text="Template")) - else: - # then modelling-kind is Instance - elm.append(_generate_element(name=NS_AAS + "kind", text="Instance")) + _extend_with_has_kind_attrs(elm, obj) if isinstance(obj, model.HasSemantics): - if obj.semantic_id: - elm.append(reference_to_xml(obj.semantic_id, tag=NS_AAS+"semanticId")) - if obj.supplemental_semantic_id: - et_supplemental_semantic_ids = _generate_element(NS_AAS + "supplementalSemanticIds") - for supplemental_semantic_id in obj.supplemental_semantic_id: - et_supplemental_semantic_ids.append(reference_to_xml(supplemental_semantic_id, NS_AAS+"reference")) - elm.append(et_supplemental_semantic_ids) + _extend_with_has_semantics_attrs(elm, obj) if isinstance(obj, model.Qualifiable): - if obj.qualifier: - et_qualifier = _generate_element(NS_AAS + "qualifiers") - for qualifier in obj.qualifier: - et_qualifier.append(qualifier_to_xml(qualifier, tag=NS_AAS+"qualifier")) - elm.append(et_qualifier) + _extend_with_qualifiable_attrs(elm, obj) if isinstance(obj, model.HasDataSpecification): - if obj.embedded_data_specifications: - et_embedded_data_specifications = _generate_element(NS_AAS + "embeddedDataSpecifications") - for eds in obj.embedded_data_specifications: - et_embedded_data_specifications.append(embedded_data_specification_to_xml(eds)) - elm.append(et_embedded_data_specifications) + _extend_with_has_data_specification_attrs(elm, obj) return elm +def _extend_with_has_extension_attrs(elm: etree.Element, obj: model.HasExtension): + if obj.extension: + et_extension = _generate_element(NS_AAS + "extensions") + for extension in obj.extension: + if isinstance(extension, model.Extension): + et_extension.append(extension_to_xml(extension, tag=NS_AAS + "extension")) + elm.append(et_extension) + + +def _extend_with_referable_attrs(elm: etree.Element, obj: model.Referable): + if obj.category: + elm.append(_generate_element(name=NS_AAS + "category", text=obj.category)) + if obj.id_short and not isinstance(obj.parent, model.SubmodelElementList): + elm.append(_generate_element(name=NS_AAS + "idShort", text=obj.id_short)) + if obj.display_name: + elm.append(lang_string_set_to_xml(obj.display_name, tag=NS_AAS + "displayName")) + if obj.description: + elm.append(lang_string_set_to_xml(obj.description, tag=NS_AAS + "description")) + + +def _extend_with_identifiable_attrs(elm: etree.Element, obj: model.Identifiable): + if obj.administration: + elm.append(administrative_information_to_xml(obj.administration)) + elm.append(_generate_element(name=NS_AAS + "id", text=obj.id)) + + +def _extend_with_has_kind_attrs(elm: etree.Element, obj: model.HasKind): + if obj.kind is model.ModellingKind.TEMPLATE: + elm.append(_generate_element(name=NS_AAS + "kind", text="Template")) + else: + # then modelling-kind is Instance + elm.append(_generate_element(name=NS_AAS + "kind", text="Instance")) + + +def _extend_with_has_semantics_attrs(elm: etree.Element, obj: model.HasSemantics): + if obj.semantic_id: + elm.append(reference_to_xml(obj.semantic_id, tag=NS_AAS + "semanticId")) + if obj.supplemental_semantic_id: + et_supplemental_semantic_ids = _generate_element(NS_AAS + "supplementalSemanticIds") + for supplemental_semantic_id in obj.supplemental_semantic_id: + et_supplemental_semantic_ids.append(reference_to_xml(supplemental_semantic_id, NS_AAS + "reference")) + elm.append(et_supplemental_semantic_ids) + + +def _extend_with_qualifiable_attrs(elm: etree.Element, obj: model.Qualifiable): + if obj.qualifier: + et_qualifier = _generate_element(NS_AAS + "qualifiers") + for qualifier in obj.qualifier: + et_qualifier.append(qualifier_to_xml(qualifier, tag=NS_AAS + "qualifier")) + elm.append(et_qualifier) + + +def _extend_with_has_data_specification_attrs(elm: etree.Element, obj: model.HasDataSpecification): + if obj.embedded_data_specifications: + et_embedded_data_specifications = _generate_element(NS_AAS + "embeddedDataSpecifications") + for eds in obj.embedded_data_specifications: + et_embedded_data_specifications.append(embedded_data_specification_to_xml(eds)) + elm.append(et_embedded_data_specifications) + + # ############################################################## # transformation functions to serialize classes from model.base # ############################################################## diff --git a/basyx/aas/backend/couchdb.py b/basyx/aas/backend/couchdb.py index ec7caec1b..8e582ee96 100644 --- a/basyx/aas/backend/couchdb.py +++ b/basyx/aas/backend/couchdb.py @@ -244,7 +244,6 @@ def __init__(self, url: str, database: str): :param url: URL to the CouchDB :param database: Name of the Database inside the CouchDB """ - super().__init__() self.url: str = url self.database_name: str = database diff --git a/basyx/aas/backend/local_file.py b/basyx/aas/backend/local_file.py index c521fa32e..867c55b8b 100644 --- a/basyx/aas/backend/local_file.py +++ b/basyx/aas/backend/local_file.py @@ -77,7 +77,6 @@ def __init__(self, directory_path: str): :param directory_path: Path to the local file backend (the path where you want to store your AAS JSON files) """ - super().__init__() self.directory_path: str = directory_path.rstrip("/") # A dictionary of weak references to local replications of stored objects. Objects are kept in this cache as diff --git a/basyx/aas/model/base.py b/basyx/aas/model/base.py index dc3079625..e3b798583 100644 --- a/basyx/aas/model/base.py +++ b/basyx/aas/model/base.py @@ -2338,7 +2338,6 @@ def __init__(self, value: Optional[ValueTypeIEC61360] = None, level_types: Iterable[IEC61360LevelType] = ()): - super().__init__() self.preferred_name: PreferredNameTypeIEC61360 = preferred_name self.short_name: Optional[ShortNameTypeIEC61360] = short_name self.data_type: Optional[DataTypeIEC61360] = data_type diff --git a/basyx/aas/model/provider.py b/basyx/aas/model/provider.py index 311618fea..d5658abc5 100644 --- a/basyx/aas/model/provider.py +++ b/basyx/aas/model/provider.py @@ -86,7 +86,6 @@ class DictObjectStore(AbstractObjectStore[_IT], Generic[_IT]): :class:`~aas.model.base.Identifier` → :class:`~aas.model.base.Identifiable` """ def __init__(self, objects: Iterable[_IT] = ()) -> None: - super().__init__() self._backend: Dict[Identifier, _IT] = {} for x in objects: self.add(x)