fix: fixed bug with enablement service

Author: matbmoser

ichub-backend/controllers/fastapi/routers/sharing_handler.py

@@ -25,14 +25,15 @@
 from services.sharing_service import SharingService
 from models.services.sharing_management import (
     SharedPartBase,
-    ShareCatalogPart
+    ShareCatalogPart,
+    SharedPartner
 )
-
+from typing import Optional, List
 router = APIRouter(prefix="/share", tags=["Sharing Functionality"])
-part_sharing_shortcut_service = SharingService()
+part_sharing_service = SharingService()
 
 @router.post("/catalog-part", response_model=SharedPartBase)
-async def twin_management_create_part_sharing_shortcut(catalog_part_to_share: ShareCatalogPart) -> SharedPartBase:
-    return part_sharing_shortcut_service.share_catalog_part(
+async def share_catalog_part(catalog_part_to_share: ShareCatalogPart) -> SharedPartBase:
+    return part_sharing_service.share_catalog_part(
         catalog_part_to_share=catalog_part_to_share
-    )
+    )

ichub-backend/controllers/fastapi/routers/twin_management.py

@@ -41,7 +41,7 @@ async def twin_management_get_catalog_part_twins(include_data_exchange_agreement
 
 @router.get("/catalog-part-twin/{global_id}", response_model=List[CatalogPartTwinDetailsRead])
 async def twin_management_get_catalog_part_twin(global_id: UUID) -> List[CatalogPartTwinDetailsRead]:
-    return twin_management_service.get_catalog_part_twin_details(global_id)
+    return twin_management_service.get_catalog_part_twin_details_id(global_id)
 
 @router.get("/catalog-part-twin/{manufacturerId}/{manufacturerPartId}", response_model=List[CatalogPartTwinDetailsRead])
 async def twin_management_get_catalog_part_twin_from_manufacturer(manufacturerId: str, manufacturerPartId: str) -> List[CatalogPartTwinDetailsRead]:
@@ -50,17 +50,3 @@ async def twin_management_get_catalog_part_twin_from_manufacturer(manufacturerId
 @router.post("/catalog-part-twin", response_model=TwinRead)
 async def twin_management_create_catalog_part_twin(catalog_part_twin_create: CatalogPartTwinCreate) -> TwinRead:
     return twin_management_service.create_catalog_part_twin(catalog_part_twin_create)
-
-@router.post("/catalog-part-twin/share", responses={
-    201: {"description": "Catalog part twin shared successfully"},
-    204: {"description": "Catalog part twin already shared"}
-})
-async def twin_management_share_catalog_part_twin(catalog_part_twin_share: CatalogPartTwinShare):
-    if twin_management_service.create_catalog_part_twin_share(catalog_part_twin_share):
-        return JSONResponse(status_code=201, content={"description":"Catalog part twin shared successfully"})
-    else:
-        return JSONResponse(status_code=204, content={"description":"Catalog part twin already shared"})
-
-@router.post("/twin-aspect", response_model=TwinAspectRead)
-async def twin_management_create_twin_aspect(twin_aspect_create: TwinAspectCreate) -> TwinAspectRead:
-    return twin_management_service.create_twin_aspect(twin_aspect_create)

ichub-backend/managers/enablement_services/dtr_manager.py

@@ -43,8 +43,9 @@
 from tools.aspect_id_tools import extract_aspect_id_name_from_urn_camelcase
 from urllib.parse import urljoin
 
-
 import json
+import logging
+logger = logging.getLogger(__name__)
 
 class DTRManager:
     def __init__(
@@ -81,89 +82,174 @@ def get_dtr_url(base_dtr_url: str = '', uri: str = '', api_path: str = '') -> st
         full_url = urljoin(base_plus_uri.rstrip('/') + '/', api_path.lstrip('/'))
         return full_url
 
+    def _reference_from_bpn_list(self, bpn_list, fallback_id=None):
+            keys = []
+            if bpn_list:
+                keys = [
+                    ReferenceKey(type=ReferenceKeyTypes.GLOBAL_REFERENCE, value=bpn)
+                    for bpn in bpn_list
+                ]
+            elif fallback_id:
+                keys = [ReferenceKey(type=ReferenceKeyTypes.GLOBAL_REFERENCE, value=fallback_id)]
+            return Reference(
+                type=ReferenceTypes.EXTERNAL_REFERENCE,
+                keys=keys,
+            )
+
+    def _add_or_update_asset_id(self, name, value, bpn_list, fallback_id=None):
+            ref = self._reference_from_bpn_list(bpn_list, fallback_id=fallback_id)
+            return SpecificAssetId(name=name, value=value, externalSubjectId=ref)
+    def create_or_update_shell_descriptor(self,
+        aas_id: UUID,
+        global_id: UUID,
+        manufacturer_id: str,
+        manufacturer_part_id: str,
+        customer_part_ids: Dict[str, str] | None,
+        part_category: str,
+        digital_twin_type: str,
+    ) -> ShellDescriptor:
+        """
+        Registers or updates a twin in the DTR.
+        """
+        try:
+            existing_shell = self.aas_service.get_asset_administration_shell_descriptor_by_id(aas_id.urn)
+            logger.info(f"Shell with ID {aas_id} already exists and will be updated.")
+            specific_asset_ids = existing_shell.specificAssetIds or []
+            existing_keys = {(id.name, id.value) for id in specific_asset_ids}
+        except Exception:
+            existing_shell = None
+            specific_asset_ids = []
+            existing_keys = set()
+
+        bpn_list = list(customer_part_ids.values()) if customer_part_ids else []
+
+        if manufacturer_id and ("manufacturerId", manufacturer_id) not in existing_keys:
+            specific_asset_ids.append(self._add_or_update_asset_id("manufacturerId", manufacturer_id, bpn_list, fallback_id=manufacturer_id))
+
+        if digital_twin_type and ("digitalTwinType", digital_twin_type) not in existing_keys:
+            specific_asset_ids.append(self._add_or_update_asset_id("digitalTwinType", digital_twin_type, bpn_list, fallback_id=manufacturer_id))
+
+        if manufacturer_part_id and ("manufacturerPartId", manufacturer_part_id) not in existing_keys:
+            specific_asset_ids.append(self._add_or_update_asset_id("manufacturerPartId", manufacturer_part_id, bpn_list, fallback_id=manufacturer_id))
+
+        if customer_part_ids:
+            for customer_part_id, bpn in customer_part_ids.items():
+                if not customer_part_id:
+                    continue
+                key = ("customerPartId", customer_part_id)
+                if key not in existing_keys:
+                    specific_customer_part_asset_id = SpecificAssetId(
+                        name="customerPartId",
+                        value=customer_part_id,
+                        externalSubjectId=self._reference_from_bpn_list([bpn]),
+                    )
+                    specific_asset_ids.append(specific_customer_part_asset_id)
+
+        shell = ShellDescriptor(
+            id=aas_id.urn,
+            globalAssetId=global_id.urn,
+            specificAssetIds=specific_asset_ids,
+        )
+
+        if existing_shell:
+            res = self.aas_service.update_asset_administration_shell_descriptor(shell)
+        else:
+            res = self.aas_service.create_asset_administration_shell_descriptor(shell)
+
+        if isinstance(res, Result):
+            raise Exception("Error creating or updating shell descriptor", res.to_json_string())
+
+        return res
+        
+    
     def create_shell_descriptor(
         self,
         aas_id: UUID,
         global_id: UUID,
         manufacturer_id: str,
         manufacturer_part_id: str,
-        customer_part_ids: Dict[str, str],
+        customer_part_ids: Dict[str, str] | None,
         part_category: str,
         digital_twin_type: str,
     ) -> ShellDescriptor:
         """
         Registers a twin in the DTR.
         """
-        # List with specific asset ids that
-        # are required by the Industry Core KIT
         specific_asset_ids = []
-        # Adds the customerPartId of the customers from which we have
-        # this information
-        for customer_part_id, bpn in customer_part_ids.items():
-            # This value is optional, so if we don't have it, we skip it
-            if not customer_part_id:
-                continue
-            specific_customer_part_asset_id = SpecificAssetId(
-                name="customerPartId",
-                value=customer_part_id,
+        # Prepare BPN list from customer_part_ids, if present
+        bpn_list = list(customer_part_ids.values()) if customer_part_ids else []
+
+        # manufacturerId
+        if manufacturer_id:
+            ref_keys = (
+                [ReferenceKey(type=ReferenceKeyTypes.GLOBAL_REFERENCE, value=bpn) for bpn in bpn_list]
+                if bpn_list else
+                [ReferenceKey(type=ReferenceKeyTypes.GLOBAL_REFERENCE, value=manufacturer_id)]
+            )
+            specific_manufacturer_asset_id = SpecificAssetId(
+                name="manufacturerId",
+                value=manufacturer_id,
                 externalSubjectId=Reference(
                     type=ReferenceTypes.EXTERNAL_REFERENCE,
-                    keys=[
-                        ReferenceKey(
-                            type=ReferenceKeyTypes.GLOBAL_REFERENCE, value=bpn
-                        ),
-                    ],
+                    keys=ref_keys,
                 ),
             )  # type: ignore
-            specific_asset_ids.append(specific_customer_part_asset_id)
+            specific_asset_ids.append(specific_manufacturer_asset_id)
 
-        # Adds the manufacturerId that we have assigned to the part
-        # The visibility of this id is restricted to partners we have
-        # a contract with, so we use set their BPN
-        specific_manufacturer_asset_id = SpecificAssetId(
-            name="manufacturerId",
-            value=manufacturer_id,
-            externalSubjectId=Reference(
-                type=ReferenceTypes.EXTERNAL_REFERENCE,
-                keys=[
-                    ReferenceKey(type=ReferenceKeyTypes.GLOBAL_REFERENCE, value=bpn)
-                    for bpn in customer_part_ids.values()
-                ],
-            ),
-        )  # type: ignore
-        specific_asset_ids.append(specific_manufacturer_asset_id)
+        # digitalTwinType
+        if digital_twin_type:
+            ref_keys = (
+                [ReferenceKey(type=ReferenceKeyTypes.GLOBAL_REFERENCE, value=bpn) for bpn in bpn_list]
+                if bpn_list else
+                [ReferenceKey(type=ReferenceKeyTypes.GLOBAL_REFERENCE, value=manufacturer_id)]
+            )
+            digital_twin_asset_id = SpecificAssetId(
+                name="digitalTwinType",
+                value=digital_twin_type,
+                externalSubjectId=Reference(
+                    type=ReferenceTypes.EXTERNAL_REFERENCE,
+                    keys=ref_keys,
+                ),
+            )  # type: ignore
+            specific_asset_ids.append(digital_twin_asset_id)
 
-        # Is added to allow data consumers to search for all digital twins of a particular type
-        digital_twin_asset_id = SpecificAssetId(
-            name="digitalTwinType",
-            value=digital_twin_type,
-            externalSubjectId=Reference(
-                type=ReferenceTypes.EXTERNAL_REFERENCE,
-                keys=[
-                    ReferenceKey(type=ReferenceKeyTypes.GLOBAL_REFERENCE, value=bpn)
-                    for bpn in customer_part_ids.values()
-                ],
-            ),
-        )  # type: ignore
-        specific_asset_ids.append(digital_twin_asset_id)
+        # manufacturerPartId
+        if manufacturer_part_id:
+            ref_keys = (
+                [ReferenceKey(type=ReferenceKeyTypes.GLOBAL_REFERENCE, value=bpn) for bpn in bpn_list]
+                if bpn_list else
+                [ReferenceKey(type=ReferenceKeyTypes.GLOBAL_REFERENCE, value=manufacturer_id)]
+            )
+            specific_manufacturer_part_asset_id = SpecificAssetId(
+                name="manufacturerPartId",
+                value=manufacturer_part_id,
+                externalSubjectId=Reference(
+                    type=ReferenceTypes.EXTERNAL_REFERENCE,
+                    keys=ref_keys,
+                ),
+            )  # type: ignore
+            specific_asset_ids.append(specific_manufacturer_part_asset_id)
 
-        # The ID of the type/catalog part from the manufacturer
-        # visble to everyone
-        specific_manufacturer_part_asset_id = SpecificAssetId(
-            name="manufacturerPartId",
-            value=manufacturer_part_id,
-            externalSubjectId=Reference(
-                type=ReferenceTypes.EXTERNAL_REFERENCE,
-                keys=[
-                    ReferenceKey(
-                        type=ReferenceKeyTypes.GLOBAL_REFERENCE, value="PUBLIC_READABLE"
+        # customerPartId(s)
+        if customer_part_ids is not None and customer_part_ids != {}:
+            for customer_part_id, bpn in customer_part_ids.items():
+                if not customer_part_id:
+                    continue
+                ref_keys = (
+                    [ReferenceKey(type=ReferenceKeyTypes.GLOBAL_REFERENCE, value=bpn)]
+                    if bpn else
+                    ([ReferenceKey(type=ReferenceKeyTypes.GLOBAL_REFERENCE, value=manufacturer_id)] if manufacturer_id else [])
+                )
+                specific_customer_part_asset_id = SpecificAssetId(
+                    name="customerPartId",
+                    value=customer_part_id,
+                    externalSubjectId=Reference(
+                        type=ReferenceTypes.EXTERNAL_REFERENCE,
+                        keys=ref_keys,
                     ),
-                ],
-            ),
-        )  # type: ignore
-        specific_asset_ids.append(specific_manufacturer_part_asset_id)
+                )  # type: ignore
+                specific_asset_ids.append(specific_customer_part_asset_id)
 
-        # ids need to be saved as urn
         shell = ShellDescriptor(
             id=aas_id.urn,
             globalAssetId=global_id.urn,

ichub-backend/models/services/sharing_management.py

@@ -47,3 +47,6 @@ class SharedPartBase(BaseModel):
     shared_at: datetime = Field(alias="sharedAt", description="The date and time when the catalog part was shared.")
     twin: Optional[CatalogPartTwinDetailsRead] = Field(alias="twin", description="The digital twin created for part that was shared.", default=None)
 
+
+class SharedPartner(SharingBase):
+    name: str = Field(description="The unique name of the business partner.")

ichub-backend/services/sharing_service.py

@@ -28,10 +28,11 @@
 from managers.metadata_database.manager import RepositoryManagerFactory, RepositoryManager
 from models.services.twin_management import CatalogPartTwinCreate, CatalogPartTwinShare, TwinAspectCreate, CatalogPartTwinDetailsRead, TwinAspectRead
 from models.metadata_database.models import BusinessPartner, DataExchangeAgreement, EnablementServiceStack, CatalogPart, Twin, PartnerCatalogPart
-from models.services.sharing_management import SharedPartBase, ShareCatalogPart
+from models.services.sharing_management import SharedPartBase, ShareCatalogPart, SharedPartner
 from models.services.partner_management import BusinessPartnerRead
 from typing import Dict, Optional, List, Any, Tuple
 
+from uuid import uuid4
 from managers.config.log_manager import LoggingManager
 
 logger = LoggingManager.get_logger(__name__)
@@ -45,14 +46,17 @@ def __init__(self):
         self.submodel_document_generator = SubmodelDocumentGenerator()
         self.twin_management_service = TwinManagementService()
 
+    def get_shared_partners(self, manufacturerId:str, manufacturerPartId:str) -> List[SharedPartner]:
+        pass
+    
     def share_catalog_part(self, catalog_part_to_share: ShareCatalogPart) -> SharedPartBase:
         shared_at = datetime.now(timezone.utc)
         with RepositoryManagerFactory.create() as repo:
             # Step 1: Retrieve the catalog part from the repository
             db_catalog_part = self._get_catalog_part(repo, catalog_part_to_share)
             # Step 2: Get or create the enablement service stack for the manufacturer
             ## Note: this is not used at the moment
-            db_enablement_service_stack = self._get_or_create_enablement_service_stack(repo, catalog_part_to_share)
+            db_enablement_service_stack = self.twin_management_service.get_or_create_enablement_stack(repo, catalog_part_to_share.manufacturer_id)
             # Step 3: Get or create the business partner entity
             db_business_partner = self._get_or_create_business_partner(repo, catalog_part_to_share)
             # Step 4: Get or create the data exchange agreement for the business partner
@@ -70,7 +74,7 @@ def share_catalog_part(self, catalog_part_to_share: ShareCatalogPart) -> SharedP
                 businessPartnerNumber=catalog_part_to_share.business_partner_number,
                 customerPartIds=db_partner_catalog_parts,
                 sharedAt=shared_at,
-                twin=self.twin_management_service.get_catalog_part_twin_details(db_twin.global_id)
+                twin=self.twin_management_service.get_catalog_part_twin_details_id(global_id=db_twin.global_id)
             )
 
     def _get_catalog_part(self, repo: RepositoryManager, catalog_part_to_share: ShareCatalogPart) -> CatalogPart:
@@ -89,21 +93,6 @@ def _get_catalog_part(self, repo: RepositoryManager, catalog_part_to_share: Shar
         db_catalog_part, _ = db_catalog_parts[0]
         return db_catalog_part
 
-    def _get_or_create_enablement_service_stack(self, repo: RepositoryManager, catalog_part_to_share: ShareCatalogPart) -> EnablementServiceStack:
-        """
-        Retrieve or create an EnablementServiceStack for the given manufacturer ID.
-        """
-        db_enablement_service_stacks = repo.enablement_service_stack_repository.find_by_legal_entity_bpnl(catalog_part_to_share.manufacturer_id)
-        if not db_enablement_service_stacks:
-            db_legal_entity = repo.legal_entity_repository.get_by_bpnl(catalog_part_to_share.manufacturer_id)
-            db_enablement_service_stack = repo.enablement_service_stack_repository.create(
-                EnablementServiceStack(name='EDC/DTR Default', legal_entity_id=db_legal_entity.id))
-            repo.commit()
-            repo.refresh(db_enablement_service_stack)
-        else:
-            db_enablement_service_stack = db_enablement_service_stacks[0]
-        return db_enablement_service_stack
-
     def _get_or_create_business_partner(self, repo: RepositoryManager, catalog_part_to_share: ShareCatalogPart) -> BusinessPartner:
         """
         Retrieve or create a BusinessPartner for the given business partner number.
@@ -221,4 +210,4 @@ def _create_part_twin_aspect(self, db_twin: Twin, db_catalog_part: CatalogPart,
             globalId=db_twin.global_id,
             semanticId=SEM_ID_PART_TYPE_INFORMATION_V1,
             payload=payload
-        ))
+        ), manufacturer_id=catalog_part_to_share.manufacturer_id)