shell_implementation.py

Go to the documentation of this file.

"""Implementation of Asset Administration Shell related API calls."""
 
import json
import logging
import mimetypes
from pathlib import Path
from typing import TYPE_CHECKING, Any
 
import requests
from pydantic import BaseModel
 
if TYPE_CHECKING:
    from aas_http_client.classes.client.aas_client import AasHttpClient
 
from aas_http_client.utilities.constants import LogIntensity
from aas_http_client.utilities.encoder import encode_base_64
from aas_http_client.utilities.http_helper import (
    STATUS_CODE_200,
    STATUS_CODE_201,
    STATUS_CODE_204,
    STATUS_CODE_404,
    log_response,
)
 
_logger = logging.getLogger(__name__)
 
 
class ShellRepoImplementation(BaseModel):
    """Implementation of Asset Administration Shell related API calls."""
 
    def __init__(self, client: "AasHttpClient"):
        """Initializes the ShellImplementation with the given parameters."""
        self._client = client
 
        session = client.get_session()
        if session is None:
            raise ValueError(
                "HTTP session is not initialized in the client. Call 'initialize()' method of the client before creating ShellRepoImplementation instance."
            )
 
        self._session: requests.Session = session
 
    # GET /shells/{aasIdentifier}
    def get_asset_administration_shell_by_id(self, aas_identifier: str) -> dict | None:
        """Returns a specific Asset Administration Shell.
 
        :param aas_identifier: The Asset Administration Shells unique id
        :return: Asset Administration Shells data or None if an error occurred
        """
        if not self._client.encoded_ids:
            aas_identifier = encode_base_64(aas_identifier)
 
        url = f"{self._client.base_url}/shells/{aas_identifier}"
 
        self._client.set_token()
 
        try:
            response = self._session.get(url, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:
                if self._client.get_log_intensity() == LogIntensity.HIGH:
                    _logger.warning(f"Asset Administration Shell with id '{aas_identifier}' not found.")
                elif self._client.get_log_intensity() == LogIntensity.STANDARD:
                    _logger.debug(f"Asset Administration Shell with id '{aas_identifier}' not found.")
 
                _logger.debug(response.text)
                return None
 
            if response.status_code != STATUS_CODE_200:
                log_response(response)
                return None
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return None
 
        content = response.content.decode("utf-8")
        return json.loads(content)
 
    # PUT /shells/{aasIdentifier}
    def put_asset_administration_shell_by_id(self, aas_identifier: str, request_body: dict) -> bool:
        """Creates or replaces an existing Asset Administration Shell.
 
        :param aas_identifier: The Asset Administration Shells unique id
        :param request_body: Json data of the Asset Administration Shell data to put
        :return: True if the update was successful, False otherwise
        """
        if not self._client.encoded_ids:
            aas_identifier = encode_base_64(aas_identifier)
 
        url = f"{self._client.base_url}/shells/{aas_identifier}"
 
        self._client.set_token()
 
        try:
            response = self._session.put(url, json=request_body, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:  #
                if self._client.get_log_intensity() == LogIntensity.HIGH:
                    _logger.warning(f"Asset Administration Shell with id '{aas_identifier}' not found.")
                elif self._client.get_log_intensity() == LogIntensity.STANDARD:
                    _logger.debug(f"Asset Administration Shell with id '{aas_identifier}' not found.")
 
                _logger.debug(response.text)
                return False
 
            if response.status_code != STATUS_CODE_204:
                log_response(response)
                return False
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return False
 
        return True
 
    # DELETE /shells/{aasIdentifier}
    def delete_asset_administration_shell_by_id(self, aas_identifier: str) -> bool:
        """Deletes an Asset Administration Shell.
 
        :param aas_identifier: The Asset Administration Shells unique id
        :return: True if the deletion was successful, False otherwise
        """
        if not self._client.encoded_ids:
            aas_identifier = encode_base_64(aas_identifier)
 
        url = f"{self._client.base_url}/shells/{aas_identifier}"
 
        self._client.set_token()
 
        try:
            response = self._session.delete(url, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:
                if self._client.get_log_intensity() == LogIntensity.HIGH:
                    _logger.warning(f"Asset Administration Shell with id '{aas_identifier}' not found.")
                elif self._client.get_log_intensity() == LogIntensity.STANDARD:
                    _logger.debug(f"Asset Administration Shell with id '{aas_identifier}' not found.")
 
                _logger.debug(response.text)
                return False
 
            if response.status_code != STATUS_CODE_204:
                log_response(response)
                return False
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return False
 
        return True
 
    # GET /shells/{aasIdentifier}/asset-information/thumbnail
    def get_thumbnail_aas_repository(self, aas_identifier: str) -> bytes | None:
        """Returns the thumbnail of the Asset Administration Shell.
 
        :param aas_identifier: The Asset Administration Shells unique id
        :return: Thumbnail file data as bytes (octet-stream) or None if an error occurred
        """
        if not self._client.encoded_ids:
            aas_identifier = encode_base_64(aas_identifier)
 
        url = f"{self._client.base_url}/shells/{aas_identifier}/asset-information/thumbnail"
 
        self._client.set_token()
 
        try:
            response = self._session.get(url, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:
                if self._client.get_log_intensity() == LogIntensity.HIGH:
                    _logger.warning(f"Asset Administration Shell with id '{aas_identifier}' or thumbnail file not found.")
                elif self._client.get_log_intensity() == LogIntensity.STANDARD:
                    _logger.debug(f"Asset Administration Shell with id '{aas_identifier}' or thumbnail file not found.")
 
                _logger.debug(response.text)
                return None
 
            if response.status_code != STATUS_CODE_200:
                log_response(response)
                return None
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return None
 
        return response.content
 
    def put_thumbnail_aas_repository(self, aas_identifier: str, file_name: str, file: Path) -> bool:
        """Creates or updates the thumbnail of the Asset Administration Shell.
 
        :param aas_identifier: The Asset Administration Shells unique id
        :param file_name: The name of the thumbnail file
        :param file: Path to the thumbnail file to upload as attachment
        :return: True if the update was successful, False otherwise
        """
        if file.exists() is False or not file.is_file():
            _logger.error(f"Attachment file '{file}' does not exist.")
            return False
 
        mime_type, _ = mimetypes.guess_type(file)
 
        with file.open("rb") as f:
            file_octet_stream = f.read()
 
        return self.put_thumbnail_aas_repository_stream(aas_identifier, file_name, file_octet_stream, mime_type or "application/octet-stream")
 
    # PUT /shells/{aasIdentifier}/asset-information/thumbnail
    def put_thumbnail_aas_repository_stream(
        self, aas_identifier: str, file_name: str, file_octet_stream: Any, mime_type: str = "application/octet-stream"
    ) -> bool:
        """Creates or updates the thumbnail of the Asset Administration Shell.
 
        :param aas_identifier: The Asset Administration Shells unique id
        :param file_name: The name of the thumbnail file
        :param file_octet_stream: The octet stream of the thumbnail file
        :param mime_type: The MIME type of the thumbnail file (e.g., "image/png")
        :return: True if the update was successful, False otherwise
        """
        if file_name is None or file_name == "" or file_octet_stream is None or mime_type is None or mime_type == "":
            _logger.error(f"Attachment file '{file_name}' does not exist.")
            return False
 
        if not self._client.encoded_ids:
            aas_identifier = encode_base_64(aas_identifier)
 
        url = f"{self._client.base_url}/shells/{aas_identifier}/asset-information/thumbnail"
 
        params = {"fileName": file_name}
 
        self._client.set_token()
 
        try:
            files: dict[str, tuple[str, Any, str]] = {"file": (file_name, file_octet_stream, mime_type)}
            response = self._session.put(url, files=files, params=params, timeout=self._client.time_out)
 
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:
                if self._client.get_log_intensity() == LogIntensity.HIGH:
                    _logger.warning(f"Asset Administration Shell with id '{aas_identifier}' not found.")
                elif self._client.get_log_intensity() == LogIntensity.STANDARD:
                    _logger.debug(f"Asset Administration Shell with id '{aas_identifier}' not found.")
 
                _logger.debug(response.text)
                return False
 
            # original dotnet server delivers 200 instead of 204
            if response.status_code not in (STATUS_CODE_200, STATUS_CODE_204):
                log_response(response)
                return False
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return False
 
        return True
 
    # DELETE /shells/{aasIdentifier}/asset-information/thumbnail
    def delete_thumbnail_aas_repository(self, aas_identifier: str) -> bool:
        """Deletes the thumbnail of the Asset Administration Shell.
 
        :param aas_identifier: The Asset Administration Shells unique id
        :return: True if the deletion was successful, False otherwise
        """
        if not self._client.encoded_ids:
            aas_identifier = encode_base_64(aas_identifier)
 
        url = f"{self._client.base_url}/shells/{aas_identifier}/asset-information/thumbnail"
 
        self._client.set_token()
 
        try:
            response = self._session.delete(url, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:
                if self._client.get_log_intensity() == LogIntensity.HIGH:
                    _logger.warning(f"Asset Administration Shell with id '{aas_identifier}' or thumbnail file not found.")
                elif self._client.get_log_intensity() == LogIntensity.STANDARD:
                    _logger.debug(f"Asset Administration Shell with id '{aas_identifier}' or thumbnail file not found.")
 
                _logger.debug(response.text)
                return False
 
            if response.status_code != STATUS_CODE_200:
                log_response(response)
                return False
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return False
 
        return True
 
    # GET /shells
    def get_all_asset_administration_shells(
        self, asset_ids: list[dict] | None = None, id_short: str = "", limit: int = 100, cursor: str = ""
    ) -> dict | None:
        """Returns all Asset Administration Shells.
 
        :param assetIds: A list of specific Asset identifiers (format: {"identifier": "string",  "encodedIdentifier": "string"})
        :param idShort: The Asset Administration Shells IdShort
        :param limit: The maximum number of elements in the response array
        :param cursor: A server-generated identifier retrieved from pagingMetadata that specifies from which position the result listing should continue
        :return: List of paginated Asset Administration Shells data or None if an error occurred
        """
        url = f"{self._client.base_url}/shells"
 
        # Build query parameters
        if asset_ids is None:
            asset_ids = []
 
        params: dict[str, Any] = {}
        if asset_ids is not None and len(asset_ids) > 0:
            params["assetIds"] = asset_ids
        if id_short:
            params["idShort"] = id_short
        if limit:
            params["limit"] = str(limit)
        if cursor:
            params["cursor"] = cursor
 
        self._client.set_token()
 
        try:
            response = self._session.get(url, timeout=self._client.time_out, params=params)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code != STATUS_CODE_200:
                log_response(response)
                return None
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return None
 
        content = response.content.decode("utf-8")
        return json.loads(content)
 
    # POST /shells
    def post_asset_administration_shell(self, request_body: dict) -> dict | None:
        """Creates a new Asset Administration Shell.
 
        :param request_body: Json data of the Asset Administration Shell to post
        :return: Response data as a dictionary or None if an error occurred
        """
        url = f"{self._client.base_url}/shells"
 
        self._client.set_token()
 
        try:
            response = self._session.post(url, json=request_body, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code != STATUS_CODE_201:
                log_response(response)
                return None
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return None
 
        content = response.content.decode("utf-8")
        return json.loads(content)
 
    # GET /shells/{aasIdentifier}/submodel-refs
    def get_all_submodel_references_aas_repository(self, aas_identifier: str, limit: int = 100, cursor: str = "") -> dict | None:
        """Returns all submodel references.
 
        :param aas_identifier: The Asset Administration Shells unique id
        :param limit: The maximum number of elements in the response array
        :param cursor: A server-generated identifier retrieved from pagingMetadata that specifies from which position the result listing should continue
        :return: List of Submodel references or None if an error occurred
        """
        if not self._client.encoded_ids:
            aas_identifier = encode_base_64(aas_identifier)
 
        url = f"{self._client.base_url}/shells/{aas_identifier}/submodel-refs"
 
        params: dict[str, str] = {}
        if limit:
            params["limit"] = str(limit)
        if cursor:
            params["cursor"] = cursor
 
        self._client.set_token()
 
        try:
            response = self._session.get(url, timeout=self._client.time_out, params=params)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:
                if self._client.get_log_intensity() == LogIntensity.HIGH:
                    _logger.warning(f"Asset Administration Shell with id '{aas_identifier}' not found.")
                elif self._client.get_log_intensity() == LogIntensity.STANDARD:
                    _logger.debug(f"Asset Administration Shell with id '{aas_identifier}' not found.")
 
                _logger.debug(response.text)
                return None
 
            if response.status_code != STATUS_CODE_200:
                log_response(response)
                return None
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return None
 
        content = response.content.decode("utf-8")
        return json.loads(content)
 
    # POST /shells/{aasIdentifier}/submodel-refs
    def post_submodel_reference_aas_repository(self, aas_identifier: str, request_body: dict) -> dict | None:
        """Creates a submodel reference at the Asset Administration Shell.
 
        :param aas_identifier: The Asset Administration Shells unique id
        :param request_body: Reference to the Submodel
        :return: Response data as a dictionary or None if an error occurred
        """
        if not self._client.encoded_ids:
            aas_identifier = encode_base_64(aas_identifier)
 
        url = f"{self._client.base_url}/shells/{aas_identifier}/submodel-refs"
 
        self._client.set_token()
 
        try:
            response = self._session.post(url, json=request_body, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:
                if self._client.get_log_intensity() == LogIntensity.HIGH:
                    _logger.warning(f"Asset Administration Shell with id '{aas_identifier}' not found.")
                elif self._client.get_log_intensity() == LogIntensity.STANDARD:
                    _logger.debug(f"Asset Administration Shell with id '{aas_identifier}' not found.")
 
                _logger.debug(response.text)
                return None
 
            if response.status_code != STATUS_CODE_201:
                log_response(response)
                return None
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return None
 
        content = response.content.decode("utf-8")
        return json.loads(content)
 
    # DELETE /shells/{aasIdentifier}/submodel-refs/{submodelIdentifier}
    def delete_submodel_reference_by_id_aas_repository(self, aas_identifier: str, submodel_identifier: str) -> bool:
        """Deletes the submodel reference from the Asset Administration Shell. Does not delete the submodel itself.
 
        :param aas_identifier: The Asset Administration Shells unique id
        :param submodel_identifier: The Submodels unique id
        :return: True if the deletion was successful, False otherwise
        """
        if not self._client.encoded_ids:
            aas_identifier = encode_base_64(aas_identifier)
            submodel_identifier = encode_base_64(submodel_identifier)
 
        url = f"{self._client.base_url}/shells/{aas_identifier}/submodel-refs/{submodel_identifier}"
 
        self._client.set_token()
 
        try:
            response = self._session.delete(url, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:
                if self._client.get_log_intensity() == LogIntensity.HIGH:
                    _logger.warning(f"Asset Administration Shell with id '{aas_identifier}' or submodel with id '{submodel_identifier}' not found.")
                elif self._client.get_log_intensity() == LogIntensity.STANDARD:
                    _logger.debug(f"Asset Administration Shell with id '{aas_identifier}' or submodel with id '{submodel_identifier}' not found.")
 
                _logger.debug(response.text)
                return False
 
            if response.status_code not in (STATUS_CODE_204, STATUS_CODE_200):
                log_response(response)
                return False
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return False
 
        return True
 
    # not supported by Java Server
 
    # PUT /shells/{aasIdentifier}/submodels/{submodelIdentifier}
    def put_submodel_by_id_aas_repository(self, aas_identifier: str, submodel_identifier: str, request_body: dict) -> bool:
        """Updates the Submodel.
 
        :param aas_identifier: ID of the AAS to update the submodel for
        :param submodel_identifier: ID of the submodel to update
        :param request_body: Json data to the Submodel to put
        :return: True if the update was successful, False otherwise
        """
        if not self._client.encoded_ids:
            aas_identifier = encode_base_64(aas_identifier)
            submodel_identifier = encode_base_64(submodel_identifier)
 
        url = f"{self._client.base_url}/shells/{aas_identifier}/submodels/{submodel_identifier}"
 
        self._client.set_token()
 
        try:
            response = self._session.put(url, json=request_body, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:
                if self._client.get_log_intensity() == LogIntensity.HIGH:
                    _logger.warning(f"Asset Administration Shell with id '{aas_identifier}' or submodel with id '{submodel_identifier}' not found.")
                elif self._client.get_log_intensity() == LogIntensity.STANDARD:
                    _logger.debug(f"Asset Administration Shell with id '{aas_identifier}' or submodel with id '{submodel_identifier}' not found.")
 
                _logger.debug(response.text)
                return False
 
            if response.status_code != STATUS_CODE_204:
                log_response(response)
                return False
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return False
 
        return True
 
    # GET /shells/{aasIdentifier}/$reference
    def get_asset_administration_shell_by_id_reference_aas_repository(self, aas_identifier: str) -> dict | None:
        """Returns a specific Asset Administration Shell as a Reference.
 
        :param aas_identifier: ID of the AAS reference to retrieve
        :return: Asset Administration Shells reference data or None if an error occurred
        """
        if not self._client.encoded_ids:
            aas_identifier = encode_base_64(aas_identifier)
 
        url = f"{self._client.base_url}/shells/{aas_identifier}/$reference"
 
        self._client.set_token()
 
        try:
            response = self._session.get(url, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:
                if self._client.get_log_intensity() == LogIntensity.HIGH:
                    _logger.warning(f"Asset Administration Shell with id '{aas_identifier}' not found.")
                elif self._client.get_log_intensity() == LogIntensity.STANDARD:
                    _logger.debug(f"Asset Administration Shell with id '{aas_identifier}' not found.")
 
                _logger.debug(response.text)
                return None
 
            if response.status_code != STATUS_CODE_200:
                log_response(response)
                return None
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return None
 
        ref_dict_string = response.content.decode("utf-8")
        return json.loads(ref_dict_string)
 
    # GET /shells/{aasIdentifier}/submodels/{submodelIdentifier}
    def get_submodel_by_id_aas_repository(self, aas_identifier: str, submodel_identifier: str) -> dict | None:
        """Returns the Submodel.
 
        :param aas_identifier: ID of the AAS to retrieve the submodel from
        :param submodel_identifier: ID of the submodel to retrieve
        :return: Submodel object or None if an error occurred
        """
        if not self._client.encoded_ids:
            aas_identifier = encode_base_64(aas_identifier)
            submodel_identifier = encode_base_64(submodel_identifier)
 
        url = f"{self._client.base_url}/shells/{aas_identifier}/submodels/{submodel_identifier}"
 
        self._client.set_token()
 
        try:
            response = self._session.get(url, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:
                if self._client.get_log_intensity() == LogIntensity.HIGH:
                    _logger.warning(f"Asset Administration Shell with id '{aas_identifier}' or submodel with id '{submodel_identifier}' not found.")
                elif self._client.get_log_intensity() == LogIntensity.STANDARD:
                    _logger.debug(f"Asset Administration Shell with id '{aas_identifier}' or submodel with id '{submodel_identifier}' not found.")
 
                _logger.debug(response.text)
                return None
 
            if response.status_code != STATUS_CODE_200:
                log_response(response)
                return None
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return None
 
        content = response.content.decode("utf-8")
        return json.loads(content)