get_value_payload.py

Go to the documentation of this file.

from typing import Any
 
from basyx.aas.model import Key, KeyTypes, ModelReference
from pydantic import BaseModel, Field, PrivateAttr
 
KEY_TYPE_MAPPING: dict[str, str] = {
    "Submodel": "SUBMODEL",
    "SubmodelElementCollection": "SUBMODEL_ELEMENT_COLLECTION",
    "Property": "PROPERTY",
}
 
 
class GetValuePayload(BaseModel):  # noqa: D101
    """Defines the model class of the payload for the `get_value` (URL `/get-value`) endpoint
    of this application using Pydantic.
 
    The JSON payload must contain exactly the following fields:
    - `Reference`
 
    Inside the `Reference`-field, a properly serialized AAS reference according to the AAS JSON serialization
    must be provided.
    Will be parsed as `basyx.aas.model.ModelReference`.
    The reference shall point to a SME in an AID submodel known to this application (refer to `add_or_update_config`).
 
    Example:
    ```
    {
      "Reference": {
        "type": "ModelReference",
        "keys": [
          {
            "type": "Submodel",
            "value": "https://fluid40.de/ids/sm/4757_4856_8464_1441"
          },
          {
              "type": "SubmodelElementCollection",
              "value": "Interface_MQTT"
          },
          {
              "type": "SubmodelElementCollection",
              "value": "InteractionMetadata"
          },
          {
              "type": "SubmodelElementCollection",
              "value": "properties"
          },
          {
              "type": "SubmodelElementCollection",
              "value": "my_endpoint"
          }
        ]
      }
    }
    ```
    """
 
    aid_ref_dict: dict = Field(..., alias="Reference")
    _aid_ref: ModelReference = PrivateAttr(default=None)
 
    def __init__(self, **data: Any):  # noqa: D107
        super().__init__(**data)
        # aid_string = json.dumps(self.aid_ref_dict)
        # self._aid_ref = json.loads(aid_string, cls=AASFromJsonDecoder)
        self._build_model_reference()
 
    def _build_model_reference(self) -> ModelReference:
        if self.aid_ref_dict is None:
            raise ValueError("AID Reference has not been initialized.")
        key_list = [Key(type_=self._get_key_type(key["type"]), value=key["value"]) for key in self.aid_ref_dict["keys"]]
        ref: ModelReference = ModelReference(key=tuple(key_list), type_=ModelReference)
        self._aid_ref_aid_ref = ref
 
    def _get_key_type(self, key_str: str) -> KeyTypes:
        key_type = KEY_TYPE_MAPPING.get(key_str)
        if key_type is None:
            raise ValueError(f"Unknown key type: {key_str}")
        return KeyTypes[key_type]