Experimental/registry and discovery server

.gitignore

@@ -32,5 +32,9 @@ compliance_tool/aas_compliance_tool/version.py
 server/app/version.py
 
 # Ignore the content of the server storage
-server/input/
+server/example_configurations/repository_standalone/input/
+server/example_configurations/repository_standalone/storage/
+server/example_configurations/registry_standalone/input/
+server/example_configurations/registry_standalone/storage/
+/storage/
 server/storage/

server/app/adapter/__init__.py

@@ -0,0 +1 @@
+from .jsonization import *

server/app/adapter/jsonization.py

@@ -0,0 +1,340 @@
+import logging
+from typing import Callable, Dict, Optional, Set, Type
+
+from basyx.aas import model
+from basyx.aas.adapter._generic import ASSET_KIND, ASSET_KIND_INVERSE, JSON_AAS_TOP_LEVEL_KEYS_TO_TYPES, PathOrIO
+from basyx.aas.adapter.json import AASToJsonEncoder
+from basyx.aas.adapter.json.json_deserialization import AASFromJsonDecoder, _get_ts, read_aas_json_file_into
+
+import app.model as server_model
+
+logger = logging.getLogger(__name__)
+
+JSON_SERVER_AAS_TOP_LEVEL_KEYS_TO_TYPES = JSON_AAS_TOP_LEVEL_KEYS_TO_TYPES + (
+    ("assetAdministrationShellDescriptors", server_model.AssetAdministrationShellDescriptor),
+    ("submodelDescriptors", server_model.SubmodelDescriptor),
+)
+
+
+class ServerAASFromJsonDecoder(AASFromJsonDecoder):
+    @classmethod
+    def _get_aas_class_parsers(cls) -> Dict[str, Callable[[Dict[str, object]], object]]:
+        aas_class_parsers = super()._get_aas_class_parsers()
+        aas_class_parsers.update(
+            {
+                "AssetAdministrationShellDescriptor": cls._construct_asset_administration_shell_descriptor,
+                "SubmodelDescriptor": cls._construct_submodel_descriptor,
+                "AssetLink": cls._construct_asset_link,
+                "ProtocolInformation": cls._construct_protocol_information,
+                "Endpoint": cls._construct_endpoint,
+            }
+        )
+        return aas_class_parsers
+
+    # ##################################################################################################
+    # Utility Methods used in constructor methods to add general attributes (from abstract base classes)
+    # ##################################################################################################
+
+    @classmethod
+    def _amend_abstract_attributes(cls, obj: object, dct: Dict[str, object]) -> None:
+        super()._amend_abstract_attributes(obj, dct)
+
+        if isinstance(obj, server_model.Descriptor):
+            if "description" in dct:
+                obj.description = cls._construct_lang_string_set(
+                    _get_ts(dct, "description", list), model.MultiLanguageTextType
+                )
+            if "displayName" in dct:
+                obj.display_name = cls._construct_lang_string_set(
+                    _get_ts(dct, "displayName", list), model.MultiLanguageNameType
+                )
+            if "extensions" in dct:
+                for extension in _get_ts(dct, "extensions", list):
+                    obj.extension.add(cls._construct_extension(extension))
+
+    @classmethod
+    def _construct_asset_administration_shell_descriptor(
+        cls, dct: Dict[str, object], object_class=server_model.AssetAdministrationShellDescriptor
+    ) -> server_model.AssetAdministrationShellDescriptor:
+        ret = object_class(id_=_get_ts(dct, "id", str))
+        cls._amend_abstract_attributes(ret, dct)
+        if "administration" in dct:
+            ret.administration = cls._construct_administrative_information(_get_ts(dct, "administration", dict))
+        if "assetKind" in dct:
+            ret.asset_kind = ASSET_KIND_INVERSE[_get_ts(dct, "assetKind", str)]
+        if "assetType" in dct:
+            ret.asset_type = _get_ts(dct, "assetType", str)
+        global_asset_id = None
+        if "globalAssetId" in dct:
+            ret.global_asset_id = _get_ts(dct, "globalAssetId", str)
+        specific_asset_id = set()
+        if "specificAssetIds" in dct:
+            for desc_data in _get_ts(dct, "specificAssetIds", list):
+                specific_asset_id.add(cls._construct_specific_asset_id(desc_data, model.SpecificAssetId))
+        if "endpoints" in dct:
+            for endpoint_dct in _get_ts(dct, "endpoints", list):
+                if "protocolInformation" in endpoint_dct:
+                    ret.endpoints.append(cls._construct_endpoint(endpoint_dct, server_model.Endpoint))
+                elif "href" in endpoint_dct:
+                    protocol_info = server_model.ProtocolInformation(
+                        href=_get_ts(endpoint_dct["href"], "href", str),
+                        endpoint_protocol=(
+                            _get_ts(endpoint_dct["href"], "endpointProtocol", str)
+                            if "endpointProtocol" in endpoint_dct["href"]
+                            else None
+                        ),
+                        endpoint_protocol_version=(
+                            _get_ts(endpoint_dct["href"], "endpointProtocolVersion", list)
+                            if "endpointProtocolVersion" in endpoint_dct["href"]
+                            else None
+                        ),
+                    )
+                    ret.endpoints.append(
+                        server_model.Endpoint(
+                            protocol_information=protocol_info, interface=_get_ts(endpoint_dct, "interface", str)
+                        )
+                    )
+        if "idShort" in dct:
+            ret.id_short = _get_ts(dct, "idShort", str)
+        if "submodelDescriptors" in dct:
+            for sm_dct in _get_ts(dct, "submodelDescriptors", list):
+                ret.submodel_descriptors.append(
+                    cls._construct_submodel_descriptor(sm_dct, server_model.SubmodelDescriptor)
+                )
+        return ret
+
+    @classmethod
+    def _construct_protocol_information(
+        cls, dct: Dict[str, object], object_class=server_model.ProtocolInformation
+    ) -> server_model.ProtocolInformation:
+        ret = object_class(
+            href=_get_ts(dct, "href", str),
+            endpoint_protocol=_get_ts(dct, "endpointProtocol", str) if "endpointProtocol" in dct else None,
+            endpoint_protocol_version=(
+                _get_ts(dct, "endpointProtocolVersion", list) if "endpointProtocolVersion" in dct else None
+            ),
+            subprotocol=_get_ts(dct, "subprotocol", str) if "subprotocol" in dct else None,
+            subprotocol_body=_get_ts(dct, "subprotocolBody", str) if "subprotocolBody" in dct else None,
+            subprotocol_body_encoding=(
+                _get_ts(dct, "subprotocolBodyEncoding", str) if "subprotocolBodyEncoding" in dct else None
+            ),
+        )
+        return ret
+
+    @classmethod
+    def _construct_endpoint(cls, dct: Dict[str, object], object_class=server_model.Endpoint) -> server_model.Endpoint:
+        ret = object_class(
+            protocol_information=cls._construct_protocol_information(
+                _get_ts(dct, "protocolInformation", dict), server_model.ProtocolInformation
+            ),
+            interface=_get_ts(dct, "interface", str),
+        )
+        cls._amend_abstract_attributes(ret, dct)
+        return ret
+
+    @classmethod
+    def _construct_submodel_descriptor(
+        cls, dct: Dict[str, object], object_class=server_model.SubmodelDescriptor
+    ) -> server_model.SubmodelDescriptor:
+        ret = object_class(id_=_get_ts(dct, "id", str), endpoints=[])
+        cls._amend_abstract_attributes(ret, dct)
+        for endpoint_dct in _get_ts(dct, "endpoints", list):
+            if "protocolInformation" in endpoint_dct:
+                ret.endpoints.append(cls._construct_endpoint(endpoint_dct, server_model.Endpoint))
+            elif "href" in endpoint_dct:
+                protocol_info = server_model.ProtocolInformation(
+                    href=_get_ts(endpoint_dct["href"], "href", str),
+                    endpoint_protocol=(
+                        _get_ts(endpoint_dct["href"], "endpointProtocol", str)
+                        if "endpointProtocol" in endpoint_dct["href"]
+                        else None
+                    ),
+                    endpoint_protocol_version=(
+                        _get_ts(endpoint_dct["href"], "endpointProtocolVersion", list)
+                        if "endpointProtocolVersion" in endpoint_dct["href"]
+                        else None
+                    ),
+                )
+                ret.endpoints.append(
+                    server_model.Endpoint(
+                        protocol_information=protocol_info, interface=_get_ts(endpoint_dct, "interface", str)
+                    )
+                )
+        if "administration" in dct:
+            ret.administration = cls._construct_administrative_information(_get_ts(dct, "administration", dict))
+        if "idShort" in dct:
+            ret.id_short = _get_ts(dct, "idShort", str)
+        if "semanticId" in dct:
+            ret.semantic_id = cls._construct_reference(_get_ts(dct, "semanticId", dict))
+        if "supplementalSemanticIds" in dct:
+            for ref in _get_ts(dct, "supplementalSemanticIds", list):
+                ret.supplemental_semantic_id.append(cls._construct_reference(ref))
+        return ret
+
+    @classmethod
+    def _construct_asset_link(
+        cls, dct: Dict[str, object], object_class=server_model.AssetLink
+    ) -> server_model.AssetLink:
+        ret = object_class(name=_get_ts(dct, "name", str), value=_get_ts(dct, "value", str))
+        return ret
+
+
+class ServerStrictAASFromJsonDecoder(ServerAASFromJsonDecoder):
+    """
+    A strict version of the AASFromJsonDecoder class for deserializing Asset Administration Shell data from the
+    official JSON format
+
+    This version has set ``failsafe = False``, which will lead to Exceptions raised for every missing attribute or wrong
+    object type.
+    """
+
+    failsafe = False
+
+
+class ServerStrippedAASFromJsonDecoder(ServerAASFromJsonDecoder):
+    """
+    Decoder for stripped JSON objects. Used in the HTTP adapter.
+    """
+
+    stripped = True
+
+
+class ServerStrictStrippedAASFromJsonDecoder(ServerStrictAASFromJsonDecoder, ServerStrippedAASFromJsonDecoder):
+    """
+    Non-failsafe decoder for stripped JSON objects.
+    """
+
+    pass
+
+
+def read_server_aas_json_file_into(
+    object_store: model.AbstractObjectStore,
+    file: PathOrIO,
+    replace_existing: bool = False,
+    ignore_existing: bool = False,
+    failsafe: bool = True,
+    stripped: bool = False,
+    decoder: Optional[Type[AASFromJsonDecoder]] = None,
+) -> Set[model.Identifier]:
+    return read_aas_json_file_into(
+        object_store=object_store,
+        file=file,
+        replace_existing=replace_existing,
+        ignore_existing=ignore_existing,
+        failsafe=failsafe,
+        stripped=stripped,
+        decoder=decoder,
+        keys_to_types=JSON_SERVER_AAS_TOP_LEVEL_KEYS_TO_TYPES,
+    )
+
+
+class ServerAASToJsonEncoder(AASToJsonEncoder):
+
+    @classmethod
+    def _get_aas_class_serializers(cls) -> Dict[Type, Callable]:
+        serializers = super()._get_aas_class_serializers()
+        serializers.update(
+            {
+                server_model.AssetAdministrationShellDescriptor: cls._asset_administration_shell_descriptor_to_json,
+                server_model.SubmodelDescriptor: cls._submodel_descriptor_to_json,
+                server_model.Endpoint: cls._endpoint_to_json,
+                server_model.ProtocolInformation: cls._protocol_information_to_json,
+                server_model.AssetLink: cls._asset_link_to_json,
+            }
+        )
+        return serializers
+
+    @classmethod
+    def _abstract_classes_to_json(cls, obj: object) -> Dict[str, object]:
+        data: Dict[str, object] = super()._abstract_classes_to_json(obj)
+        if isinstance(obj, server_model.Descriptor):
+            if obj.description:
+                data["description"] = obj.description
+            if obj.display_name:
+                data["displayName"] = obj.display_name
+            if obj.extension:
+                data["extensions"] = list(obj.extension)
+        return data
+
+    @classmethod
+    def _asset_administration_shell_descriptor_to_json(
+        cls, obj: server_model.AssetAdministrationShellDescriptor
+    ) -> Dict[str, object]:
+        """
+        serialization of an object from class AssetAdministrationShell to json
+
+        :param obj: object of class AssetAdministrationShell
+        :return: dict with the serialized attributes of this object
+        """
+        data = cls._abstract_classes_to_json(obj)
+        data.update(cls._namespace_to_json(obj))
+        data["id"] = obj.id
+        if obj.administration:
+            data["administration"] = obj.administration
+        if obj.asset_kind:
+            data["assetKind"] = ASSET_KIND[obj.asset_kind]
+        if obj.asset_type:
+            data["assetType"] = obj.asset_type
+        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.endpoints:
+            data["endpoints"] = list(obj.endpoints)
+        if obj.id_short:
+            data["idShort"] = obj.id_short
+        if obj.submodel_descriptors:
+            data["submodelDescriptors"] = list(obj.submodel_descriptors)
+        return data
+
+    @classmethod
+    def _protocol_information_to_json(cls, obj: server_model.ProtocolInformation) -> Dict[str, object]:
+        data = cls._abstract_classes_to_json(obj)
+
+        data["href"] = obj.href
+        if obj.endpoint_protocol:
+            data["endpointProtocol"] = obj.endpoint_protocol
+        if obj.endpoint_protocol_version:
+            data["endpointProtocolVersion"] = obj.endpoint_protocol_version
+        if obj.subprotocol:
+            data["subprotocol"] = obj.subprotocol
+        if obj.subprotocol_body:
+            data["subprotocolBody"] = obj.subprotocol_body
+        if obj.subprotocol_body_encoding:
+            data["subprotocolBodyEncoding"] = obj.subprotocol_body_encoding
+        return data
+
+    @classmethod
+    def _endpoint_to_json(cls, obj: server_model.Endpoint) -> Dict[str, object]:
+        data = cls._abstract_classes_to_json(obj)
+        data["protocolInformation"] = cls._protocol_information_to_json(obj.protocol_information)
+        data["interface"] = obj.interface
+        return data
+
+    @classmethod
+    def _submodel_descriptor_to_json(cls, obj: server_model.SubmodelDescriptor) -> Dict[str, object]:
+        """
+        serialization of an object from class Submodel to json
+
+        :param obj: object of class Submodel
+        :return: dict with the serialized attributes of this object
+        """
+        data = cls._abstract_classes_to_json(obj)
+        data["id"] = obj.id
+        data["endpoints"] = [cls._endpoint_to_json(ep) for ep in obj.endpoints]
+        if obj.id_short:
+            data["idShort"] = obj.id_short
+        if obj.administration:
+            data["administration"] = obj.administration
+        if obj.semantic_id:
+            data["semanticId"] = obj.semantic_id
+        if obj.supplemental_semantic_id:
+            data["supplementalSemanticIds"] = list(obj.supplemental_semantic_id)
+        return data
+
+    @classmethod
+    def _asset_link_to_json(cls, obj: server_model.AssetLink) -> Dict[str, object]:
+        data = cls._abstract_classes_to_json(obj)
+        data["name"] = obj.name
+        data["value"] = obj.value
+        return data

server/app/backend/__init__.py

@@ -0,0 +1 @@
+from .local_file import *