diff --git a/sdk/basyx/aas/adapter/_ordering.py b/sdk/basyx/aas/adapter/_ordering.py new file mode 100644 index 000000000..86d60db36 --- /dev/null +++ b/sdk/basyx/aas/adapter/_ordering.py @@ -0,0 +1,103 @@ +# Copyright (c) 2026 the Eclipse BaSyx Authors +# +# This program and the accompanying materials are made available under the terms of the MIT License, available in +# the LICENSE file of this project. +# +# SPDX-License-Identifier: MIT +from typing import Optional, Tuple + +from basyx.aas import model + +from . import _generic + + +def reference_sort_key(obj: Optional[model.Reference]) -> Tuple: + """ + Sort key of a Reference, consisting of its type, its keys and its referredSemanticId + + :param obj: object of class Reference or ``None`` + :return: tuple which totally orders References, ``()`` for ``None`` + """ + if obj is None: + return () + return ( + _generic.REFERENCE_TYPES[obj.__class__], + tuple((_generic.KEY_TYPES[key.type], key.value) for key in obj.key), + reference_sort_key(obj.referred_semantic_id), + ) + + +def specific_asset_id_sort_key(obj: model.SpecificAssetId) -> Tuple: + """ + Sort key of a SpecificAssetId, consisting of all its attributes, since its name is not unique + + :param obj: object of class SpecificAssetId + :return: tuple which totally orders SpecificAssetIds + """ + return ( + obj.name, + obj.value, + reference_sort_key(obj.external_subject_id), + reference_sort_key(obj.semantic_id), + tuple(sorted(reference_sort_key(ref) for ref in obj.supplemental_semantic_id)), + ) + + +def embedded_data_specification_sort_key(obj: model.EmbeddedDataSpecification) -> Tuple: + """ + Sort key of an EmbeddedDataSpecification, consisting of its dataSpecification + + :param obj: object of class EmbeddedDataSpecification + :return: tuple which totally orders EmbeddedDataSpecifications + """ + return reference_sort_key(obj.data_specification) + + +def value_reference_pair_sort_key(obj: model.ValueReferencePair) -> Tuple: + """ + Sort key of a ValueReferencePair, consisting of its value and its valueId, since its value is not unique + + :param obj: object of class ValueReferencePair + :return: tuple which totally orders ValueReferencePairs + """ + return model.datatypes.xsd_repr(obj.value), reference_sort_key(obj.value_id) + + +def extension_sort_key(obj: model.Extension) -> str: + """ + Sort key of an Extension, its name, which is unique (Constraint AASd-077) + + :param obj: object of class Extension + :return: name of the Extension + """ + return obj.name + + +def qualifier_sort_key(obj: model.Qualifier) -> str: + """ + Sort key of a Qualifier, its type, which is unique (Constraint AASd-021) + + :param obj: object of class Qualifier + :return: type of the Qualifier + """ + return obj.type + + +def referable_sort_key(obj: model.Referable) -> str: + """ + Sort key of a Referable within its namespace, its idShort, which is unique (Constraint AASd-022) + + :param obj: object of class Referable + :return: idShort of the Referable, ``""`` if it has none + """ + return obj.id_short or "" + + +def identifiable_sort_key(obj: model.Identifiable) -> str: + """ + Sort key of an Identifiable, its globally unique id + + :param obj: object of class Identifiable + :return: id of the Identifiable + """ + return obj.id diff --git a/sdk/basyx/aas/adapter/json/__init__.py b/sdk/basyx/aas/adapter/json/__init__.py index d2431c211..89e3b3f0e 100644 --- a/sdk/basyx/aas/adapter/json/__init__.py +++ b/sdk/basyx/aas/adapter/json/__init__.py @@ -17,6 +17,10 @@ :class:`ObjectStore `. """ +from .formats.metadata_serialization import MetadataJsonEncoder +from .formats.path_serialization import PathJsonEncoder +from .formats.reference_serialization import ReferenceJsonEncoder +from .formats.value_only_serialization import ValueOnlyJsonEncoder from .json_deserialization import ( AASFromJsonDecoder, StrictAASFromJsonDecoder, @@ -43,4 +47,8 @@ "StrippedAASToJsonEncoder", "object_store_to_json", "write_aas_json_file", + "PathJsonEncoder", + "ValueOnlyJsonEncoder", + "MetadataJsonEncoder", + "ReferenceJsonEncoder", ] diff --git a/sdk/basyx/aas/adapter/json/formats/metadata_serialization.py b/sdk/basyx/aas/adapter/json/formats/metadata_serialization.py new file mode 100644 index 000000000..1ce714c77 --- /dev/null +++ b/sdk/basyx/aas/adapter/json/formats/metadata_serialization.py @@ -0,0 +1,140 @@ +# Copyright (c) 2026 the Eclipse BaSyx Authors +# +# This program and the accompanying materials are made available under the terms of the MIT License, available in +# the LICENSE file of this project. +# +# SPDX-License-Identifier: MIT +from typing import Callable, Dict, Type + +from basyx.aas import model + +from ..json_serialization import AASToJsonEncoder + + +class MetadataJsonEncoder(AASToJsonEncoder): + """ + Custom JSONEncoder that serializes BaSyx Python SDK objects into the JSON format "Metadata". This equals + the normal serialization format but omits certain keys. + """ + + @classmethod + def _get_aas_class_serializers(cls) -> Dict[Type, Callable]: + mapping = super()._get_aas_class_serializers() + + mapping.pop(model.AssetInformation, None) + mapping.pop(model.Resource, None) + mapping.pop(model.SpecificAssetId, None) + mapping.pop(model.ConceptDescription, None) + + return mapping + + @classmethod + def _asset_administration_shell_to_json( + cls, obj: model.AssetAdministrationShell + ) -> Dict[str, object]: + data = super()._asset_administration_shell_to_json(obj) + data.pop("assetInformation", None) + data.pop("submodels", None) + return data + + @classmethod + def _submodel_to_json( + cls, obj: model.Submodel + ) -> Dict[str, object]: + data = super()._submodel_to_json(obj) + data.pop("submodelElements", None) + return data + + @classmethod + def _submodel_element_collection_to_json( + cls, obj: model.SubmodelElementCollection + ) -> Dict[str, object]: + data = super()._submodel_element_collection_to_json(obj) + data.pop("value", None) + return data + + @classmethod + def _submodel_element_list_to_json( + cls, obj: model.SubmodelElementList + ) -> Dict[str, object]: + data = super()._submodel_element_list_to_json(obj) + data.pop("value", None) + return data + + @classmethod + def _entity_to_json(cls, obj: model.Entity) -> Dict[str, object]: + data = super()._entity_to_json(obj) + data.pop("statements", None) + data.pop("globalAssetId", None) + data.pop("specificAssetIds", None) + return data + + @classmethod + def _basic_event_element_to_json( + cls, obj: model.BasicEventElement + ) -> Dict[str, object]: + data = super()._basic_event_element_to_json(obj) + data.pop("observed", None) + return data + + @classmethod + def _property_to_json(cls, obj: model.Property) -> Dict[str, object]: + data = super()._property_to_json(obj) + data.pop("value", None) + data.pop("valueId", None) + return data + + @classmethod + def _multi_language_property_to_json(cls, obj: model.MultiLanguageProperty) -> Dict[str, object]: + data = super()._multi_language_property_to_json(obj) + data.pop("value", None) + data.pop("valueId", None) + return data + + @classmethod + def _range_to_json(cls, obj: model.Range) -> Dict[str, object]: + data = super()._range_to_json(obj) + data.pop("min", None) + data.pop("max", None) + return data + + @classmethod + def _reference_element_to_json( + cls, obj: model.ReferenceElement + ) -> Dict[str, object]: + data = super()._reference_element_to_json(obj) + data.pop("value", None) + return data + + @classmethod + def _relationship_element_to_json( + cls, obj: model.RelationshipElement + ) -> Dict[str, object]: + data = super()._relationship_element_to_json(obj) + data.pop("first", None) + data.pop("second", None) + return data + + @classmethod + def _annotated_relationship_element_to_json( + cls, obj: model.AnnotatedRelationshipElement + ) -> Dict[str, object]: + data = super()._annotated_relationship_element_to_json(obj) + data.pop("first", None) + data.pop("second", None) + data.pop("annotations", None) + return data + + @classmethod + def _blob_to_json(cls, obj: model.Blob) -> Dict[str, object]: + data = super()._blob_to_json(obj) + data.pop("value", None) + data.pop("contentType", None) + return data + + @classmethod + def _file_to_json(cls, obj: model.File) -> Dict[str, object]: + data = super()._file_to_json(obj) + data.pop("value", None) + data.pop("contentType", None) + return data diff --git a/sdk/basyx/aas/adapter/json/formats/path_serialization.py b/sdk/basyx/aas/adapter/json/formats/path_serialization.py new file mode 100644 index 000000000..564737c5e --- /dev/null +++ b/sdk/basyx/aas/adapter/json/formats/path_serialization.py @@ -0,0 +1,108 @@ +# Copyright (c) 2026 the Eclipse BaSyx Authors +# +# This program and the accompanying materials are made available under the terms of the MIT License, available in +# the LICENSE file of this project. +# +# SPDX-License-Identifier: MIT + +from typing import Iterable, List, Tuple + +from basyx.aas import model +from basyx.aas.adapter import _ordering + +from ..json_serialization import AASToJsonEncoder + +# Types for which the $path content modifier returns the idShort of the requested element and a list +# of idShort paths to its child elements (AAS API Part 2). +_PATH_SUPPORTED_TYPES: Tuple[type, ...] = ( + model.Submodel, + model.SubmodelElementCollection, + model.SubmodelElementList, + model.AnnotatedRelationshipElement, + model.Entity, +) + + +class PathJsonEncoder(AASToJsonEncoder): + """ + Custom JSONEncoder that serializes selected AAS objects as the list of idShortPaths reachable + from them (``$path`` content modifier). Supported inputs are :class:`~basyx.aas.model.submodel.Submodel`, + :class:`~basyx.aas.model.submodel.SubmodelElementCollection`, + :class:`~basyx.aas.model.submodel.SubmodelElementList`, + :class:`~basyx.aas.model.submodel.AnnotatedRelationshipElement`, and :class:`~basyx.aas.model.submodel.Entity`. + Any other type raises :class:`TypeError`. + + The idShortPaths of children are listed in a canonical order: Elements of a SubmodelElementList in the order of + the list, all other children sorted by their idShort. + + :param deep: If True (default), all descendants on all sublevels are included. If False, only + the requested element and its direct children are included. Compare to + SerializationModifier level. + """ + + def __init__(self, *args, deep: bool = True, **kwargs): + super().__init__(*args, **kwargs) + self.deep = deep + + def default(self, obj: object) -> object: + if isinstance(obj, _PATH_SUPPORTED_TYPES): + return self._referable_to_path(obj) + raise TypeError( + f"Object of type {type(obj).__name__} has no $path representation; " + f"supported types are: {', '.join(t.__name__ for t in _PATH_SUPPORTED_TYPES)}." + ) + + def _referable_to_path(self, obj: model.Referable) -> List[str]: + if isinstance(obj, model.Submodel): + # Submodel is Identifiable; its idShort is optional and never part of an idShortPath. + paths: List[str] = [] + self._append_children(paths, obj.submodel_element, prefix=[]) + return paths + # For the remaining supported types the idShort of the requested element is the root of + # every returned idShortPath. If it is missing, no path can be computed from this element. + if obj.id_short is None: + raise ValueError(f"{obj!r} has no idShort, so no idShortPath can be computed for it!") + prefix = [obj.id_short] + paths = [model.Referable.build_id_short_path(prefix)] + if isinstance(obj, model.SubmodelElementList): + self._append_indexed_children(paths, obj.value, prefix) + elif isinstance(obj, model.SubmodelElementCollection): + self._append_children(paths, obj.value, prefix) + elif isinstance(obj, model.Entity): + self._append_children(paths, obj.statement, prefix) + elif isinstance(obj, model.AnnotatedRelationshipElement): + self._append_children(paths, obj.annotation, prefix) + return paths + + def _append_children( + self, paths: List[str], children: Iterable[model.SubmodelElement], prefix: List[str] + ) -> None: + for child in sorted(children, key=_ordering.referable_sort_key): + if child.id_short is None: + raise ValueError( + f"{child!r} has no idShort, so no idShortPath can be computed for it!" + ) + child_prefix = prefix + [child.id_short] + paths.append(model.Referable.build_id_short_path(child_prefix)) + if self.deep: + self._recurse(paths, child, child_prefix) + + def _append_indexed_children( + self, paths: List[str], children: Iterable[model.SubmodelElement], prefix: List[str] + ) -> None: + for index, child in enumerate(children): + child_prefix = prefix + [str(index)] + paths.append(model.Referable.build_id_short_path(child_prefix)) + if self.deep: + self._recurse(paths, child, child_prefix) + + def _recurse(self, paths: List[str], element: object, prefix: List[str]) -> None: + if isinstance(element, model.SubmodelElementList): + self._append_indexed_children(paths, element.value, prefix) + elif isinstance(element, model.SubmodelElementCollection): + self._append_children(paths, element.value, prefix) + elif isinstance(element, model.Entity): + self._append_children(paths, element.statement, prefix) + elif isinstance(element, model.AnnotatedRelationshipElement): + self._append_children(paths, element.annotation, prefix) + # Leaf elements have no children to traverse into. diff --git a/sdk/basyx/aas/adapter/json/formats/reference_serialization.py b/sdk/basyx/aas/adapter/json/formats/reference_serialization.py new file mode 100644 index 000000000..227112771 --- /dev/null +++ b/sdk/basyx/aas/adapter/json/formats/reference_serialization.py @@ -0,0 +1,42 @@ +# Copyright (c) 2026 the Eclipse BaSyx Authors +# +# This program and the accompanying materials are made available under the terms of the MIT License, available in +# the LICENSE file of this project. +# +# SPDX-License-Identifier: MIT +from typing import Callable, Dict, Type + +from basyx.aas import model + +from ..json_serialization import AASToJsonEncoder + + +class ReferenceJsonEncoder(AASToJsonEncoder): + """ + Custom JSONEncoder that serializes :class:`~basyx.aas.model.base.Referable` objects, which are contained + in an :class:`~basyx.aas.model.base.Identifiable`, as their :class:`~basyx.aas.model.base.ModelReference`. + """ + + @classmethod + def _get_aas_class_serializers(cls) -> Dict[Type, Callable]: + normal_serializers = super()._get_aas_class_serializers() + serializers: Dict[Type, Callable] = {} + for typ in normal_serializers: + if not issubclass(typ, model.Referable): + serializers[typ] = cls._no_referable + + serializers.update({ + model.Reference: normal_serializers[model.Reference], + model.Key: normal_serializers[model.Key], + model.Referable: cls._referable_to_reference, + }) + + return serializers + + @classmethod + def _referable_to_reference(cls, obj: model.Referable) -> model.Reference: + return model.ModelReference.from_referable(obj) + + @classmethod + def _no_referable(cls, obj: object): + raise TypeError(f"Object of type {obj.__class__.__name__} has no Reference serialization") diff --git a/sdk/basyx/aas/adapter/json/formats/value_only_serialization.py b/sdk/basyx/aas/adapter/json/formats/value_only_serialization.py new file mode 100644 index 000000000..6bb66a3bb --- /dev/null +++ b/sdk/basyx/aas/adapter/json/formats/value_only_serialization.py @@ -0,0 +1,295 @@ +# Copyright (c) 2026 the Eclipse BaSyx Authors +# +# This program and the accompanying materials are made available under the terms of the MIT License, available in +# the LICENSE file of this project. +# +# SPDX-License-Identifier: MIT +import base64 +import math +from typing import Callable, Iterable, Union + +from basyx.aas import model +from basyx.aas.adapter import _ordering, _generic + +from ..json_serialization import AASToJsonEncoder + +TYPES_WITHOUT_VALUE_ONLY = (model.Capability, model.Operation) +CONTAINER_TYPES = (model.SubmodelElementCollection, model.SubmodelElementList) + + +class ValueOnlyJsonEncoder(AASToJsonEncoder): + """ + Custom JSONEncoder that serializes BaSyx Python SDK objects into the JSON format "ValueOnly" as defined by the + specification. + + .. note:: + + The specification contradicts itself at some points. In these cases the following assumptions were made: + + * ``annotations`` of an AnnotatedRelationshipElement and ``statements`` of an Entity are serialized as JSON + object with the idShorts as names, as defined in the schema and the Entity example, instead of an array, + as stated in the text. + * ``specificAssetIds`` of an Entity are serialized as array of SpecificAssetIds in format "Normal", as stated + in the text, instead of objects with string values only, as defined in the schema. + * ``globalAssetId`` of an Entity is serialized as string, as defined in the schema and the metamodel, instead + of a Reference, as shown in the Entity example. + + Like in :class:`~basyx.aas.adapter.json.json_serialization.AASToJsonEncoder`, unordered collections are serialized + in a canonical order: The JSON objects with idShorts as names are sorted by idShort and the ``specificAssetIds`` + of an Entity are sorted by all their attributes. The value of a SubmodelElementList keeps its order. + + :cvar with_blob_value: Inherited from :class:`~basyx.aas.adapter.json.json_serialization.AASToJsonEncoder`. + :param deep: Inherited from :class:`~basyx.aas.adapter.json.json_serialization.AASToJsonEncoder`. + """ + + @classmethod + def _get_aas_class_serializers(cls) -> dict[type, Callable]: + serializers = super()._get_aas_class_serializers() + + serializers.update( + { + model.Submodel: cls._submodel_to_value_only, + model.SubmodelElementList: cls._submodel_element_list_to_value_only, + model.SubmodelElementCollection: cls._submodel_element_collection_to_value_only, + model.Property: cls._property_to_value_only, + model.MultiLanguageProperty: cls._multi_language_property_to_value_only, + model.Range: cls._range_to_value_only, + model.File: cls._file_to_value_only, + model.Blob: cls._blob_to_value_only, + model.ReferenceElement: cls._reference_element_to_value_only, + model.RelationshipElement: cls._relationship_element_to_value_only, + model.AnnotatedRelationshipElement: cls._annotated_relationship_element_to_value_only, + model.Entity: cls._entity_to_value_only, + model.BasicEventElement: cls._basic_event_element_to_value_only, + } + ) + # Capability and Operation have no ValueOnly representation + for key in TYPES_WITHOUT_VALUE_ONLY: + serializers.pop(key, None) + + return serializers + + @classmethod + def _get_aas_class_no_children_serializers(cls) -> dict[type, Callable]: + serializers = super()._get_aas_class_no_children_serializers() + + serializers.update( + { + model.Submodel: cls._submodel_to_value_only_no_children, + model.SubmodelElementList: cls._submodel_element_list_to_value_only_no_children, + model.SubmodelElementCollection: cls._submodel_element_collection_to_value_only_no_children, + model.AnnotatedRelationshipElement: cls._annotated_relationship_element_to_value_only_no_children, + model.Entity: cls._entity_to_value_only_no_children, + } + ) + + return serializers + + @classmethod + def _has_value(cls, obj: model.SubmodelElement) -> bool: + if isinstance(obj, TYPES_WITHOUT_VALUE_ONLY): + return False + if isinstance(obj, model.Property): + return obj.value is not None + if isinstance(obj, model.MultiLanguageProperty): + return bool(obj.value) + if isinstance(obj, model.Range): + return obj.min is not None or obj.max is not None + if isinstance(obj, model.File): + return obj.value is not None and obj.content_type is not None + if isinstance(obj, model.Blob): + return obj.content_type is not None + if isinstance(obj, model.ReferenceElement): + return obj.value is not None + if isinstance(obj, model.RelationshipElement): + return obj.first is not None or obj.second is not None + if isinstance(obj, model.Entity): + return (obj.global_asset_id is not None or bool(obj.specific_asset_id) + or any(cls._has_value(statement) for statement in obj.statement)) + if isinstance(obj, CONTAINER_TYPES): + return any(cls._has_value(element) for element in obj.value) + return True + + @classmethod + def _submodel_to_value_only(cls, obj: model.Submodel) -> dict[str, object]: + return cls._submodel_element_iterable_to_value_only(obj.submodel_element) + + @classmethod + def _submodel_element_list_to_value_only( + cls, obj: model.SubmodelElementList + ) -> list[object]: + return [element for element in obj.value] + + @classmethod + def _submodel_element_iterable_to_value_only(cls, it: Iterable[model.SubmodelElement]) -> dict[str, object]: + return { + element.id_short: element + for element in sorted(it, key=_ordering.referable_sort_key) + if cls._has_value(element) and element.id_short is not None + } + + @classmethod + def _submodel_element_collection_to_value_only(cls, obj: model.SubmodelElementCollection) -> dict[str, object]: + data: dict[str, object] = cls._submodel_element_iterable_to_value_only(obj.value) + return data + + @classmethod + def _value_to_json_type(cls, value: model.datatypes.AnyXSDType) -> Union[bool, str, int, float]: + if isinstance(value, (bool, model.datatypes.Boolean)): + return bool(value) + if isinstance(value, int): + return value + if isinstance(value, (float, model.datatypes.Decimal)): + if not math.isfinite(float(value)): + return model.datatypes.xsd_repr(value) + return float(value) + return model.datatypes.xsd_repr(value) + + @classmethod + def _property_to_value_only(cls, obj: model.Property) -> object: + if obj.value is not None: + return cls._value_to_json_type(obj.value) + return None + + @classmethod + def _multi_language_property_to_value_only(cls, obj: model.MultiLanguageProperty) -> list[object]: + value = obj.value + if not value: + return [] + return [{language: text} for language, text in sorted(value.items())] + + @classmethod + def _range_to_value_only(cls, obj: model.Range) -> Union[object, dict[str, object]]: + data: dict[str, object] = {} + if obj.min is not None: + data["min"] = cls._value_to_json_type(obj.min) + if obj.max is not None: + data["max"] = cls._value_to_json_type(obj.max) + return data + + @classmethod + def _file_to_value_only(cls, obj: model.File) -> dict[str, object]: + data: dict[str, object] = {} + data["contentType"] = obj.content_type + if obj.value is not None: + data["value"] = obj.value + return data + + @classmethod + def _blob_to_value_only(cls, obj: model.Blob) -> dict[str, object]: + data: dict[str, object] = {} + data["contentType"] = obj.content_type + if cls.with_blob_value and obj.value is not None: + data["value"] = base64.b64encode(obj.value).decode() + return data + + @classmethod + def _reference_element_to_value_only(cls, obj: model.ReferenceElement) -> object: + return obj.value + + @classmethod + def _relationship_element_to_value_only(cls, obj: model.RelationshipElement) -> dict[str, object]: + data: dict[str, object] = {} + if obj.first is not None: + data["first"] = obj.first + if obj.second is not None: + data["second"] = obj.second + return data + + @classmethod + def _annotated_relationship_element_to_value_only( + cls, obj: model.AnnotatedRelationshipElement + ) -> dict[str, object]: + data = cls._relationship_element_to_value_only(obj) + if obj.annotation: + data["annotations"] = cls._submodel_element_iterable_to_value_only(obj.annotation) + return data + + @classmethod + def _entity_to_value_only(cls, obj: model.Entity) -> object: + data: dict[str, object] = {} + if obj.statement: + data["statements"] = cls._submodel_element_iterable_to_value_only(obj.statement) + if obj.global_asset_id: + data["globalAssetId"] = obj.global_asset_id + if obj.specific_asset_id: + data["specificAssetIds"] = sorted(obj.specific_asset_id, key=_ordering.specific_asset_id_sort_key) + if obj.entity_type: + data["entityType"] = _generic.ENTITY_TYPES[obj.entity_type] + return data + + @classmethod + def _basic_event_element_to_value_only(cls, obj: model.BasicEventElement) -> dict[str, object]: + data: dict[str, object] = {"observed": obj.observed} + return data + + # ############################################################# + # transformation functions to serialize with deep=False + # ############################################################# + + @classmethod + def _remove_children(cls, obj: object) -> object: + # The ValueOnly representation of SubmodelElementCollections and SubmodelElementLists consists only of their + # children + if isinstance(obj, model.SubmodelElementCollection): + return {} + if isinstance(obj, model.SubmodelElementList): + return [] + return super()._remove_children(obj) + + @classmethod + def _remove_children_of_elements(cls, data: dict[str, object]) -> dict[str, object]: + # Sorted again by idShort, since the callers append SubmodelElementCollections and SubmodelElementLists + # without a value at the end + return {id_short: cls._remove_children(element) for id_short, element in sorted(data.items())} + + @classmethod + def _submodel_to_value_only_no_children(cls, obj: model.Submodel) -> dict[str, object]: + data = cls._get_aas_class_serializers()[model.Submodel](obj) + # Direct child SubmodelElementCollections and SubmodelElementLists are always serialized with an empty value, + # even if they contain no elements with a value + for element in obj.submodel_element: + if isinstance(element, CONTAINER_TYPES) and element.id_short is not None: + data.setdefault(element.id_short, element) + return cls._remove_children_of_elements(data) + + @classmethod + def _submodel_element_list_to_value_only_no_children(cls, obj: model.SubmodelElementList) -> list[object]: + data = cls._get_aas_class_serializers()[model.SubmodelElementList](obj) + return [cls._remove_children(element) for element in data] + + @classmethod + def _submodel_element_collection_to_value_only_no_children( + cls, obj: model.SubmodelElementCollection + ) -> dict[str, object]: + data = cls._get_aas_class_serializers()[model.SubmodelElementCollection](obj) + # Direct child SubmodelElementCollections and SubmodelElementLists are always serialized with an empty value, + # even if they contain no elements with a value + for element in obj.value: + if isinstance(element, CONTAINER_TYPES) and element.id_short is not None: + data.setdefault(element.id_short, element) + return cls._remove_children_of_elements(data) + + @classmethod + def _annotated_relationship_element_to_value_only_no_children( + cls, obj: model.AnnotatedRelationshipElement + ) -> dict[str, object]: + data = cls._get_aas_class_serializers()[model.AnnotatedRelationshipElement](obj) + annotations = data.get("annotations") + if isinstance(annotations, dict): + # No checking for child SubmodelElementCollections and SubmodelElementLists necessary + data["annotations"] = cls._remove_children_of_elements(annotations) + return data + + @classmethod + def _entity_to_value_only_no_children(cls, obj: model.Entity) -> dict[str, object]: + data = cls._get_aas_class_serializers()[model.Entity](obj) + statements = data.get("statements") + if isinstance(statements, dict): + # Direct child SubmodelElementCollections and SubmodelElementLists are always serialized with an empty + # value, even if they contain no elements with a value + for element in obj.statement: + if isinstance(element, CONTAINER_TYPES) and element.id_short is not None: + statements.setdefault(element.id_short, element) + data["statements"] = cls._remove_children_of_elements(statements) + return data diff --git a/sdk/basyx/aas/adapter/json/json_serialization.py b/sdk/basyx/aas/adapter/json/json_serialization.py index 155aced55..10508d654 100644 --- a/sdk/basyx/aas/adapter/json/json_serialization.py +++ b/sdk/basyx/aas/adapter/json/json_serialization.py @@ -35,18 +35,16 @@ Callable, ContextManager, Dict, - Iterable, List, Optional, TextIO, - Tuple, Type, get_args, ) from basyx.aas import model -from .. import _generic +from .. import _generic, _ordering from .._generic import JSON_AAS_TOP_LEVEL_KEYS_TO_TYPES @@ -64,12 +62,31 @@ class AASToJsonEncoder(json.JSONEncoder): json_string = json.dumps(data, cls=AASToJsonEncoder) + To extend the encoder, override the respective ``__to_json()`` methods or the mapping returned by + ``_get_aas_class_serializers()``. Override ``_get_aas_class_no_children_serializers()`` to adapt ``deep=False``. + + Collections which are unordered according to the specification (every multi-valued attribute not marked + ``<>``) are serialized in a canonical order, so that equal objects always result in identical JSON + strings. Only the keys of a :class:`~basyx.aas.model.base.Reference` and the value of a + :class:`~basyx.aas.model.submodel.SubmodelElementList` keep their order. + :cvar stripped: If True, the JSON objects will be serialized in a stripped manner, excluding some attributes. Defaults to ``False``. See https://git.rwth-aachen.de/acplt/pyi40aas/-/issues/91 + :cvar with_blob_value: If True, objects of type :class:`~basyx.aas.model.submodel.Blob` will be serialized + containing their value in base64 encoding. Otherwise, the ``value`` key is omitted. + Default to ``True``. + :param deep: If True (default), all children of the serialized object are included on all sublevels. If False, + only the serialized object and its direct children are included, while the children of these are + omitted. This corresponds to the serialization level ``core`` of the AAS API specification. """ stripped = False + with_blob_value = True + + def __init__(self, *args, deep: bool = True, **kwargs): + super().__init__(*args, **kwargs) + self.deep = deep @classmethod def _get_aas_class_serializers(cls) -> Dict[Type, Callable]: @@ -105,6 +122,28 @@ def _get_aas_class_serializers(cls) -> Dict[Type, Callable]: } return mapping + @classmethod + def _get_aas_class_no_children_serializers(cls) -> Dict[Type, Callable]: + mapping: Dict[Type, Callable] = { + model.AnnotatedRelationshipElement: cls._annotated_relationship_element_to_json_no_children, + model.Entity: cls._entity_to_json_no_children, + model.Submodel: cls._submodel_to_json_no_children, + model.SubmodelElementCollection: cls._submodel_element_collection_to_json_no_children, + model.SubmodelElementList: cls._submodel_element_list_to_json_no_children, + } + return mapping + + @classmethod + def _get_aas_class_children_keys(cls) -> Dict[Type, str]: + mapping: Dict[Type, str] = { + model.AnnotatedRelationshipElement: "annotations", + model.Entity: "statements", + model.Submodel: "submodelElements", + model.SubmodelElementCollection: "value", + model.SubmodelElementList: "value", + } + return mapping + def default(self, obj: object) -> object: """ The overwritten ``default`` method for :class:`json.JSONEncoder` @@ -112,6 +151,12 @@ def default(self, obj: object) -> object: :param obj: The object to serialize to json :return: The serialized object """ + if not self.deep: + no_children_mapping = self._get_aas_class_no_children_serializers() + for typ in no_children_mapping: + if isinstance(obj, typ): + no_children_mapping_method = no_children_mapping[typ] + return no_children_mapping_method(obj) mapping = self._get_aas_class_serializers() for typ in mapping: if isinstance(obj, typ): @@ -128,9 +173,29 @@ def _abstract_classes_to_json(cls, obj: object) -> Dict[str, object]: :return: dict with the serialized attributes of the abstract classes this object inherits from """ data: Dict[str, object] = {} + if isinstance(obj, model.Referable): + try: + ref_type = model.resolve_referable_class_in_key_types(obj) + 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__ + 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 + if isinstance(obj, model.HasExtension) and not cls.stripped: if obj.extension: - data["extensions"] = list(obj.extension) + data["extensions"] = sorted(obj.extension, key=_ordering.extension_sort_key) + if isinstance(obj, model.HasDataSpecification) and not cls.stripped: if obj.embedded_data_specifications: data["embeddedDataSpecifications"] = [ @@ -138,27 +203,12 @@ def _abstract_classes_to_json(cls, obj: object) -> Dict[str, object]: "dataSpecification": spec.data_specification, "dataSpecificationContent": spec.data_specification_content, } - for spec in obj.embedded_data_specifications + for spec in sorted( + obj.embedded_data_specifications, + key=lambda s: _ordering.embedded_data_specification_sort_key(s), + ) ] - 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 = model.resolve_referable_class_in_key_types(obj) - 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__ if isinstance(obj, model.Identifiable): data["id"] = obj.id if obj.administration: @@ -167,13 +217,18 @@ def _abstract_classes_to_json(cls, obj: object) -> Dict[str, object]: if obj.semantic_id: data["semanticId"] = obj.semantic_id if obj.supplemental_semantic_id: - data["supplementalSemanticIds"] = list(obj.supplemental_semantic_id) + data["supplementalSemanticIds"] = sorted( + obj.supplemental_semantic_id, key=_ordering.reference_sort_key + ) + if isinstance(obj, model.HasKind): if obj.kind is model.ModellingKind.TEMPLATE: data["kind"] = _generic.MODELLING_KIND[obj.kind] + if isinstance(obj, model.Qualifiable) and not cls.stripped: if obj.qualifier: - data["qualifiers"] = list(obj.qualifier) + data["qualifiers"] = sorted(obj.qualifier, key=_ordering.qualifier_sort_key) + return data # ############################################################# @@ -184,7 +239,7 @@ def _abstract_classes_to_json(cls, obj: object) -> Dict[str, object]: def _lang_string_set_to_json( cls, obj: model.LangStringSet ) -> List[Dict[str, object]]: - return [{"language": k, "text": v} for k, v in obj.items()] + return [{"language": k, "text": v} for k, v in sorted(obj.items())] @classmethod def _key_to_json(cls, obj: model.Key) -> Dict[str, object]: @@ -283,7 +338,7 @@ def _extension_to_json(cls, obj: model.Extension) -> Dict[str, object]: model.datatypes.xsd_repr(obj.value) if obj.value is not None else None ) if obj.refers_to: - data["refersTo"] = list(obj.refers_to) + data["refersTo"] = sorted(obj.refers_to, key=_ordering.reference_sort_key) if obj.value_type: data["valueType"] = model.datatypes.XSD_TYPE_NAMES[obj.value_type] data["name"] = obj.name @@ -313,7 +368,11 @@ def _value_list_to_json(cls, obj: model.ValueList) -> Dict[str, object]: :param obj: object of class ValueList :return: dict with the serialized attributes of this object """ - return {"valueReferencePairs": list(obj)} + return { + "valueReferencePairs": sorted( + obj, key=_ordering.value_reference_pair_sort_key + ) + } # ############################################################ # transformation functions to serialize classes from model.aas @@ -351,7 +410,9 @@ def _asset_information_to_json( if obj.global_asset_id: data["globalAssetId"] = obj.global_asset_id if obj.specific_asset_id: - data["specificAssetIds"] = list(obj.specific_asset_id) + data["specificAssetIds"] = sorted( + obj.specific_asset_id, key=_ordering.specific_asset_id_sort_key + ) if obj.asset_type: data["assetType"] = obj.asset_type if obj.default_thumbnail: @@ -370,7 +431,7 @@ def _concept_description_to_json( """ data = cls._abstract_classes_to_json(obj) if obj.is_case_of: - data["isCaseOf"] = list(obj.is_case_of) + data["isCaseOf"] = sorted(obj.is_case_of, key=_ordering.reference_sort_key) return data @classmethod @@ -431,7 +492,7 @@ def _asset_administration_shell_to_json( if obj.asset_information: data["assetInformation"] = obj.asset_information if not cls.stripped and obj.submodel: - data["submodels"] = list(obj.submodel) + data["submodels"] = sorted(obj.submodel, key=_ordering.reference_sort_key) return data # ################################################################# @@ -450,7 +511,9 @@ def _submodel_to_json( """ data = cls._abstract_classes_to_json(obj) if not cls.stripped and obj.submodel_element != set(): - data["submodelElements"] = list(obj.submodel_element) + data["submodelElements"] = sorted( + obj.submodel_element, key=_ordering.referable_sort_key + ) return data @classmethod @@ -525,7 +588,7 @@ def _blob_to_json(cls, obj: model.Blob) -> Dict[str, object]: data = cls._abstract_classes_to_json(obj) if obj.content_type is not None: data["contentType"] = obj.content_type - if obj.value is not None: + if cls.with_blob_value and obj.value is not None: data["value"] = base64.b64encode(obj.value).decode() return data @@ -585,7 +648,7 @@ def _submodel_element_collection_to_json( """ data = cls._abstract_classes_to_json(obj) if not cls.stripped and len(obj.value) > 0: - data["value"] = list(obj.value) + data["value"] = sorted(obj.value, key=_ordering.referable_sort_key) return data @classmethod @@ -644,7 +707,9 @@ def _annotated_relationship_element_to_json( """ data = cls._relationship_element_to_json(obj) if not cls.stripped and obj.annotation: - data["annotations"] = list(obj.annotation) + data["annotations"] = sorted( + obj.annotation, key=_ordering.referable_sort_key + ) return data @classmethod @@ -678,7 +743,10 @@ def _operation_to_json(cls, obj: model.Operation) -> Dict[str, object]: ("inoutputVariables", obj.in_output_variable), ): if nss: - data[tag] = [cls._operation_variable_to_json(obj) for obj in nss] + data[tag] = [ + cls._operation_variable_to_json(variable) + for variable in sorted(nss, key=_ordering.referable_sort_key) + ] return data @classmethod @@ -703,13 +771,15 @@ def _entity_to_json(cls, obj: model.Entity) -> Dict[str, object]: """ data = cls._abstract_classes_to_json(obj) if not cls.stripped and obj.statement: - data["statements"] = list(obj.statement) + data["statements"] = sorted(obj.statement, key=_ordering.referable_sort_key) if obj.entity_type is not None: data["entityType"] = _generic.ENTITY_TYPES[obj.entity_type] if obj.global_asset_id is not None: data["globalAssetId"] = obj.global_asset_id if obj.specific_asset_id: - data["specificAssetIds"] = list(obj.specific_asset_id) + data["specificAssetIds"] = sorted( + obj.specific_asset_id, key=_ordering.specific_asset_id_sort_key + ) return data @classmethod @@ -750,6 +820,105 @@ def _basic_event_element_to_json( data["maxInterval"] = model.datatypes.xsd_repr(obj.max_interval) return data + # ############################################################# + # transformation functions to serialize with deep=False + # ############################################################# + + @classmethod + def _remove_children(cls, obj: object) -> object: + """ + Serialization of a child element without its own children + + :param obj: The child element to serialize + :return: dict with the serialized attributes of the child element without its children or the child element + itself, if it cannot contain children + """ + mapping = cls._get_aas_class_serializers() + children_keys = cls._get_aas_class_children_keys() + for typ in children_keys: + if isinstance(obj, typ): + data = mapping[typ](obj) + data.pop(children_keys[typ], None) + return data + return obj + + @classmethod + def _submodel_to_json_no_children(cls, obj: model.Submodel) -> Dict[str, object]: + """ + serialization of an object from class Submodel to json, omitting the children of its submodel elements + + :param obj: object of class Submodel + :return: dict with the serialized attributes of this object + """ + data = cls._get_aas_class_serializers()[model.Submodel](obj) + children = data.get("submodelElements") + if isinstance(children, list): + data["submodelElements"] = [cls._remove_children(child) for child in children] + return data + + @classmethod + def _submodel_element_collection_to_json_no_children( + cls, obj: model.SubmodelElementCollection + ) -> Dict[str, object]: + """ + serialization of an object from class SubmodelElementCollection to json, omitting the children of its value + + :param obj: object of class SubmodelElementCollection + :return: dict with the serialized attributes of this object + """ + data = cls._get_aas_class_serializers()[model.SubmodelElementCollection](obj) + children = data.get("value") + if isinstance(children, list): + data["value"] = [cls._remove_children(child) for child in children] + return data + + @classmethod + def _submodel_element_list_to_json_no_children( + cls, obj: model.SubmodelElementList + ) -> Dict[str, object]: + """ + serialization of an object from class SubmodelElementList to json, omitting the children of its value + + :param obj: object of class SubmodelElementList + :return: dict with the serialized attributes of this object + """ + data = cls._get_aas_class_serializers()[model.SubmodelElementList](obj) + children = data.get("value") + if isinstance(children, list): + data["value"] = [cls._remove_children(child) for child in children] + return data + + @classmethod + def _annotated_relationship_element_to_json_no_children( + cls, obj: model.AnnotatedRelationshipElement + ) -> Dict[str, object]: + """ + serialization of an object from class AnnotatedRelationshipElement to json, omitting the children of its + annotations + + :param obj: object of class AnnotatedRelationshipElement + :return: dict with the serialized attributes of this object + """ + data = cls._get_aas_class_serializers()[model.AnnotatedRelationshipElement](obj) + children = data.get("annotations") + if isinstance(children, list): + data["annotations"] = [cls._remove_children(child) for child in children] + return data + + @classmethod + def _entity_to_json_no_children(cls, obj: model.Entity) -> Dict[str, object]: + """ + serialization of an object from class Entity to json, omitting the children of its statements + + :param obj: object of class Entity + :return: dict with the serialized attributes of this object + """ + data = cls._get_aas_class_serializers()[model.Entity](obj) + children = data.get("statements") + if isinstance(children, list): + data["statements"] = [cls._remove_children(child) for child in children] + return data + class StrippedAASToJsonEncoder(AASToJsonEncoder): """ @@ -776,35 +945,30 @@ def _select_encoder( return AASToJsonEncoder if not stripped else StrippedAASToJsonEncoder -def _create_dict( - data: model.AbstractObjectStore, - keys_to_types: Iterable[Tuple[str, Type]] = JSON_AAS_TOP_LEVEL_KEYS_TO_TYPES, -) -> Dict[str, List[model.Identifiable]]: +def _create_dict(data: model.AbstractObjectStore) -> Dict[str, List[model.Identifiable]]: """ Categorizes objects from an AbstractObjectStore into a dictionary based on their types. This function iterates over the objects in the provided AbstractObjectStore and groups them into lists - based on their types, as defined in the `keys_to_types` mapping. The resulting dictionary contains - keys corresponding to the names in `keys_to_types` and values as lists of objects of the respective types. + based on their types, as defined in ``JSON_AAS_TOP_LEVEL_KEYS_TO_TYPES``. The resulting dictionary contains + the top-level keys of the JSON format and values as lists of objects of the respective types. + The keys are ordered as in ``JSON_AAS_TOP_LEVEL_KEYS_TO_TYPES`` and each list is sorted by the id of the objects, + so that the order does not depend on the iteration order of the AbstractObjectStore. :param data: An AbstractObjectStore containing objects to be categorized. - :param keys_to_types: An iterable of tuples where each tuple contains: - - A string key representing the category name. - - A type to match objects against. :return: A dictionary where keys are category names and values are lists of objects of the corresponding types. """ - objects: Dict[str, List[model.Identifiable]] = {} + objects: Dict[str, List[model.Identifiable]] = {name: [] for name, _ in JSON_AAS_TOP_LEVEL_KEYS_TO_TYPES} for obj in data: # Iterate through the mapping of category names to expected types - for name, expected_type in keys_to_types: + for name, expected_type in JSON_AAS_TOP_LEVEL_KEYS_TO_TYPES: # Check if the object matches the expected type if isinstance(obj, expected_type): # Add the object to the appropriate category in the dictionary - objects.setdefault(name, []) objects[name].append(obj) break # Exit the inner loop once a match is found - return objects + return {name: sorted(objs, key=_ordering.identifiable_sort_key) for name, objs in objects.items() if objs} def object_store_to_json( diff --git a/sdk/test/adapter/json/formats/__init__.py b/sdk/test/adapter/json/formats/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/sdk/test/adapter/json/formats/test_core_serialization.py b/sdk/test/adapter/json/formats/test_core_serialization.py new file mode 100644 index 000000000..0069dc4ab --- /dev/null +++ b/sdk/test/adapter/json/formats/test_core_serialization.py @@ -0,0 +1,276 @@ +# Copyright (c) 2026 the Eclipse BaSyx Authors +# +# This program and the accompanying materials are made available under the terms of the MIT License, available in +# the LICENSE file of this project. +# +# SPDX-License-Identifier: MIT +import json +import unittest + +from basyx.aas import model +from basyx.aas.adapter.json import AASToJsonEncoder, StrippedAASToJsonEncoder +from basyx.aas.examples.data import example_aas, example_aas_mandatory_attributes + + +class TestAASToJsonEncoderCore(unittest.TestCase): + def setUp(self): + self.submodel = example_aas.create_example_submodel() + self.deep_submodel = json.loads(json.dumps(self.submodel, cls=AASToJsonEncoder)) + + def test_deep_is_default_ok(self): + deep_output = json.loads(json.dumps(self.submodel, cls=AASToJsonEncoder, deep=True)) + + out_str = json.dumps(self.submodel, cls=AASToJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual(deep_output, parsed_output) + + def test_deep_keeps_grandchildren_ok(self): + smc = self.submodel.get_referable("ExampleSubmodelCollection") + + out_str = json.dumps(smc, cls=AASToJsonEncoder, deep=True) + parsed_output = json.loads(out_str) + + children = {element["idShort"]: element for element in parsed_output["value"]} + self.assertEqual(2, len(children["ExampleSubmodelList"]["value"])) + + def test_submodel_ok(self): + out_str = json.dumps(self.submodel, cls=AASToJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertCountEqual([ + "ExampleRelationshipElement", + "ExampleAnnotatedRelationshipElement", + "ExampleOperation", + "ExampleCapability", + "ExampleBasicEventElement", + "ExampleSubmodelCollection", + ], [element["idShort"] for element in parsed_output["submodelElements"]]) + + def test_submodel_keeps_own_attributes_ok(self): + expected = {key: value for key, value in self.deep_submodel.items() if key != "submodelElements"} + + out_str = json.dumps(self.submodel, cls=AASToJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + parsed_output.pop("submodelElements") + self.assertEqual(expected, parsed_output) + + def test_submodel_omits_submodel_element_collection_children_ok(self): + deep_children = {element["idShort"]: element for element in self.deep_submodel["submodelElements"]} + expected = deep_children["ExampleSubmodelCollection"] + expected.pop("value") + + out_str = json.dumps(self.submodel, cls=AASToJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + children = {element["idShort"]: element for element in parsed_output["submodelElements"]} + self.assertEqual(expected, children["ExampleSubmodelCollection"]) + + def test_submodel_omits_annotated_relationship_element_annotations_ok(self): + deep_children = {element["idShort"]: element for element in self.deep_submodel["submodelElements"]} + expected = deep_children["ExampleAnnotatedRelationshipElement"] + expected.pop("annotations") + + out_str = json.dumps(self.submodel, cls=AASToJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + children = {element["idShort"]: element for element in parsed_output["submodelElements"]} + self.assertEqual(expected, children["ExampleAnnotatedRelationshipElement"]) + + def test_submodel_keeps_leaf_children_ok(self): + deep_children = {element["idShort"]: element for element in self.deep_submodel["submodelElements"]} + + out_str = json.dumps(self.submodel, cls=AASToJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + children = {element["idShort"]: element for element in parsed_output["submodelElements"]} + self.assertEqual(deep_children["ExampleRelationshipElement"], children["ExampleRelationshipElement"]) + self.assertEqual(deep_children["ExampleOperation"], children["ExampleOperation"]) + self.assertEqual(deep_children["ExampleCapability"], children["ExampleCapability"]) + self.assertEqual(deep_children["ExampleBasicEventElement"], children["ExampleBasicEventElement"]) + + def test_submodel_omits_entity_statements_ok(self): + submodel = example_aas.create_example_bill_of_material_submodel() + deep_output = json.loads(json.dumps(submodel, cls=AASToJsonEncoder)) + deep_children = {element["idShort"]: element for element in deep_output["submodelElements"]} + expected = deep_children["ExampleEntity"] + expected.pop("statements") + + out_str = json.dumps(submodel, cls=AASToJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + children = {element["idShort"]: element for element in parsed_output["submodelElements"]} + self.assertEqual(expected, children["ExampleEntity"]) + self.assertEqual(deep_children["ExampleEntity2"], children["ExampleEntity2"]) + + def test_empty_submodel_ok(self): + submodel = example_aas_mandatory_attributes.create_example_empty_submodel() + + out_str = json.dumps(submodel, cls=AASToJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertNotIn("submodelElements", parsed_output) + + def test_submodel_element_collection_ok(self): + smc = self.submodel.get_referable("ExampleSubmodelCollection") + deep_output = json.loads(json.dumps(smc, cls=AASToJsonEncoder)) + deep_children = {element["idShort"]: element for element in deep_output["value"]} + deep_children["ExampleSubmodelList"].pop("value") + + out_str = json.dumps(smc, cls=AASToJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + children = {element["idShort"]: element for element in parsed_output["value"]} + self.assertEqual(deep_children, children) + + def test_submodel_element_list_ok(self): + submodel = example_aas_mandatory_attributes.create_example_submodel() + sml = submodel.get_referable("ExampleSubmodelList") + deep_output = json.loads(json.dumps(sml, cls=AASToJsonEncoder)) + expected = deep_output["value"] + expected[0].pop("value") + + out_str = json.dumps(sml, cls=AASToJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertEqual(expected, parsed_output["value"]) + + def test_empty_submodel_element_list_ok(self): + submodel = example_aas_mandatory_attributes.create_example_submodel() + sml = submodel.get_referable("ExampleSubmodelList2") + + out_str = json.dumps(sml, cls=AASToJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertNotIn("value", parsed_output) + + def test_entity_ok(self): + submodel = example_aas.create_example_bill_of_material_submodel() + entity = submodel.get_referable("ExampleEntity") + deep_output = json.loads(json.dumps(entity, cls=AASToJsonEncoder)) + + out_str = json.dumps(entity, cls=AASToJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertEqual(deep_output, parsed_output) + + def test_annotated_relationship_element_ok(self): + are = self.submodel.get_referable("ExampleAnnotatedRelationshipElement") + deep_output = json.loads(json.dumps(are, cls=AASToJsonEncoder)) + + out_str = json.dumps(are, cls=AASToJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertEqual(deep_output, parsed_output) + + def test_leaf_element_ok(self): + range_element = self.submodel.get_referable("ExampleSubmodelCollection.ExampleRange") + deep_output = json.loads(json.dumps(range_element, cls=AASToJsonEncoder)) + + out_str = json.dumps(range_element, cls=AASToJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertEqual(deep_output, parsed_output) + + def test_core_is_relative_to_each_top_level_object_ok(self): + smc = self.submodel.get_referable("ExampleSubmodelCollection") + + out_str = json.dumps([self.submodel, smc], cls=AASToJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + submodel_output, smc_output = parsed_output + submodel_children = {element["idShort"]: element for element in submodel_output["submodelElements"]} + self.assertNotIn("value", submodel_children["ExampleSubmodelCollection"]) + self.assertEqual(7, len(smc_output["value"])) + + def test_core_with_indent_ok(self): + compact_output = json.loads(json.dumps(self.submodel, cls=AASToJsonEncoder, deep=False)) + + out_str = json.dumps(self.submodel, cls=AASToJsonEncoder, deep=False, indent=2) + parsed_output = json.loads(out_str) + + self.assertEqual(compact_output, parsed_output) + + def test_stripped_core_ok(self): + stripped_output = json.loads(json.dumps(self.submodel, cls=StrippedAASToJsonEncoder)) + + out_str = json.dumps(self.submodel, cls=StrippedAASToJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertEqual(stripped_output, parsed_output) + + def test_subclass_override_applies_to_children_ok(self): + class CustomCoreEncoder(AASToJsonEncoder): + @classmethod + def _submodel_element_collection_to_json(cls, obj): + data = super()._submodel_element_collection_to_json(obj) + data["custom"] = True + return data + + out_str = json.dumps(self.submodel, cls=CustomCoreEncoder, deep=False) + parsed_output = json.loads(out_str) + + children = {element["idShort"]: element for element in parsed_output["submodelElements"]} + self.assertTrue(children["ExampleSubmodelCollection"]["custom"]) + self.assertNotIn("value", children["ExampleSubmodelCollection"]) + + def test_subclass_override_applies_to_requested_object_ok(self): + class CustomCoreEncoder(AASToJsonEncoder): + @classmethod + def _submodel_element_collection_to_json(cls, obj): + data = super()._submodel_element_collection_to_json(obj) + data["custom"] = True + return data + + smc = self.submodel.get_referable("ExampleSubmodelCollection") + + out_str = json.dumps(smc, cls=CustomCoreEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertTrue(parsed_output["custom"]) + self.assertEqual(7, len(parsed_output["value"])) + + def test_subclass_deep_override_ok(self): + class CustomEncoder(AASToJsonEncoder): + @classmethod + def _submodel_element_collection_to_json(cls, obj): + data = super()._submodel_element_collection_to_json(obj) + data["childTypes"] = [type(element).__name__ for element in data["value"]] + return data + + smc = self.submodel.get_referable("ExampleSubmodelCollection") + + out_str = json.dumps(smc, cls=CustomEncoder) + parsed_output = json.loads(out_str) + + self.assertCountEqual([ + "Blob", + "File", + "File", + "MultiLanguageProperty", + "Range", + "ReferenceElement", + "SubmodelElementList", + ], parsed_output["childTypes"]) + + def test_subclass_mapping_override_applies_to_children_ok(self): + class CustomCoreEncoder(AASToJsonEncoder): + @classmethod + def _get_aas_class_serializers(cls): + mapping = super()._get_aas_class_serializers() + mapping[model.SubmodelElementCollection] = cls._custom_collection_to_json + return mapping + + @classmethod + def _custom_collection_to_json(cls, obj): + data = cls._submodel_element_collection_to_json(obj) + data["custom"] = True + return data + + out_str = json.dumps(self.submodel, cls=CustomCoreEncoder, deep=False) + parsed_output = json.loads(out_str) + + children = {element["idShort"]: element for element in parsed_output["submodelElements"]} + self.assertTrue(children["ExampleSubmodelCollection"]["custom"]) + self.assertNotIn("value", children["ExampleSubmodelCollection"]) diff --git a/sdk/test/adapter/json/formats/test_metadata_serialization.py b/sdk/test/adapter/json/formats/test_metadata_serialization.py new file mode 100644 index 000000000..706f283e4 --- /dev/null +++ b/sdk/test/adapter/json/formats/test_metadata_serialization.py @@ -0,0 +1,279 @@ +# Copyright (c) 2026 the Eclipse BaSyx Authors +# +# This program and the accompanying materials are made available under the terms of the MIT License, available in +# the LICENSE file of this project. +# +# SPDX-License-Identifier: MIT +import json +import unittest + +from basyx.aas import model +from basyx.aas.adapter.json import AASToJsonEncoder, MetadataJsonEncoder +from basyx.aas.examples.data import example_aas, example_aas_mandatory_attributes + + +class TestMetadataJsonEncoder(unittest.TestCase): + """ + Tests follow Table Metadata Attributes of the AAS API spec. For every class, verify that the fields + listed as "not available in metadata representation" are dropped, while other attributes are retained. + """ + + def setUp(self): + self.submodel = example_aas.create_example_submodel() + + # ================================================================== Identifiables + + # ----- AssetAdministrationShell: drops assetInformation, submodels + def test_asset_administration_shell_drops_asset_information_and_submodels_ok(self): + aas = example_aas.create_example_asset_administration_shell() + + base_output = json.loads(json.dumps(aas, cls=AASToJsonEncoder)) + parsed_output = json.loads(json.dumps(aas, cls=MetadataJsonEncoder)) + + self.assertNotIn("assetInformation", parsed_output) + self.assertNotIn("submodels", parsed_output) + self.assertEqual({"assetInformation", "submodels"}, set(base_output.keys()) - set(parsed_output.keys())) + self.assertEqual("AssetAdministrationShell", parsed_output["modelType"]) + + # ----- Submodel: drops submodelElements + def test_submodel_drops_submodel_elements_ok(self): + base_output = json.loads(json.dumps(self.submodel, cls=AASToJsonEncoder)) + parsed_output = json.loads(json.dumps(self.submodel, cls=MetadataJsonEncoder)) + + self.assertNotIn("submodelElements", parsed_output) + self.assertEqual({"submodelElements"}, set(base_output.keys()) - set(parsed_output.keys())) + + def test_submodel_keeps_identifiable_fields_ok(self): + parsed_output = json.loads(json.dumps(self.submodel, cls=MetadataJsonEncoder)) + + self.assertEqual(self.submodel.id, parsed_output["id"]) + self.assertEqual(self.submodel.id_short, parsed_output["idShort"]) + self.assertEqual("Submodel", parsed_output["modelType"]) + + def test_empty_submodel_ok(self): + submodel = example_aas_mandatory_attributes.create_example_empty_submodel() + + parsed_output = json.loads(json.dumps(submodel, cls=MetadataJsonEncoder)) + + self.assertNotIn("submodelElements", parsed_output) + self.assertEqual(submodel.id, parsed_output["id"]) + + # ================================================================== SubmodelElements + + # ----- SubmodelElementCollection: drops value + def test_submodel_element_collection_drops_value_ok(self): + smc = self.submodel.get_referable("ExampleSubmodelCollection") + + base_output = json.loads(json.dumps(smc, cls=AASToJsonEncoder)) + parsed_output = json.loads(json.dumps(smc, cls=MetadataJsonEncoder)) + + self.assertNotIn("value", parsed_output) + self.assertEqual({"value"}, set(base_output.keys()) - set(parsed_output.keys())) + self.assertEqual("SubmodelElementCollection", parsed_output["modelType"]) + + # ----- SubmodelElementList: drops value + def test_submodel_element_list_drops_value_ok(self): + sml = self.submodel.get_referable("ExampleSubmodelCollection.ExampleSubmodelList") + + base_output = json.loads(json.dumps(sml, cls=AASToJsonEncoder)) + parsed_output = json.loads(json.dumps(sml, cls=MetadataJsonEncoder)) + + self.assertNotIn("value", parsed_output) + self.assertEqual({"value"}, set(base_output.keys()) - set(parsed_output.keys())) + self.assertEqual("SubmodelElementList", parsed_output["modelType"]) + # List-level metadata (type/semantics of elements) must be retained + self.assertIn("typeValueListElement", parsed_output) + + # ----- Entity: drops statements, globalAssetId, specificAssetId + def test_entity_drops_statements_and_asset_identifiers_ok(self): + submodel = example_aas.create_example_bill_of_material_submodel() + entity = submodel.get_referable("ExampleEntity") + + base_output = json.loads(json.dumps(entity, cls=AASToJsonEncoder)) + parsed_output = json.loads(json.dumps(entity, cls=MetadataJsonEncoder)) + + self.assertNotIn("statements", parsed_output) + self.assertNotIn("globalAssetId", parsed_output) + self.assertNotIn("specificAssetIds", parsed_output) + self.assertEqual( + {"statements", "globalAssetId", "specificAssetIds"}, + set(base_output.keys()) - set(parsed_output.keys()), + ) + self.assertEqual("Entity", parsed_output["modelType"]) + self.assertEqual("SelfManagedEntity", parsed_output["entityType"]) + + def test_entity_without_statements_ok(self): + submodel = example_aas.create_example_bill_of_material_submodel() + entity = submodel.get_referable("ExampleEntity2") + + parsed_output = json.loads(json.dumps(entity, cls=MetadataJsonEncoder)) + + self.assertNotIn("statements", parsed_output) + + # ----- BasicEventElement: drops observed + def test_basic_event_element_drops_observed_ok(self): + basic_event_element = self.submodel.get_referable("ExampleBasicEventElement") + + base_output = json.loads(json.dumps(basic_event_element, cls=AASToJsonEncoder)) + parsed_output = json.loads(json.dumps(basic_event_element, cls=MetadataJsonEncoder)) + + self.assertNotIn("observed", parsed_output) + self.assertEqual({"observed"}, set(base_output.keys()) - set(parsed_output.keys())) + # Other BasicEventElement metadata fields must be kept + self.assertIn("direction", parsed_output) + self.assertIn("state", parsed_output) + + # ----- Capability: no fields dropped + def test_capability_ok(self): + capability = self.submodel.get_referable("ExampleCapability") + + base_output = json.loads(json.dumps(capability, cls=AASToJsonEncoder)) + parsed_output = json.loads(json.dumps(capability, cls=MetadataJsonEncoder)) + + self.assertEqual(set(base_output.keys()), set(parsed_output.keys())) + self.assertEqual("Capability", parsed_output["modelType"]) + + # ----- Operation: no fields dropped + def test_operation_ok(self): + operation = self.submodel.get_referable("ExampleOperation") + + base_output = json.loads(json.dumps(operation, cls=AASToJsonEncoder)) + parsed_output = json.loads(json.dumps(operation, cls=MetadataJsonEncoder)) + + # Operation metadata keeps input/output/inoutput variables (they describe shape, not value) + self.assertEqual(set(base_output.keys()), set(parsed_output.keys())) + self.assertEqual("Operation", parsed_output["modelType"]) + + # ================================================================== DataElements + + # ----- Property: drops value, valueId + def test_property_drops_value_and_value_id_ok(self): + prop = model.Property( + "Prop", + model.datatypes.String, + value="exampleValue", + value_id=model.ExternalReference( + (model.Key(model.KeyTypes.GLOBAL_REFERENCE, "https://example.org/ValueId"),) + ), + ) + + base_output = json.loads(json.dumps(prop, cls=AASToJsonEncoder)) + parsed_output = json.loads(json.dumps(prop, cls=MetadataJsonEncoder)) + + self.assertNotIn("value", parsed_output) + self.assertNotIn("valueId", parsed_output) + self.assertEqual({"value", "valueId"}, set(base_output.keys()) - set(parsed_output.keys())) + self.assertIn("valueType", parsed_output) + + # ----- MultiLanguageProperty: drops value, valueId + def test_multi_language_property_drops_value_and_value_id_ok(self): + mlp = self.submodel.get_referable("ExampleSubmodelCollection.ExampleMultiLanguageProperty") + + base_output = json.loads(json.dumps(mlp, cls=AASToJsonEncoder)) + parsed_output = json.loads(json.dumps(mlp, cls=MetadataJsonEncoder)) + + self.assertNotIn("value", parsed_output) + self.assertNotIn("valueId", parsed_output) + self.assertLessEqual( + set(base_output.keys()) - set(parsed_output.keys()), + {"value", "valueId"}, + ) + self.assertEqual("MultiLanguageProperty", parsed_output["modelType"]) + + # ----- Range: drops min, max + def test_range_drops_min_and_max_ok(self): + range_element = self.submodel.get_referable("ExampleSubmodelCollection.ExampleRange") + + base_output = json.loads(json.dumps(range_element, cls=AASToJsonEncoder)) + parsed_output = json.loads(json.dumps(range_element, cls=MetadataJsonEncoder)) + + self.assertNotIn("min", parsed_output) + self.assertNotIn("max", parsed_output) + self.assertEqual({"min", "max"}, set(base_output.keys()) - set(parsed_output.keys())) + self.assertIn("valueType", parsed_output) + + # ----- ReferenceElement: drops value + def test_reference_element_drops_value_ok(self): + reference_element = self.submodel.get_referable("ExampleSubmodelCollection.ExampleReferenceElement") + + base_output = json.loads(json.dumps(reference_element, cls=AASToJsonEncoder)) + parsed_output = json.loads(json.dumps(reference_element, cls=MetadataJsonEncoder)) + + self.assertNotIn("value", parsed_output) + self.assertEqual({"value"}, set(base_output.keys()) - set(parsed_output.keys())) + self.assertEqual("ReferenceElement", parsed_output["modelType"]) + + # ----- RelationshipElement: drops first, second + def test_relationship_element_drops_first_and_second_ok(self): + relationship_element = self.submodel.get_referable("ExampleRelationshipElement") + + base_output = json.loads(json.dumps(relationship_element, cls=AASToJsonEncoder)) + parsed_output = json.loads(json.dumps(relationship_element, cls=MetadataJsonEncoder)) + + self.assertNotIn("first", parsed_output) + self.assertNotIn("second", parsed_output) + self.assertEqual({"first", "second"}, set(base_output.keys()) - set(parsed_output.keys())) + self.assertEqual("RelationshipElement", parsed_output["modelType"]) + + # ----- AnnotatedRelationshipElement: drops first, second, annotations + def test_annotated_relationship_element_drops_endpoints_and_annotations_ok(self): + are = self.submodel.get_referable("ExampleAnnotatedRelationshipElement") + + base_output = json.loads(json.dumps(are, cls=AASToJsonEncoder)) + parsed_output = json.loads(json.dumps(are, cls=MetadataJsonEncoder)) + + self.assertNotIn("first", parsed_output) + self.assertNotIn("second", parsed_output) + self.assertNotIn("annotations", parsed_output) + self.assertEqual( + {"first", "second", "annotations"}, + set(base_output.keys()) - set(parsed_output.keys()), + ) + self.assertEqual("AnnotatedRelationshipElement", parsed_output["modelType"]) + + # ----- Blob: drops value, contentType + def test_blob_drops_value_and_content_type_ok(self): + blob = self.submodel.get_referable("ExampleSubmodelCollection.ExampleBlob") + + base_output = json.loads(json.dumps(blob, cls=AASToJsonEncoder)) + parsed_output = json.loads(json.dumps(blob, cls=MetadataJsonEncoder)) + + self.assertNotIn("value", parsed_output) + self.assertNotIn("contentType", parsed_output) + self.assertEqual({"value", "contentType"}, set(base_output.keys()) - set(parsed_output.keys())) + self.assertEqual("Blob", parsed_output["modelType"]) + + # ----- File: drops value, contentType + def test_file_drops_value_and_content_type_ok(self): + file = self.submodel.get_referable("ExampleSubmodelCollection.ExampleFile") + + base_output = json.loads(json.dumps(file, cls=AASToJsonEncoder)) + parsed_output = json.loads(json.dumps(file, cls=MetadataJsonEncoder)) + + self.assertNotIn("value", parsed_output) + self.assertNotIn("contentType", parsed_output) + self.assertEqual({"value", "contentType"}, set(base_output.keys()) - set(parsed_output.keys())) + self.assertEqual("File", parsed_output["modelType"]) + + # ================================================================== Unsupported types + + def test_concept_description_fail(self): + concept_description = example_aas.create_example_concept_description() + + with self.assertRaises(TypeError): + json.dumps(concept_description, cls=MetadataJsonEncoder) + + # ================================================================== Composite + + def test_list_of_elements_ok(self): + elements = [ + self.submodel.get_referable("ExampleRelationshipElement"), + self.submodel.get_referable("ExampleSubmodelCollection"), + ] + + parsed_output = json.loads(json.dumps(elements, cls=MetadataJsonEncoder)) + + self.assertEqual(2, len(parsed_output)) + self.assertNotIn("first", parsed_output[0]) + self.assertNotIn("second", parsed_output[0]) + self.assertNotIn("value", parsed_output[1]) diff --git a/sdk/test/adapter/json/formats/test_path_serialization.py b/sdk/test/adapter/json/formats/test_path_serialization.py new file mode 100644 index 000000000..13e31ffea --- /dev/null +++ b/sdk/test/adapter/json/formats/test_path_serialization.py @@ -0,0 +1,311 @@ +# Copyright (c) 2026 the Eclipse BaSyx Authors +# +# This program and the accompanying materials are made available under the terms of the MIT License, available in +# the LICENSE file of this project. +# +# SPDX-License-Identifier: MIT +import json +import unittest + +from basyx.aas import model +from basyx.aas.adapter.json import PathJsonEncoder +from basyx.aas.examples.data import example_aas, example_aas_mandatory_attributes + + +class TestPathJsonEncoder(unittest.TestCase): + def setUp(self): + self.submodel = example_aas.create_example_submodel() + + def test_submodel_deep_ok(self): + out_str = json.dumps(self.submodel, cls=PathJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertCountEqual([ + "ExampleRelationshipElement", + "ExampleAnnotatedRelationshipElement", + "ExampleAnnotatedRelationshipElement.ExampleAnnotatedRange", + "ExampleAnnotatedRelationshipElement.ExampleAnnotatedProperty", + "ExampleOperation", + "ExampleCapability", + "ExampleBasicEventElement", + "ExampleSubmodelCollection", + "ExampleSubmodelCollection.ExampleBlob", + "ExampleSubmodelCollection.ExampleFile", + "ExampleSubmodelCollection.ExampleFileURI", + "ExampleSubmodelCollection.ExampleMultiLanguageProperty", + "ExampleSubmodelCollection.ExampleRange", + "ExampleSubmodelCollection.ExampleReferenceElement", + "ExampleSubmodelCollection.ExampleSubmodelList", + "ExampleSubmodelCollection.ExampleSubmodelList[0]", + "ExampleSubmodelCollection.ExampleSubmodelList[1]", + ], parsed_output) + + def test_submodel_core_ok(self): + out_str = json.dumps(self.submodel, cls=PathJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertCountEqual([ + "ExampleRelationshipElement", + "ExampleAnnotatedRelationshipElement", + "ExampleOperation", + "ExampleCapability", + "ExampleBasicEventElement", + "ExampleSubmodelCollection", + ], parsed_output) + + def test_deep_is_default_ok(self): + deep_output = json.loads(json.dumps(self.submodel, cls=PathJsonEncoder, deep=True)) + + out_str = json.dumps(self.submodel, cls=PathJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual(deep_output, parsed_output) + + def test_submodel_without_id_short_ok(self): + submodel = example_aas_mandatory_attributes.create_example_submodel() + + out_str = json.dumps(submodel, cls=PathJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertCountEqual([ + "ExampleRelationshipElement", + "ExampleAnnotatedRelationshipElement", + "ExampleOperation", + "ExampleCapability", + "ExampleBasicEventElement", + "ExampleSubmodelList", + "ExampleSubmodelList[0]", + "ExampleSubmodelList[0].ExampleBlob", + "ExampleSubmodelList[0].ExampleFile", + "ExampleSubmodelList[0].ExampleMultiLanguageProperty", + "ExampleSubmodelList[0].ExampleProperty", + "ExampleSubmodelList[0].ExampleRange", + "ExampleSubmodelList[0].ExampleReferenceElement", + "ExampleSubmodelList[1]", + "ExampleSubmodelList2", + ], parsed_output) + + def test_empty_submodel_ok(self): + submodel = example_aas_mandatory_attributes.create_example_empty_submodel() + + out_str = json.dumps(submodel, cls=PathJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual([], parsed_output) + + def test_submodel_element_collection_deep_ok(self): + smc = self.submodel.get_referable("ExampleSubmodelCollection") + + out_str = json.dumps(smc, cls=PathJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertCountEqual([ + "ExampleSubmodelCollection", + "ExampleSubmodelCollection.ExampleBlob", + "ExampleSubmodelCollection.ExampleFile", + "ExampleSubmodelCollection.ExampleFileURI", + "ExampleSubmodelCollection.ExampleMultiLanguageProperty", + "ExampleSubmodelCollection.ExampleRange", + "ExampleSubmodelCollection.ExampleReferenceElement", + "ExampleSubmodelCollection.ExampleSubmodelList", + "ExampleSubmodelCollection.ExampleSubmodelList[0]", + "ExampleSubmodelCollection.ExampleSubmodelList[1]", + ], parsed_output) + + def test_submodel_element_collection_core_ok(self): + smc = self.submodel.get_referable("ExampleSubmodelCollection") + + out_str = json.dumps(smc, cls=PathJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertCountEqual([ + "ExampleSubmodelCollection", + "ExampleSubmodelCollection.ExampleBlob", + "ExampleSubmodelCollection.ExampleFile", + "ExampleSubmodelCollection.ExampleFileURI", + "ExampleSubmodelCollection.ExampleMultiLanguageProperty", + "ExampleSubmodelCollection.ExampleRange", + "ExampleSubmodelCollection.ExampleReferenceElement", + "ExampleSubmodelCollection.ExampleSubmodelList", + ], parsed_output) + + def test_submodel_element_list_deep_ok(self): + submodel = example_aas_mandatory_attributes.create_example_submodel() + sml = submodel.get_referable("ExampleSubmodelList") + + out_str = json.dumps(sml, cls=PathJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertCountEqual([ + "ExampleSubmodelList", + "ExampleSubmodelList[0]", + "ExampleSubmodelList[0].ExampleBlob", + "ExampleSubmodelList[0].ExampleFile", + "ExampleSubmodelList[0].ExampleMultiLanguageProperty", + "ExampleSubmodelList[0].ExampleProperty", + "ExampleSubmodelList[0].ExampleRange", + "ExampleSubmodelList[0].ExampleReferenceElement", + "ExampleSubmodelList[1]", + ], parsed_output) + + def test_submodel_element_list_core_ok(self): + submodel = example_aas_mandatory_attributes.create_example_submodel() + sml = submodel.get_referable("ExampleSubmodelList") + + out_str = json.dumps(sml, cls=PathJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertCountEqual([ + "ExampleSubmodelList", + "ExampleSubmodelList[0]", + "ExampleSubmodelList[1]", + ], parsed_output) + + def test_empty_submodel_element_list_ok(self): + submodel = example_aas_mandatory_attributes.create_example_submodel() + sml = submodel.get_referable("ExampleSubmodelList2") + + out_str = json.dumps(sml, cls=PathJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual(["ExampleSubmodelList2"], parsed_output) + + def test_entity_ok(self): + submodel = example_aas.create_example_bill_of_material_submodel() + entity = submodel.get_referable("ExampleEntity") + + out_str = json.dumps(entity, cls=PathJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertCountEqual([ + "ExampleEntity", + "ExampleEntity.ExampleProperty", + "ExampleEntity.ExampleProperty2", + ], parsed_output) + + def test_submodel_with_entity_deep_ok(self): + submodel = example_aas.create_example_bill_of_material_submodel() + + out_str = json.dumps(submodel, cls=PathJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertCountEqual([ + "ExampleEntity", + "ExampleEntity.ExampleProperty", + "ExampleEntity.ExampleProperty2", + "ExampleEntity2", + ], parsed_output) + + def test_submodel_with_entity_core_ok(self): + submodel = example_aas.create_example_bill_of_material_submodel() + + out_str = json.dumps(submodel, cls=PathJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertCountEqual(["ExampleEntity", "ExampleEntity2"], parsed_output) + + def test_empty_entity_ok(self): + submodel = example_aas.create_example_bill_of_material_submodel() + entity = submodel.get_referable("ExampleEntity2") + + out_str = json.dumps(entity, cls=PathJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual(["ExampleEntity2"], parsed_output) + + def test_annotated_relationship_element_ok(self): + are = self.submodel.get_referable("ExampleAnnotatedRelationshipElement") + + out_str = json.dumps(are, cls=PathJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertCountEqual([ + "ExampleAnnotatedRelationshipElement", + "ExampleAnnotatedRelationshipElement.ExampleAnnotatedProperty", + "ExampleAnnotatedRelationshipElement.ExampleAnnotatedRange", + ], parsed_output) + + def test_annotated_relationship_element_without_annotation_ok(self): + submodel = example_aas_mandatory_attributes.create_example_submodel() + are = submodel.get_referable("ExampleAnnotatedRelationshipElement") + + out_str = json.dumps(are, cls=PathJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual(["ExampleAnnotatedRelationshipElement"], parsed_output) + + def test_operation_fail(self): + operation = self.submodel.get_referable("ExampleOperation") + + with self.assertRaises(TypeError): + json.dumps(operation, cls=PathJsonEncoder) + + def test_leaf_data_element_fail(self): + range_element = self.submodel.get_referable("ExampleSubmodelCollection.ExampleRange") + + with self.assertRaises(TypeError): + json.dumps(range_element, cls=PathJsonEncoder) + + def test_property_fail(self): + prop = model.Property("Prop", model.datatypes.String) + + with self.assertRaises(TypeError): + json.dumps(prop, cls=PathJsonEncoder) + + def test_unsupported_type_fail(self): + unsupported = object() + + with self.assertRaises(TypeError): + json.dumps(unsupported, cls=PathJsonEncoder) + + def test_asset_administration_shell_fail(self): + aas = example_aas.create_example_asset_administration_shell() + + with self.assertRaises(TypeError): + json.dumps(aas, cls=PathJsonEncoder) + + def test_concept_description_fail(self): + concept_description = example_aas.create_example_concept_description() + + with self.assertRaises(TypeError): + json.dumps(concept_description, cls=PathJsonEncoder) + + +class TestPathJsonEncoderOrder(unittest.TestCase): + def test_submodel_children_sorted_by_id_short_ok(self): + submodel = model.Submodel("https://example.org/Submodel", submodel_element=[ + model.SubmodelElementCollection("b", [ + model.Property("d", model.datatypes.String), + model.Property("c", model.datatypes.String), + ]), + model.Property("a", model.datatypes.String), + ]) + + out_str = json.dumps(submodel, cls=PathJsonEncoder) + + self.assertEqual('["a", "b", "b.c", "b.d"]', out_str) + + def test_annotations_sorted_by_id_short_ok(self): + relationship = model.AnnotatedRelationshipElement( + "Relationship", + first=model.ExternalReference((model.Key(model.KeyTypes.GLOBAL_REFERENCE, "first"),)), + second=model.ExternalReference((model.Key(model.KeyTypes.GLOBAL_REFERENCE, "second"),)), + annotation=[ + model.Property("b", model.datatypes.String), + model.Property("a", model.datatypes.String), + ], + ) + + out_str = json.dumps(relationship, cls=PathJsonEncoder) + + self.assertEqual('["Relationship", "Relationship.a", "Relationship.b"]', out_str) + + def test_entity_statements_sorted_by_id_short_ok(self): + entity = model.Entity("Entity", model.EntityType.CO_MANAGED_ENTITY, statement=[ + model.Property("b", model.datatypes.String), + model.Property("a", model.datatypes.String), + ]) + + out_str = json.dumps(entity, cls=PathJsonEncoder) + + self.assertEqual('["Entity", "Entity.a", "Entity.b"]', out_str) diff --git a/sdk/test/adapter/json/formats/test_reference_serialization.py b/sdk/test/adapter/json/formats/test_reference_serialization.py new file mode 100644 index 000000000..da4668851 --- /dev/null +++ b/sdk/test/adapter/json/formats/test_reference_serialization.py @@ -0,0 +1,43 @@ +# Copyright (c) 2026 the Eclipse BaSyx Authors +# +# This program and the accompanying materials are made available under the terms of the MIT License, available in +# the LICENSE file of this project. +# +# SPDX-License-Identifier: MIT +import json +import unittest + +from basyx.aas import model +from basyx.aas.adapter.json import ReferenceJsonEncoder +from basyx.aas.examples.data import example_aas + + +class TestReferenceSerialization(unittest.TestCase): + + def test_submodel_element(self): + sm = example_aas.create_example_submodel() + element = sm.get_referable("ExampleRelationshipElement") + + out_str = json.dumps(element, cls=ReferenceJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertIsInstance(parsed_output, dict) + self.assertEqual({"keys", "type"}, parsed_output.keys()) + self.assertGreater(len(parsed_output["keys"]), 0) + + def test_non_referable(self): + obj = model.AssetInformation(global_asset_id="https://example.or/TestAssetId") + + with self.assertRaises(TypeError): + json.dumps(obj, cls=ReferenceJsonEncoder) + + def test_list_of_elements(self): + objects = [example_aas.create_example_submodel(), example_aas.create_example_asset_administration_shell()] + + out_str = json.dumps(objects, cls=ReferenceJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertIsInstance(parsed_output, list) + for ref in parsed_output: + self.assertEqual({"keys", "type"}, ref.keys()) + self.assertGreater(len(ref["keys"]), 0) diff --git a/sdk/test/adapter/json/formats/test_value_only_serialization.py b/sdk/test/adapter/json/formats/test_value_only_serialization.py new file mode 100644 index 000000000..fe8345481 --- /dev/null +++ b/sdk/test/adapter/json/formats/test_value_only_serialization.py @@ -0,0 +1,694 @@ +# Copyright (c) 2026 the Eclipse BaSyx Authors +# +# This program and the accompanying materials are made available under the terms of the MIT License, available in +# the LICENSE file of this project. +# +# SPDX-License-Identifier: MIT +import decimal +import json +import math +import unittest + +from basyx.aas import model +from basyx.aas.adapter.json import ValueOnlyJsonEncoder +from basyx.aas.examples.data import example_aas, example_aas_mandatory_attributes + +EXAMPLE_SUBMODEL_ID = "http://example.org/Test_Submodel" + + +class TestValueOnlyJsonEncoder(unittest.TestCase): + def setUp(self): + self.submodel = example_aas.create_example_submodel() + + # ------------------------------------------------------------------ Submodel + + def test_submodel_ok(self): + submodel = example_aas.create_example_asset_identification_submodel() + + out_str = json.dumps(submodel, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual({"ManufacturerName": "ACPLT", "InstanceId": "978-8234-234-342"}, parsed_output) + + def test_submodel_omits_operation_and_capability_ok(self): + out_str = json.dumps(self.submodel, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertCountEqual([ + "ExampleRelationshipElement", + "ExampleAnnotatedRelationshipElement", + "ExampleBasicEventElement", + "ExampleSubmodelCollection", + ], parsed_output.keys()) + + def test_submodel_omits_elements_without_value_ok(self): + submodel = model.Submodel("https://example.org/Submodel", submodel_element=[ + model.Property("FilledProperty", model.datatypes.String, "exampleValue"), + model.Property("EmptyProperty", model.datatypes.String), + model.MultiLanguageProperty("EmptyMultiLanguageProperty"), + model.Range("EmptyRange", model.datatypes.Int), + model.File("EmptyFile", "application/pdf"), + model.ReferenceElement("EmptyReferenceElement"), + model.RelationshipElement("EmptyRelationshipElement"), + ]) + + out_str = json.dumps(submodel, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual({"FilledProperty": "exampleValue"}, parsed_output) + + def test_submodel_omits_containers_of_omitted_elements_ok(self): + submodel = model.Submodel("https://example.org/Submodel", submodel_element=[ + model.Property("FilledProperty", model.datatypes.String, "exampleValue"), + model.SubmodelElementCollection("CollectionOfOperations", [model.Operation("Operation")]), + model.SubmodelElementList("ListOfOperations", model.Operation, [model.Operation(None)]), + ]) + + out_str = json.dumps(submodel, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual({"FilledProperty": "exampleValue"}, parsed_output) + + def test_empty_submodel_ok(self): + submodel = example_aas_mandatory_attributes.create_example_empty_submodel() + + out_str = json.dumps(submodel, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual({}, parsed_output) + + # ------------------------------------------------------------------ SubmodelElementCollection / -List + + def test_submodel_element_collection_ok(self): + smc = self.submodel.get_referable("ExampleSubmodelCollection") + + out_str = json.dumps(smc, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertCountEqual([ + "ExampleBlob", + "ExampleFile", + "ExampleFileURI", + "ExampleMultiLanguageProperty", + "ExampleRange", + "ExampleReferenceElement", + "ExampleSubmodelList", + ], parsed_output.keys()) + self.assertEqual({"min": 0, "max": 100}, parsed_output["ExampleRange"]) + self.assertEqual(["exampleValue", "exampleValue"], parsed_output["ExampleSubmodelList"]) + + def test_submodel_element_list_ok(self): + sml = self.submodel.get_referable("ExampleSubmodelCollection.ExampleSubmodelList") + + out_str = json.dumps(sml, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual(["exampleValue", "exampleValue"], parsed_output) + + def test_submodel_element_list_keeps_index_of_element_without_value_ok(self): + sml = model.SubmodelElementList("List", model.Property, [ + model.Property(None, model.datatypes.String, "first"), + model.Property(None, model.datatypes.String), + model.Property(None, model.datatypes.String, "third"), + ], value_type_list_element=model.datatypes.String) + + out_str = json.dumps(sml, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual(["first", None, "third"], parsed_output) + + def test_empty_submodel_element_list_ok(self): + submodel = example_aas_mandatory_attributes.create_example_submodel() + sml = submodel.get_referable("ExampleSubmodelList2") + + out_str = json.dumps(sml, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual([], parsed_output) + + # ------------------------------------------------------------------ Property data type to value mapping + + def test_property_ok(self): + submodel = example_aas.create_example_asset_identification_submodel() + prop = submodel.get_referable("ManufacturerName") + + out_str = json.dumps(prop, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual("ACPLT", parsed_output) + + def test_property_without_value_ok(self): + prop = model.Property("EmptyProperty", model.datatypes.String) + + out_str = json.dumps(prop, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertIsNone(parsed_output) + + def test_property_boolean_types_ok(self): + smc = model.SubmodelElementCollection("Booleans", [ + model.Property("True", model.datatypes.Boolean, True), + model.Property("False", model.datatypes.Boolean, False), + ]) + + out_str = json.dumps(smc, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual({"True": True, "False": False}, parsed_output) + + def test_property_number_types_ok(self): + smc = model.SubmodelElementCollection("Numbers", [ + model.Property("Int", model.datatypes.Int, -1), + model.Property("Long", model.datatypes.Long, 9223372036854775807), + model.Property("UnsignedLong", model.datatypes.UnsignedLong, 18446744073709551615), + model.Property("Zero", model.datatypes.NonNegativeInteger, 0), + # xs:integer is unbounded, this value cannot be represented as a float + model.Property("HugeInteger", model.datatypes.Integer, 10**400), + model.Property("Decimal", model.datatypes.Decimal, decimal.Decimal("-1.23")), + model.Property("Double", model.datatypes.Double, 234.567e8), + model.Property("Float", model.datatypes.Float, 1.5), + ]) + + out_str = json.dumps(smc, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual({ + "Int": -1, + "Long": 9223372036854775807, + "UnsignedLong": 18446744073709551615, + "Zero": 0, + "HugeInteger": 10**400, + "Decimal": -1.23, + "Double": 234.567e8, + "Float": 1.5, + }, parsed_output) + + def test_property_non_finite_double_ok(self): + # JSON has no representation for NaN, INF and -INF, so the XSD representation is used instead + smc = model.SubmodelElementCollection("NonFinite", [ + model.Property("NaN", model.datatypes.Double, math.nan), + model.Property("Inf", model.datatypes.Double, math.inf), + model.Property("NegativeInf", model.datatypes.Double, -math.inf), + ]) + + out_str = json.dumps(smc, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual({"NaN": "NaN", "Inf": "INF", "NegativeInf": "-INF"}, parsed_output) + + def test_property_string_types_ok(self): + smc = model.SubmodelElementCollection("Strings", [ + model.Property("String", model.datatypes.String, "Hello world"), + model.Property("EmptyString", model.datatypes.String, ""), + model.Property("Date", model.datatypes.Date, model.datatypes.Date(2000, 1, 1)), + model.Property("Time", model.datatypes.Time, model.datatypes.Time(14, 23)), + model.Property("DateTime", model.datatypes.DateTime, model.datatypes.DateTime(2000, 1, 1, 14, 23)), + model.Property("GYear", model.datatypes.GYear, model.datatypes.GYear(2000)), + model.Property("Duration", model.datatypes.Duration, model.datatypes.Duration(days=30)), + model.Property("HexBinary", model.datatypes.HexBinary, model.datatypes.HexBinary(b"kumowashere")), + model.Property("Base64Binary", model.datatypes.Base64Binary, + model.datatypes.Base64Binary(b"Hello, World!")), + model.Property("AnyURI", model.datatypes.AnyURI, model.datatypes.AnyURI("urn:example:company:1.0.0")), + ]) + + out_str = json.dumps(smc, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual({ + "String": "Hello world", + "EmptyString": "", + "Date": "2000-01-01", + "Time": "14:23:00", + "DateTime": "2000-01-01T14:23:00", + "GYear": "2000", + "Duration": "P30D", + "HexBinary": "6b756d6f77617368657265", + "Base64Binary": "SGVsbG8sIFdvcmxkIQ==", + "AnyURI": "urn:example:company:1.0.0", + }, parsed_output) + + # ------------------------------------------------------------------ other SubmodelElements + + def test_multi_language_property_ok(self): + mlp = self.submodel.get_referable("ExampleSubmodelCollection.ExampleMultiLanguageProperty") + + out_str = json.dumps(mlp, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertCountEqual([ + {"de": "Beispielwert für ein MultiLanguageProperty-Element"}, + {"en-US": "Example value of a MultiLanguageProperty element"}, + ], parsed_output) + + def test_multi_language_property_without_value_ok(self): + mlp = model.MultiLanguageProperty("EmptyMultiLanguageProperty") + + out_str = json.dumps(mlp, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual([], parsed_output) + + def test_range_ok(self): + range_element = self.submodel.get_referable("ExampleSubmodelCollection.ExampleRange") + + out_str = json.dumps(range_element, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual({"min": 0, "max": 100}, parsed_output) + + def test_blob_ok(self): + blob = self.submodel.get_referable("ExampleSubmodelCollection.ExampleBlob") + + out_str = json.dumps(blob, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + # the value is base64 encoded b"\x01\x02\x03\x04\x05" + self.assertEqual({"contentType": "application/pdf", "value": "AQIDBAU="}, parsed_output) + + def test_file_ok(self): + file = self.submodel.get_referable("ExampleSubmodelCollection.ExampleFile") + + out_str = json.dumps(file, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual({"contentType": "application/pdf", "value": "/TestFile.pdf"}, parsed_output) + + def test_reference_element_ok(self): + reference_element = self.submodel.get_referable("ExampleSubmodelCollection.ExampleReferenceElement") + + out_str = json.dumps(reference_element, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual({ + "type": "ModelReference", + "keys": [ + {"type": "Submodel", "value": EXAMPLE_SUBMODEL_ID}, + {"type": "Property", "value": "ExampleProperty"}, + ], + }, parsed_output) + + def test_relationship_element_ok(self): + relationship_element = self.submodel.get_referable("ExampleRelationshipElement") + + out_str = json.dumps(relationship_element, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual({ + "first": { + "type": "ModelReference", + "keys": [ + {"type": "Submodel", "value": EXAMPLE_SUBMODEL_ID}, + {"type": "Property", "value": "ExampleProperty"}, + ], + }, + "second": { + "type": "ModelReference", + "keys": [ + {"type": "Submodel", "value": EXAMPLE_SUBMODEL_ID}, + {"type": "Property", "value": "ExampleProperty2"}, + ], + }, + }, parsed_output) + + def test_annotated_relationship_element_ok(self): + are = self.submodel.get_referable("ExampleAnnotatedRelationshipElement") + + out_str = json.dumps(are, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual({ + "first": { + "type": "ModelReference", + "keys": [ + {"type": "Submodel", "value": EXAMPLE_SUBMODEL_ID}, + {"type": "Property", "value": "ExampleProperty"}, + ], + }, + "second": { + "type": "ModelReference", + "keys": [ + {"type": "Submodel", "value": EXAMPLE_SUBMODEL_ID}, + {"type": "Property", "value": "ExampleProperty2"}, + ], + }, + "annotations": { + "ExampleAnnotatedProperty": "exampleValue", + "ExampleAnnotatedRange": {"min": 1, "max": 5}, + }, + }, parsed_output) + + def test_annotated_relationship_element_without_annotation_ok(self): + submodel = example_aas_mandatory_attributes.create_example_submodel() + are = submodel.get_referable("ExampleAnnotatedRelationshipElement") + + out_str = json.dumps(are, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertCountEqual(["first", "second"], parsed_output.keys()) + + def test_entity_ok(self): + submodel = example_aas.create_example_bill_of_material_submodel() + entity = submodel.get_referable("ExampleEntity") + + out_str = json.dumps(entity, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual({ + "statements": { + "ExampleProperty": "exampleValue", + "ExampleProperty2": "exampleValue2", + }, + "entityType": "SelfManagedEntity", + "globalAssetId": "http://example.org/TestAsset/", + "specificAssetIds": [{ + "name": "TestKey", + "value": "TestValue", + "externalSubjectId": { + "type": "ExternalReference", + "keys": [{"type": "GlobalReference", "value": "http://example.org/SpecificAssetId/"}], + }, + }], + }, parsed_output) + + def test_entity_without_statements_ok(self): + submodel = example_aas.create_example_bill_of_material_submodel() + entity = submodel.get_referable("ExampleEntity2") + + out_str = json.dumps(entity, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertNotIn("statements", parsed_output) + + def test_submodel_keeps_co_managed_entity_with_statements_ok(self): + submodel = model.Submodel("https://example.org/Submodel", submodel_element=[ + model.Entity("CoManagedEntity", model.EntityType.CO_MANAGED_ENTITY, statement=[ + model.Property("Statement", model.datatypes.String, "exampleValue"), + ]), + ]) + + out_str = json.dumps(submodel, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual({ + "CoManagedEntity": { + "statements": {"Statement": "exampleValue"}, + "entityType": "CoManagedEntity", + }, + }, parsed_output) + + def test_basic_event_element_ok(self): + basic_event_element = self.submodel.get_referable("ExampleBasicEventElement") + + out_str = json.dumps(basic_event_element, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual({ + "observed": { + "type": "ModelReference", + "keys": [ + {"type": "Submodel", "value": EXAMPLE_SUBMODEL_ID}, + {"type": "Property", "value": "ExampleProperty"}, + ], + }, + }, parsed_output) + + def test_operation_fail(self): + operation = self.submodel.get_referable("ExampleOperation") + + with self.assertRaises(TypeError): + json.dumps(operation, cls=ValueOnlyJsonEncoder) + + def test_capability_fail(self): + capability = self.submodel.get_referable("ExampleCapability") + + with self.assertRaises(TypeError): + json.dumps(capability, cls=ValueOnlyJsonEncoder) + +class TestValueOnlyJsonEncoderCore(unittest.TestCase): + + def setUp(self): + self.submodel = example_aas.create_example_submodel() + + def test_deep_is_default_ok(self): + deep_output = json.loads(json.dumps(self.submodel, cls=ValueOnlyJsonEncoder, deep=True)) + + out_str = json.dumps(self.submodel, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual(deep_output, parsed_output) + + def test_submodel_core_empties_submodel_element_collection_ok(self): + out_str = json.dumps(self.submodel, cls=ValueOnlyJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertEqual({}, parsed_output["ExampleSubmodelCollection"]) + + def test_submodel_element_collection_core_empties_submodel_element_list_ok(self): + smc = self.submodel.get_referable("ExampleSubmodelCollection") + + out_str = json.dumps(smc, cls=ValueOnlyJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertEqual([], parsed_output["ExampleSubmodelList"]) + self.assertEqual({"min": 0, "max": 100}, parsed_output["ExampleRange"]) + + def test_submodel_core_keeps_leaf_values_ok(self): + submodel = example_aas.create_example_asset_identification_submodel() + + out_str = json.dumps(submodel, cls=ValueOnlyJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertEqual({"ManufacturerName": "ACPLT", "InstanceId": "978-8234-234-342"}, parsed_output) + + def test_submodel_element_list_core_empties_submodel_element_collections_ok(self): + submodel = example_aas_mandatory_attributes.create_example_submodel() + sml = submodel.get_referable("ExampleSubmodelList") + + out_str = json.dumps(sml, cls=ValueOnlyJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertEqual([{}, {}], parsed_output) + + def test_submodel_core_empties_submodel_element_list_ok(self): + submodel = example_aas_mandatory_attributes.create_example_submodel() + + out_str = json.dumps(submodel, cls=ValueOnlyJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertEqual([], parsed_output["ExampleSubmodelList"]) + + def test_submodel_core_omits_entity_statements_ok(self): + submodel = example_aas.create_example_bill_of_material_submodel() + deep_output = json.loads(json.dumps(submodel, cls=ValueOnlyJsonEncoder)) + expected = deep_output["ExampleEntity"] + expected.pop("statements") + + out_str = json.dumps(submodel, cls=ValueOnlyJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertEqual(expected, parsed_output["ExampleEntity"]) + + def test_entity_core_keeps_statements_ok(self): + submodel = example_aas.create_example_bill_of_material_submodel() + entity = submodel.get_referable("ExampleEntity") + + out_str = json.dumps(entity, cls=ValueOnlyJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertEqual({"ExampleProperty": "exampleValue", "ExampleProperty2": "exampleValue2"}, + parsed_output["statements"]) + + def test_entity_core_empties_statement_submodel_element_collection_ok(self): + collection = model.SubmodelElementCollection( + "ExampleCollection", + value=[model.Property("ExampleProperty", model.datatypes.String, value="exampleValue")], + ) + entity = model.Entity("ExampleEntity", model.EntityType.CO_MANAGED_ENTITY, statement=[collection]) + + out_str = json.dumps(entity, cls=ValueOnlyJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertEqual({"ExampleCollection": {}}, parsed_output["statements"]) + + def test_submodel_core_omits_annotated_relationship_element_annotations_ok(self): + deep_output = json.loads(json.dumps(self.submodel, cls=ValueOnlyJsonEncoder)) + expected = deep_output["ExampleAnnotatedRelationshipElement"] + expected.pop("annotations") + + out_str = json.dumps(self.submodel, cls=ValueOnlyJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertEqual(expected, parsed_output["ExampleAnnotatedRelationshipElement"]) + + def test_annotated_relationship_element_core_keeps_annotations_ok(self): + are = self.submodel.get_referable("ExampleAnnotatedRelationshipElement") + deep_output = json.loads(json.dumps(are, cls=ValueOnlyJsonEncoder)) + + out_str = json.dumps(are, cls=ValueOnlyJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertEqual(deep_output, parsed_output) + + def test_leaf_element_core_ok(self): + range_element = self.submodel.get_referable("ExampleSubmodelCollection.ExampleRange") + + out_str = json.dumps(range_element, cls=ValueOnlyJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertEqual({"min": 0, "max": 100}, parsed_output) + + def test_submodel_core_keeps_containers_of_omitted_elements_ok(self): + submodel = model.Submodel("https://example.org/Submodel", submodel_element=[ + model.Property("FilledProperty", model.datatypes.String, "exampleValue"), + model.SubmodelElementCollection("CollectionOfOperations", [model.Operation("Operation")]), + model.SubmodelElementList("ListOfOperations", model.Operation, [model.Operation(None)]), + ]) + + out_str = json.dumps(submodel, cls=ValueOnlyJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertEqual({ + "FilledProperty": "exampleValue", + "CollectionOfOperations": {}, + "ListOfOperations": [], + }, parsed_output) + + +class WithoutBlobValueValueOnlyJsonEncoder(ValueOnlyJsonEncoder): + with_blob_value = False + + +class TestValueOnlyJsonEncoderBlobValue(unittest.TestCase): + def setUp(self): + self.submodel = example_aas.create_example_submodel() + self.blob = self.submodel.get_referable("ExampleSubmodelCollection.ExampleBlob") + + def test_with_blob_value_is_default_ok(self): + out_str = json.dumps(self.blob, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertTrue(ValueOnlyJsonEncoder.with_blob_value) + # the value is base64 encoded b"\x01\x02\x03\x04\x05" + self.assertEqual({"contentType": "application/pdf", "value": "AQIDBAU="}, parsed_output) + + def test_without_blob_value_omits_value_ok(self): + out_str = json.dumps(self.blob, cls=WithoutBlobValueValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual({"contentType": "application/pdf"}, parsed_output) + + def test_without_blob_value_keeps_nested_blob_ok(self): + out_str = json.dumps(self.submodel, cls=WithoutBlobValueValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual({"contentType": "application/pdf"}, parsed_output["ExampleSubmodelCollection"]["ExampleBlob"]) + + def test_without_blob_value_core_omits_blob_value_ok(self): + smc = self.submodel.get_referable("ExampleSubmodelCollection") + + out_str = json.dumps(smc, cls=WithoutBlobValueValueOnlyJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertEqual({"contentType": "application/pdf"}, parsed_output["ExampleBlob"]) + + def test_with_blob_value_empty_value_ok(self): + blob = model.Blob("EmptyBlob", "application/octet-stream", value=b"") + + out_str = json.dumps(blob, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual({"contentType": "application/octet-stream", "value": ""}, parsed_output) + + def test_with_blob_value_without_value_ok(self): + blob = model.Blob("BlobWithoutValue", "application/octet-stream") + + out_str = json.dumps(blob, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual({"contentType": "application/octet-stream"}, parsed_output) + + +class TestValueOnlyJsonEncoderOrder(unittest.TestCase): + def test_submodel_elements_sorted_by_id_short_ok(self): + submodel = model.Submodel("https://example.org/Submodel", submodel_element=[ + model.Property("c", model.datatypes.String, "c"), + model.Property("a", model.datatypes.String, "a"), + model.Property("b", model.datatypes.String, "b"), + ]) + + out_str = json.dumps(submodel, cls=ValueOnlyJsonEncoder) + + self.assertEqual('{"a": "a", "b": "b", "c": "c"}', out_str) + + def test_submodel_element_collection_value_sorted_by_id_short_ok(self): + collection = model.SubmodelElementCollection("Collection", [ + model.Property("b", model.datatypes.String, "b"), + model.Property("a", model.datatypes.String, "a"), + ]) + + out_str = json.dumps(collection, cls=ValueOnlyJsonEncoder) + + self.assertEqual('{"a": "a", "b": "b"}', out_str) + + def test_annotations_sorted_by_id_short_ok(self): + relationship = model.AnnotatedRelationshipElement( + "Relationship", + first=model.ExternalReference((model.Key(model.KeyTypes.GLOBAL_REFERENCE, "first"),)), + second=model.ExternalReference((model.Key(model.KeyTypes.GLOBAL_REFERENCE, "second"),)), + annotation=[ + model.Property("b", model.datatypes.String, "b"), + model.Property("a", model.datatypes.String, "a"), + ], + ) + + out_str = json.dumps(relationship, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual(["a", "b"], list(parsed_output["annotations"])) + + def test_entity_statements_sorted_by_id_short_ok(self): + entity = model.Entity("Entity", model.EntityType.CO_MANAGED_ENTITY, statement=[ + model.Property("b", model.datatypes.String, "b"), + model.Property("a", model.datatypes.String, "a"), + ]) + + out_str = json.dumps(entity, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual(["a", "b"], list(parsed_output["statements"])) + + def test_entity_specific_asset_ids_sorted_ok(self): + entity = model.Entity("Entity", model.EntityType.SELF_MANAGED_ENTITY, specific_asset_id=[ + model.SpecificAssetId("b", "1"), + model.SpecificAssetId("a", "2"), + model.SpecificAssetId("a", "1"), + ]) + + out_str = json.dumps(entity, cls=ValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual( + [("a", "1"), ("a", "2"), ("b", "1")], + [(asset_id["name"], asset_id["value"]) for asset_id in parsed_output["specificAssetIds"]], + ) + + def test_submodel_core_containers_without_value_sorted_by_id_short_ok(self): + submodel = model.Submodel("https://example.org/Submodel", submodel_element=[ + model.SubmodelElementCollection("a", [model.Operation("Operation")]), + model.Property("b", model.datatypes.String, "b"), + ]) + + out_str = json.dumps(submodel, cls=ValueOnlyJsonEncoder, deep=False) + + self.assertEqual('{"a": {}, "b": "b"}', out_str) + + def test_entity_core_containers_without_value_sorted_by_id_short_ok(self): + entity = model.Entity("Entity", model.EntityType.CO_MANAGED_ENTITY, statement=[ + model.SubmodelElementCollection("a", [model.Operation("Operation")]), + model.Property("b", model.datatypes.String, "b"), + ]) + + out_str = json.dumps(entity, cls=ValueOnlyJsonEncoder, deep=False) + parsed_output = json.loads(out_str) + + self.assertEqual(["a", "b"], list(parsed_output["statements"])) diff --git a/sdk/test/adapter/json/test_json_serialization.py b/sdk/test/adapter/json/test_json_serialization.py index b3d689d15..0e749910b 100644 --- a/sdk/test/adapter/json/test_json_serialization.py +++ b/sdk/test/adapter/json/test_json_serialization.py @@ -14,6 +14,8 @@ from basyx.aas.adapter.json import ( AASToJsonEncoder, StrippedAASToJsonEncoder, + object_store_to_json, + read_aas_json_file, write_aas_json_file, ) from basyx.aas.examples.data import ( @@ -314,3 +316,232 @@ def test_stripped_asset_administration_shell(self) -> None: ) self._checkNormalAndStripped({"submodels"}, aas) + + +class WithoutBlobValueAASToJsonEncoder(AASToJsonEncoder): + with_blob_value = False + + +class TestAASToJsonEncoderBlobValue(unittest.TestCase): + def setUp(self): + self.submodel = example_aas.create_example_submodel() + self.blob = self.submodel.get_referable("ExampleSubmodelCollection.ExampleBlob") + + def test_with_blob_value_is_default_ok(self): + out_str = json.dumps(self.blob, cls=AASToJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertTrue(AASToJsonEncoder.with_blob_value) + # the value is base64 encoded b"\x01\x02\x03\x04\x05" + self.assertEqual("AQIDBAU=", parsed_output["value"]) + + def test_without_blob_value_omits_value_ok(self): + expected = json.loads(json.dumps(self.blob, cls=AASToJsonEncoder)) + expected.pop("value") + + out_str = json.dumps(self.blob, cls=WithoutBlobValueAASToJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual(expected, parsed_output) + + def test_with_blob_value_empty_value_ok(self): + blob = model.Blob("EmptyBlob", "application/octet-stream", value=b"") + + out_str = json.dumps(blob, cls=AASToJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual("", parsed_output["value"]) + + def test_with_blob_value_without_value_ok(self): + blob = model.Blob("BlobWithoutValue", "application/octet-stream") + + out_str = json.dumps(blob, cls=AASToJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertNotIn("value", parsed_output) + + +def _external_reference(*values: str) -> model.ExternalReference: + return model.ExternalReference( + tuple(model.Key(model.KeyTypes.GLOBAL_REFERENCE, value) for value in values) + ) + + +def _submodel_reference(identifier: str) -> model.ModelReference: + return model.ModelReference( + (model.Key(model.KeyTypes.SUBMODEL, identifier),), model.Submodel + ) + + +class TestAASToJsonEncoderOrder(unittest.TestCase): + """ + Collections that are unordered according to the specification must be serialized in a canonical order, so that + equal objects always produce identical JSON strings, independent of insertion order and hash randomization. + The elements are always inserted in reverse order of the expected output. + """ + + def test_referables_start_with_modelType_ok(self) -> None: + referables = [ + model.AssetAdministrationShell( + model.AssetInformation(global_asset_id="urn:asset"), "urn:aas" + ), + model.Submodel("urn:sm"), + model.ConceptDescription("urn:cd"), + model.Property("prop", model.datatypes.String), + model.MultiLanguageProperty("mlp"), + model.Range("range", model.datatypes.Int), + model.Blob("blob", "application/octet-stream"), + model.File("file", "text/plain"), + model.ReferenceElement("ref"), + model.SubmodelElementCollection("smc"), + model.SubmodelElementList("sml", model.Capability), + model.RelationshipElement("rel"), + model.AnnotatedRelationshipElement("arel"), + model.Operation("op"), + model.Capability("cap"), + model.Entity("entity", model.EntityType.CO_MANAGED_ENTITY), + model.BasicEventElement( + "event", + _submodel_reference("urn:sm"), + model.Direction.OUTPUT, + model.StateOfEvent.ON, + ), + ] + + data = [json.loads(json.dumps(r, cls=AASToJsonEncoder)) for r in referables] + + self.assertEqual( + [(type(r).__name__, "modelType") for r in referables], + [(type(r).__name__, next(iter(d))) for r, d in zip(referables, data)], + ) + + def test_extensions_sorted_by_name_ok(self) -> None: + submodel = model.Submodel( + "urn:sm", + extension=[model.Extension(name) for name in ["e2", "e1", "e0"]], + ) + + data = json.loads(json.dumps(submodel, cls=AASToJsonEncoder)) + + self.assertEqual(["e0", "e1", "e2"], [e["name"] for e in data["extensions"]]) + + def test_embedded_data_specifications_sorted_by_data_specification_ok(self) -> None: + content = model.DataSpecificationIEC61360( + model.PreferredNameTypeIEC61360({"en": "name"}) + ) + submodel = model.Submodel( + "urn:sm", + embedded_data_specifications=[ + model.EmbeddedDataSpecification(_external_reference(value), content) + for value in ["urn:ds2", "urn:ds1", "urn:ds0"] + ], + ) + + data = json.loads(json.dumps(submodel, cls=AASToJsonEncoder)) + + self.assertEqual( + ["urn:ds0", "urn:ds1", "urn:ds2"], + [ + spec["dataSpecification"]["keys"][0]["value"] + for spec in data["embeddedDataSpecifications"] + ], + ) + + + def test_lang_string_sets_sorted_by_language_ok(self) -> None: + prop = model.Property( + "prop", + model.datatypes.String, + display_name=model.MultiLanguageNameType({"fr": "n", "en": "n", "de": "n"}), + description=model.MultiLanguageTextType({"fr": "t", "en": "t", "de": "t"}), + ) + + data = json.loads(json.dumps(prop, cls=AASToJsonEncoder)) + + self.assertEqual( + ["de", "en", "fr"], [s["language"] for s in data["displayName"]] + ) + self.assertEqual( + ["de", "en", "fr"], [s["language"] for s in data["description"]] + ) + + def test_qualifiers_sorted_by_type_ok(self) -> None: + submodel = model.Submodel( + "urn:sm", + qualifier=[ + model.Qualifier(type_, model.datatypes.String) + for type_ in ["q2", "q1", "q0"] + ], + ) + + data = json.loads(json.dumps(submodel, cls=AASToJsonEncoder)) + + self.assertEqual(["q0", "q1", "q2"], [q["type"] for q in data["qualifiers"]]) + + def test_entity_specific_asset_ids_sorted_ok(self) -> None: + entity = model.Entity( + "entity", + model.EntityType.SELF_MANAGED_ENTITY, + specific_asset_id=[ + model.SpecificAssetId(name, "v") for name in ["n2", "n1", "n0"] + ], + ) + + data = json.loads(json.dumps(entity, cls=AASToJsonEncoder)) + + self.assertEqual( + ["n0", "n1", "n2"], [s["name"] for s in data["specificAssetIds"]] + ) + + def test_asset_administration_shell_submodels_sorted_ok(self) -> None: + aas = model.AssetAdministrationShell( + model.AssetInformation(global_asset_id="urn:asset"), + "urn:aas", + submodel={_submodel_reference(i) for i in ["urn:sm2", "urn:sm1", "urn:sm0"]}, + ) + + data = json.loads(json.dumps(aas, cls=AASToJsonEncoder)) + + self.assertEqual( + ["urn:sm0", "urn:sm1", "urn:sm2"], + [ref["keys"][0]["value"] for ref in data["submodels"]], + ) + + def test_submodel_elements_sorted_by_id_short_ok(self) -> None: + submodel = model.Submodel( + "urn:sm", + submodel_element=[ + model.Property(id_short, model.datatypes.String) + for id_short in ["p2", "p1", "p0"] + ], + ) + + data = json.loads(json.dumps(submodel, cls=AASToJsonEncoder)) + + self.assertEqual( + ["p0", "p1", "p2"], [e["idShort"] for e in data["submodelElements"]] + ) + + def test_submodel_element_collection_sorted_by_id_short_ok(self) -> None: + collection = model.SubmodelElementCollection( + "smc", + value=[ + model.Property(id_short, model.datatypes.String) + for id_short in ["p2", "p1", "p0"] + ], + ) + + data = json.loads(json.dumps(collection, cls=AASToJsonEncoder)) + + self.assertEqual(["p0", "p1", "p2"], [e["idShort"] for e in data["value"]]) + + def test_object_store_identifiables_sorted_by_id_ok(self) -> None: + store: model.DictIdentifiableStore[model.Identifiable] = model.DictIdentifiableStore( + [model.Submodel(identifier) for identifier in ["urn:sm2", "urn:sm1", "urn:sm0"]] + ) + + data = json.loads(object_store_to_json(store)) + + self.assertEqual( + ["urn:sm0", "urn:sm1", "urn:sm2"], [sm["id"] for sm in data["submodels"]] + ) \ No newline at end of file diff --git a/server/README.md b/server/README.md index c09d565a1..4bf0a61df 100644 --- a/server/README.md +++ b/server/README.md @@ -130,8 +130,8 @@ Several features and routes are currently not supported: - `/description` These routes are not implemented at this time. -5. Value, Path, and PATCH Routes: - - All `/…/value$`, `/…/path$`, and `PATCH` routes are currently not implemented. +5. Path and PATCH Routes: + - All `/…/$path` and `PATCH` routes are currently not implemented. 6. Operation Invocation Routes: The following routes are not implemented because operation invocation is not yet supported by the `basyx-python-sdk`: diff --git a/server/app/adapter/__init__.py b/server/app/adapter/__init__.py index 10c99aa69..65ad6b609 100644 --- a/server/app/adapter/__init__.py +++ b/server/app/adapter/__init__.py @@ -1 +1,3 @@ from .jsonization import * +from .modifiers import * +from .value_only import * diff --git a/server/app/adapter/modifiers.py b/server/app/adapter/modifiers.py new file mode 100644 index 000000000..7eef67ce3 --- /dev/null +++ b/server/app/adapter/modifiers.py @@ -0,0 +1,70 @@ +# Copyright (c) 2026 the Eclipse BaSyx Authors +# +# This program and the accompanying materials are made available under the terms of the MIT License, available in +# the LICENSE file of this project. +# +# SPDX-License-Identifier: MIT +""" +Central parsing and validation of serialization modifiers +""" + +from dataclasses import dataclass +from enum import Enum +from typing import Optional + +from werkzeug import Request +from werkzeug.exceptions import BadRequest + +__all__ = ["Content", "Modifiers", "parse_modifiers"] + + +class Content(str, Enum): + NORMAL = "normal" + METADATA = "metadata" + VALUE = "value" + REFERENCE = "reference" + PATH = "path" + + +@dataclass(frozen=True) +class Modifiers: + content: Content + is_deep: bool + is_with_blob: bool + + +def parse_modifiers(request: Request, content: Optional[Content] = None) -> Modifiers: + # content is None when URL carried no $... suffix. Default: normal + resolved = content if content is not None else Content.NORMAL + is_deep = _is_deep_request(request, resolved) + is_with_blob = _is_with_blob_request(request, resolved) + return Modifiers(content=resolved, is_deep=is_deep, is_with_blob=is_with_blob) + + +def _is_deep_request(request: Request, content: Content) -> bool: + raw = request.args.get("level") + if raw is None: + # Spec default is deep + return True + if content is Content.METADATA: + raise BadRequest("Parameter 'level' is not allowed with /$metadata.") + if content is Content.REFERENCE and raw == "deep": + raise BadRequest("Parameter 'level=deep' is not allowed with /$reference.") + if raw not in {"deep", "core"}: + raise BadRequest(f"Invalid value for 'level': {raw!r}. Expected 'deep' or 'core'.") + return raw == "deep" + + +def _is_with_blob_request(request: Request, content: Content) -> bool: + raw = request.args.get("extent") + if raw is None: + return False + norm = raw.lower() + if norm not in {"withblobvalue", "withoutblobvalue"}: + raise BadRequest( + f"Invalid value for 'extent': {raw!r}. Expected 'WithBLOBValue' or 'WithoutBLOBValue'." + ) + want_blob = norm == "withblobvalue" + if want_blob and content in {Content.METADATA, Content.REFERENCE, Content.PATH}: + raise BadRequest(f"Parameter 'extent=WithBLOBValue' is not allowed with /${content.value}.") + return want_blob diff --git a/server/app/adapter/value_only.py b/server/app/adapter/value_only.py new file mode 100644 index 000000000..8eeb61a7a --- /dev/null +++ b/server/app/adapter/value_only.py @@ -0,0 +1,223 @@ +# Copyright (c) 2026 the Eclipse BaSyx Authors +# +# This program and the accompanying materials are made available under the terms of the MIT License, available in +# the LICENSE file of this project. +# +# SPDX-License-Identifier: MIT +""" +ValueOnly serialization of Submodels and SubmodelElements, as defined in "Specification of the Asset Administration +Shell Part 1", section "Value-Only Serialization in JSON". + +The ValueOnly serialization cannot be derived from the JSON schema of the metamodel, since it defines individual rules +per class. Therefore, it is implemented by hand here, on top of the object model of the ``basyx-python-sdk``. + +The returned structures are JSON-serializable with :class:`~basyx.aas.adapter.json.AASToJsonEncoder` (or any subclass, +e.g. the ``ResultToJsonEncoder`` of this server): :class:`~basyx.aas.model.base.Reference` and +:class:`~basyx.aas.model.base.SpecificAssetId` objects are passed through as-is instead of being converted here, to +avoid duplicating the serialization rules of the SDK. +""" + +import base64 +import decimal +import math +from typing import Any, Dict, Iterable, List, Optional + +from basyx.aas import model +from basyx.aas.adapter._generic import ENTITY_TYPES + +__all__ = [ + "has_value_only_representation", + "submodel_element_to_named_value_only", + "submodel_element_to_value_only", + "submodel_to_value_only", +] + +# SubmodelElement types that don't have a ValueOnly representation. Operations only have one in the context of their +# invocation, which is not supported by this server. +_TYPES_WITHOUT_VALUE_ONLY = (model.Operation, model.Capability) + + +def has_value_only_representation(element: model.SubmodelElement) -> bool: + """ + Whether a SubmodelElement has a ValueOnly representation at all. + + :param element: The SubmodelElement to check + :return: ``False`` for Operations and Capabilities, ``True`` otherwise + """ + return not isinstance(element, _TYPES_WITHOUT_VALUE_ONLY) + + +def submodel_to_value_only(submodel: model.Submodel, deep: bool = True) -> Dict[str, Any]: + """ + Serialize a Submodel in its ValueOnly representation. + + :param submodel: The Submodel to serialize + :param deep: If ``False`` (i.e. ``level=core``), only the first level of SubmodelElements is serialized and nested + containers are returned empty + :return: A dictionary, mapping the idShort of each SubmodelElement to its value + """ + return _namespace_to_value_only(submodel.submodel_element, _initial_depth(deep)) + + +def submodel_element_to_value_only(element: model.SubmodelElement, deep: bool = True) -> Any: + """ + Serialize a SubmodelElement in its ValueOnly representation. + + :param element: The SubmodelElement to serialize + :param deep: If ``False`` (i.e. ``level=core``), only the first level of child elements is serialized and nested + containers are returned empty + :raises TypeError: If the given element doesn't have a ValueOnly representation, see + :func:`has_value_only_representation` + :return: The value of the element, ``None`` if it doesn't have one + """ + return _element_to_value_only(element, _initial_depth(deep)) + + +def _initial_depth(deep: bool) -> float: + # The number of container levels that are serialized below the requested resource. level=core includes the direct + # children of the requested resource, but returns their children as empty containers. + return math.inf if deep else 1 + + +def _namespace_to_value_only(elements: Iterable[model.SubmodelElement], depth: float) -> Dict[str, Any]: + data: Dict[str, Any] = {} + if depth <= 0: + return data + for element in elements: + if not has_value_only_representation(element): + continue + value = _element_to_value_only(element, depth - 1) + # Elements without a value are omitted + if value is None: + continue + # The idShort is always set for elements that aren't contained in a SubmodelElementList + assert element.id_short is not None + data[element.id_short] = value + return data + + +def _element_to_value_only(element: model.SubmodelElement, depth: float) -> Any: + if isinstance(element, model.Property): + return _typed_value_to_value_only(element.value, element.value_type) + if isinstance(element, model.MultiLanguageProperty): + return _lang_string_set_to_value_only(element.value) + if isinstance(element, model.Range): + return _range_to_value_only(element) + if isinstance(element, model.Blob): + return { + "contentType": element.content_type, + "value": None if element.value is None else base64.b64encode(element.value).decode(), + } + if isinstance(element, model.File): + return {"contentType": element.content_type, "value": element.value} + if isinstance(element, model.ReferenceElement): + return element.value + # AnnotatedRelationshipElement is a subclass of RelationshipElement and must therefore be checked first + if isinstance(element, model.AnnotatedRelationshipElement): + return { + "first": element.first, + "second": element.second, + "annotations": _annotations_to_value_only(element.annotation, depth), + } + if isinstance(element, model.RelationshipElement): + return {"first": element.first, "second": element.second} + if isinstance(element, model.Entity): + return _entity_to_value_only(element, depth) + if isinstance(element, model.BasicEventElement): + return {"observed": element.observed} + if isinstance(element, model.SubmodelElementList): + return _submodel_element_list_to_value_only(element, depth) + if isinstance(element, model.SubmodelElementCollection): + return _namespace_to_value_only(element.value, depth) + raise TypeError(f"{element!r} doesn't have a ValueOnly representation!") + + +def _typed_value_to_value_only(value: Optional[model.ValueDataType], value_type: model.DataTypeDefXsd) -> Any: + """ + Map a typed value to its JSON representation: booleans become JSON booleans, numeric types become JSON numbers and + everything else becomes a JSON string containing the XSD representation of the value. + """ + if value is None: + return None + if value_type is model.datatypes.Boolean or isinstance(value, bool): + return bool(value) + if isinstance(value, int): + # int subclasses are serialized as JSON numbers as-is. xs:integer is unbounded, so no conversion to float may + # happen here, as that would raise an OverflowError for large values. + return value + if isinstance(value, (float, decimal.Decimal)): + # JSON doesn't support NaN, INF and -INF. The specification acknowledges this gap without resolving it, so the + # XSD representation is returned as a string instead of emitting invalid JSON. This also catches values of + # decimal.Decimal that are finite, but too large to be represented as a float. + if not math.isfinite(float(value)): + return model.datatypes.xsd_repr(value) + # decimal.Decimal cannot be serialized as JSON and may lose precision here + return float(value) if isinstance(value, decimal.Decimal) else value + return model.datatypes.xsd_repr(value) + + +def _lang_string_set_to_value_only(value: Optional[model.MultiLanguageTextType]) -> List[Dict[str, str]]: + if value is None: + return [] + # Sorted by language, to keep the order of the languages deterministic across requests + return [{language: text} for language, text in sorted(value.items())] + + +def _range_to_value_only(element: model.Range) -> Dict[str, Any]: + data: Dict[str, Any] = {} + if element.min is not None: + data["min"] = _typed_value_to_value_only(element.min, element.value_type) + if element.max is not None: + data["max"] = _typed_value_to_value_only(element.max, element.value_type) + return data + + +def _annotations_to_value_only(annotations: Iterable[model.DataElement], depth: float) -> List[Dict[str, Any]]: + if depth <= 0: + return [] + return [ + {annotation.id_short: _element_to_value_only(annotation, depth - 1)} + for annotation in annotations + if annotation.id_short is not None + ] + + +def _entity_to_value_only(element: model.Entity, depth: float) -> Dict[str, Any]: + data: Dict[str, Any] = {"statements": _namespace_to_value_only(element.statement, depth)} + # entityType is mandatory in the specification, but optional in the object model of the SDK + if element.entity_type is not None: + data["entityType"] = ENTITY_TYPES[element.entity_type] + if element.global_asset_id is not None: + data["globalAssetId"] = element.global_asset_id + if len(element.specific_asset_id) > 0: + data["specificAssetIds"] = list(element.specific_asset_id) + return data + + +def _submodel_element_list_to_value_only(element: model.SubmodelElementList, depth: float) -> List[Any]: + if depth <= 0: + return [] + # In contrast to the other containers, elements without a value are not omitted here, to keep the indices of the + # remaining elements intact + return [ + _element_to_value_only(child, depth - 1) if has_value_only_representation(child) else None + for child in element.value + ] + + +def submodel_element_to_named_value_only(element: model.SubmodelElement, deep: bool = True) -> Dict[str, Any]: + """ + Serialize a SubmodelElement as a single-entry dictionary, mapping its idShort to its ValueOnly representation. + + This is the representation used for the entries of ``GET /submodels/{submodelIdentifier}/submodel-elements/$value``. + The specification leaves the shape of these entries open: a bare ``SubmodelElementValue`` would lose the + association between a value and the element it belongs to, hence the elements are named here, just like they are + within the ValueOnly representation of their parent. + + :param element: The SubmodelElement to serialize + :param deep: If ``False`` (i.e. ``level=core``), only the first level of child elements is serialized + :return: A dictionary with a single entry, mapping the idShort of the given element to its value + """ + # The idShort is always set for elements that aren't contained in a SubmodelElementList + assert element.id_short is not None + return {element.id_short: submodel_element_to_value_only(element, deep)} diff --git a/server/app/interfaces/base.py b/server/app/interfaces/base.py index ac974cbad..4cd19216e 100644 --- a/server/app/interfaces/base.py +++ b/server/app/interfaces/base.py @@ -17,6 +17,7 @@ import werkzeug.utils from basyx.aas import model from basyx.aas.adapter._generic import XML_NS_MAP +from basyx.aas.adapter.json import MetadataJsonEncoder, PathJsonEncoder, ReferenceJsonEncoder, ValueOnlyJsonEncoder from basyx.aas.adapter.xml import XMLConstructables, read_aas_xml_element, xml_serialization from basyx.aas.model import AbstractObjectStore from basyx.aas.model.datatypes import NonNegativeInteger @@ -26,7 +27,13 @@ from werkzeug.routing import MapAdapter import app.model -from app.adapter import ServerAASToJsonEncoder, ServerStrictAASFromJsonDecoder, ServerStrictStrippedAASFromJsonDecoder +from app.adapter import ( + Content, + Modifiers, + ServerAASToJsonEncoder, + ServerStrictAASFromJsonDecoder, + ServerStrictStrippedAASFromJsonDecoder, +) from app.model import AssetAdministrationShellDescriptor, AssetLink, SubmodelDescriptor from app.util.converters import base64url_decode @@ -110,17 +117,22 @@ def serialize(self, obj: ResponseData, paging_metadata: Optional[PagingMetadata] class JsonResponse(APIResponse): - def __init__(self, *args, content_type="application/json", **kwargs): + def __init__(self, *args, modifiers: Optional[Modifiers] = None, content_type="application/json", **kwargs): + self._modifiers = modifiers super().__init__(*args, **kwargs, content_type=content_type) def serialize(self, obj: ResponseData, paging_metadata: Optional[PagingMetadata], stripped: bool) -> str: + # TODO: stripped is still handled here, needs to be replaced see #650 + if self._modifiers is not None: + encoder_cls, encoder_kwargs = pick_encoder(self._modifiers) + else: + encoder_cls = StrippedResultToJsonEncoder if stripped else ResultToJsonEncoder + encoder_kwargs = {} if paging_metadata is None: data = obj else: data = {"paging_metadata": paging_metadata, "result": obj} - return json.dumps( - data, cls=StrippedResultToJsonEncoder if stripped else ResultToJsonEncoder, separators=(",", ":") - ) + return json.dumps(data, cls=encoder_cls, separators=(",", ":"), **encoder_kwargs) class XmlResponse(APIResponse): @@ -221,6 +233,43 @@ class StrippedResultToJsonEncoder(ResultToJsonEncoder): stripped = True +# TODO need to be added + +class ResultMetadataJsonEncoder(ResultToJsonEncoder, MetadataJsonEncoder): + pass + + +class ResultReferenceJsonEncoder(ResultToJsonEncoder, ReferenceJsonEncoder): + pass + + +class ResultValueOnlyJsonEncoder(ResultToJsonEncoder, ValueOnlyJsonEncoder): + pass + + +class ResultPathJsonEncoder(ResultToJsonEncoder, PathJsonEncoder): + pass + + +def pick_encoder(mods: Modifiers) -> Tuple[Type["ResultToJsonEncoder"], Dict[str, Any]]: + """Pick a JSON encoder class + encoder-constructor kwargs for the given modifier combination. + + Encoders that take per-request axis state (e.g. ``deep``) receive it via kwargs here; + ``json.dumps`` forwards them into the encoder's ``__init__``. + """ + match mods.content: + case Content.NORMAL: + return ResultToJsonEncoder, {"deep": mods.is_deep} + case Content.METADATA: + return ResultMetadataJsonEncoder, {} + case Content.VALUE: + return ResultValueOnlyJsonEncoder, {"deep": mods.is_deep} + case Content.REFERENCE: + return ResultReferenceJsonEncoder, {} + case Content.PATH: + return ResultPathJsonEncoder, {"deep": mods.is_deep} + + class BaseWSGIApp: url_map: werkzeug.routing.Map diff --git a/server/app/interfaces/repository.py b/server/app/interfaces/repository.py index fc7b4b0a0..bdd4fa45b 100644 --- a/server/app/interfaces/repository.py +++ b/server/app/interfaces/repository.py @@ -23,11 +23,23 @@ from werkzeug.routing import MapAdapter, Rule, Submount from app._config import API_BASE_PATH +from app.adapter import ( + has_value_only_representation, + parse_modifiers, + submodel_element_to_named_value_only, + submodel_element_to_value_only, + submodel_to_value_only, +) from app.interfaces.base import PagingMetadata from app.model import ServiceDescription, ServiceSpecificationProfileEnum -from app.util.converters import IdentifierToBase64URLConverter, IdShortPathConverter, base64url_decode +from app.util.converters import ( + ContentModifierConverter, + IdentifierToBase64URLConverter, + IdShortPathConverter, + base64url_decode, +) -from .base import APIResponse, HTTPApiDecoder, ObjectStoreWSGIApp, T, is_stripped_request +from .base import APIResponse, HTTPApiDecoder, JsonResponse, ObjectStoreWSGIApp, T, is_stripped_request SUPPORTED_PROFILES: ServiceDescription = ServiceDescription( [ @@ -123,21 +135,22 @@ def __init__( [ Rule("/$metadata", methods=["GET"], endpoint=self.get_submodel_all_metadata), Rule("/$reference", methods=["GET"], endpoint=self.get_submodel_all_reference), - Rule("/$value", methods=["GET"], endpoint=self.not_implemented), + Rule("/$value", methods=["GET"], endpoint=self.get_submodel_all_value), Rule("/$path", methods=["GET"], endpoint=self.not_implemented), Rule("/", methods=["GET"], endpoint=self.get_submodel), + Rule( + "//", + methods=["GET"], + endpoint=self.get_submodel, + ), Rule("/", methods=["PUT"], endpoint=self.put_submodel), Rule("/", methods=["DELETE"], endpoint=self.delete_submodel), Rule("/", methods=["PATCH"], endpoint=self.not_implemented), Submount( "/", [ - Rule("/$metadata", methods=["GET"], endpoint=self.get_submodels_metadata), Rule("/$metadata", methods=["PATCH"], endpoint=self.not_implemented), - Rule("/$value", methods=["GET"], endpoint=self.not_implemented), Rule("/$value", methods=["PATCH"], endpoint=self.not_implemented), - Rule("/$reference", methods=["GET"], endpoint=self.get_submodels_reference), - Rule("/$path", methods=["GET"], endpoint=self.not_implemented), Rule( "/submodel-elements", methods=["GET"], @@ -161,7 +174,11 @@ def __init__( methods=["GET"], endpoint=self.get_submodel_submodel_elements_reference, ), - Rule("/$value", methods=["GET"], endpoint=self.not_implemented), + Rule( + "/$value", + methods=["GET"], + endpoint=self.get_submodel_submodel_elements_value, + ), Rule("/$path", methods=["GET"], endpoint=self.not_implemented), Rule( "/", @@ -206,7 +223,11 @@ def __init__( methods=["GET"], endpoint=self.get_submodel_submodel_elements_id_short_path_reference, ), - Rule("/$value", methods=["GET"], endpoint=self.not_implemented), + Rule( + "/$value", + methods=["GET"], + endpoint=self.get_submodel_submodel_elements_id_short_path_value, + ), Rule( "/$value", methods=["PATCH"], endpoint=self.not_implemented ), @@ -349,7 +370,11 @@ def __init__( ], ) ], - converters={"base64url": IdentifierToBase64URLConverter, "id_short_path": IdShortPathConverter}, + converters={ + "base64url": IdentifierToBase64URLConverter, + "content_modifier": ContentModifierConverter, + "id_short_path": IdShortPathConverter, + }, strict_slashes=False, ) @@ -520,6 +545,29 @@ def _get_submodel_submodel_elements_id_short_path(self, url_args: Dict) -> model def _get_concept_description(self, url_args): return self._get_obj_ts(url_args["concept_id"], model.ConceptDescription) + @classmethod + def _assert_json_response(cls, response_t: Type[APIResponse]) -> None: + """ + The ValueOnly serialization is only defined for JSON, thus the ``$value`` routes cannot serve XML. + """ + if not issubclass(response_t, JsonResponse): + raise werkzeug.exceptions.NotAcceptable( + "The ValueOnly serialization is only available as application/json!" + ) + + @classmethod + def _assert_value_only_representation(cls, element: model.SubmodelElement) -> None: + if not has_value_only_representation(element): + raise BadRequest(f"{element.id_short} does not allow the content modifier value!") + + @classmethod + def _value_only_response(cls, response_t: Type[APIResponse], value: object) -> Response: + if value is None: + # An element without a value has `null` as its ValueOnly representation. This cannot be passed to the + # response classes, as they interpret `None` as an empty body and respond with `204 No Content`. + return Response(json.dumps(None), content_type="application/json") + return response_t(value) + # ------ all not implemented ROUTES ------- def not_implemented(self, request: Request, url_args: Dict, **_kwargs) -> Response: raise werkzeug.exceptions.NotImplemented("This route is not implemented!") @@ -688,6 +736,15 @@ def get_submodel_all_metadata( submodels, paging_metadata = self._get_submodels(request) return response_t(list(submodels), paging_metadata=paging_metadata, stripped=True) + def get_submodel_all_value( + self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs + ) -> Response: + self._assert_json_response(response_t) + deep = not is_stripped_request(request) + submodels, paging_metadata = self._get_submodels(request) + values = [submodel_to_value_only(submodel, deep) for submodel in submodels] + return response_t(values, paging_metadata=paging_metadata) + def get_submodel_all_reference( self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs ) -> Response: @@ -704,23 +761,9 @@ def delete_submodel(self, request: Request, url_args: Dict, response_t: Type[API return response_t() def get_submodel(self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs) -> Response: + mods = parse_modifiers(request, url_args.get("content")) submodel = self._get_submodel(url_args) - return response_t(submodel, stripped=is_stripped_request(request)) - - def get_submodels_metadata( - self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs - ) -> Response: - if "level" in request.args: - raise BadRequest("level cannot be used when retrieving metadata!") - submodel = self._get_submodel(url_args) - return response_t(submodel, stripped=True) - - def get_submodels_reference( - self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs - ) -> Response: - submodel = self._get_submodel(url_args) - reference = model.ModelReference.from_referable(submodel) - return response_t(reference, stripped=is_stripped_request(request)) + return response_t(submodel, modifiers=mods) def put_submodel(self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs) -> Response: submodel = self._get_submodel(url_args) @@ -744,6 +787,19 @@ def get_submodel_submodel_elements_metadata( submodel_elements, paging_metadata = self._get_submodel_submodel_elements(request, url_args) return response_t(list(submodel_elements), paging_metadata=paging_metadata, stripped=True) + def get_submodel_submodel_elements_value( + self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs + ) -> Response: + self._assert_json_response(response_t) + deep = not is_stripped_request(request) + submodel_elements, paging_metadata = self._get_submodel_submodel_elements(request, url_args) + values = [ + submodel_element_to_named_value_only(submodel_element, deep) + for submodel_element in submodel_elements + if has_value_only_representation(submodel_element) + ] + return response_t(values, paging_metadata=paging_metadata) + def get_submodel_submodel_elements_reference( self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs ) -> Response: @@ -769,6 +825,15 @@ def get_submodel_submodel_elements_id_short_path_metadata( raise BadRequest(f"{submodel_element.id_short} does not allow the content modifier metadata!") return response_t(submodel_element, stripped=True) + def get_submodel_submodel_elements_id_short_path_value( + self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs + ) -> Response: + self._assert_json_response(response_t) + submodel_element = self._get_submodel_submodel_elements_id_short_path(url_args) + self._assert_value_only_representation(submodel_element) + value = submodel_element_to_value_only(submodel_element, not is_stripped_request(request)) + return self._value_only_response(response_t, value) + def get_submodel_submodel_elements_id_short_path_reference( self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs ) -> Response: diff --git a/server/app/util/converters.py b/server/app/util/converters.py index 3b98cd5d4..d7f5b3f5d 100644 --- a/server/app/util/converters.py +++ b/server/app/util/converters.py @@ -20,6 +20,8 @@ from basyx.aas import model from werkzeug.exceptions import BadRequest +from app.adapter.modifiers import Content + BASE64URL_ENCODING = "utf-8" @@ -57,6 +59,28 @@ def to_python(self, value: str) -> model.Identifier: return decoded +class ContentModifierConverter(werkzeug.routing.BaseConverter): + """ + URL converter matching the ``$metadata``, ``$value``, ``$reference`` and ``$path`` suffixes to convert + to :class:`~app.adapter.modifiers.Content` value + """ + + _TO_PYTHON = { + "$metadata": Content.METADATA, + "$value": Content.VALUE, + "$reference": Content.REFERENCE, + "$path": Content.PATH, + } + _TO_URL = {v: k for k, v in _TO_PYTHON.items()} + regex = r"\$(?:metadata|value|reference|path)" + + def to_python(self, value: str) -> Content: + return self._TO_PYTHON[value] + + def to_url(self, value: Content) -> str: + return self._TO_URL[value] + + class IdShortPathConverter(werkzeug.routing.UnicodeConverter): """ A custom Werkzeug URL converter for handling dot-separated idShort paths and indexes. diff --git a/server/test/interfaces/repository/test_submodels.py b/server/test/interfaces/repository/test_submodels.py index 622b80600..4d425d5a8 100644 --- a/server/test/interfaces/repository/test_submodels.py +++ b/server/test/interfaces/repository/test_submodels.py @@ -5,13 +5,17 @@ # # SPDX-License-Identifier: MIT +import base64 import io import json +import math +from typing import Any from unittest import mock from app.util.converters import base64url_encode from basyx.aas import model from basyx.aas.adapter.json import AASToJsonEncoder +from basyx.aas.examples.data.example_aas import create_example_bill_of_material_submodel from basyx.aas.examples.data.example_aas_missing_attributes import create_example_submodel from ..format_utils import ( @@ -32,6 +36,49 @@ def _encode_reference(reference: model.Reference) -> str: (model.Key(model.KeyTypes.GLOBAL_REFERENCE, "http://example.org/SubmodelTemplates/ExampleSubmodel"),) ) +# Top-level submodel elements of ``create_example_submodel()`` that have a ValueOnly representation, in document +# order. ExampleOperation and ExampleCapability have none and are omitted by the ``$value`` routes. +VALUE_ONLY_ID_SHORTS = ( + "ExampleRelationshipElement", + "ExampleAnnotatedRelationshipElement", + "ExampleBasicEventElement", + "ExampleSubmodelCollection", +) + +TYPED_SUBMODEL_ID = "http://example.org/Typed_Submodel" + + +def create_typed_submodel() -> model.Submodel: + """ + A Submodel containing the value types and edge cases of the ValueOnly serialization that aren't part of the + example data. + """ + return model.Submodel( + id_=TYPED_SUBMODEL_ID, + submodel_element=( + model.Property(id_short="IntProperty", value_type=model.datatypes.Int, value=42), + model.Property(id_short="BoolProperty", value_type=model.datatypes.Boolean, value=False), + model.Property(id_short="DoubleProperty", value_type=model.datatypes.Double, value=1.5), + model.Property(id_short="NanProperty", value_type=model.datatypes.Double, value=math.nan), + # xs:integer is unbounded, this value cannot be represented as a float + model.Property(id_short="HugeIntProperty", value_type=model.datatypes.Integer, value=10**400), + model.Property( + id_short="DateProperty", value_type=model.datatypes.Date, value=model.datatypes.Date(2026, 8, 31) + ), + model.Property(id_short="EmptyProperty", value_type=model.datatypes.String, value=None), + model.SubmodelElementList( + id_short="ListWithEmptyElement", + type_value_list_element=model.Property, + value_type_list_element=model.datatypes.String, + value=( + model.Property(id_short=None, value_type=model.datatypes.String, value="first"), + model.Property(id_short=None, value_type=model.datatypes.String, value=None), + model.Property(id_short=None, value_type=model.datatypes.String, value="third"), + ), + ), + ), + ) + @inject_format_clients class SubmodelsEndpointsTest(RepositoryEndpointTestBase): @@ -214,6 +261,28 @@ def test_submodels_reference_get_supports_pagination(self, format_client: Format self.assertEqual([1, 1], [len(page) for page in pages]) + # ------------------------------------------------------------------ GET /submodels/$value + + @with_json_client + def test_submodels_value_get(self, format_client: FormatClient): + self.object_store.update(self.two_submodels_store()) + + response = format_client.get("/submodels/$value") + + self.assert_ok(response) + values = format_client.parse_collection(response) + self.assertEqual(2, len(values)) + for value in values: + self.assertEqual(set(VALUE_ONLY_ID_SHORTS), set(value)) + + @with_json_client + def test_submodels_value_get_supports_pagination(self, format_client: FormatClient): + self.object_store.update(self.two_submodels_store()) + + pages = format_client.get_paginated("/submodels/$value", limit=1, max_pages=2) + + self.assertEqual([1, 1], [len(page) for page in pages]) + # ------------------------------------------------------------------ GET /submodels/ @with_json_client @@ -356,6 +425,99 @@ def test_submodel_metadata_get_not_found(self): self.assert_error(response, 404) + # ------------------------------------------------------------------ GET /submodels//$value + + @with_json_client + def test_submodel_value_get(self, format_client: FormatClient): + example_submodel = create_example_submodel() + self.object_store.add(example_submodel) + + response = format_client.get(f"/submodels/{base64url_encode(example_submodel.id)}/$value") + + self.assert_ok(response) + value = format_client.parse_object(response) + # ExampleOperation and ExampleCapability have no ValueOnly representation and are omitted + self.assertEqual(set(VALUE_ONLY_ID_SHORTS), set(value)) + self.assertEqual("exampleValue", value["ExampleSubmodelCollection"]["ExampleProperty"]) + + @with_json_client + def test_submodel_value_get_level_core_empties_nested_containers(self, format_client: FormatClient): + example_submodel = create_example_submodel() + self.object_store.add(example_submodel) + + response = format_client.get(f"/submodels/{base64url_encode(example_submodel.id)}/$value?level=core") + + self.assert_ok(response) + value = format_client.parse_object(response) + # the direct children are present, their children are not + self.assertEqual({}, value["ExampleSubmodelCollection"]) + self.assertEqual([], value["ExampleAnnotatedRelationshipElement"]["annotations"]) + + @with_json_client + def test_submodel_value_get_empty_submodel(self, format_client: FormatClient): + self.object_store.add(model.Submodel(id_=TYPED_SUBMODEL_ID)) + + response = format_client.get(f"/submodels/{base64url_encode(TYPED_SUBMODEL_ID)}/$value") + + self.assert_ok(response) + self.assertEqual({}, format_client.parse_object(response)) + + @with_json_client + def test_submodel_value_get_huge_integer(self, format_client: FormatClient): + self.object_store.add(create_typed_submodel()) + + response = format_client.get(f"/submodels/{base64url_encode(TYPED_SUBMODEL_ID)}/$value") + + self.assert_ok(response) + value = format_client.parse_object(response) + # integers that exceed the range of a float are also serialized when the whole Submodel is requested + self.assertEqual(10**400, value["HugeIntProperty"]) + self.assertEqual(42, value["IntProperty"]) + + @with_json_client + def test_submodel_value_get_omits_property_without_value(self, format_client: FormatClient): + self.object_store.add(create_typed_submodel()) + + response = format_client.get(f"/submodels/{base64url_encode(TYPED_SUBMODEL_ID)}/$value") + + self.assert_ok(response) + self.assertNotIn("EmptyProperty", format_client.parse_object(response)) + + @with_json_client + def test_submodel_value_get_not_found(self, format_client: FormatClient): + response = format_client.get(f"/submodels/{base64url_encode('https://example.org/unknown')}/$value") + + self.assert_error(response, 404) + + @with_xml_client + def test_submodel_value_get_xml_not_acceptable(self, format_client: FormatClient): + example_submodel = create_example_submodel() + self.object_store.add(example_submodel) + + response = format_client.get(f"/submodels/{base64url_encode(example_submodel.id)}/$value") + + self.assert_error(response, 406) + + @with_json_client + def test_submodel_value_get_invalid_level_returns_400(self, format_client: FormatClient): + example_submodel = create_example_submodel() + self.object_store.add(example_submodel) + + response = format_client.get(f"/submodels/{base64url_encode(example_submodel.id)}/$value?level=invalid") + + self.assert_error(response, 400) + + @with_json_client + def test_submodel_value_get_extent_not_implemented(self, format_client: FormatClient): + example_submodel = create_example_submodel() + self.object_store.add(example_submodel) + + response = format_client.get( + f"/submodels/{base64url_encode(example_submodel.id)}/$value?extent=withBlobValue" + ) + + self.assert_error(response, 501) + # ------------------------------------------------------------------ GET /submodels//$reference @with_json_client @@ -443,6 +605,11 @@ def _nested_blob(self, submodel: model.Submodel, id_short_path: str) -> model.Bl assert isinstance(element, model.Blob) return element + def _get_element_value(self, format_client: FormatClient, submodel_id: str, id_short_path: str) -> Any: + response = format_client.get(f"{self.elements_path(submodel_id, id_short_path)}/$value") + self.assert_ok(response) + return format_client.parse_object(response) + # ------------------------------------------------------------------ GET .../submodel-elements @with_json_client @@ -570,6 +737,29 @@ def test_submodel_elements_reference_get_supports_pagination(self, format_client self.assertEqual([3, 3], [len(page) for page in pages]) + # ------------------------------------------------------------------ GET .../submodel-elements/$value + + @with_json_client + def test_submodel_elements_value_get(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + response = format_client.get(f"{self.elements_path(submodel.id)}/$value") + + self.assert_ok(response) + values = format_client.parse_collection(response) + # each element is an {idShort: value} object, ExampleOperation and ExampleCapability are omitted + self.assertEqual(list(VALUE_ONLY_ID_SHORTS), [id_short for value in values for id_short in value]) + + @with_json_client + def test_submodel_elements_value_get_supports_pagination(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + pages = format_client.get_paginated(f"{self.elements_path(submodel.id)}/$value", limit=3, max_pages=2) + + self.assertEqual( + list(VALUE_ONLY_ID_SHORTS), [id_short for page in pages for value in page for id_short in value] + ) + # ------------------------------------------------------------------ GET .../submodel-elements/ @with_json_client @@ -781,6 +971,145 @@ def test_submodel_element_reference_get_not_found(self, format_client: FormatCli self.assert_error(response, 404) + # ------------------------------------------------------------------ GET ...//$value + + @with_json_client + def test_submodel_element_value_get_property_value_types(self, format_client: FormatClient): + self.object_store.add(create_typed_submodel()) + + def get_value(id_short_path: str) -> Any: + return self._get_element_value(format_client, TYPED_SUBMODEL_ID, id_short_path) + + self.assertEqual(42, get_value("IntProperty")) + self.assertEqual(False, get_value("BoolProperty")) + self.assertEqual(1.5, get_value("DoubleProperty")) + # JSON has no representation for NaN, INF and -INF, so the XSD representation is returned instead + self.assertEqual("NaN", get_value("NanProperty")) + # integers are not converted to float, as that would raise an OverflowError for large values + self.assertEqual(10**400, get_value("HugeIntProperty")) + self.assertEqual("2026-08-31", get_value("DateProperty")) + self.assertIsNone(get_value("EmptyProperty")) + + @with_json_client + def test_submodel_element_value_get_list_keeps_indices(self, format_client: FormatClient): + self.object_store.add(create_typed_submodel()) + + value = self._get_element_value(format_client, TYPED_SUBMODEL_ID, "ListWithEmptyElement") + + self.assertEqual(["first", None, "third"], value) + + @with_json_client + def test_submodel_element_value_get_multi_language_property(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + value = self._get_element_value( + format_client, submodel.id, f"{NESTED_COLLECTION}.ExampleMultiLanguageProperty" + ) + + self.assertEqual(2, len(value)) + self.assertEqual([1, 1], [len(entry) for entry in value]) + self.assertEqual({"de", "en-US"}, {next(iter(entry)) for entry in value}) + + @with_json_client + def test_submodel_element_value_get_range(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + value = self._get_element_value(format_client, submodel.id, f"{NESTED_COLLECTION}.ExampleRange") + + self.assertEqual({"min": 0, "max": 100}, value) + + @with_json_client + def test_submodel_element_value_get_blob(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + value = self._get_element_value(format_client, submodel.id, NESTED_BLOB) + + self.assertEqual( + {"contentType": "application/pdf", "value": base64.b64encode(b"\x01\x02\x03\x04\x05").decode()}, value + ) + + @with_json_client + def test_submodel_element_value_get_file(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + value = self._get_element_value(format_client, submodel.id, NESTED_FILE) + + self.assertEqual({"contentType": "application/pdf", "value": "/TestFile.pdf"}, value) + + @with_json_client + def test_submodel_element_value_get_reference_element(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + value = self._get_element_value(format_client, submodel.id, f"{NESTED_COLLECTION}.ExampleReferenceElement") + + self.assertEqual("ModelReference", value["type"]) + self.assertEqual("Submodel", value["keys"][0]["type"]) + + @with_json_client + def test_submodel_element_value_get_relationship_element(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + value = self._get_element_value(format_client, submodel.id, "ExampleRelationshipElement") + + self.assertEqual({"first", "second"}, set(value)) + self.assertEqual("ModelReference", value["first"]["type"]) + + @with_json_client + def test_submodel_element_value_get_annotated_relationship_element(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + value = self._get_element_value(format_client, submodel.id, "ExampleAnnotatedRelationshipElement") + + self.assertEqual({"first", "second", "annotations"}, set(value)) + self.assertIn({"ExampleAnnotatedProperty": "exampleValue"}, value["annotations"]) + + @with_json_client + def test_submodel_element_value_get_basic_event_element(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + value = self._get_element_value(format_client, submodel.id, "ExampleBasicEventElement") + + self.assertEqual(["observed"], list(value)) + self.assertEqual("ModelReference", value["observed"]["type"]) + + @with_json_client + def test_submodel_element_value_get_entity(self, format_client: FormatClient): + submodel = create_example_bill_of_material_submodel() + self.object_store.add(submodel) + + value = self._get_element_value(format_client, submodel.id, "ExampleEntity") + + self.assertEqual("SelfManagedEntity", value["entityType"]) + self.assertEqual("http://example.org/TestAsset/", value["globalAssetId"]) + self.assertEqual("exampleValue", value["statements"]["ExampleProperty"]) + self.assertEqual("TestKey", value["specificAssetIds"][0]["name"]) + + @with_json_client + def test_submodel_element_value_get_collection(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + value = self._get_element_value(format_client, submodel.id, NESTED_COLLECTION) + + self.assertIn("ExampleBlob", value) + self.assertEqual("exampleValue", value["ExampleProperty"]) + + @with_json_client + def test_submodel_element_value_get_rejects_operation_and_capability(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + for id_short in ("ExampleOperation", "ExampleCapability"): + with self.subTest(id_short=id_short): + response = format_client.get(f"{self.elements_path(submodel.id, id_short)}/$value") + self.assert_error(response, 400) + + @with_json_client + def test_submodel_element_value_get_not_found(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + response = format_client.get(f"{self.elements_path(submodel.id, 'DoesNotExist')}/$value") + + self.assert_error(response, 404) + # ------------------------------------------------------------------ ...//attachment def test_submodel_element_attachment_get_blob(self): diff --git a/server/test/interfaces/test_base.py b/server/test/interfaces/test_base.py index afeed463f..29c5a0ea7 100644 --- a/server/test/interfaces/test_base.py +++ b/server/test/interfaces/test_base.py @@ -12,6 +12,7 @@ from typing import Optional from unittest import mock +from app.adapter import Content from app.interfaces import base from basyx.aas import model from basyx.aas.adapter._generic import XML_NS_MAP @@ -265,3 +266,30 @@ def test_result(self): self.assertEqual("Error", messages[0].findtext("messageType")) isodatetime = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{6}$") self.assertIsNotNone(isodatetime.match(messages[0].findtext("timestamp") or "")) + +class TestJsonReferenceResponse(unittest.TestCase): + + def test_paginated_list(self): + # Test that Reference Serialization and Result Serialization is combined correctly + response = base.JsonResponse( + [ + model.Submodel(id_="https://example.org/Example_Submodel"), + model.Submodel(id_="https://example.org/Second_Submodel"), + ], + modifiers=base.Modifiers(Content.REFERENCE, True, False), + paging_metadata=base.PagingMetadata(cursor="asdf"), + ) + + self.assertEqual("application/json", response.content_type) + self.assertEqual(200, response.status_code) + + parsed_body = json.loads(response.get_data(as_text=True)) + if not isinstance(parsed_body, dict): + self.fail("Response is no JSON dict") + self.assertEqual({"paging_metadata", "result"}, parsed_body.keys()) + + result = parsed_body["result"] + self.assertIsInstance(result, list) + for ref in result: + self.assertEqual({"keys", "type"}, ref.keys()) + self.assertGreater(len(ref["keys"]), 0)