aas_helper.py

# Copyright (c) 2022 - for information on the respective copyright owner
# see the NOTICE file and/or the repository
# https://github.com/catenax-ng/product-testdata-2-edc
#
# SPDX-License-Identifier: Apache-2.0

from uuid import uuid4
from urllib.parse import urlparse

from aas.registry.models.asset_administration_shell_descriptor import AssetAdministrationShellDescriptor
from aas.registry.models.identifier_key_value_pair import IdentifierKeyValuePair
from aas.registry.models.submodel_descriptor import SubmodelDescriptor
from aas.registry.models.protocol_information import ProtocolInformation
from aas.registry.models.endpoint import Endpoint
from aas.registry.models.reference import Reference

"""
Generic helper for the generated aas models
Everything going into here must not have any relationship to the testdata set!
"""

def build_aas(local_identifiers:dict, aas_id: str = '', cx_id: str = '', id_short: str = '') -> AssetAdministrationShellDescriptor:
    """
    Generates a basic aas (AAS Descriptor) (without submodels in it).
    Only mandatory is the local_identifiers to make the shell discoverable in the registry.
    cx_id and aas_id will be autogenerated if not given.

    sm_type: e.g. serialPartTypization
    local_identifiers / specificAssetIds: key/value list
    """

    if not aas_id:
        aas_id = generate_uuid()
    if not cx_id:
        cx_id = generate_uuid()
    if not id_short:
        id_short = cx_id

    specific_asset_ids = list(map(lambda x: IdentifierKeyValuePair(key=x[0], value=x[1]), local_identifiers.items()))
    
    aas_descriptor = AssetAdministrationShellDescriptor(
        identification=aas_id,
        id_short=id_short,
        global_asset_id=Reference(value=[cx_id]),
        specific_asset_ids=specific_asset_ids,
    )
    return aas_descriptor

def build_submodel(endpoint: str, semantic_id_schema:str = '', submodel_id: str = '', id_short: str = '') -> SubmodelDescriptor:
    """
    
    """
    if not submodel_id:
        submodel_id = generate_uuid()
    if not id_short:
        id_short = 'assemblyPartRelationship' # TODO
    if not semantic_id_schema:
        #TODO
        semantic_id_schema = 'urn:bamm:io.catenax.assembly_part_relationship:1.1.0#AssemblyPartRelationship'

    # typically only 1 per submodel!
    endpoints = [
        Endpoint(
            interface='EDC',
            protocol_information=ProtocolInformation(
                endpoint_address=endpoint,
                endpointProtocol="IDS/ECLIPSE DATASPACE CONNECTOR",
                endpointProtocolVersion="0.0.0"
            )
        )]

    submodel = SubmodelDescriptor(
        identification=submodel_id,
        id_short=id_short,
        semantic_id=Reference(
            value=[semantic_id_schema]
        ),
        endpoints=endpoints,
    )
    return submodel


def generate_uuid():
    return f"urn:uuid:{uuid4()}"

def get_global_asset_id(aas) -> str:
    """
    Tries to return the 1 (!) of the list (which is typically the only one in our use cases!)
    """
    try:
        return aas['globalAssetId']['value'][0]
    except:
        pass
    return None

def extract_edc_information(submodel_descriptor_endpoint_url: str):
    """
    Extracts the asset_id from a given URL
    """
    url = urlparse(submodel_descriptor_endpoint_url)
    bpn = url.path.split('/')[1]
    edc_asset_id = url.path.split('/')[2]
    connector_endpoint = url.scheme + '://' + url.netloc
    return {'bpn': bpn, 'edc_asset_id': edc_asset_id, 'connector_ednpoint': connector_endpoint}

def get_submodel_id_for_endpoint(aas: AssetAdministrationShellDescriptor, endpoint: str) -> str:
    """
    Go through all endpoints and find the given endpoint and return its submodel 'identification'
    """
    for sm in aas.submodel_descriptors:
        for ep in sm.endpoints:
            if ep.protocol_information.endpoint_address == endpoint:
                return sm.identification
    return None