diff --git a/sdk/basyx/aas/adapter/json/path_serialization.py b/sdk/basyx/aas/adapter/json/path_serialization.py new file mode 100644 index 000000000..8bb996df0 --- /dev/null +++ b/sdk/basyx/aas/adapter/json/path_serialization.py @@ -0,0 +1,132 @@ +# 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, Iterable, List + +from basyx.aas import model +from basyx.aas.adapter.json import AASToJsonEncoder + + +class PathJsonEncoder(AASToJsonEncoder): + """ + Custom JSONEncoder that serializes :class:`~basyx.aas.model.Referable` objects as the list of + idShortPaths reachable from them (``$path`` content modifier). + + :cvar 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. + """ + + deep: bool = True + + @classmethod + def _get_aas_class_serializers(cls) -> dict[type, Callable]: + serializers = super()._get_aas_class_serializers() + serializers.update( + { + model.Submodel: cls._submodel_to_path, + model.SubmodelElementCollection: cls._submodel_element_collection_to_path, + model.SubmodelElementList: cls._submodel_element_list_to_path, + model.Entity: cls._entity_to_path, + model.AnnotatedRelationshipElement: cls._annotated_relationship_element_to_path, + model.Property: cls._leaf_to_path, + model.MultiLanguageProperty: cls._leaf_to_path, + model.Range: cls._leaf_to_path, + model.Blob: cls._leaf_to_path, + model.File: cls._leaf_to_path, + model.ReferenceElement: cls._leaf_to_path, + model.RelationshipElement: cls._leaf_to_path, + model.BasicEventElement: cls._leaf_to_path, + model.Capability: cls._leaf_to_path, + model.Operation: cls._leaf_to_path, + } + ) + return serializers + + @classmethod + def _submodel_to_path(cls, obj: model.Submodel) -> List[str]: + paths: List[str] = [] + cls._append_children(paths, obj.submodel_element, prefix=[]) + return paths + + @classmethod + def _submodel_element_collection_to_path(cls, obj: model.SubmodelElementCollection) -> List[str]: + paths, prefix = cls._self_path(obj) + cls._append_children(paths, obj.value, prefix) + return paths + + @classmethod + def _submodel_element_list_to_path(cls, obj: model.SubmodelElementList) -> List[str]: + paths, prefix = cls._self_path(obj) + cls._append_indexed_children(paths, obj.value, prefix) + return paths + + @classmethod + def _entity_to_path(cls, obj: model.Entity) -> List[str]: + paths, prefix = cls._self_path(obj) + cls._append_children(paths, obj.statement, prefix) + return paths + + @classmethod + def _annotated_relationship_element_to_path(cls, obj: model.AnnotatedRelationshipElement) -> List[str]: + paths, prefix = cls._self_path(obj) + cls._append_children(paths, obj.annotation, prefix) + return paths + + @classmethod + def _leaf_to_path(cls, obj: model.Referable) -> List[str]: + paths, _ = cls._self_path(obj) + return paths + + @staticmethod + def _self_path(obj: model.Referable) -> tuple[List[str], List[str]]: + """Return (initial paths list containing the element's own path, prefix for children).""" + 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] + return [model.Referable.build_id_short_path(prefix)], prefix + + @classmethod + def _append_children( + cls, paths: List[str], children: Iterable[model.SubmodelElement], prefix: List[str] + ) -> None: + for child in children: + child_prefix = prefix + [child.id_short] + paths.append(model.Referable.build_id_short_path(child_prefix)) + if cls.deep: + cls._recurse(paths, child, child_prefix) + + @classmethod + def _append_indexed_children( + cls, 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 cls.deep: + cls._recurse(paths, child, child_prefix) + + @classmethod + def _recurse(cls, paths: List[str], element: object, prefix: List[str]) -> None: + if isinstance(element, model.SubmodelElementList): + cls._append_indexed_children(paths, element.value, prefix) + elif isinstance(element, model.SubmodelElementCollection): + cls._append_children(paths, element.value, prefix) + elif isinstance(element, model.Entity): + cls._append_children(paths, element.statement, prefix) + elif isinstance(element, model.AnnotatedRelationshipElement): + cls._append_children(paths, element.annotation, prefix) + # Leaf elements have no children to traverse into. + + +class ShallowPathJsonEncoder(PathJsonEncoder): + """ + :class:`PathJsonEncoder` variant that emits only the requested element and its direct children + (matches ``?level=core``). + """ + + deep: bool = False diff --git a/sdk/basyx/aas/adapter/json/value_only_serialization.py b/sdk/basyx/aas/adapter/json/value_only_serialization.py new file mode 100644 index 000000000..1699eb483 --- /dev/null +++ b/sdk/basyx/aas/adapter/json/value_only_serialization.py @@ -0,0 +1,179 @@ +# 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 _generic +from basyx.aas.adapter.json import AASToJsonEncoder + +TYPES_WITHOUT_VALUE_ONLY = (model.Capability, model.Operation) + +class AASToValueOnlyJsonEncoder(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_or_blob_to_value_only, + model.Blob: cls._file_or_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 _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) + if isinstance(obj, (model.SubmodelElementList, model.SubmodelElementCollection)): + 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]: + data: dict[str, object] = {} + + for element in obj.submodel_element: + if cls._has_value(element): + data[element.id_short] = element + + return data + + @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 it + 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_or_blob_to_value_only(cls, obj: Union[model.Blob, model.File]) -> dict[str, object]: + data: dict[str, object] = {} + data["contentType"] = obj.content_type + if obj.value is not None: + data["value"] = base64.b64encode(obj.value).decode() if isinstance(obj, model.Blob) else obj.value + 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"] = list(obj.specific_asset_id) + 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 diff --git a/sdk/test/adapter/json/test_value_only_serialization.py b/sdk/test/adapter/json/test_value_only_serialization.py new file mode 100644 index 000000000..085777ba5 --- /dev/null +++ b/sdk/test/adapter/json/test_value_only_serialization.py @@ -0,0 +1,285 @@ +import json +import unittest + +from basyx.aas import model +from basyx.aas.adapter.json.value_only_serialization import AASToValueOnlyJsonEncoder + + +class TestValueOnlyEncoder(unittest.TestCase): + def test_omits_empty(self): + sm = model.Submodel(id_="http://example.org/TestingSM", id_short="TestingSM") + + prop1 = model.Property(id_short="TestProperty1", value_type=model.datatypes.String, value="TestValue1") + prop2 = model.Property(id_short="TestProperty2", value_type=model.datatypes.Boolean, value=True) + collection1 = model.SubmodelElementCollection(id_short="TestCollection", value=[prop1, prop2]) + + prop3 = model.Property(id_short="TestProperty3", value_type=model.datatypes.String) + mlp = model.MultiLanguageProperty(id_short="TestMLP") + rng1 = model.Range(id_short="TestRange", value_type=model.datatypes.Int) + file = model.File(id_short="TestFile", content_type="application/pdf") + blob = model.Blob(id_short="TestBlob") + ref = model.ReferenceElement(id_short="TestRef") + rel = model.RelationshipElement(id_short="TestRelationship") + entity = model.Entity(id_short="TestEntity", entity_type=model.EntityType.CO_MANAGED_ENTITY) + + sm_list = model.SubmodelElementList( + id_short="EmptyList", + type_value_list_element=model.Property, + value_type_list_element=model.datatypes.String, + value=[prop3], + ) + collection2 = model.SubmodelElementCollection( + id_short="EmptyCollection", value=[mlp, rng1, file, blob, ref, rel, entity] + ) + + sm.submodel_element.add(collection1) + sm.submodel_element.add(collection2) + sm.submodel_element.add(sm_list) + + sm_str = json.dumps(sm, cls=AASToValueOnlyJsonEncoder) + + sm_dict = json.loads(sm_str) + self.assertIsInstance(sm_dict, dict) + self.assertNotIn("EmptyList", sm_dict) + self.assertNotIn("EmptyCollection", sm_dict) + + def test_property(self): + prop = model.Property(id_short="TestProperty", value_type=model.datatypes.Int, value=5000) + + out_str = json.dumps(prop, cls=AASToValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertEqual(5000, parsed_output) + + def test_submodel_element_collection(self): + prop1 = model.Property( + id_short="ProductClassificationSystem", value_type=model.datatypes.String, value="ECLASS" + ) + prop2 = model.Property(id_short="ProductOwnership", value_type=model.datatypes.Boolean, value=True) + collection = model.SubmodelElementCollection(id_short="TestCollection", value=[prop1, prop2]) + + out_str = json.dumps(collection, cls=AASToValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertIsInstance(parsed_output, dict) + self.assertEqual({"ProductClassificationSystem", "ProductOwnership"}, set(parsed_output.keys())) + self.assertEqual("ECLASS", parsed_output["ProductClassificationSystem"]) + self.assertEqual(True, parsed_output["ProductOwnership"]) + + def test_submodel_element_list(self): + prop1 = model.Property(id_short="Prop1", value_type=model.datatypes.String, value="Martha") + prop2 = model.Property(id_short="Prop2", value_type=model.datatypes.String, value="Jonathan") + prop3 = model.Property(id_short="Prop3", value_type=model.datatypes.String, value="Clark") + sme_list = model.SubmodelElementList( + id_short="Authors", + type_value_list_element=model.Property, + value_type_list_element=model.datatypes.String, + value=[prop1, prop2, prop3], + ) + + out_str = json.dumps(sme_list, cls=AASToValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertIsInstance(parsed_output, list) + self.assertEqual(["Martha", "Jonathan", "Clark"], parsed_output) + + def test_submodel_element_list_preserves_order(self): + prop1 = model.Property(id_short="Prop1", value_type=model.datatypes.String, value="Martha") + prop2 = model.Property(id_short="Prop2", value_type=model.datatypes.String) + prop3 = model.Property(id_short="Prop3", value_type=model.datatypes.String, value="Clark") + sme_list = model.SubmodelElementList( + id_short="Authors", + type_value_list_element=model.Property, + value_type_list_element=model.datatypes.String, + value=[prop1, prop2, prop3], + ) + + out_str = json.dumps(sme_list, cls=AASToValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertIsInstance(parsed_output, list) + self.assertEqual(["Martha", None, "Clark"], parsed_output) + + def test_multi_language_property(self): + mlp = model.MultiLanguageProperty( + id_short="MLP", + value=model.MultiLanguageTextType( + {"de": "Das ist ein deutscher Bezeichner", "en": "That's an English label"} + ), + ) + + out_str = json.dumps(mlp, cls=AASToValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertIsInstance(parsed_output, list) + self.assertEqual(2, len(parsed_output)) + self.assertIn({"de": "Das ist ein deutscher Bezeichner"}, parsed_output) + self.assertIn({"en": "That's an English label"}, parsed_output) + + def test_range(self): + rng = model.Range(id_short="TorqueRange", value_type=model.datatypes.Int, min=3, max=15) + + out_str = json.dumps(rng, cls=AASToValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertIsInstance(parsed_output, dict) + self.assertEqual({"min", "max"}, parsed_output.keys()) + self.assertEqual(3, parsed_output["min"]) + self.assertEqual(15, parsed_output["max"]) + + def test_reference_element(self): + ref = model.ReferenceElement( + id_short="MaxRotationSpeedReference", + value=model.ExternalReference( + key=(model.Key(type_=model.KeyTypes.GLOBAL_REFERENCE, value="0173-1#02-BAA120#008"),) + ), + ) + + out_str = json.dumps(ref, cls=AASToValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertIsInstance(parsed_output, dict) + self.assertEqual({"type", "keys"}, parsed_output.keys()) + self.assertIsInstance(parsed_output["keys"], list) + self.assertEqual(1, len(parsed_output["keys"])) + self.assertEqual({"type", "value"}, parsed_output["keys"][0].keys()) + self.assertEqual("GlobalReference", parsed_output["keys"][0]["type"]) + self.assertEqual("0173-1#02-BAA120#008", parsed_output["keys"][0]["value"]) + + def test_file(self): + file = model.File( + id_short="Document", content_type="application/pdf", value="https://example.org/SafetyInstructions.pdf" + ) + + out_str = json.dumps(file, cls=AASToValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertIsInstance(parsed_output, dict) + self.assertEqual({"contentType", "value"}, parsed_output.keys()) + self.assertEqual("application/pdf", parsed_output["contentType"]) + self.assertEqual("https://example.org/SafetyInstructions.pdf", parsed_output["value"]) + + def test_file_without_value(self): + file = model.File(id_short="Document", content_type="application/pdf") + + out_str = json.dumps(file, cls=AASToValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertIsInstance(parsed_output, dict) + self.assertEqual({"contentType"}, parsed_output.keys()) + self.assertEqual("application/pdf", parsed_output["contentType"]) + + def test_blob(self): + blob = model.Blob(id_short="Logo", content_type="application/octet-stream", value=b"\x00\x01BaSyx") + + out_str = json.dumps(blob, cls=AASToValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertIsInstance(parsed_output, dict) + self.assertEqual({"contentType", "value"}, parsed_output.keys()) + self.assertEqual("application/octet-stream", parsed_output["contentType"]) + self.assertEqual("AAFCYVN5eA==", parsed_output["value"]) + + def test_relationship_element(self): + rel = model.RelationshipElement( + id_short="CurrentFlowsFrom", + first=model.ModelReference( + type_=model.Property, + key=( + model.Key(type_=model.KeyTypes.SUBMODEL, value="http://customer.com/demo/aas/1/1/1234859590"), + model.Key(type_=model.KeyTypes.PROPERTY, value="PlusPole"), + ), + ), + second=model.ModelReference( + type_=model.Property, + key=( + model.Key(type_=model.KeyTypes.SUBMODEL, value="http://customer.com/demo/aas/1/0/1234859123490"), + model.Key(type_=model.KeyTypes.PROPERTY, value="MinusPole"), + ), + ), + ) + + out_str = json.dumps(rel, cls=AASToValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertIsInstance(parsed_output, dict) + self.assertEqual({"first", "second"}, parsed_output.keys()) + self.assertEqual("PlusPole", parsed_output["first"]["keys"][1]["value"]) + self.assertEqual("MinusPole", parsed_output["second"]["keys"][1]["value"]) + + def test_annotated_relationship_element(self): + anrel = model.AnnotatedRelationshipElement( + id_short="CurrentFlowsFrom", + first=model.ModelReference( + type_=model.Property, + key=( + model.Key(type_=model.KeyTypes.SUBMODEL, value="http://customer.com/demo/aas/1/1/1234859590"), + model.Key(type_=model.KeyTypes.PROPERTY, value="PlusPole"), + ), + ), + second=model.ModelReference( + type_=model.Property, + key=( + model.Key(type_=model.KeyTypes.SUBMODEL, value="http://customer.com/demo/aas/1/0/1234859123490"), + model.Key(type_=model.KeyTypes.PROPERTY, value="MinusPole"), + ), + ), + annotation=[ + model.Property( + id_short="AppliedRule", value_type=model.datatypes.String, value="TechnicalCurrentFlowDirection" + ) + ], + ) + + out_str = json.dumps(anrel, cls=AASToValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertIsInstance(parsed_output, dict) + self.assertEqual({"first", "second", "annotations"}, parsed_output.keys()) + self.assertEqual("PlusPole", parsed_output["first"]["keys"][1]["value"]) + self.assertEqual("MinusPole", parsed_output["second"]["keys"][1]["value"]) + self.assertIsInstance(parsed_output["annotations"], dict) + self.assertEqual(1, len(parsed_output["annotations"])) + self.assertEqual("TechnicalCurrentFlowDirection", parsed_output["annotations"]["AppliedRule"]) + + def test_entity(self): + entity = model.Entity( + id_short="MySubAssetEntity", + entity_type=model.EntityType.SELF_MANAGED_ENTITY, + global_asset_id="http://customer.com/demo/asset/1/1/MySubAsset", + specific_asset_id=[ + model.SpecificAssetId(name="SpecificAsset", value="http://customer.com/demo/asset/2/2/TestAsset") + ], + statement=[ + model.Property(id_short="MaxRotationSpeed", value_type=model.datatypes.Int, value=5000) + ] + ) + + out_str = json.dumps(entity, cls=AASToValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertIsInstance(parsed_output, dict) + self.assertEqual({"entityType", "globalAssetId", "specificAssetIds", "statements"}, parsed_output.keys()) + self.assertEqual(5000, parsed_output["statements"]["MaxRotationSpeed"]) + self.assertEqual("http://customer.com/demo/asset/1/1/MySubAsset", parsed_output["globalAssetId"]) + + def test_basic_event_element(self): + event = model.BasicEventElement( + id_short="MyBasicEvent", + observed=model.ModelReference( + type_=model.Property, + key=( + model.Key(type_=model.KeyTypes.SUBMODEL, value="http://customer.com/demo/aas/1/1/1234859590"), + model.Key(type_=model.KeyTypes.PROPERTY, value="MaxRotation"), + ) + ), + direction=model.Direction.INPUT, + state=model.StateOfEvent.ON + ) + + out_str = json.dumps(event, cls=AASToValueOnlyJsonEncoder) + parsed_output = json.loads(out_str) + + self.assertIsInstance(parsed_output, dict) + self.assertEqual({"observed"}, parsed_output.keys()) + self.assertEqual("MaxRotation", parsed_output["observed"]["keys"][1]["value"]) 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..f1ab4d04e --- /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 + deep: bool + 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 + deep = _parse_level(request, resolved) + with_blob = _parse_extent(request, resolved) + return Modifiers(content=resolved, deep=deep, with_blob=with_blob) + + +def _parse_level(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 _parse_extent(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..189bbe00a 100644 --- a/server/app/interfaces/base.py +++ b/server/app/interfaces/base.py @@ -17,6 +17,8 @@ import werkzeug.utils from basyx.aas import model from basyx.aas.adapter._generic import XML_NS_MAP +from basyx.aas.adapter.json.path_serialization import PathJsonEncoder, ShallowPathJsonEncoder +from basyx.aas.adapter.json.value_only_serialization import AASToValueOnlyJsonEncoder 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 +28,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 +118,21 @@ 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 = pick_encoder(self._modifiers) + else: + encoder_cls = StrippedResultToJsonEncoder if stripped else ResultToJsonEncoder 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=(",", ":")) class XmlResponse(APIResponse): @@ -221,6 +233,47 @@ class StrippedResultToJsonEncoder(ResultToJsonEncoder): stripped = True +# TODO need to be added + +class ResultMetadataJsonEncoder(ResultToJsonEncoder): + pass + + +class ResultReferenceJsonEncoder(ResultToJsonEncoder): + pass + + +class ResultValueOnlyJsonEncoder(ResultToJsonEncoder, AASToValueOnlyJsonEncoder): + pass + + +class ResultShallowValueOnlyJsonEncoder(ResultValueOnlyJsonEncoder): + pass + + +class ResultPathJsonEncoder(ResultToJsonEncoder, PathJsonEncoder): + pass + + +class ResultShallowPathJsonEncoder(ResultToJsonEncoder, ShallowPathJsonEncoder): + pass + + +def pick_encoder(mods: Modifiers) -> Type["ResultToJsonEncoder"]: + """Pick a JSON encoder class for the given modifier combination.""" + match mods.content: + case Content.NORMAL: + return StrippedResultToJsonEncoder if not mods.deep else ResultToJsonEncoder + case Content.METADATA: + return ResultMetadataJsonEncoder + case Content.VALUE: + return ResultShallowValueOnlyJsonEncoder if not mods.deep else ResultValueOnlyJsonEncoder + case Content.REFERENCE: + return ResultReferenceJsonEncoder + case Content.PATH: + return ResultShallowPathJsonEncoder if not mods.deep else ResultPathJsonEncoder + + 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):