Model API

core

Identifiable (BaseModel)

Base class for all identifiable classes that have an identifier, that allows to identify these objects.

If no id is set, the id function of the python object is used. Otherwise, a uuid is generated.

Parameters:

Name	Type	Description	Default
id	str	Global id of the object.	required

Source code in aas_middleware\model\core.py

class Identifiable(BaseModel):
    """
    Base class for all identifiable classes that have an identifier, that allows to identify these objects.

    If no id is set, the id function of the python object is used. Otherwise, a uuid is generated.

    Args:
        id (str): Global id of the object.
    """

    id: IdString

    @model_validator(mode="before")
    @classmethod
    def check_id_and_id_short(cls, data: Any) -> Any:
        potential_id = get_id(data)
        assert potential_id, "Either id or id_short must be set"
        return {"id": potential_id}

data_model

DataModel (BaseModel)

The data model is a container that allows to store all models of a data model and provides methods to access them easily by their id or type.

Parameters:

Name	Type	Description	Default
**data	Dict[str, Any]	The data to load into the data model (used by pydantic).	{}

Attributes:

Name	Type	Description
_models_key_id	Dict[str, Identifiable]	The dictionary of models with their id as key.
_top_level_models	Dict[str, List[str]]	The dictionary of top level models with their type as key.
_models_key_type	Dict[str, List[str]]	The dictionary of models with their type as key.
_reference_info_dict_for_referencing	Dict[str, Dict[str, ReferenceInfo]]	The dictionary of reference infos with keys from the referencing model to the referenced model.
_reference_info_dict_for_referenced	Dict[str, Dict[str, ReferenceInfo]]	The dictionary of reference infos with keys from the referenced model to the referencing model.

Source code in aas_middleware\model\data_model.py

class DataModel(BaseModel):
    """
    The data model is a container that allows to store all models of a data model and provides methods to access them easily by their id or type.

    Args:
        **data (Dict[str, Any]): The data to load into the data model (used by pydantic).

    Attributes:
        _models_key_id (Dict[str, Identifiable]): The dictionary of models with their id as key.
        _top_level_models (Dict[str, List[str]]): The dictionary of top level models with their type as key.
        _models_key_type (Dict[str, List[str]]): The dictionary of models with their type as key.
        _reference_info_dict_for_referencing (Dict[str, Dict[str, ReferenceInfo]]): The dictionary of reference infos with keys from the referencing model to the referenced model.
        _reference_info_dict_for_referenced (Dict[str, Dict[str, ReferenceInfo]]): The dictionary of reference infos with keys from the referenced model to the referencing model.
    """

    _key_ids_models: Dict[str, Identifiable] = {}
    _top_level_models: Dict[str, List[str]] = {}
    _models_key_type: Dict[str, List[str]] = {}
    _reference_infos: Set[ReferenceInfo] = set()
    _reference_info_dict_for_referencing: Dict[str, Dict[str, ReferenceInfo]] = {}
    _reference_info_dict_for_referenced: Dict[str, Dict[str, ReferenceInfo]] = {}

    _schemas: Dict[str, Type[Any]] = {}
    _top_level_schemas: Set[str] = set()
    _schema_reference_infos: Set[ReferenceInfo] = set()
    _schema_reference_info_for_referencing: Dict[str, Dict[str, ReferenceInfo]] = {}
    _schema_reference_info_for_referenced: Dict[str, Dict[str, ReferenceInfo]] = {}

    def __init__(self, **data: Dict[str, Any]):
        super().__init__(**data)
        try:
            Identifiable.model_validate(self)
            self.add_model(self)
        except ValidationError:
            for attribute_value in get_value_attributes(self).values():
                if is_identifiable_container(attribute_value):
                    self.add(*attribute_value)
                else:
                    self.add(attribute_value)

    @classmethod
    def from_models(
        cls, *models: Tuple[Identifiable], **data: Dict[str, Any]
    ) -> DataModel:
        """
        Method to create a data model from a list of provided models.

        Args:
            models (Tuple[Identifiable]): The models to load into the data model.
            data (Dict[str, Any]): The data to load into the data model.

        Returns:
            DataModel: The data model with loaded models
        """
        data_model = cls(**data)
        data_model.add(*models)
        return data_model

    @classmethod
    def from_model_types(cls, *model_types: Tuple[Type[Identifiable]], **data: Dict[str, Any]) -> DataModel:
        """
        Method to create a data model a provided list of model types.

        Args:
            model_types (Tuple[Type[Identifiable]]): The model types to load into the data model.
            data (Dict[str, Any]): The data to load into the data model.

        Returns:
            DataModel: The data model with loaded models
        """
        data_model = cls(**data)
        for model_type in model_types:
            data_model.add_schema(model_type)
        return data_model

    @property
    def model_ids(self) -> Set[str]:
        """
        Property to get the ids of all contained models.

        Returns:
            Set[str]: The set of ids.
        """
        return set(str(key) for key in self._key_ids_models.keys())

    def get_contained_ids(self) -> Set[str]:
        """
        Method to get all ids of contained models.

        Returns:
            Set[str]: The set of ids.
        """
        return self.model_ids

    def add(self, *models: Identifiable) -> None:
        """
        Method to add models to the data model.

        Args:
            *models (Tuple[Identifiable]): The models to load into the data model.
        """
        for model in models:
            self.add_model(model)

    def add_schema(self, schema: Type[Identifiable]) -> None:
        """
        Method to add a schema of the data model.

        Args:
            schema (Type[Identifiable]): The schema to load.
        """
        all_schemas, schema_reference_infos = ReferenceFinder.find_schema_references(schema)
        self._add_contained_schemas(all_schemas)
        self._add_top_level_schema(schema)
        self._add_schema_references_to_referencing_schemas_dict(schema_reference_infos)

    def add_model(self, model: Identifiable) -> None:
        """
        Method to load a model of the data model.

        Args:
            model (Identifiable): The model to load.
        """
        # Identifiable.model_validate(model)
        model_id = get_id_with_patch(model)
        if model_id in self.model_ids:
            raise ValueError(f"Model with id {model_id} already loaded.")
        self.add_schema(type(model))
        all_identifiables, reference_infos = ReferenceFinder.find(model)
        self._add_contained_models(model, all_identifiables)
        self._add_top_level_model(model)
        self._add_references_to_referencing_models_dict(reference_infos)

    def check_different_model_with_same_id_contained(self, model: Identifiable) -> bool:
        """
        Method to check if a model is already contained in the data model.

        Args:
            model (Identifiable): The model to check.

        Returns:
            bool: True if the model is already contained, False otherwise.
        """
        model_id = get_id_with_patch(model)
        if model_id in self.model_ids:
            same_id_model = self.get_model(model_id)
            if not same_id_model == model:
                return True
        return False

    def _add_contained_models(
        self, top_level_model: Identifiable, contained_models: List[Identifiable]
    ) -> None:
        """
        Method to load all contained models of a model.

        Args:
            top_level_model (Identifiable): The top level model to load.
            contained_models (List[Identifiable]): The contained models to load.
        """
        for contained_model in contained_models:
            contained_model_id = get_id_with_patch(contained_model)
            if contained_model_id in self.model_ids:
                same_id_model = self.get_model(contained_model_id)
                if not models_are_equal(same_id_model, contained_model):
                    raise ValueError(
                        f"Model with id {contained_model_id} already loaded but with different content. Make sure to only load models with unique ids."
                    )
                replace_attribute_with_model(top_level_model, same_id_model)
                continue
            self._add_model(contained_model)

    def _add_contained_schemas(
        self, contained_schemas: List[Type[Identifiable]]
    ) -> None:
        """
        Method to load all contained schemas of a schema.

        Args:
            top_level_schema (Type[Identifiable]): The top level schema to load.
            contained_schemas (List[Type[Identifiable]]): The contained schemas to load.
        """
        for contained_schema in contained_schemas:
            self._schemas[contained_schema.__name__] = contained_schema


    def patch_schema_references(self) -> None:
        """
        Function tries to patch schema reference infos to represent a more realistic data model.

        ReferenceInfos of Type Reference are patched so that another ReferenceInfo is added to every schema that has a class name that contains the reference id.
        ReferenceInfos of Type Attribute are patched so that another ReferenceInfo is added to every schema that has a class name that contains the reference id.
        """
        reference_infos = patch_references(self._schema_reference_infos, self._schemas.values())
        self._schema_reference_infos = reference_infos
        self._add_schema_references_to_referencing_schemas_dict(reference_infos)

    def _add_references_to_referencing_models_dict(
        self, reference_infos: Set[ReferenceInfo]
    ) -> None:
        """
        Method to add information about referencing model ids of the input model.

        Args:
            model (Identifiable): The model to add the information for.
        """
        self._reference_infos = self._reference_infos | reference_infos
        for reference_info in reference_infos:
            referencing_model_id = reference_info.identifiable_id
            referenced_model_id = reference_info.reference_id
            if not referencing_model_id in self._reference_info_dict_for_referencing:
                self._reference_info_dict_for_referencing[referencing_model_id] = {}
            self._reference_info_dict_for_referencing[referencing_model_id][
                referenced_model_id
            ] = reference_info
            if not referenced_model_id in self._reference_info_dict_for_referenced:
                self._reference_info_dict_for_referenced[referenced_model_id] = {}
            self._reference_info_dict_for_referenced[referenced_model_id][
                referencing_model_id
            ] = reference_info

    def _add_schema_references_to_referencing_schemas_dict(self, reference_infos: Set[ReferenceInfo]) -> None:
        """
        Method to add information about referencing schema ids of the input schema.

        Args:
            schema (Type[Identifiable]): The schema to add the information for.
        """
        self._schema_reference_infos = self._schema_reference_infos | reference_infos
        for reference_info in reference_infos:
            referencing_schema_id = reference_info.identifiable_id
            referenced_schema_id = reference_info.reference_id
            if not referencing_schema_id in self._schema_reference_info_for_referencing:
                self._schema_reference_info_for_referencing[referencing_schema_id] = {}
            self._schema_reference_info_for_referencing[referencing_schema_id][
                referenced_schema_id
            ] = reference_info
            if not referenced_schema_id in self._schema_reference_info_for_referenced:
                self._schema_reference_info_for_referenced[referenced_schema_id] = {}
            self._schema_reference_info_for_referenced[referenced_schema_id][
                referencing_schema_id
            ] = reference_info


    def _add_model(self, model: Identifiable) -> None:
        """
        Method to add a model to the data model.

        Args:
            model (Identifiable): The model to add.
        """
        model_id = get_id_with_patch(model)
        if model_id in self.model_ids:
            raise ValueError(f"Model with id {model_id} already loaded.")
        self._key_ids_models[model_id] = model
        type_name = model.__class__.__name__.split(".")[-1]
        if not type_name in self._models_key_type:
            self._models_key_type[type_name] = []
        self._models_key_type[type_name].append(model_id)

    def _add_top_level_model(self, model: Identifiable) -> None:
        """
        Method to add a model to the data model.

        Args:
            model (Identifiable): The model to add.
        """
        type_name = model.__class__.__name__.split(".")[-1]
        underscore_type_name = convert_camel_case_to_underscrore_str(type_name)
        if not underscore_type_name in self._top_level_models:
            self._top_level_models[underscore_type_name] = []
        self._top_level_models[underscore_type_name].append(get_id_with_patch(model))

    def _add_top_level_schema(self, schema: Type[Identifiable]) -> None:
        """
        Method to add a schema to the data model.

        Args:
            schema (Type[Identifiable]): The schema to add.
        """
        schema_name = schema.__name__
        if not schema_name in self._top_level_schemas:
            self._top_level_schemas.add(schema_name)

    def from_dict(self, data: NESTED_DICT, types: List[Type]) -> None:
        """
        Method to load a data model from a dict.

        Args:
            data (NESTED_DICT): The dict to load the data model from.
        """
        for attribute_name, attribute_value in data.items():
            class_name = convert_under_score_to_camel_case_str(attribute_name)
            for type_ in types:
                if type_.__name__ == class_name:
                    type_for_attribute_values = type_
                    break
            else:
                raise ValueError(f"Type {class_name} not supported.")
            for model_dict in attribute_value:
                model = type_for_attribute_values(**model_dict)
                self.add(model)

    def dict(self) -> NESTED_DICT:
        """
        Method to get the dict of the data model.

        Returns:
            NESTED_DICT: The dict of the data model.
        """
        nested_dict = {}
        for attribute_name, attribute_value in self.get_top_level_models().items():
            nested_dict[attribute_name] = [
                model.model_dump() for model in attribute_value
            ]
        return nested_dict

    def json(self) -> str:
        """
        Method to get the json of the data model.

        Returns:
            str: The json of the data model.
        """
        nested_dict = {}
        for attribute_name, attribute_value in self.get_top_level_models().items():
            # TODO: if a non-BaseModel object is loaded, this breakds down -> adjust this
            nested_dict[attribute_name] = [
                model.model_dump() for model in attribute_value
            ]
        return json.dumps(nested_dict, indent=4, cls=DateTimeEncoder)

    def get_top_level_models(self) -> Dict[str, List[Identifiable]]:
        """
        Method to get all models of the data model.

        Returns:
            Dict[str, List[Identifiable]]: The dictionary of models.
        """
        top_level_models = {}
        for top_level_model_name, top_level_model_ids in self._top_level_models.items():
            top_level_models[top_level_model_name] = [
                self.get_model(model_id) for model_id in top_level_model_ids
            ]
        return top_level_models

    def get_top_level_types(self) -> List[Type[Identifiable]]:
        """
        Method to get all types of the top level models in the data model.

        Returns:
            List[Type[Identifiable]]: The types of the top level models in the data model
        """
        top_level_types = []
        for schema_name in self._top_level_schemas:
            top_level_types.append(self._schemas[schema_name])
        return top_level_types

    def get_models_of_type_name(self, model_type_name: str) -> List[Identifiable]:
        """
        Method to get all models of a specific type.

        Args:
            model_type (str): The type of the models to get.

        Returns:
            List[Identifiable]: The list of models of the type.
        """
        if not model_type_name in self._models_key_type:
            raise ValueError(f"Model type {model_type_name} not supported.")
        return [
            self.get_model(model_id)
            for model_id in self._models_key_type[model_type_name]
        ]

    def get_models_of_type(self, model_type: Type[T]) -> List[T]:
        """
        Method to get all models of a specific type.

        Args:
            model_type (Type[T]): The type of the models to get.

        Returns:
            List[T]: The list of models of the type.
        """
        type_name = model_type.__name__.split(".")[-1]
        return self.get_models_of_type_name(type_name)

    def get_contained_models(self) -> List[Identifiable]:
        """
        Method to get all models that are contained in the data model.

        Returns:
            List[Identifiable]: The list of models.
        """
        return list(self._key_ids_models.values())

    def get_referencing_info(
        self, referenced_model: Identifiable
    ) -> List[ReferenceInfo]:
        """
        Method to get all reference infos of a model.

        Args:
            referenced_model (Identifiable): The model to get the reference infos for.

        Returns:
            List[ReferenceInfo]: The list of reference infos.
        """
        referenced_model_id = get_id_with_patch(referenced_model)
        if not referenced_model_id in self._reference_info_dict_for_referenced:
            return []
        return list(
            self._reference_info_dict_for_referenced[referenced_model_id].values()
        )

    def get_referencing_models(
        self, referenced_model: Identifiable
    ) -> List[Identifiable]:
        """
        Method to get all models that reference a specific model directly as an attribute or by its id.

        Args:
            referenced_model (Identifiable): The model to get the referencing models for.

        Returns:
            List[Identifiable]: The list of referencing models of the model.
        """
        referenced_model_id = get_id_with_patch(referenced_model)
        if not referenced_model_id in self._reference_info_dict_for_referenced:
            return []
        referencing_model_dict = self._reference_info_dict_for_referenced[
            referenced_model_id
        ]
        return [self.get_model(model_id) for model_id in referencing_model_dict]

    def get_referencing_models_of_type(
        self, referenced_model: Identifiable, referencing_model_type: Type[T]
    ) -> List[T]:
        """
        Method to get all models that reference a specific model directly as an attribute or by its id.

        Args:
            referenced_model (Identifiable): The model to get the referencing models for.
            referencing_model_type (Type[T]): The type of the referencing models to get.

        Returns:
            List[T]: The list of referencing models of the model.
        """
        referenced_model_id = get_id_with_patch(referenced_model)
        if not referenced_model_id in self._reference_info_dict_for_referenced:
            return []
        referencing_model_dict = self._reference_info_dict_for_referenced[
            referenced_model_id
        ]
        return [
            self.get_model(model_id)
            for model_id in referencing_model_dict
            if isinstance(self.get_model(model_id), referencing_model_type)
        ]

    def get_referenced_info(
        self, referencing_model: Identifiable
    ) -> List[ReferenceInfo]:
        """
        Method to get all reference infos of a model.

        Args:
            referencing_model (Identifiable): The model to get the reference infos for.

        Returns:
            List[ReferenceInfo]: The list of reference infos.
        """
        referencing_model_id = get_id_with_patch(referencing_model)
        if not referencing_model_id in self._reference_info_dict_for_referencing:
            return []
        return list(
            self._reference_info_dict_for_referencing[referencing_model_id].values()
        )

    def get_referenced_models(
        self, referencing_model: Identifiable
    ) -> List[Identifiable]:
        """
        Method to get all models that are referenced by a specific model directly as an attribute or by its id.

        Args:
            referencing_model (Identifiable): The model to get the referenced models for.

        Returns:
            List[Identifiable]: The list of referenced models of the model.
        """
        referencing_model_id = get_id_with_patch(referencing_model)
        if not referencing_model_id in self._reference_info_dict_for_referencing:
            return []
        referenced_model_dict = self._reference_info_dict_for_referencing[
            referencing_model_id
        ]
        return [self.get_model(model_id) for model_id in referenced_model_dict if model_id in self.model_ids]

    def get_referenced_models_of_type(
        self, referencing_model: Identifiable, referenced_model_type: Type[T]
    ) -> List[T]:
        """
        Method to get all models that are referenced by a specific model directly as an attribute or by its id.

        Args:
            referencing_model (Identifiable): The model to get the referenced models for.
            referenced_model_type (Type[T]): The type of the referenced models to get.

        Returns:
            List[T]: The list of referenced models of the model.
        """
        referencing_model_id = get_id_with_patch(referencing_model)
        if not referencing_model_id in self._reference_info_dict_for_referencing:
            return []
        referenced_model_dict = self._reference_info_dict_for_referencing[
            referencing_model_id
        ]
        return [
            self.get_model(model_id)
            for model_id in referenced_model_dict
            if isinstance(self.get_model(model_id), referenced_model_type)
        ]

    def get_model(self, model_id: str) -> Identifiable:
        """
        Method to get a model by its id.

        Args:
            model_id (str): The id of the model to get.

        Returns:
            Identifiable: The model.
        """
        if model_id not in self.model_ids:
            return None
        return self._key_ids_models[model_id]

    def contains_model(self, model_id: str) -> bool:
        """
        Method to check if a model is contained in the data model.

        Args:
            model_id (str): The id of the model to check.

        Returns:
            bool: True if the model is contained, False otherwise.
        """
        if self.get_model(model_id) is not None:
            return True
        return False

model_ids: Set[str] property readonly

Property to get the ids of all contained models.

Returns:

Type	Description
Set[str]	The set of ids.

add(self, *models)

Method to add models to the data model.

Parameters:

Name	Type	Description	Default
*models	Tuple[Identifiable]	The models to load into the data model.	()

Source code in aas_middleware\model\data_model.py

def add(self, *models: Identifiable) -> None:
    """
    Method to add models to the data model.

    Args:
        *models (Tuple[Identifiable]): The models to load into the data model.
    """
    for model in models:
        self.add_model(model)

add_model(self, model)

Method to load a model of the data model.

Parameters:

Name	Type	Description	Default
model	Identifiable	The model to load.	required

Source code in aas_middleware\model\data_model.py

def add_model(self, model: Identifiable) -> None:
    """
    Method to load a model of the data model.

    Args:
        model (Identifiable): The model to load.
    """
    # Identifiable.model_validate(model)
    model_id = get_id_with_patch(model)
    if model_id in self.model_ids:
        raise ValueError(f"Model with id {model_id} already loaded.")
    self.add_schema(type(model))
    all_identifiables, reference_infos = ReferenceFinder.find(model)
    self._add_contained_models(model, all_identifiables)
    self._add_top_level_model(model)
    self._add_references_to_referencing_models_dict(reference_infos)

add_schema(self, schema)

Method to add a schema of the data model.

Parameters:

Name	Type	Description	Default
schema	Type[Identifiable]	The schema to load.	required

Source code in aas_middleware\model\data_model.py

def add_schema(self, schema: Type[Identifiable]) -> None:
    """
    Method to add a schema of the data model.

    Args:
        schema (Type[Identifiable]): The schema to load.
    """
    all_schemas, schema_reference_infos = ReferenceFinder.find_schema_references(schema)
    self._add_contained_schemas(all_schemas)
    self._add_top_level_schema(schema)
    self._add_schema_references_to_referencing_schemas_dict(schema_reference_infos)

check_different_model_with_same_id_contained(self, model)

Method to check if a model is already contained in the data model.

Parameters:

Name	Type	Description	Default
model	Identifiable	The model to check.	required

Returns:

Type	Description
bool	True if the model is already contained, False otherwise.

Source code in aas_middleware\model\data_model.py

def check_different_model_with_same_id_contained(self, model: Identifiable) -> bool:
    """
    Method to check if a model is already contained in the data model.

    Args:
        model (Identifiable): The model to check.

    Returns:
        bool: True if the model is already contained, False otherwise.
    """
    model_id = get_id_with_patch(model)
    if model_id in self.model_ids:
        same_id_model = self.get_model(model_id)
        if not same_id_model == model:
            return True
    return False

contains_model(self, model_id)

Method to check if a model is contained in the data model.

Parameters:

Name	Type	Description	Default
model_id	str	The id of the model to check.	required

Returns:

Type	Description
bool	True if the model is contained, False otherwise.

Source code in aas_middleware\model\data_model.py

def contains_model(self, model_id: str) -> bool:
    """
    Method to check if a model is contained in the data model.

    Args:
        model_id (str): The id of the model to check.

    Returns:
        bool: True if the model is contained, False otherwise.
    """
    if self.get_model(model_id) is not None:
        return True
    return False

dict(self)

Method to get the dict of the data model.

Returns:

Type	Description
NESTED_DICT	The dict of the data model.

Source code in aas_middleware\model\data_model.py

def dict(self) -> NESTED_DICT:
    """
    Method to get the dict of the data model.

    Returns:
        NESTED_DICT: The dict of the data model.
    """
    nested_dict = {}
    for attribute_name, attribute_value in self.get_top_level_models().items():
        nested_dict[attribute_name] = [
            model.model_dump() for model in attribute_value
        ]
    return nested_dict

from_dict(self, data, types)

Method to load a data model from a dict.

Parameters:

Name	Type	Description	Default
data	NESTED_DICT	The dict to load the data model from.	required

Source code in aas_middleware\model\data_model.py

def from_dict(self, data: NESTED_DICT, types: List[Type]) -> None:
    """
    Method to load a data model from a dict.

    Args:
        data (NESTED_DICT): The dict to load the data model from.
    """
    for attribute_name, attribute_value in data.items():
        class_name = convert_under_score_to_camel_case_str(attribute_name)
        for type_ in types:
            if type_.__name__ == class_name:
                type_for_attribute_values = type_
                break
        else:
            raise ValueError(f"Type {class_name} not supported.")
        for model_dict in attribute_value:
            model = type_for_attribute_values(**model_dict)
            self.add(model)

from_model_types(*model_types, **data) classmethod

Method to create a data model a provided list of model types.

Parameters:

Name	Type	Description	Default
model_types	Tuple[Type[Identifiable]]	The model types to load into the data model.	()
data	Dict[str, Any]	The data to load into the data model.	{}

Returns:

Type	Description
DataModel	The data model with loaded models

Source code in aas_middleware\model\data_model.py

@classmethod
def from_model_types(cls, *model_types: Tuple[Type[Identifiable]], **data: Dict[str, Any]) -> DataModel:
    """
    Method to create a data model a provided list of model types.

    Args:
        model_types (Tuple[Type[Identifiable]]): The model types to load into the data model.
        data (Dict[str, Any]): The data to load into the data model.

    Returns:
        DataModel: The data model with loaded models
    """
    data_model = cls(**data)
    for model_type in model_types:
        data_model.add_schema(model_type)
    return data_model

from_models(*models, **data) classmethod

Method to create a data model from a list of provided models.

Parameters:

Name	Type	Description	Default
models	Tuple[Identifiable]	The models to load into the data model.	()
data	Dict[str, Any]	The data to load into the data model.	{}

Returns:

Type	Description
DataModel	The data model with loaded models

Source code in aas_middleware\model\data_model.py

@classmethod
def from_models(
    cls, *models: Tuple[Identifiable], **data: Dict[str, Any]
) -> DataModel:
    """
    Method to create a data model from a list of provided models.

    Args:
        models (Tuple[Identifiable]): The models to load into the data model.
        data (Dict[str, Any]): The data to load into the data model.

    Returns:
        DataModel: The data model with loaded models
    """
    data_model = cls(**data)
    data_model.add(*models)
    return data_model

get_contained_ids(self)

Method to get all ids of contained models.

Returns:

Type	Description
Set[str]	The set of ids.

Source code in aas_middleware\model\data_model.py

def get_contained_ids(self) -> Set[str]:
    """
    Method to get all ids of contained models.

    Returns:
        Set[str]: The set of ids.
    """
    return self.model_ids

get_contained_models(self)

Method to get all models that are contained in the data model.

Returns:

Type	Description
List[Identifiable]	The list of models.

Source code in aas_middleware\model\data_model.py

def get_contained_models(self) -> List[Identifiable]:
    """
    Method to get all models that are contained in the data model.

    Returns:
        List[Identifiable]: The list of models.
    """
    return list(self._key_ids_models.values())

get_model(self, model_id)

Method to get a model by its id.

Parameters:

Name	Type	Description	Default
model_id	str	The id of the model to get.	required

Returns:

Type	Description
Identifiable	The model.

Source code in aas_middleware\model\data_model.py

def get_model(self, model_id: str) -> Identifiable:
    """
    Method to get a model by its id.

    Args:
        model_id (str): The id of the model to get.

    Returns:
        Identifiable: The model.
    """
    if model_id not in self.model_ids:
        return None
    return self._key_ids_models[model_id]

get_models_of_type(self, model_type)

Method to get all models of a specific type.

Parameters:

Name	Type	Description	Default
model_type	Type[T]	The type of the models to get.	required

Returns:

Type	Description
List[T]	The list of models of the type.

Source code in aas_middleware\model\data_model.py

def get_models_of_type(self, model_type: Type[T]) -> List[T]:
    """
    Method to get all models of a specific type.

    Args:
        model_type (Type[T]): The type of the models to get.

    Returns:
        List[T]: The list of models of the type.
    """
    type_name = model_type.__name__.split(".")[-1]
    return self.get_models_of_type_name(type_name)

get_models_of_type_name(self, model_type_name)

Method to get all models of a specific type.

Parameters:

Name	Type	Description	Default
model_type	str	The type of the models to get.	required

Returns:

Type	Description
List[Identifiable]	The list of models of the type.

Source code in aas_middleware\model\data_model.py

def get_models_of_type_name(self, model_type_name: str) -> List[Identifiable]:
    """
    Method to get all models of a specific type.

    Args:
        model_type (str): The type of the models to get.

    Returns:
        List[Identifiable]: The list of models of the type.
    """
    if not model_type_name in self._models_key_type:
        raise ValueError(f"Model type {model_type_name} not supported.")
    return [
        self.get_model(model_id)
        for model_id in self._models_key_type[model_type_name]
    ]

get_referenced_info(self, referencing_model)

Method to get all reference infos of a model.

Parameters:

Name	Type	Description	Default
referencing_model	Identifiable	The model to get the reference infos for.	required

Returns:

Type	Description
List[ReferenceInfo]	The list of reference infos.

Source code in aas_middleware\model\data_model.py

def get_referenced_info(
    self, referencing_model: Identifiable
) -> List[ReferenceInfo]:
    """
    Method to get all reference infos of a model.

    Args:
        referencing_model (Identifiable): The model to get the reference infos for.

    Returns:
        List[ReferenceInfo]: The list of reference infos.
    """
    referencing_model_id = get_id_with_patch(referencing_model)
    if not referencing_model_id in self._reference_info_dict_for_referencing:
        return []
    return list(
        self._reference_info_dict_for_referencing[referencing_model_id].values()
    )

get_referenced_models(self, referencing_model)

Method to get all models that are referenced by a specific model directly as an attribute or by its id.

Parameters:

Name	Type	Description	Default
referencing_model	Identifiable	The model to get the referenced models for.	required

Returns:

Type	Description
List[Identifiable]	The list of referenced models of the model.

Source code in aas_middleware\model\data_model.py

def get_referenced_models(
    self, referencing_model: Identifiable
) -> List[Identifiable]:
    """
    Method to get all models that are referenced by a specific model directly as an attribute or by its id.

    Args:
        referencing_model (Identifiable): The model to get the referenced models for.

    Returns:
        List[Identifiable]: The list of referenced models of the model.
    """
    referencing_model_id = get_id_with_patch(referencing_model)
    if not referencing_model_id in self._reference_info_dict_for_referencing:
        return []
    referenced_model_dict = self._reference_info_dict_for_referencing[
        referencing_model_id
    ]
    return [self.get_model(model_id) for model_id in referenced_model_dict if model_id in self.model_ids]

get_referenced_models_of_type(self, referencing_model, referenced_model_type)

Method to get all models that are referenced by a specific model directly as an attribute or by its id.

Parameters:

Name	Type	Description	Default
referencing_model	Identifiable	The model to get the referenced models for.	required
referenced_model_type	Type[T]	The type of the referenced models to get.	required

Returns:

Type	Description
List[T]	The list of referenced models of the model.

Source code in aas_middleware\model\data_model.py

def get_referenced_models_of_type(
    self, referencing_model: Identifiable, referenced_model_type: Type[T]
) -> List[T]:
    """
    Method to get all models that are referenced by a specific model directly as an attribute or by its id.

    Args:
        referencing_model (Identifiable): The model to get the referenced models for.
        referenced_model_type (Type[T]): The type of the referenced models to get.

    Returns:
        List[T]: The list of referenced models of the model.
    """
    referencing_model_id = get_id_with_patch(referencing_model)
    if not referencing_model_id in self._reference_info_dict_for_referencing:
        return []
    referenced_model_dict = self._reference_info_dict_for_referencing[
        referencing_model_id
    ]
    return [
        self.get_model(model_id)
        for model_id in referenced_model_dict
        if isinstance(self.get_model(model_id), referenced_model_type)
    ]

get_referencing_info(self, referenced_model)

Method to get all reference infos of a model.

Parameters:

Name	Type	Description	Default
referenced_model	Identifiable	The model to get the reference infos for.	required

Returns:

Type	Description
List[ReferenceInfo]	The list of reference infos.

Source code in aas_middleware\model\data_model.py

def get_referencing_info(
    self, referenced_model: Identifiable
) -> List[ReferenceInfo]:
    """
    Method to get all reference infos of a model.

    Args:
        referenced_model (Identifiable): The model to get the reference infos for.

    Returns:
        List[ReferenceInfo]: The list of reference infos.
    """
    referenced_model_id = get_id_with_patch(referenced_model)
    if not referenced_model_id in self._reference_info_dict_for_referenced:
        return []
    return list(
        self._reference_info_dict_for_referenced[referenced_model_id].values()
    )

get_referencing_models(self, referenced_model)

Method to get all models that reference a specific model directly as an attribute or by its id.

Parameters:

Name	Type	Description	Default
referenced_model	Identifiable	The model to get the referencing models for.	required

Returns:

Type	Description
List[Identifiable]	The list of referencing models of the model.

Source code in aas_middleware\model\data_model.py

def get_referencing_models(
    self, referenced_model: Identifiable
) -> List[Identifiable]:
    """
    Method to get all models that reference a specific model directly as an attribute or by its id.

    Args:
        referenced_model (Identifiable): The model to get the referencing models for.

    Returns:
        List[Identifiable]: The list of referencing models of the model.
    """
    referenced_model_id = get_id_with_patch(referenced_model)
    if not referenced_model_id in self._reference_info_dict_for_referenced:
        return []
    referencing_model_dict = self._reference_info_dict_for_referenced[
        referenced_model_id
    ]
    return [self.get_model(model_id) for model_id in referencing_model_dict]

get_referencing_models_of_type(self, referenced_model, referencing_model_type)

Method to get all models that reference a specific model directly as an attribute or by its id.

Parameters:

Name	Type	Description	Default
referenced_model	Identifiable	The model to get the referencing models for.	required
referencing_model_type	Type[T]	The type of the referencing models to get.	required

Returns:

Type	Description
List[T]	The list of referencing models of the model.

Source code in aas_middleware\model\data_model.py

def get_referencing_models_of_type(
    self, referenced_model: Identifiable, referencing_model_type: Type[T]
) -> List[T]:
    """
    Method to get all models that reference a specific model directly as an attribute or by its id.

    Args:
        referenced_model (Identifiable): The model to get the referencing models for.
        referencing_model_type (Type[T]): The type of the referencing models to get.

    Returns:
        List[T]: The list of referencing models of the model.
    """
    referenced_model_id = get_id_with_patch(referenced_model)
    if not referenced_model_id in self._reference_info_dict_for_referenced:
        return []
    referencing_model_dict = self._reference_info_dict_for_referenced[
        referenced_model_id
    ]
    return [
        self.get_model(model_id)
        for model_id in referencing_model_dict
        if isinstance(self.get_model(model_id), referencing_model_type)
    ]

get_top_level_models(self)

Method to get all models of the data model.

Returns:

Type	Description
Dict[str, List[Identifiable]]	The dictionary of models.

Source code in aas_middleware\model\data_model.py

def get_top_level_models(self) -> Dict[str, List[Identifiable]]:
    """
    Method to get all models of the data model.

    Returns:
        Dict[str, List[Identifiable]]: The dictionary of models.
    """
    top_level_models = {}
    for top_level_model_name, top_level_model_ids in self._top_level_models.items():
        top_level_models[top_level_model_name] = [
            self.get_model(model_id) for model_id in top_level_model_ids
        ]
    return top_level_models

get_top_level_types(self)

Method to get all types of the top level models in the data model.

Returns:

Type	Description
List[Type[Identifiable]]	The types of the top level models in the data model

Source code in aas_middleware\model\data_model.py

def get_top_level_types(self) -> List[Type[Identifiable]]:
    """
    Method to get all types of the top level models in the data model.

    Returns:
        List[Type[Identifiable]]: The types of the top level models in the data model
    """
    top_level_types = []
    for schema_name in self._top_level_schemas:
        top_level_types.append(self._schemas[schema_name])
    return top_level_types

json(self)

Method to get the json of the data model.

Returns:

Type	Description
str	The json of the data model.

Source code in aas_middleware\model\data_model.py

def json(self) -> str:
    """
    Method to get the json of the data model.

    Returns:
        str: The json of the data model.
    """
    nested_dict = {}
    for attribute_name, attribute_value in self.get_top_level_models().items():
        # TODO: if a non-BaseModel object is loaded, this breakds down -> adjust this
        nested_dict[attribute_name] = [
            model.model_dump() for model in attribute_value
        ]
    return json.dumps(nested_dict, indent=4, cls=DateTimeEncoder)

model_post_init(/, self, context)

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that's what pydantic-core passes when calling it.

Parameters:

Name	Type	Description	Default
self	BaseModel	The BaseModel instance.	required
context	Any	The context.	required

Source code in aas_middleware\model\data_model.py

def init_private_attributes(self: BaseModel, context: Any, /) -> None:
    """This function is meant to behave like a BaseModel method to initialise private attributes.

    It takes context as an argument since that's what pydantic-core passes when calling it.

    Args:
        self: The BaseModel instance.
        context: The context.
    """
    if getattr(self, '__pydantic_private__', None) is None:
        pydantic_private = {}
        for name, private_attr in self.__private_attributes__.items():
            default = private_attr.get_default()
            if default is not PydanticUndefined:
                pydantic_private[name] = default
        object_setattr(self, '__pydantic_private__', pydantic_private)

patch_schema_references(self)

Function tries to patch schema reference infos to represent a more realistic data model.

ReferenceInfos of Type Reference are patched so that another ReferenceInfo is added to every schema that has a class name that contains the reference id. ReferenceInfos of Type Attribute are patched so that another ReferenceInfo is added to every schema that has a class name that contains the reference id.

Source code in aas_middleware\model\data_model.py

def patch_schema_references(self) -> None:
    """
    Function tries to patch schema reference infos to represent a more realistic data model.

    ReferenceInfos of Type Reference are patched so that another ReferenceInfo is added to every schema that has a class name that contains the reference id.
    ReferenceInfos of Type Attribute are patched so that another ReferenceInfo is added to every schema that has a class name that contains the reference id.
    """
    reference_infos = patch_references(self._schema_reference_infos, self._schemas.values())
    self._schema_reference_infos = reference_infos
    self._add_schema_references_to_referencing_schemas_dict(reference_infos)

DateTimeEncoder (JSONEncoder)

Source code in aas_middleware\model\data_model.py

class DateTimeEncoder(json.JSONEncoder):
    def default(self, o):
        if isinstance(o, datetime):
            return o.isoformat()

        return super().default(o)

default(self, o)

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this::

def default(self, o):
    !!! try
        iterable = iter(o)
    except TypeError:
        pass
    !!! else
        return list(iterable)
    # Let the base class default method raise the TypeError
    return JSONEncoder.default(self, o)

Source code in aas_middleware\model\data_model.py

def default(self, o):
    if isinstance(o, datetime):
        return o.isoformat()

    return super().default(o)

data_model_rebuilder

DataModelRebuilder

Source code in aas_middleware\model\data_model_rebuilder.py

class DataModelRebuilder:
    def __init__(self, data_model: DataModel):
        """
        Rebuilds a data model with either direct or indirect references.

        Args:
            data_model (DataModel): The data model to rebuild.
        """
        self.data_model = data_model

    def rebuild_data_model_with_associations(self) -> DataModel:
        """
        Rebuilds all models in the data model with assosiations.

        Returns:
            DataModel: The rebuilt data model.
        """
        raise NotImplementedError

    def rebuild_data_model_for_AAS_structure(self) -> DataModel:
        """
        Rebuilds the data model for AAS meta model structure by adjusting the associations and references and infering correct AAS types.

        Returns:
            DataModel: The rebuilt data model.
        """
        aas_candidates = []
        submodel_candidates = []
        top_level_models_list = []
        for models in self.data_model.get_top_level_models().values():
            top_level_models_list += models

        for model in top_level_models_list:
            if isinstance(model, aas_model.AAS):
                aas_candidates.append(model)
                continue
            if any(
                reference_info.reference_type == ReferenceType.ASSOCIATION
                for reference_info in self.data_model.get_referencing_info(model)
            ):
                continue
            if not all(
                is_identifiable(attribute_value)
                for attribute_value in get_value_attributes(model).values()
            ):
                continue
            aas_candidates.append(model)

        for model in top_level_models_list:
            if model in aas_candidates:
                continue
            submodel_candidates.append(model)

        submodel_objects = []
        for submodel_candidate in submodel_candidates:
            # TODO: remove the need to patch models that are already subclasses of submodel...
            patched_submodel_object = get_patched_aas_object(
                submodel_candidate, patch_type=aas_model.Submodel
            )
            submodel_objects.append(patched_submodel_object)

        aas_objects = []
        for aas_candidate in aas_candidates:
            # TODO: remove the need to patch models that are already subclasses of aas...
            patched_aas_object = get_patched_aas_object(
                aas_candidate, patch_type=aas_model.AAS
            )
            aas_objects.append(patched_aas_object)

        aas_data_model = DataModel.from_models(*aas_objects + submodel_objects)
        return aas_data_model

    def rebuild_data_model_with_references(self) -> DataModel:
        """
        Rebuilds all models in the data model with references.

        Returns:
            DataModel: The rebuilt data model.
        """
        raise NotImplementedError

__init__(self, data_model) special

Rebuilds a data model with either direct or indirect references.

Parameters:

Name	Type	Description	Default
data_model	DataModel	The data model to rebuild.	required

Source code in aas_middleware\model\data_model_rebuilder.py

def __init__(self, data_model: DataModel):
    """
    Rebuilds a data model with either direct or indirect references.

    Args:
        data_model (DataModel): The data model to rebuild.
    """
    self.data_model = data_model

rebuild_data_model_for_AAS_structure(self)

Rebuilds the data model for AAS meta model structure by adjusting the associations and references and infering correct AAS types.

Returns:

Type	Description
DataModel	The rebuilt data model.

Source code in aas_middleware\model\data_model_rebuilder.py

def rebuild_data_model_for_AAS_structure(self) -> DataModel:
    """
    Rebuilds the data model for AAS meta model structure by adjusting the associations and references and infering correct AAS types.

    Returns:
        DataModel: The rebuilt data model.
    """
    aas_candidates = []
    submodel_candidates = []
    top_level_models_list = []
    for models in self.data_model.get_top_level_models().values():
        top_level_models_list += models

    for model in top_level_models_list:
        if isinstance(model, aas_model.AAS):
            aas_candidates.append(model)
            continue
        if any(
            reference_info.reference_type == ReferenceType.ASSOCIATION
            for reference_info in self.data_model.get_referencing_info(model)
        ):
            continue
        if not all(
            is_identifiable(attribute_value)
            for attribute_value in get_value_attributes(model).values()
        ):
            continue
        aas_candidates.append(model)

    for model in top_level_models_list:
        if model in aas_candidates:
            continue
        submodel_candidates.append(model)

    submodel_objects = []
    for submodel_candidate in submodel_candidates:
        # TODO: remove the need to patch models that are already subclasses of submodel...
        patched_submodel_object = get_patched_aas_object(
            submodel_candidate, patch_type=aas_model.Submodel
        )
        submodel_objects.append(patched_submodel_object)

    aas_objects = []
    for aas_candidate in aas_candidates:
        # TODO: remove the need to patch models that are already subclasses of aas...
        patched_aas_object = get_patched_aas_object(
            aas_candidate, patch_type=aas_model.AAS
        )
        aas_objects.append(patched_aas_object)

    aas_data_model = DataModel.from_models(*aas_objects + submodel_objects)
    return aas_data_model

rebuild_data_model_with_associations(self)

Rebuilds all models in the data model with assosiations.

Returns:

Type	Description
DataModel	The rebuilt data model.

Source code in aas_middleware\model\data_model_rebuilder.py

def rebuild_data_model_with_associations(self) -> DataModel:
    """
    Rebuilds all models in the data model with assosiations.

    Returns:
        DataModel: The rebuilt data model.
    """
    raise NotImplementedError

rebuild_data_model_with_references(self)

Rebuilds all models in the data model with references.

Returns:

Type	Description
DataModel	The rebuilt data model.

Source code in aas_middleware\model\data_model_rebuilder.py

def rebuild_data_model_with_references(self) -> DataModel:
    """
    Rebuilds all models in the data model with references.

    Returns:
        DataModel: The rebuilt data model.
    """
    raise NotImplementedError

get_patched_aas_object(model, patch_type)

Rebuilds an Identifiable object to an AAS object.

Parameters:

Name	Type	Description	Default
model	object	The object to rebuild.	required

Returns:

Type	Description
aas_model.AAS	The rebuilt AAS object.

Source code in aas_middleware\model\data_model_rebuilder.py

def get_patched_aas_object(model: Any, patch_type: Type[T]) -> T:
    """
    Rebuilds an Identifiable object to an AAS object.

    Args:
        model (object): The object to rebuild.

    Returns:
        aas_model.AAS: The rebuilt AAS object.
    """
    model_id = get_id_with_patch(model)
    dict_dynamic_model_creation = {}
    dict_model_instantiation = {}
    dict_model_instantiation["id_short"] = model_id
    if not patch_type == aas_model.SubmodelElementCollection:
        dict_model_instantiation["id"] = model_id

    for attribute_name, attribute_value in get_value_attributes(model).items():
        below_patch_type = get_below_patch_type(patch_type)

        if is_identifiable(attribute_value):
            patched_attribute_value = get_patched_aas_object(
                attribute_value, patch_type=below_patch_type
            )
        elif is_identifiable_container(attribute_value):
            patched_attribute_value = []
            for element in attribute_value:
                patched_element = get_patched_aas_object(
                    element, patch_type=below_patch_type
                )
                patched_attribute_value.append(patched_element)
        else:
            patched_attribute_value = attribute_value

        attribute_patch_type = get_patched_type(model, attribute_name, patched_attribute_value)

        dict_dynamic_model_creation.update(
            {
                attribute_name: (
                    attribute_patch_type,
                    Field(examples=[patched_attribute_value]),
                )
            }
        )
        dict_model_instantiation.update({attribute_name: patched_attribute_value})
    new_model_type = create_model(
        model.__class__.__name__, **dict_dynamic_model_creation, __base__=patch_type
    )
    return new_model_type.model_validate(dict_model_instantiation)

formatting special

aas special

aas_json_formatter

AasJsonFormatter

Allows to serialize and deserialize Basyx AAS objects (AssetAdministrationShells, Submodels or Containers of both) to a DataModel.

Source code in aas_middleware\model\formatting\aas\aas_json_formatter.py

class AasJsonFormatter:
    """
    Allows to serialize and deserialize Basyx AAS objects (AssetAdministrationShells, Submodels or Containers of both) to a DataModel.
    """

    def serialize(self, data: DataModel) -> Dict[Literal["assetAdministrationShells", "submodels"], List[str]]:
        """
        Serialize a DataModel object to the specific format of the formatter.

        Args:
            data (DataModel): A data model

        Returns:
            Objectstore: the basyx object store contain all AAS elements
        """
        basyx_dict_obj_store = BasyxFormatter().serialize(data)
        return json.loads(basyx.aas.adapter.json.object_store_to_json(basyx_dict_obj_store))


    def deserialize(self, data: Dict[Literal["assetAdministrationShells", "submodels"], List[str]]) -> DataModel:
        """
        Deserialize the specific format of the formater to a DataModel object.

        Args:
            data (Any): The specific format of the formatter.

        Returns:
            DataModel: A data model that holds the objects that were deserialized
        """
        object_store = DictObjectStore()
        for key, items in data.items():
            if key == "assetAdministrationShells":
                for aas_item in items:
                    aas = json.loads(json.dumps(aas_item), cls=basyx.aas.adapter.json.AASFromJsonDecoder)
                    object_store.add(aas)
            elif key == "submodels":
                for submodel_item in items:
                    submodel = json.loads(json.dumps(submodel_item), cls=basyx.aas.adapter.json.AASFromJsonDecoder)
                    object_store.add(submodel)
        return BasyxFormatter().deserialize(object_store)

deserialize(self, data)

Deserialize the specific format of the formater to a DataModel object.

Parameters:

Name	Type	Description	Default
data	Any	The specific format of the formatter.	required

Returns:

Type	Description
DataModel	A data model that holds the objects that were deserialized

Source code in aas_middleware\model\formatting\aas\aas_json_formatter.py

def deserialize(self, data: Dict[Literal["assetAdministrationShells", "submodels"], List[str]]) -> DataModel:
    """
    Deserialize the specific format of the formater to a DataModel object.

    Args:
        data (Any): The specific format of the formatter.

    Returns:
        DataModel: A data model that holds the objects that were deserialized
    """
    object_store = DictObjectStore()
    for key, items in data.items():
        if key == "assetAdministrationShells":
            for aas_item in items:
                aas = json.loads(json.dumps(aas_item), cls=basyx.aas.adapter.json.AASFromJsonDecoder)
                object_store.add(aas)
        elif key == "submodels":
            for submodel_item in items:
                submodel = json.loads(json.dumps(submodel_item), cls=basyx.aas.adapter.json.AASFromJsonDecoder)
                object_store.add(submodel)
    return BasyxFormatter().deserialize(object_store)

serialize(self, data)

Serialize a DataModel object to the specific format of the formatter.

Parameters:

Name	Type	Description	Default
data	DataModel	A data model	required

Returns:

Type	Description
Objectstore	the basyx object store contain all AAS elements

Source code in aas_middleware\model\formatting\aas\aas_json_formatter.py

def serialize(self, data: DataModel) -> Dict[Literal["assetAdministrationShells", "submodels"], List[str]]:
    """
    Serialize a DataModel object to the specific format of the formatter.

    Args:
        data (DataModel): A data model

    Returns:
        Objectstore: the basyx object store contain all AAS elements
    """
    basyx_dict_obj_store = BasyxFormatter().serialize(data)
    return json.loads(basyx.aas.adapter.json.object_store_to_json(basyx_dict_obj_store))

aas_middleware_util

convert_under_score_to_camel_case_str(underscore_str)

Convert a underscore seperated string to a camel case string.

Parameters:

Name	Type	Description	Default
class_name	str	The underscore seperated string to convert.	required

Returns:

Type	Description
str	The camel case string.

Source code in aas_middleware\model\formatting\aas\aas_middleware_util.py

def convert_under_score_to_camel_case_str(underscore_str: str) -> str:
    """
    Convert a underscore seperated string to a camel case string.

    Args:
        class_name (str): The underscore seperated string to convert.

    Returns:
        str: The camel case string.
    """
    words = underscore_str.split("_")
    camel_case_str = "".join(word.title() for word in words)
    return camel_case_str

core_model_check(fieldinfo)

Checks if a pydantic model field is a core model.

Parameters:

Name	Type	Description	Default
fieldinfo	FieldInfo	Pydantic model field.	required

Returns:

Type	Description
bool	If the model field is a core model.

Source code in aas_middleware\model\formatting\aas\aas_middleware_util.py

def core_model_check(fieldinfo: FieldInfo) -> bool:
    """
    Checks if a pydantic model field is a core model.

    Args:
        fieldinfo (FieldInfo): Pydantic model field.

    Returns:
        bool: If the model field is a core model.
    """
    if isinstance(fieldinfo.default, BaseModel):
        return True
    if typing.get_origin(fieldinfo.annotation) is typing.Union:
        args = typing.get_args(fieldinfo.annotation)
        if all(issubclass(arg, BaseModel) for arg in args):
            return True
    else:
        if issubclass(fieldinfo.annotation, BaseModel):
            return True

get_all_submodel_elements_from_submodel(model)

Function to get all submodel elements from a pydantic submodel

Parameters:

Name	Type	Description	Default
model	Type[BaseModel]	The pydantic submodel to get the submodel elements from	required

Returns:

Type	Description
List[aas_model.SubmodelElementCollection | list | str | bool | float | int]	A list of all submodel elements in the pydantic submodel

Source code in aas_middleware\model\formatting\aas\aas_middleware_util.py

def get_all_submodel_elements_from_submodel(
    model: Type[aas_model.Submodel],
) -> Dict[
    str, Type[aas_model.SubmodelElementCollection | list | str | bool | float | int]
]:
    """
    Function to get all submodel elements from a pydantic submodel

    Args:
        model (Type[BaseModel]): The pydantic submodel to get the submodel elements from

    Returns:
        List[aas_model.SubmodelElementCollection | list | str | bool | float | int]: A list of all submodel elements in the pydantic submodel
    """
    submodel_elements = {}
    for field_name, field_info in model.model_fields.items():
        if (
            field_name != "description"
            and field_name != "id_short"
            and field_name != "semantic_id"
            and field_name != "id"
        ):
            submodel_elements[field_name] = field_info.annotation
    return submodel_elements

get_all_submodels_from_object_store(obj_store)

Function to get all basyx submodels from an object store

Parameters:

Name	Type	Description	Default
obj_store	model.DictObjectStore	Object store to get submodels from	required

Returns:

Type	Description
List[model.Submodel]	List of basyx submodels

Source code in aas_middleware\model\formatting\aas\aas_middleware_util.py

def get_all_submodels_from_object_store(
    obj_store: model.DictObjectStore,
) -> List[model.Submodel]:
    """
    Function to get all basyx submodels from an object store
    Args:
        obj_store (model.DictObjectStore): Object store to get submodels from
    Returns:
        List[model.Submodel]: List of basyx submodels
    """
    submodels = []
    for item in obj_store:
        if isinstance(item, model.Submodel):
            submodels.append(item)
    return submodels

get_contained_models_attribute_info(model)

Function to get all submodels from a pydantic model

Parameters:

Name	Type	Description	Default
model	Type[BaseModel]	The pydantic model to get the submodels from	required

Returns:

Type	Description
List[Tuple[str, Type[aas_model.Submodel]]]	List of attribute name and value of all submodels in the pydantic model

Source code in aas_middleware\model\formatting\aas\aas_middleware_util.py

def get_contained_models_attribute_info(
    model: Type[BaseModel],
) -> List[Tuple[str, Type[aas_model.Submodel]]]:
    """
    Function to get all submodels from a pydantic model
    Args:
        model (Type[BaseModel]): The pydantic model to get the submodels from
    Returns:
        List[Tuple[str, Type[aas_model.Submodel]]]: List of attribute name and value of all submodels in the pydantic model
    """
    submodels = []
    for attribute_name, fieldinfo in model.model_fields.items():
        if typing.get_args(fieldinfo.annotation) != () and any(issubclass(arg, aas_model.Submodel) for arg in typing.get_args(fieldinfo.annotation)):
            submodels.append((attribute_name, fieldinfo.annotation))
        elif issubclass(fieldinfo.annotation, aas_model.Submodel):
            submodels.append((attribute_name, fieldinfo.annotation))
    return submodels

get_field_default_value(fieldinfo)

Function to get the default values of a pydantic model field. If no default is given, the function tries to infer a default cored on the type.

Parameters:

Name	Type	Description	Default
fieldinfo	FieldInfo	Pydantic model field.	required

Returns:

Type	Description
Any	Missing default value.

Source code in aas_middleware\model\formatting\aas\aas_middleware_util.py

def get_field_default_value(fieldinfo: FieldInfo) -> typing.Any:
    """
    Function to get the default values of a pydantic model field. If no default is given, the function tries to infer a default cored on the type.

    Args:
        fieldinfo (FieldInfo): Pydantic model field.

    Returns:
        typing.Any: Missing default value.
    """
    if fieldinfo.default:
        return fieldinfo.default
    elif fieldinfo.default_factory:
        return fieldinfo.default_factory()
    elif fieldinfo.annotation == str:
        return "string"
    elif fieldinfo.annotation == bool:
        return False
    elif fieldinfo.annotation == int:
        return 1
    elif fieldinfo.annotation == float:
        return 1.0
    elif fieldinfo.annotation == list:
        return []

get_pydantic_model_from_dict(dict_values, model_name, all_fields_required=False)

Functions that creates pydantic model from dict.

Parameters:

Name	Type	Description	Default
dict_values	dict	Dictionary of values.	required
model_name	str	Name of the model.	required
all_fields_required	bool	If all fields should be required (non-Optional) in the model. Defaults to False.	False

Returns:

Type	Description
Type[BaseModel]	Pydantic model.

Source code in aas_middleware\model\formatting\aas\aas_middleware_util.py

def get_pydantic_model_from_dict(
    dict_values: dict, model_name: str, all_fields_required: bool = False
) -> Type[BaseModel]:
    """
    Functions that creates pydantic model from dict.

    Args:
        dict_values (dict): Dictionary of values.
        model_name (str): Name of the model.
        all_fields_required (bool, optional): If all fields should be required (non-Optional) in the model. Defaults to False.
    Returns:
        Type[BaseModel]: Pydantic model.
    """
    # TODO: check function below if needed...
    pydantic_model = recusrive_model_creation(model_name, dict_values)
    pydantic_model = set_example_values(pydantic_model)
    if all_fields_required:
        for field_name, field_info in pydantic_model.model_fields.items():
            field_info.default = None
    return pydantic_model

get_pydantic_models_from_instances(instances)

Functions that creates pydantic models from instances.

Parameters:

Name	Type	Description	Default
instances	List[BaseModel]	List of pydantic model instances.	required

Returns:

Type	Description
List[Type[BaseModel]]	List of pydantic models.

Source code in aas_middleware\model\formatting\aas\aas_middleware_util.py

def get_pydantic_models_from_instances(
    instances: List[BaseModel],
) -> List[Type[BaseModel]]:
    """
    Functions that creates pydantic models from instances.

    Args:
        instances (typing.List[BaseModel]): List of pydantic model instances.

    Returns:
        List[Type[BaseModel]]: List of pydantic models.
    """
    # TODO: update method with pydantic v2 arguments of create_model
    models = []
    for instance in instances:
        model_name = type(instance).__name__
        pydantic_model = create_model(model_name, **vars(instance))
        pydantic_model = set_example_values(pydantic_model)
        pydantic_model = set_required_fields(pydantic_model, instance.__class__)
        pydantic_model = set_default_values(pydantic_model, instance.__class__)
        models.append(pydantic_model)
    return models

recusrive_model_creation(model_name, dict_values, depth=0)

Function that creates a pydantic model from a dict.

Parameters:

Name	Type	Description	Default
model_name	_type_	description	required
dict_values	_type_	description	required

Returns:

Type	Description
_type_	description

Source code in aas_middleware\model\formatting\aas\aas_middleware_util.py

def recusrive_model_creation(model_name, dict_values, depth=0):
    """
    Function that creates a pydantic model from a dict.

    Args:
        model_name (_type_): _description_
        dict_values (_type_): _description_

    Returns:
        _type_: _description_
    """
    # TODO: check function below if needed
    for attribute_name, attribute_values in dict_values.items():
        if isinstance(attribute_values, dict):
            class_name = convert_under_score_to_camel_case_str(attribute_name)
            created_model = recusrive_model_creation(
                class_name, attribute_values, depth=depth + 1
            )
            dict_values[attribute_name] = created_model(**attribute_values)
    if depth == 0:
        core_class = aas_model.AAS
    elif depth == 1:
        core_class = aas_model.Submodel
    else:
        core_class = aas_model.SubmodelElementCollection
    return create_model(model_name, **dict_values, __core__=core_class)

save_model_list_with_schema(model_list, path)

Saves a list of pydantic models to a json file.

Parameters:

Name	Type	Description	Default
model_list	List[aas_model.AAS]	List of pydantic models	required
path	str	Path to the json file	required

Source code in aas_middleware\model\formatting\aas\aas_middleware_util.py

def save_model_list_with_schema(model_list: typing.List[BaseModel], path: str):
    """
    Saves a list of pydantic models to a json file.
    Args:
        model_list (typing.List[aas_model.AAS]): List of pydantic models
        path (str): Path to the json file
    """
    save_dict = {
        "models": [model.model_dump() for model in model_list],
        "schema": [model.model_json_schema() for model in model_list],
    }

    with open(path, "w", encoding="utf-8") as json_file:
        json.dump(save_dict, json_file, indent=4)

set_default_values(model, origin_model)

Sets the default values and default factory of a pydantic model cored on a original model.

Parameters:

Name	Type	Description	Default
model	Type[BaseModel]	Pydantic model where default values should be removed.	required

Returns:

Type	Description
Type[BaseModel]	Pydantic model with the default values set.

Source code in aas_middleware\model\formatting\aas\aas_middleware_util.py

def set_default_values(
    model: Type[BaseModel], origin_model: Type[BaseModel]
) -> Type[BaseModel]:
    """
    Sets the default values and default factory of a pydantic model cored on a original model.

    Args:
        model (Type[BaseModel]): Pydantic model where default values should be removed.

    Returns:
        Type[BaseModel]: Pydantic model with the default values set.
    """
    # TODO: validate if this method is still needed in pydantic 2.0
    for field_name, fieldinfo in origin_model.model_fields.items():
        if union_type_field_check(fieldinfo):
            original_sub_types = typing.get_args(fieldinfo.annotation)
            model_sub_types = typing.get_args(model.model_fields[field_name].annotation)
            new_types = []
            for original_sub_type, model_sub_type in zip(
                original_sub_types, model_sub_types
            ):
                new_type = set_default_values(model_sub_type, original_sub_type)
                new_types.append(new_type)
            model.model_fields[field_name].annotation = typing.Union[tuple(new_types)]
        elif core_model_check(fieldinfo):
            new_type = set_default_values(
                model.model_fields[field_name].annotation, fieldinfo.annotation
            )
            model.model_fields[field_name].annotation = new_type
        if not fieldinfo.is_required() and (
            fieldinfo.default
            or fieldinfo.default == ""
            or fieldinfo.default == 0
            or fieldinfo.default == 0.0
            or fieldinfo.default == False
            or fieldinfo.default == []
            or fieldinfo.default == {}
        ):
            model.model_fields[field_name].default = fieldinfo.default
        else:
            model.model_fields[field_name].default = None

        if not fieldinfo.is_required() and fieldinfo.default_factory:
            model.model_fields[field_name].default_factory = fieldinfo.default_factory
        else:
            model.model_fields[field_name].default_factory = None
    return model

set_example_values(model)

Sets the example values of a pydantic model cored on its default values.

Parameters:

Name	Type	Description	Default
model	Type[BaseModel]	Pydantic model.	required

Returns:

Type	Description
Type[BaseModel]	Pydantic model with the example values set.

Source code in aas_middleware\model\formatting\aas\aas_middleware_util.py

def set_example_values(model: Type[BaseModel]) -> Type[BaseModel]:
    """
    Sets the example values of a pydantic model cored on its default values.

    Args:
        model (Type[BaseModel]): Pydantic model.

    Returns:
        Type[BaseModel]: Pydantic model with the example values set.
    """
    # TODO: potentially delete this method, since not required...
    example_dict = {}
    for field_name, fieldinfo in model.model_fields.items():
        if issubclass(fieldinfo.annotation, BaseModel):
            config_dict = ConfigDict(
                json_schema_extra={"examples": [fieldinfo.default.model_dump_json()]}
            )
            fieldinfo.annotation.model_config = config_dict
        example_dict[field_name] = get_field_default_value(fieldinfo)
    serialized_example = model(**example_dict).model_dump_json()
    config_dict = ConfigDict(json_schema_extra={"examples": [serialized_example]})
    model.model_config = config_dict
    return model

set_required_fields(model, origin_model)

Sets the required fields of a pydantic model.

Parameters:

Name	Type	Description	Default
model	Type[BaseModel]	Pydantic model.	required
origin_model	Type[BaseModel]	Pydantic model from which the required fields should be copied.	required

Returns:

Type	Description
Type[BaseModel]	Pydantic model with the required fields set.

Source code in aas_middleware\model\formatting\aas\aas_middleware_util.py

def set_required_fields(
    model: Type[BaseModel], origin_model: Type[BaseModel]
) -> Type[BaseModel]:
    """
    Sets the required fields of a pydantic model.

    Args:
        model (Type[BaseModel]): Pydantic model.
        origin_model (Type[BaseModel]): Pydantic model from which the required fields should be copied.

    Returns:
        Type[BaseModel]: Pydantic model with the required fields set.
    """
    # TODO: potentially delete this method, since not required in pydantic v2
    for field_name, fieldinfo in origin_model.model_fields.items():
        if union_type_field_check(fieldinfo):
            original_sub_types = typing.get_args(fieldinfo.annotation)
            model_sub_types = typing.get_args(model.model_fields[field_name].annotation)
            new_types = []
            for original_sub_type, model_sub_type in zip(
                original_sub_types, model_sub_types
            ):
                new_type = set_required_fields(model_sub_type, original_sub_type)
                new_types.append(new_type)
            model.model_fields[field_name].annotation = typing.Union[tuple(new_types)]
        elif core_model_check(fieldinfo):
            new_type = set_required_fields(
                model.model_fields[field_name].annotation, fieldinfo.annotation
            )
            model.model_fields[field_name].annotation = new_type
        if fieldinfo.is_required():
            model.model_fields[field_name].default = None
            model.model_fields[field_name].default_factory = True
    return model

union_type_check(model)

Checks if a type is a union type.

Parameters:

Name	Type	Description	Default
model	Type	Type.	required

Returns:

Type	Description
bool	If the type is a union type.

Source code in aas_middleware\model\formatting\aas\aas_middleware_util.py

def union_type_check(model: Type) -> bool:
    """
    Checks if a type is a union type.

    Args:
        model (Type): Type.

    Returns:
        bool: If the type is a union type.
    """
    if typing.get_origin(model) == typing.Union:
        args = typing.get_args(model)
        if all(issubclass(arg, BaseModel) for arg in args):
            return True
        else:
            False
    else:
        return False

union_type_field_check(fieldinfo)

Checks if a pydantic model field is a union type.

Parameters:

Name	Type	Description	Default
fieldinfo	FieldInfo	Pydantic model field.	required

Returns:

Type	Description
bool	If the model field is a union type.

Source code in aas_middleware\model\formatting\aas\aas_middleware_util.py

def union_type_field_check(fieldinfo: FieldInfo) -> bool:
    """
    Checks if a pydantic model field is a union type.

    Args:
        fieldinfo (FieldInfo): Pydantic model field.

    Returns:
        bool: If the model field is a union type.
    """
    return union_type_check(fieldinfo.annotation)

aas_model

AAS (Identifiable)

Base class for all Asset Administration Shells (AAS).

Parameters:

Name	Type	Description	Default
id	str	Global id of the object.	required
id_short	str	Local id of the object.	required
description	str	Description of the object. Defaults to None.	required

Source code in aas_middleware\model\formatting\aas\aas_model.py

class AAS(Identifiable):
    """
    Base class for all Asset Administration Shells (AAS).

    Args:
        id (str): Global id of the object.
        id_short (str): Local id of the object.
        description (str, optional): Description of the object. Defaults to None.
    """

    @model_validator(mode="before")
    @classmethod
    def set_optional_fields_to_None(cls, data):
        if isinstance(data, BaseModel):
            data = data.model_dump()
        for field_name, field_info in cls.model_fields.items():
            if field_name in data:
                continue
            if typing.get_origin(field_info.annotation) == Union and type(None) in typing.get_args(field_info.annotation):
                data[field_name] = None
        return data

    @model_validator(mode="after")
    def check_submodels(self) -> Self:
        for field_name, field_info in self.model_fields.items():
            if field_name in ["id", "id_short", "description"]:
                continue
            elif typing.get_origin(field_info.annotation) == Union and type(None) in typing.get_args(field_info.annotation) and getattr(self, field_name) is None:
                continue
            try:
                Submodel.model_validate(getattr(self, field_name))
            except ValidationError:
                assert (
                    False
                ), f"All attributes of an AAS must be of type Submodel or inherit from Submodel"
        return self

HasSemantics (BaseModel)

Base class for all classes that have semantics of the AAS meta model. Semantics are defined by a semantic id, which reference the semantic definition of the object.

Parameters:

Name	Type	Description	Default
semantic_id	str	Semantic id of the object. Defaults to None.	required

Source code in aas_middleware\model\formatting\aas\aas_model.py

class HasSemantics(BaseModel):
    """
    Base class for all classes that have semantics of the AAS meta model. Semantics are defined by a semantic id, which reference the semantic definition of the object.

    Args:
        semantic_id (str): Semantic id of the object. Defaults to None.
    """

    semantic_id: str = ""

Identifiable (Referable)

Base class for all identifiable classes in the AAS Meta model. An Identifiable is a Referable with a global id (id_).

Parameters:

Name	Type	Description	Default
id	str	Global id of the object.	required
id_short	str	Local id of the object.	required
description	str	Description of the object. Defaults to None.	required

Source code in aas_middleware\model\formatting\aas\aas_model.py

class Identifiable(Referable):
    """
    Base class for all identifiable classes in the AAS Meta model. An Identifiable is a Referable with a global id (id_).

    Args:
        id (str): Global id of the object.
        id_short (str): Local id of the object.
        description (str, optional): Description of the object. Defaults to None.
    """

    id: AasIdString

    @model_validator(mode="before")
    @classmethod
    def check_id_and_id_short(cls, data: Any) -> Any:
        if isinstance(data, BaseModel):
            data = data.model_dump()
        elif not isinstance(data, dict):
            data = {
                "id": getattr(data, "id", ""),
                "id_short": getattr(data, "id_short", ""),
            }
        assert "id" in data or "id_short" in data, "Either id or id_short must be set"
        if "id_short" not in data:
            data["id_short"] = data["id"]
        if "id" not in data:
            data["id"] = data["id_short"]
        return data

Referable (BaseModel)

Base class for all referable classes of the AAS meta model. A Referable is an object with a local id (id_short) and a description.

Parameters:

Name	Type	Description	Default
id_short	IdString	Local id of the object.	required
description	str	Description of the object. Defaults to None.	required

Source code in aas_middleware\model\formatting\aas\aas_model.py

class Referable(BaseModel):
    """
    Base class for all referable classes of the AAS meta model. A Referable is an object with a local id (id_short) and a description.

    Args:
        id_short (IdString): Local id of the object.
        description (str): Description of the object. Defaults to None.
    """

    id_short: AasIdString
    description: str = ""

Submodel (HasSemantics, Identifiable)

Base class for all submodels.

Parameters:

Name	Type	Description	Default
id	str	Global id of the object.	required
id_short	str	Local id of the object.	required
description	str	Description of the object. Defaults to None.	required
semantic_id	str	Semantic id of the object. Defaults to None.	required

Source code in aas_middleware\model\formatting\aas\aas_model.py

class Submodel(HasSemantics, Identifiable):
    """
    Base class for all submodels.

    Args:
        id (str): Global id of the object.
        id_short (str): Local id of the object.
        description (str, optional): Description of the object. Defaults to None.
        semantic_id (str, optional): Semantic id of the object. Defaults to None.
    """

    @model_validator(mode="after")
    def check_submodel_elements(self) -> Self:
        for field_name in self.model_fields:
            if field_name in ["id", "id_short", "description", "semantic_id"]:
                continue
            assert is_valid_submodel_element(
                getattr(self, field_name)
            ), f"All attributes of a Submodel must be valid SubmodelElements."
        return self

SubmodelElementCollection (HasSemantics, Referable)

Base class for all submodel element collections.

Parameters:

Name	Type	Description	Default
id_short	str	Local id of the object.	required
description	str	Description of the object. Defaults to None.	required
semantic_id	str	Semantic id of the object. Defaults to None.	required

Source code in aas_middleware\model\formatting\aas\aas_model.py

class SubmodelElementCollection(HasSemantics, Referable):
    """
    Base class for all submodel element collections.

    Args:
        id_short (str): Local id of the object.
        description (str, optional): Description of the object. Defaults to None.
        semantic_id (str, optional): Semantic id of the object. Defaults to None.
    """

    @model_validator(mode="after")
    def check_submodel_elements(self) -> Self:
        for field_name in self.model_fields:
            if field_name in ["id", "id_short", "description", "semantic_id"]:
                continue
            assert is_valid_submodel_element(
                getattr(self, field_name)
            ), f"All attributes of a SubmodelElementCollection must be valid SubmodelElements. Field {field_name} is not valid."
        return self

basyx_formatter

BasyxFormatter

Allows to serialize and deserialize Basyx AAS objects (AssetAdministrationShells, Submodels or Containers of both) to a DataModel.

Source code in aas_middleware\model\formatting\aas\basyx_formatter.py

class BasyxFormatter:
    """
    Allows to serialize and deserialize Basyx AAS objects (AssetAdministrationShells, Submodels or Containers of both) to a DataModel.
    """

    def serialize(self, data: DataModel) -> DictObjectStore[model.Identifiable]:
        """
        Serialize a DataModel object to the specific format of the formatter.

        Args:
            data (DataModel): A data model

        Returns:
            Objectstore: the basyx object store contain all AAS elements
        """
        aas_models, submodel_models = infere_aas_structure(data)
        obj_store = DictObjectStore()
        for aas in aas_models:
            obj_store_to_add = convert_model_to_aas(aas)
            for identifiable in obj_store_to_add:
                if obj_store.get(identifiable.id_short) is not None:
                    continue
                obj_store.add(identifiable)
        for submodel in submodel_models:
            submodel_to_add = convert_model_to_submodel(submodel)
            if obj_store.get(submodel_to_add.id_short) is not None:
                continue
            obj_store.add(submodel_to_add)
        return obj_store

    def deserialize(self, data: BasyxModels) -> DataModel:
        """
        Deserialize the specific format of the formater to a DataModel object.

        Args:
            data (Any): The specific format of the formatter.

        Returns:
            DataModel: A data model that holds the objects that were deserialized
        """
        if not isinstance(data, DictObjectStore):
            data = DictObjectStore(data)
        models = convert_object_store_to_pydantic_models(data)
        return DataModel.from_models(*models)

deserialize(self, data)

Deserialize the specific format of the formater to a DataModel object.

Parameters:

Name	Type	Description	Default
data	Any	The specific format of the formatter.	required

Returns:

Type	Description
DataModel	A data model that holds the objects that were deserialized

Source code in aas_middleware\model\formatting\aas\basyx_formatter.py

def deserialize(self, data: BasyxModels) -> DataModel:
    """
    Deserialize the specific format of the formater to a DataModel object.

    Args:
        data (Any): The specific format of the formatter.

    Returns:
        DataModel: A data model that holds the objects that were deserialized
    """
    if not isinstance(data, DictObjectStore):
        data = DictObjectStore(data)
    models = convert_object_store_to_pydantic_models(data)
    return DataModel.from_models(*models)

serialize(self, data)

Serialize a DataModel object to the specific format of the formatter.

Parameters:

Name	Type	Description	Default
data	DataModel	A data model	required

Returns:

Type	Description
Objectstore	the basyx object store contain all AAS elements

Source code in aas_middleware\model\formatting\aas\basyx_formatter.py

def serialize(self, data: DataModel) -> DictObjectStore[model.Identifiable]:
    """
    Serialize a DataModel object to the specific format of the formatter.

    Args:
        data (DataModel): A data model

    Returns:
        Objectstore: the basyx object store contain all AAS elements
    """
    aas_models, submodel_models = infere_aas_structure(data)
    obj_store = DictObjectStore()
    for aas in aas_models:
        obj_store_to_add = convert_model_to_aas(aas)
        for identifiable in obj_store_to_add:
            if obj_store.get(identifiable.id_short) is not None:
                continue
            obj_store.add(identifiable)
    for submodel in submodel_models:
        submodel_to_add = convert_model_to_submodel(submodel)
        if obj_store.get(submodel_to_add.id_short) is not None:
            continue
        obj_store.add(submodel_to_add)
    return obj_store

convert_aas

convert_aas_to_pydantic_model(aas, pydantic_submodels)

Converts an AAS to a Pydantic model.

Parameters:

Name	Type	Description	Default
aas	model.AssetAdministrationShell	AAS to convert	required

Returns:

Type	Description
aas_model.AAS	Pydantic model of the asset administration shell

Source code in aas_middleware\model\formatting\aas\convert_aas.py

def convert_aas_to_pydantic_model(
    aas: model.AssetAdministrationShell,
    pydantic_submodels: typing.List[aas_model.Submodel],
) -> aas_model.AAS:
    """
    Converts an AAS to a Pydantic model.

    Args:
        aas (model.AssetAdministrationShell): AAS to convert

    Returns:
        aas_model.AAS: Pydantic model of the asset administration shell
    """
    aas_class_name = convert_util.get_class_name_from_basyx_model(aas)
    dict_dynamic_model_creation = get_initial_dict_for_dynamic_model_creation(aas)
    dict_model_instantiation = get_initial_dict_for_model_instantiation(aas)
    aas_submodel_ids = [sm.get_identifier() for sm in aas.submodel]

    for sm in pydantic_submodels:
        if not sm.id in aas_submodel_ids:
            continue
        attribute_name_of_submodel = convert_util.get_attribute_name_from_basyx_model(
            aas, sm.id
        )
        dict_dynamic_model_creation.update(
            {
                attribute_name_of_submodel: typing.Annotated[
                    type(sm), Field(examples=[sm])
                ]
            }
        )
        dict_model_instantiation.update({attribute_name_of_submodel: sm.model_dump()})
    model_type = create_model(
        aas_class_name, **dict_dynamic_model_creation, __base__=aas_model.AAS
    )
    return model_type(**dict_model_instantiation)

convert_object_store_to_pydantic_models(obj_store)

Converts an object store with AAS and submodels to pydantic models, representing the original data structure.

Parameters:

Name	Type	Description	Default
obj_store	model.DictObjectStore	Object store with AAS and submodels	required

Returns:

Type	Description
List[aas_model.AAS]	List of pydantic models

Source code in aas_middleware\model\formatting\aas\convert_aas.py

def convert_object_store_to_pydantic_models(
    obj_store: model.DictObjectStore,
) -> typing.List[aas_model.AAS]:
    """
    Converts an object store with AAS and submodels to pydantic models, representing the original data structure.

    Args:
        obj_store (model.DictObjectStore): Object store with AAS and submodels

    Returns:
        typing.List[aas_model.AAS]: List of pydantic models
    """
    pydantic_submodels: typing.List[aas_model.Submodel] = []
    for identifiable in obj_store:
        if isinstance(identifiable, model.Submodel):
            pydantic_submodel = convert_submodel_to_model(identifiable)
            pydantic_submodels.append(pydantic_submodel)

    pydantic_aas_list: typing.List[aas_model.AAS] = []
    for identifiable in obj_store:
        if isinstance(identifiable, model.AssetAdministrationShell):
            pydantic_aas = convert_aas_to_pydantic_model(
                identifiable, pydantic_submodels
            )
            pydantic_aas_list.append(pydantic_aas)

    return pydantic_aas_list

convert_property_to_pydantic_model(sm_element)

Converts a Property to a Pydantic model.

Parameters:

Name	Type	Description	Default
sm_element	model.Property	Property to convert.	required

Returns:

Type	Description
aas_model.PrimitiveSubmodelElement	Value of the Property.

Source code in aas_middleware\model\formatting\aas\convert_aas.py

def convert_property_to_pydantic_model(
    sm_element: model.Property,
) -> aas_model.PrimitiveSubmodelElement:
    """
    Converts a Property to a Pydantic model.

    Args:
        sm_element (model.Property): Property to convert.

    Returns:
        aas_model.PrimitiveSubmodelElement: Value of the Property.
    """
    return sm_element.value

convert_reference_element_to_pydantic_model(sm_element)

Converts a ReferenceElement to a Pydantic model.

Parameters:

Name	Type	Description	Default
sm_element	model.ReferenceElement	ReferenceElement to convert.	required

Returns:

Type	Description
str	Value of the ReferenceElement.

Source code in aas_middleware\model\formatting\aas\convert_aas.py

def convert_reference_element_to_pydantic_model(
    sm_element: model.ReferenceElement,
) -> str:
    """
    Converts a ReferenceElement to a Pydantic model.

    Args:
        sm_element (model.ReferenceElement): ReferenceElement to convert.

    Returns:
        str: Value of the ReferenceElement.
    """
    return sm_element.value.key[0].value

convert_submodel_collection_to_pydantic_model(sm_element)

Converts a SubmodelElementCollection to a Pydantic model.

Parameters:

Name	Type	Description	Default
sm_element	model.SubmodelElementCollection	SubmodelElementCollection to convert.	required

Returns:

Type	Description
aas_model.SubmodelElementCollection	Pydantic model of the submodel element collection.

Source code in aas_middleware\model\formatting\aas\convert_aas.py

def convert_submodel_collection_to_pydantic_model(
    sm_element: model.SubmodelElementCollection,
) -> aas_model.SubmodelElementCollection:
    """
    Converts a SubmodelElementCollection to a Pydantic model.

    Args:
        sm_element (model.SubmodelElementCollection): SubmodelElementCollection to convert.

    Returns:
        aas_model.SubmodelElementCollection: Pydantic model of the submodel element collection.
    """
    class_name = convert_util.get_class_name_from_basyx_model(sm_element)
    dict_dynamic_model_creation = get_initial_dict_for_dynamic_model_creation(
        sm_element
    )
    dict_model_instantiation = get_initial_dict_for_model_instantiation(sm_element)

    for sub_sm_element in sm_element.value:
        attribute_name = convert_util.get_attribute_name_from_basyx_model(
            sm_element, sub_sm_element.id_short
        )
        immutable = is_attribute_from_basyx_model_immutable(
            sm_element, sub_sm_element.id_short
        )
        attribute_value = get_submodel_element_value(sub_sm_element, immutable)
        dict_sme = get_dynamic_model_creation_dict_from_submodel_element(
            attribute_name, attribute_value
        )
        dict_dynamic_model_creation.update(dict_sme)
        dict_sme_instantiation = get_model_instantiation_dict_from_submodel_element(
            attribute_name, attribute_value
        )
        dict_model_instantiation.update(dict_sme_instantiation)
    model_type = create_model(
        class_name,
        **dict_dynamic_model_creation,
        __base__=aas_model.SubmodelElementCollection,
    )
    return model_type(**dict_model_instantiation)

convert_submodel_list_to_pydantic_model(sm_element, immutable=False)

Converts a SubmodelElementList to a Pydantic model.

Parameters:

Name	Type	Description	Default
sm_element	model.SubmodelElementList	SubmodelElementList to convert.	required

Returns:

Type	Description
List[aas_model.SubmodelElement]	List of Pydantic models of the submodel elements.

Source code in aas_middleware\model\formatting\aas\convert_aas.py

def convert_submodel_list_to_pydantic_model(
    sm_element: model.SubmodelElementList, immutable: bool = False
) -> typing.Union[typing.List[aas_model.SubmodelElement], typing.Set[aas_model.SubmodelElement], typing.Tuple[aas_model.SubmodelElement]]:
    """
    Converts a SubmodelElementList to a Pydantic model.

    Args:
        sm_element (model.SubmodelElementList): SubmodelElementList to convert.

    Returns:
        typing.List[aas_model.SubmodelElement]: List of Pydantic models of the submodel elements.
    """
    sme_pydantic_models = []
    for sme in sm_element.value:
        if isinstance(sme, model.SubmodelElementCollection):
            sme = unpatch_id_short_from_temp_attribute(sme)
            sme_pydantic_models.append(
                convert_submodel_collection_to_pydantic_model(sme)
            )
        elif isinstance(sme, model.SubmodelElementList):
            sme_pydantic_models.append(convert_submodel_list_to_pydantic_model(sme))
        elif isinstance(sme, model.ReferenceElement):
            sme_pydantic_models.append(convert_reference_element_to_pydantic_model(sme))
        elif isinstance(sme, model.Property):
            sme_pydantic_models.append(sme.value)
        else:
            raise NotImplementedError("Type not implemented:", type(sme))
    if not sm_element.order_relevant:
        return set(sme_pydantic_models)
    if immutable:
        return tuple(sme_pydantic_models)
    return sme_pydantic_models

convert_submodel_to_model(sm)

Converts a Submodel to a Pydantic model.

Parameters:

Name	Type	Description	Default
sm	model.Submodel	Submodel to convert.	required

Returns:

Type	Description
aas_model.Submodel	Pydantic model of the submodel.

Source code in aas_middleware\model\formatting\aas\convert_aas.py

def convert_submodel_to_model(sm: model.Submodel) -> aas_model.Submodel:
    """
    Converts a Submodel to a Pydantic model.

    Args:
        sm (model.Submodel): Submodel to convert.

    Returns:
        aas_model.Submodel: Pydantic model of the submodel.
    """
    class_name = convert_util.get_class_name_from_basyx_model(sm)
    dict_dynamic_model_creation = get_initial_dict_for_dynamic_model_creation(sm)
    dict_model_instantiation = get_initial_dict_for_model_instantiation(sm)

    for sm_element in sm.submodel_element:
        attribute_name = convert_util.get_attribute_name_from_basyx_model(
            sm, sm_element.id_short
        )
        immutable = is_attribute_from_basyx_model_immutable(sm, sm_element.id_short)
        attribute_value = get_submodel_element_value(sm_element, immutable)
        sme_model_creation_dict = get_dynamic_model_creation_dict_from_submodel_element(
            attribute_name, attribute_value
        )
        dict_dynamic_model_creation.update(sme_model_creation_dict)
        sme_model_instantiation_dict = (
            get_model_instantiation_dict_from_submodel_element(
                attribute_name, attribute_value
            )
        )
        dict_model_instantiation.update(sme_model_instantiation_dict)
    model_type = create_model(
        class_name, **dict_dynamic_model_creation, __base__=aas_model.Submodel
    )
    return model_type(**dict_model_instantiation)

get_dynamic_model_creation_dict_from_submodel_element(attribute_name, attribute_value)

Converts a SubmodelElement to a dict.

Parameters:

Name	Type	Description	Default
attribute_name	str	Name of the attribute to create in the dictionary.	required
sm_element	model.SubmodelElement	SubmodelElement to convert.	required

Returns:

Type	Description
dict	Dictionary that can be used to create a Pydantic model, with Annoated types for the attributes and examples.

Source code in aas_middleware\model\formatting\aas\convert_aas.py

def get_dynamic_model_creation_dict_from_submodel_element(
    attribute_name: str, attribute_value: typing.Any
) -> typing.Dict[str, typing.Any]:
    """
    Converts a SubmodelElement to a dict.

    Args:
        attribute_name (str): Name of the attribute to create in the dictionary.
        sm_element (model.SubmodelElement): SubmodelElement to convert.

    Returns:
        dict: Dictionary that can be used to create a Pydantic model, with Annoated types for the attributes and examples.
    """
    if isinstance(attribute_value, list) and attribute_value:
        inner_type = type(attribute_value[0])
        attribute_type = typing.List[inner_type]
    elif isinstance(attribute_value, set) and attribute_value:
        inner_type = type(next(iter(attribute_value)))
        attribute_type = typing.Set[inner_type]
    elif isinstance(attribute_value, tuple) and attribute_value:
        inner_type = type(attribute_value[0])
        attribute_type = typing.Tuple[inner_type, ...]
    else:
        attribute_type = type(attribute_value)
    return {
        attribute_name: typing.Annotated[
            attribute_type, Field(examples=[attribute_value])
        ]
    }

get_initial_dict_for_dynamic_model_creation(basyx_model)

Returns a dictionary that can be used to create a Pydantic model based on a provided basyx submodel.

Parameters:

Name	Type	Description	Default
basyx_model	model.Submodel | model.AssetAdministrationShell | model.SubmodelElementCollection	Basyx model to create the dictionary from.	required

Returns:

Type	Description
Dict[str, Any]	Dictionary that can be used to create a Pydantic model.

Source code in aas_middleware\model\formatting\aas\convert_aas.py

def get_initial_dict_for_dynamic_model_creation(
    basyx_model: (
        model.Submodel
        | model.AssetAdministrationShell
        | model.SubmodelElementCollection
    ),
) -> typing.Dict[str, typing.Any]:
    """
    Returns a dictionary that can be used to create a Pydantic model based on a provided basyx submodel.

    Args:
        basyx_model (model.Submodel | model.AssetAdministrationShell | model.SubmodelElementCollection): Basyx model to create the dictionary from.

    Returns:
        typing.Dict[str, typing.Any]: Dictionary that can be used to create a Pydantic model.
    """
    model_creation_dict = {
        "id_short": typing.Annotated[str, Field(examples=[basyx_model.id_short])],
        "description": typing.Annotated[
            str,
            Field(examples=[convert_util.get_str_description(basyx_model.description)]),
        ],
    }
    if isinstance(basyx_model, model.Identifiable):
        model_creation_dict["id"] = typing.Annotated[
            str, Field(examples=[str(basyx_model.id)])
        ]
    if isinstance(basyx_model, model.HasSemantics):
        model_creation_dict["semantic_id"] = typing.Annotated[
            str, Field(examples=[get_semantic_id_value_of_model(basyx_model)])
        ]
    return model_creation_dict

get_initial_dict_for_model_instantiation(basyx_model)

Returns a dictionary that can be used to instantiate a Pydantic model based on a provided basyx submodel.

Parameters:

Name	Type	Description	Default
basyx_model	model.Submodel | model.AssetAdministrationShell | model.SubmodelElementCollection	Basyx model to create the dictionary from.	required

Returns:

Type	Description
Dict[str, Any]	Dictionary that can be used to instantiate a Pydantic model.

Source code in aas_middleware\model\formatting\aas\convert_aas.py

def get_initial_dict_for_model_instantiation(
    basyx_model: (
        model.Submodel
        | model.AssetAdministrationShell
        | model.SubmodelElementCollection
    ),
) -> typing.Dict[str, typing.Any]:
    """
    Returns a dictionary that can be used to instantiate a Pydantic model based on a provided basyx submodel.

    Args:
        basyx_model (model.Submodel | model.AssetAdministrationShell | model.SubmodelElementCollection): Basyx model to create the dictionary from.

    Returns:
        typing.Dict[str, typing.Any]: Dictionary that can be used to instantiate a Pydantic model.
    """
    model_instantiation_dict = {
        "id_short": basyx_model.id_short,
        "description": convert_util.get_str_description(basyx_model.description),
    }
    if isinstance(basyx_model, model.Identifiable):
        model_instantiation_dict["id"] = str(basyx_model.id)
    if isinstance(basyx_model, model.HasSemantics):
        model_instantiation_dict["semantic_id"] = get_semantic_id_value_of_model(
            basyx_model
        )
    return model_instantiation_dict

get_model_instantiation_dict_from_submodel_element(attribute_name, attribute_value)

Converts a SubmodelElement to a dict.

Parameters:

Name	Type	Description	Default
attribute_name	str	Name of the attribute to create in the dictionary.	required
sm_element	model.SubmodelElement	SubmodelElement to convert.	required

Returns:

Type	Description
dict	Dictionary that can be used to instantiate a Pydantic model.

Source code in aas_middleware\model\formatting\aas\convert_aas.py

def get_model_instantiation_dict_from_submodel_element(
    attribute_name: str, attribute_value: typing.Any
) -> typing.Dict[str, typing.Any]:
    """
    Converts a SubmodelElement to a dict.

    Args:
        attribute_name (str): Name of the attribute to create in the dictionary.
        sm_element (model.SubmodelElement): SubmodelElement to convert.

    Returns:
        dict: Dictionary that can be used to instantiate a Pydantic model.
    """
    if isinstance(attribute_value, BaseModel):
        attribute_value = attribute_value.model_dump()
    elif isinstance(attribute_value, (list, set, tuple)) and any(
        isinstance(element, BaseModel) for element in attribute_value
    ):
        attribute_value = [element.model_dump() for element in attribute_value]
    return {attribute_name: attribute_value}

get_submodel_element_value(sm_element, immutable=False)

Returns the value of a SubmodelElement.

Parameters:

Name	Type	Description	Default
sm_element	model.SubmodelElement	SubmodelElement to get the value from.	required

Returns:

Type	Description
aas_model.SubmodelElement	Value of the SubmodelElement.

Source code in aas_middleware\model\formatting\aas\convert_aas.py

def get_submodel_element_value(
    sm_element: model.SubmodelElement, immutable: bool = False
) -> aas_model.SubmodelElement:
    """
    Returns the value of a SubmodelElement.

    Args:
        sm_element (model.SubmodelElement): SubmodelElement to get the value from.

    Returns:
        aas_model.SubmodelElement: Value of the SubmodelElement.
    """
    if isinstance(sm_element, model.SubmodelElementCollection):
        return convert_submodel_collection_to_pydantic_model(sm_element)
    elif isinstance(sm_element, model.SubmodelElementList):
        return convert_submodel_list_to_pydantic_model(sm_element, immutable)
    elif isinstance(sm_element, model.ReferenceElement):
        return convert_reference_element_to_pydantic_model(sm_element)
    elif isinstance(sm_element, model.Property):
        return convert_property_to_pydantic_model(sm_element)
    else:
        raise NotImplementedError("Type not implemented:", type(sm_element))

unpatch_id_short_from_temp_attribute(smec)

Unpatches the id_short attribute of a SubmodelElementCollection from the temporary attribute.

Parameters:

Name	Type	Description	Default
sm_element	model.SubmodelElementCollection	SubmodelElementCollection to unpatch.	required

Source code in aas_middleware\model\formatting\aas\convert_aas.py

def unpatch_id_short_from_temp_attribute(smec: model.SubmodelElementCollection):
    """
    Unpatches the id_short attribute of a SubmodelElementCollection from the temporary attribute.

    Args:
        sm_element (model.SubmodelElementCollection): SubmodelElementCollection to unpatch.
    """
    if not smec.id_short.startswith("generated_submodel_list_hack_"):
        return smec
    if not any(isinstance(sm_element, model.Property) and sm_element.id_short.startswith("temp_id_short_attribute") for sm_element in smec.value):
        raise ValueError("No temporary id_short attribute found in SubmodelElementCollection.")
    no_temp_values = []
    id_short = None
    for sm_element in smec.value:
        if isinstance(sm_element, model.Property) and sm_element.id_short.startswith("temp_id_short_attribute"):
            id_short = sm_element.value
            continue
        no_temp_values.append(sm_element)

    for value in no_temp_values:
        smec.value.remove(value)
    new_smec = model.SubmodelElementCollection(
        id_short=id_short, value=no_temp_values,
        embedded_data_specifications=smec.embedded_data_specifications,
    )
    # new_smec.value.remove(contained_sm_element)
    return new_smec

convert_pydantic

convert_model_to_aas(model_aas)

Convert a model aas to an Basyx AssetAdministrationShell and return it as a DictObjectStore with all Submodels

Parameters:

Name	Type	Description	Default
model_aas	aas_model.AAS	model aas to convert	required

Returns:

Type	Description
model.DictObjectStore[model.Identifiable]	DictObjectStore with all Submodels

Source code in aas_middleware\model\formatting\aas\convert_pydantic.py

def convert_model_to_aas(
    model_aas: aas_model.AAS,
) -> model.DictObjectStore[model.Identifiable]:
    """
    Convert a model aas to an Basyx AssetAdministrationShell and return it as a DictObjectStore with all Submodels

    Args:
        model_aas (aas_model.AAS): model aas to convert

    Returns:
        model.DictObjectStore[model.Identifiable]: DictObjectStore with all Submodels
    """
    aas_attributes = get_attribute_dict(model_aas)
    aas_submodels = []  # placeholder for submodels created
    aas_submodel_data_specifications = []
    for attribute_name, attribute_value in aas_attributes.items():
        if isinstance(attribute_value, aas_model.Submodel):
            tempsubmodel = convert_model_to_submodel(model_submodel=attribute_value)
            aas_submodels.append(tempsubmodel)
            attribute_data_specifications = convert_util.get_data_specification_for_attribute(
                    attribute_name, attribute_value.id, attribute_value

            )
            aas_submodel_data_specifications += attribute_data_specifications

    asset_information = model.AssetInformation(
        global_asset_id=model.Identifier(model_aas.id),
    )

    basyx_aas = model.AssetAdministrationShell(
        asset_information=asset_information,
        id_short=get_id_short(model_aas),
        id_=model.Identifier(model_aas.id),
        description=convert_util.get_basyx_description_from_model(model_aas),
        submodel={
            model.ModelReference.from_referable(submodel) for submodel in aas_submodels
        },
        embedded_data_specifications=convert_util.get_data_specification_for_model(model_aas) + aas_submodel_data_specifications,
    )
    obj_store: model.DictObjectStore[model.Identifiable] = model.DictObjectStore()
    obj_store.add(basyx_aas)
    for sm in aas_submodels:
        obj_store.add(sm)
    return obj_store

create_submodel_element(attribute_name, attribute_value)

Create a basyx SubmodelElement from a model SubmodelElementCollection or a primitive type

Parameters:

Name	Type	Description	Default
attribute_name	str	Name of the attribute that is used for ID and id_short	required
attribute_value	Union[ aas_model.SubmodelElementCollection, str, float, int, bool, tuple, list, set ]	Value of the attribute	required

Returns:

Type	Description
model.SubmodelElement	basyx SubmodelElement

Source code in aas_middleware\model\formatting\aas\convert_pydantic.py

def create_submodel_element(
    attribute_name: str,
    attribute_value: Union[
        aas_model.SubmodelElementCollection, str, float, int, bool, tuple, list, set
    ],
) -> model.SubmodelElement:
    """
    Create a basyx SubmodelElement from a model SubmodelElementCollection or a primitive type

    Args:
        attribute_name (str): Name of the attribute that is used for ID and id_short
        attribute_value (Union[ aas_model.SubmodelElementCollection, str, float, int, bool, tuple, list, set ]): Value of the attribute


    Returns:
        model.SubmodelElement: basyx SubmodelElement
    """
    if isinstance(attribute_value, aas_model.SubmodelElementCollection):
        smc = create_submodel_element_collection(attribute_value)
        return smc
    elif isinstance(attribute_value, list):
        sml = create_submodel_element_list(attribute_name, attribute_value)
        return sml
    elif isinstance(attribute_value, tuple):
        attribute_value_as_list = list(attribute_value)
        sml = create_submodel_element_list(attribute_name, attribute_value_as_list)
        return sml
    elif isinstance(attribute_value, set):
        attribute_value_as_list = list(attribute_value)	
        sml = create_submodel_element_list(
            attribute_name, attribute_value_as_list, ordered=False
        )
        return sml
    elif (isinstance(attribute_value, str)) and (
        (
            parse.urlparse(attribute_value).scheme
            and parse.urlparse(attribute_value).netloc
        )
        or (attribute_value.split("_")[-1] in ["id", "ids"])
    ):
        key = model.Key(
            type_=model.KeyTypes.ASSET_ADMINISTRATION_SHELL,
            value=attribute_value,
        )
        reference = model.ModelReference(key=(key,), type_="")
        reference_element = model.ReferenceElement(
            id_short=attribute_name,
            value=reference,
        )
        return reference_element
    else:
        property = create_property(attribute_name, attribute_value)

        return property

infere_aas_structure(data)

The function assert that the data contained in the data model fulfills the aas meta model structure.

Parameters:

Name	Type	Description	Default
data	DataModel	The Data Model containing the objects that should be transformed to AAS models	required

Returns:

Type	Description
Tuple[List[aas_model.AAS], List[aas_model.Submodel]]	Tuple with AAS models and Submodel models

Source code in aas_middleware\model\formatting\aas\convert_pydantic.py

def infere_aas_structure(
    data: DataModel,
) -> Tuple[List[aas_model.AAS], List[aas_model.Submodel]]:
    """
    The function assert that the data contained in the data model fulfills the aas meta model structure.

    Args:
        data (DataModel): The Data Model containing the objects that should be transformed to AAS models

    Returns:
        Tuple[List[aas_model.AAS], List[aas_model.Submodel]]: Tuple with AAS models and Submodel models
    """
    if all(isinstance(model, aas_model.AAS) for model in data.get_top_level_models()):
        return data.get_top_level_models(), []
    logger.warning(
        "The data model does not contain only AAS models. Trying to infer the AAS structure by rebuilding the data model."
    )
    new_data_model = DataModelRebuilder(data).rebuild_data_model_for_AAS_structure()
    top_level_models_list = []
    for models in new_data_model.get_top_level_models().values():
        top_level_models_list += models
    aas_models = [
        model for model in top_level_models_list if isinstance(model, aas_model.AAS)
    ]
    submodel_models = [
        model
        for model in top_level_models_list
        if isinstance(model, aas_model.Submodel)
    ]
    return aas_models, submodel_models

patch_id_short_with_temp_attribute(submodel_element_collection)

Patch the id_short of a SubmodelElementCollection as an attribute in the value of the SubmodelElementCollection, to make it accesible after retrieving from the value list.

Parameters:

Name	Type	Description	Default
submodel_element_collection	model.SubmodelElementCollection	SubmodelElementCollection to patch	required

Source code in aas_middleware\model\formatting\aas\convert_pydantic.py

def patch_id_short_with_temp_attribute(
        submodel_element_collection: model.SubmodelElementCollection
    ) -> None:
    """
    Patch the id_short of a SubmodelElementCollection as an attribute in the value of the SubmodelElementCollection, to make it accesible after retrieving from the value list.

    Args:
        submodel_element_collection (model.SubmodelElementCollection): SubmodelElementCollection to patch
    """
    temp_id_short_property = model.Property(
        id_short="temp_id_short_attribute_" + uuid.uuid4().hex,
        value_type=get_value_type_of_attribute(str),
        value=submodel_element_collection.id_short,
    )
    submodel_element_collection.value.add(temp_id_short_property)

convert_util

get_attribute_dict(obj)

Returns a dictionary of all attributes of an object that are not None, do not start with an underscore and are not standard attributes of the aas object.

Parameters:

Name	Type	Description	Default
obj	Union[aas_model.AAS, aas_model.Submodel, aas_model.SubmodelElementCollection]	Object to get the attributes from	required

Returns:

Type	Description
Dict[str, Union[aas_model.Submodel, aas_model.SubmodelElement]]	Dictionary of all attributes of the object and their respective values

Source code in aas_middleware\model\formatting\aas\convert_util.py

def get_attribute_dict(
    obj: Union[aas_model.AAS, aas_model.Submodel, aas_model.SubmodelElementCollection]
) -> Dict[str, Union[aas_model.Submodel, aas_model.SubmodelElement]]:
    """
    Returns a dictionary of all attributes of an object that are not None, do not start with an underscore and are not standard attributes of the aas object.

    Args:
        obj (Union[aas_model.AAS, aas_model.Submodel, aas_model.SubmodelElementCollection]): Object to get the attributes from

    Returns:
        Dict[str, Union[aas_model.Submodel, aas_model.SubmodelElement]]: Dictionary of all attributes of the object and their respective values
    """
    vars_dict = vars(obj)
    vars_dict = {key: value for key, value in vars_dict.items() if key[0] != "_"}
    vars_dict = {key: value for key, value in vars_dict.items() if value is not None}
    vars_dict = {key: value for key, value in vars_dict.items() if key != "id"}
    vars_dict = {key: value for key, value in vars_dict.items() if key != "description"}
    vars_dict = {key: value for key, value in vars_dict.items() if key != "id_short"}
    vars_dict = {key: value for key, value in vars_dict.items() if key != "semantic_id"}
    return vars_dict

get_attribute_name_from_basyx_model(item, referenced_item_id)

Returns the attribute name of the referenced element of the item.

Parameters:

Name	Type	Description	Default
item	Union[model.AssetAdministrationShell, model.Submodel, model.SubmodelElementCollection]	The container of the refernced item	required
referenced_item_id	str	The id of the referenced item	required

Exceptions:

Type	Description
ValueError	If not data specifications are found in the item or if no attribute name is found

Returns:

Type	Description
str	The attribute name of the referenced item

Source code in aas_middleware\model\formatting\aas\convert_util.py

def get_attribute_name_from_basyx_model(
    item: typing.Union[
        model.AssetAdministrationShell, model.Submodel, model.SubmodelElementCollection
    ],
    referenced_item_id: str,
) -> str:
    """
    Returns the attribute name of the referenced element of the item.

    Args:
        item (typing.Union[model.AssetAdministrationShell, model.Submodel, model.SubmodelElementCollection]): The container of the refernced item
        referenced_item_id (str): The id of the referenced item

    Raises:
        ValueError: If not data specifications are found in the item or if no attribute name is found

    Returns:
        str: The attribute name of the referenced item
    """
    if not item.embedded_data_specifications:
        raise ValueError("No data specifications found in item:", item)
    for data_spec in item.embedded_data_specifications:
        content = data_spec.data_specification_content
        if not isinstance(content, model.DataSpecificationIEC61360):
            continue
        if not any(
            key.value == referenced_item_id for key in data_spec.data_specification.key
        ):
            continue
        if not content.preferred_name.get("en") == "attribute":
            continue
        return content.value
    raise ValueError(
        f"Attribute reference to {referenced_item_id} could not be found in {item.id_short} of type {type(item)}"
    )

get_basyx_description_from_model(model_object)

Creates a LangStringSet from an aas model.

Parameters:

Name	Type	Description	Default
model_object	aas_model.AAS | aas_model.Submodel | aas_model.SubmodelElementCollection	The model to get the description from.	required

Returns:

Type	Description
model.LangStringSet	LangStringSet description representation of the model object

Exceptions:

Type	Description
ValueError	If the description of the model object is not a dict or a string

Source code in aas_middleware\model\formatting\aas\convert_util.py

def get_basyx_description_from_model(
    model_object: (
        aas_model.AAS | aas_model.Submodel | aas_model.SubmodelElementCollection
    ),
) -> model.LangStringSet:
    """
    Creates a LangStringSet from an aas model.
    Args:
        model_object (aas_model.AAS | aas_model.Submodel | aas_model.SubmodelElementCollection): The model to get the description from.
    Returns:
        model.LangStringSet: LangStringSet description representation of the model object
    Raises:
        ValueError: If the description of the model object is not a dict or a string
    """
    if not model_object.description:
        return None
    try:
        dict_description = json.loads(model_object.description)
        if not isinstance(dict_description, dict):
            raise ValueError
    except ValueError:
        dict_description = {"en": model_object.description}
    return model.LangStringSet(dict_description)

get_class_name_from_basyx_model(item)

Returns the class name of an basyx model from the data specifications.

Parameters:

Name	Type	Description	Default
item	model.HasDataSpecification	Basyx model to get the class name from	required

Exceptions:

Type	Description
ValueError	If no data specifications are found in the item or if no class name is found

Returns:

Type	Description
str	Class name of the basyx model

Source code in aas_middleware\model\formatting\aas\convert_util.py

def get_class_name_from_basyx_model(
    item: typing.Union[
        model.AssetAdministrationShell, model.Submodel, model.SubmodelElementCollection
    ]
) -> str:
    """
    Returns the class name of an basyx model from the data specifications.

    Args:
        item (model.HasDataSpecification): Basyx model to get the class name from

    Raises:
        ValueError: If no data specifications are found in the item or if no class name is found

    Returns:
        str: Class name of the basyx model
    """
    if not item.embedded_data_specifications:
        raise ValueError("No data specifications found in item:", item)
    for data_spec in item.embedded_data_specifications:
        content = data_spec.data_specification_content
        if not isinstance(content, model.DataSpecificationIEC61360):
            continue
        if not any(
            key.value == item.id_short for key in data_spec.data_specification.key
        ):
            continue
        if not content.preferred_name.get("en") == "class":
            continue
        return content.value
    raise ValueError(
        f"No class name found in item with id {item.id_short} and type {type(item)}"
    )

get_semantic_id_value_of_model(basyx_model)

Returns the semantic id of a submodel or submodel element.

Parameters:

Name	Type	Description	Default
basyx_model	model.Submodel | model.SubmodelElement	Basyx model to get the semantic id from.	required

Returns:

Type	Description
str	Semantic id of the model.

Source code in aas_middleware\model\formatting\aas\convert_util.py

def get_semantic_id_value_of_model(
    basyx_model: typing.Union[model.Submodel, model.SubmodelElement]
) -> str:
    """
    Returns the semantic id of a submodel or submodel element.

    Args:
        basyx_model (model.Submodel | model.SubmodelElement): Basyx model to get the semantic id from.

    Returns:
        str: Semantic id of the model.
    """
    if not isinstance(basyx_model, model.HasSemantics):
        raise NotImplementedError("Type not implemented:", type(basyx_model))
    if not basyx_model.semantic_id:
        return ""
    return basyx_model.semantic_id.key[0].value

get_str_description(langstring_set)

Converts a LangStringSet to a string.

Parameters:

Name	Type	Description	Default
langstring_set	model.LangStringSet	LangStringSet to convert	required

Returns:

Type	Description
str	String representation of the LangStringSet

Source code in aas_middleware\model\formatting\aas\convert_util.py

def get_str_description(langstring_set: model.LangStringSet) -> str:
    """
    Converts a LangStringSet to a string.
    Args:
        langstring_set (model.LangStringSet): LangStringSet to convert
    Returns:
        str: String representation of the LangStringSet
    """
    if not langstring_set:
        return ""
    dict_description = {}
    if "en" in langstring_set:
        return str(langstring_set.get("en"))
    elif "ger" in langstring_set:
        return str(langstring_set.get("ger"))
    elif "de" in langstring_set:
        return str(langstring_set.get("de"))
    else:
        return str(langstring_set.get(list(langstring_set.keys())[0]))

is_attribute_from_basyx_model_immutable(item, referenced_item_id)

Returns if the referenced item of the item is immutable.

Parameters:

Name	Type	Description	Default
item	Union[model.AssetAdministrationShell, model.Submodel, model.SubmodelElementCollection]	The container of the refernced item	required
referenced_item_id	str	The id of the referenced item	required

Exceptions:

Type	Description
ValueError	If not data specifications are found in the item or if no attribute name is found

Returns:

Type	Description
bool	If the referenced item is immutable

Source code in aas_middleware\model\formatting\aas\convert_util.py

def is_attribute_from_basyx_model_immutable(
    item: typing.Union[
        model.AssetAdministrationShell, model.Submodel, model.SubmodelElementCollection
    ],
    referenced_item_id: str,
) -> bool:
    """
    Returns if the referenced item of the item is immutable.

    Args:
        item (typing.Union[model.AssetAdministrationShell, model.Submodel, model.SubmodelElementCollection]): The container of the refernced item
        referenced_item_id (str): The id of the referenced item

    Raises:
        ValueError: If not data specifications are found in the item or if no attribute name is found

    Returns:
        bool: If the referenced item is immutable
    """
    if not item.embedded_data_specifications:
        raise ValueError("No data specifications found in item:", item)
    for data_spec in item.embedded_data_specifications:
        content = data_spec.data_specification_content
        if not isinstance(content, model.DataSpecificationIEC61360):
            continue
        if not any(
            key.value == referenced_item_id for key in data_spec.data_specification.key
        ):
            continue
        if not content.preferred_name.get("en") == "immutable":
            continue
        return content.value == "true"
    raise ValueError(
        f"Attribute reference to {referenced_item_id} could not be found in {item.id_short} of type {type(item)}"
    )

formatter

Formatter (Protocol)

Protocol for all formatters that are used to serialize and deserialize data models.

Source code in aas_middleware\model\formatting\formatter.py

class Formatter(Protocol):
    """
    Protocol for all formatters that are used to serialize and deserialize data models.
    """

    def serialize(self, data: DataModel) -> Any:
        """
        Serialize a DataModel object to the specific format of the formatter.

        Args:
            data (DataModel): A data model

        Returns:
            Any: A string in the specific format of the formatter.
        """
        ...

    def deserialize(self, data: Any) -> DataModel:
        """
        Deserialize the specific format of the formater to a DataModel object.

        Args:
            data (Any): The specific format of the formatter.

        Returns:
            DataModel: A data model that holds the objects that were deserialized
        """
        ...

deserialize(self, data)

Deserialize the specific format of the formater to a DataModel object.

Parameters:

Name	Type	Description	Default
data	Any	The specific format of the formatter.	required

Returns:

Type	Description
DataModel	A data model that holds the objects that were deserialized

Source code in aas_middleware\model\formatting\formatter.py

def deserialize(self, data: Any) -> DataModel:
    """
    Deserialize the specific format of the formater to a DataModel object.

    Args:
        data (Any): The specific format of the formatter.

    Returns:
        DataModel: A data model that holds the objects that were deserialized
    """
    ...

serialize(self, data)

Serialize a DataModel object to the specific format of the formatter.

Parameters:

Name	Type	Description	Default
data	DataModel	A data model	required

Returns:

Type	Description
Any	A string in the specific format of the formatter.

Source code in aas_middleware\model\formatting\formatter.py

def serialize(self, data: DataModel) -> Any:
    """
    Serialize a DataModel object to the specific format of the formatter.

    Args:
        data (DataModel): A data model

    Returns:
        Any: A string in the specific format of the formatter.
    """
    ...

mapping special

mapper

Mapper (Protocol)

Protocol for all mappers that are used to map data models to other data models.

Source code in aas_middleware\model\mapping\mapper.py

@typing.runtime_checkable
class Mapper(Protocol, Generic[S, T]):
    """
    Protocol for all mappers that are used to map data models to other data models.
    """

    def map(self, data: S) -> T:
        """
        Map a DataModel object to another DataModel object with different types and structure.

        Args:
            data (DataModel): A data model

        Returns:
            DataModel: The mapped data model.
        """
        ...

map(self, data)

Map a DataModel object to another DataModel object with different types and structure.

Parameters:

Name	Type	Description	Default
data	DataModel	A data model	required

Returns:

Type	Description
DataModel	The mapped data model.

Source code in aas_middleware\model\mapping\mapper.py

def map(self, data: S) -> T:
    """
    Map a DataModel object to another DataModel object with different types and structure.

    Args:
        data (DataModel): A data model

    Returns:
        DataModel: The mapped data model.
    """
    ...

reference_finder

ReferenceFinder

Source code in aas_middleware\model\reference_finder.py

class ReferenceFinder:
    model: Identifiable
    contained_models: List[Identifiable] = []
    references: List[ReferenceInfo] = []

    contained_schemas: List[Type[Identifiable]] = []
    schema_references: List[ReferenceInfo] = []

    @classmethod
    def find(
        cls, model: Identifiable
    ) -> Tuple[List[Identifiable], Set[ReferenceInfo]]:
        """
        Method to find all contained models (inclusive the model itself) and references in a given model.

        Args:
            model (Identifiable): The model to find contained models and references in.

        Returns:
            Tuple[List[Identifiable], List[ReferenceInfo]]: A tuple containing the list of contained models and the list of references.
        """
        finder = cls()
        finder.model = model
        finder.find_contained_identifiables_and_references()
        return finder.contained_models, finder.references

    def find_contained_identifiables_and_references(self):
        """
        Method to find all contained identifiables (inclusive the model itself) and references in the model.
        """
        self.contained_models = get_all_contained_identifiables(self.model)
        self.references = get_reference_infos(self.contained_models)

    @classmethod
    def find_schema_references(
            cls,
            model: Identifiable,
    ) -> Tuple[List[Identifiable], Set[ReferenceInfo]]:
        """
        Method to find all contained models (inclusive the model itself) and references in a given model.

        Args:
            model (Identifiable): The model to find contained models and references in.

        Returns:
            Tuple[List[Identifiable], List[ReferenceInfo]]: A tuple containing the list of contained models and the list of references.
        """
        finder = cls()
        finder.model = model
        finder.find_contained_schemas_and_references()
        return finder.contained_schemas, finder.schema_references

    def find_contained_schemas_and_references(self):
        """
        Method to find all contained identifiables (inclusive the model itself) and references in the model.
        """
        self.contained_schemas = get_all_contained_schemas(self.model)
        self.schema_references = get_schema_reference_infos(self.contained_schemas)
        # FIXME: resolve empty referenced nodes without an associated type -> either find subclasses or classes that contain the referenced name (e.g. "acticePoleHousing" for class "PoleHousing")

find(model) classmethod

Method to find all contained models (inclusive the model itself) and references in a given model.

Parameters:

Name	Type	Description	Default
model	Identifiable	The model to find contained models and references in.	required

Returns:

Type	Description
Tuple[List[Identifiable], List[ReferenceInfo]]	A tuple containing the list of contained models and the list of references.

Source code in aas_middleware\model\reference_finder.py

@classmethod
def find(
    cls, model: Identifiable
) -> Tuple[List[Identifiable], Set[ReferenceInfo]]:
    """
    Method to find all contained models (inclusive the model itself) and references in a given model.

    Args:
        model (Identifiable): The model to find contained models and references in.

    Returns:
        Tuple[List[Identifiable], List[ReferenceInfo]]: A tuple containing the list of contained models and the list of references.
    """
    finder = cls()
    finder.model = model
    finder.find_contained_identifiables_and_references()
    return finder.contained_models, finder.references

find_contained_identifiables_and_references(self)

Method to find all contained identifiables (inclusive the model itself) and references in the model.

Source code in aas_middleware\model\reference_finder.py

def find_contained_identifiables_and_references(self):
    """
    Method to find all contained identifiables (inclusive the model itself) and references in the model.
    """
    self.contained_models = get_all_contained_identifiables(self.model)
    self.references = get_reference_infos(self.contained_models)

find_contained_schemas_and_references(self)

Method to find all contained identifiables (inclusive the model itself) and references in the model.

Source code in aas_middleware\model\reference_finder.py

def find_contained_schemas_and_references(self):
    """
    Method to find all contained identifiables (inclusive the model itself) and references in the model.
    """
    self.contained_schemas = get_all_contained_schemas(self.model)
    self.schema_references = get_schema_reference_infos(self.contained_schemas)
    # FIXME: resolve empty referenced nodes without an associated type -> either find subclasses or classes that contain the referenced name (e.g. "acticePoleHousing" for class "PoleHousing")

find_schema_references(model) classmethod

Method to find all contained models (inclusive the model itself) and references in a given model.

Parameters:

Name	Type	Description	Default
model	Identifiable	The model to find contained models and references in.	required

Returns:

Type	Description
Tuple[List[Identifiable], List[ReferenceInfo]]	A tuple containing the list of contained models and the list of references.

Source code in aas_middleware\model\reference_finder.py

@classmethod
def find_schema_references(
        cls,
        model: Identifiable,
) -> Tuple[List[Identifiable], Set[ReferenceInfo]]:
    """
    Method to find all contained models (inclusive the model itself) and references in a given model.

    Args:
        model (Identifiable): The model to find contained models and references in.

    Returns:
        Tuple[List[Identifiable], List[ReferenceInfo]]: A tuple containing the list of contained models and the list of references.
    """
    finder = cls()
    finder.model = model
    finder.find_contained_schemas_and_references()
    return finder.contained_schemas, finder.schema_references

ReferenceInfo (BaseModel)

Object reference to a model in the data model.

Parameters:

Name	Type	Description	Default
identifiable_id	str	The id of the identifiable.	required
reference_id	str	The id of the referenced identifiable.	required
reference_type	ReferenceType	The type of the reference.	required

Source code in aas_middleware\model\reference_finder.py

class ReferenceInfo(BaseModel):
    """
    Object reference to a model in the data model.

    Args:
        identifiable_id (str): The id of the identifiable.
        reference_id (str): The id of the referenced identifiable.
        reference_type (ReferenceType): The type of the reference.
    """

    identifiable_id: str
    reference_id: str
    reference_type: ReferenceType

    model_config = ConfigDict(frozen=True)

ReferenceType (Enum)

Enum for the reference types. There exist three types of references: - Association: The reference is done by an association between two objects, where the model has the referenced object as an attribute. - Reference: The reference element is an object and the reference in the model is done by referencing the id of the referenced object. - Attribute: The referenced element is an primitive attribute of the model.

Source code in aas_middleware\model\reference_finder.py

class ReferenceType(Enum):
    """
    Enum for the reference types. There exist three types of references:
    - Association: The reference is done by an association between two objects, where the model has the referenced object as an attribute.
    - Reference: The reference element is an object and the reference in the model is done by referencing the id of the referenced object.
    - Attribute: The referenced element is an primitive attribute of the model.
    """
    ASSOCIATION = "association"
    REFERENCE = "reference"
    ATTRIBUTE = "attribute"

get_reference_infos(identifiables)

Method to get all reference infos of a list of identifiables.

Parameters:

Name	Type	Description	Default
identifiables	List[Identifiable]	The list of identifiables.	required

Returns:

Type	Description
Set[ReferenceInfo]	The list of reference infos.

Source code in aas_middleware\model\reference_finder.py

def get_reference_infos(identifiables: List[Identifiable]) -> Set[ReferenceInfo]:
    """
    Method to get all reference infos of a list of identifiables.

    Args:
        identifiables (List[Identifiable]): The list of identifiables.

    Returns:
        Set[ReferenceInfo]: The list of reference infos.
    """
    reference_infos = set()
    for identifiable in identifiables:
        reference_infos = reference_infos | get_reference_infos_of_model(identifiable)
    return reference_infos

get_reference_infos_of_model(model)

Method to add information about referencing model ids of the input model.

Parameters:

Name	Type	Description	Default
model	Referable	The model to add the information for.	required

Returns:

Type	Description
Set[ReferenceInfo]	The list of reference infos.

Source code in aas_middleware\model\reference_finder.py

def get_reference_infos_of_model(model: Identifiable) -> Set[ReferenceInfo]:
    """
    Method to add information about referencing model ids of the input model.

    Args:
        model (Referable): The model to add the information for.

    Returns:
        Set[ReferenceInfo]: The list of reference infos.
    """
    reference_infos = set()
    identifiables_of_model = get_identifiable_attributes_of_model(model)
    for identifiable in identifiables_of_model:
        if identifiable == model:
            continue
        reference_info = ReferenceInfo(
            identifiable_id=get_id_with_patch(model),
            reference_id=get_id_with_patch(identifiable),
            reference_type=ReferenceType.ASSOCIATION,
        )
        reference_infos.add(reference_info)
    indirect_references = get_referenced_ids_of_model(model)
    for indirect_reference in indirect_references:
        reference_info = ReferenceInfo(
            identifiable_id=get_id_with_patch(model),
            reference_id=indirect_reference,
            reference_type=ReferenceType.REFERENCE,
        )
        reference_infos.add(reference_info)

    unidentifiable_attributes = get_unidentifiable_attributes_of_model(model)
    for attribute_name, attribute_value in unidentifiable_attributes.items():
        reference_info = ReferenceInfo(
            identifiable_id=get_id_with_patch(model),
            reference_id=f"{attribute_name}={attribute_value}",
            reference_type=ReferenceType.ATTRIBUTE,
        )
        reference_infos.add(reference_info)
    return reference_infos

get_reference_infos_of_schema(schema)

Method to add information about referencing schema ids of the input schema.

Parameters:

Name	Type	Description	Default
schema	Type[Identifiable]	The schema to add the information for.	required

Returns:

Type	Description
Set[ReferenceInfo]	The list of reference infos.

Source code in aas_middleware\model\reference_finder.py

def get_reference_infos_of_schema(schema: Type[Identifiable]) -> Set[ReferenceInfo]:
    """
    Method to add information about referencing schema ids of the input schema.

    Args:
        schema (Type[Identifiable]): The schema to add the information for.

    Returns:
        Set[ReferenceInfo]: The list of reference infos.
    """
    reference_infos = set()
    attribute_dict_of_schema = get_attribute_dict_of_schema(schema)
    for attribute_name, attribute_type in attribute_dict_of_schema.items():
        attribute_types = get_identifiable_types(attribute_type)

        for arg in attribute_types:
            reference_info = get_reference_info_for_schema(schema, attribute_name, arg)
            if not reference_info:
                continue
            reference_infos.add(reference_info)
    return reference_infos

get_schema_reference_infos(schemas)

Method to get all reference infos of a list of schemas.

Parameters:

Name	Type	Description	Default
schemas	List[Type[Identifiable]]	The list of schemas.	required

Returns:

Type	Description
List[ReferenceInfo]	The list of reference infos.

Source code in aas_middleware\model\reference_finder.py

def get_schema_reference_infos(schemas: List[Type[Identifiable]]) -> Set[ReferenceInfo]:
    """
    Method to get all reference infos of a list of schemas.

    Args:
        schemas (List[Type[Identifiable]]): The list of schemas.

    Returns:
        List[ReferenceInfo]: The list of reference infos.
    """
    reference_infos = set()
    for schema in schemas:
        reference_infos = reference_infos | get_reference_infos_of_schema(schema)
    return reference_infos

schema_util

add_non_redundant_schema(schema, schemas)

Method to add a schema to a list of schemas if it is not already in the list.

Parameters:

Name	Type	Description	Default
schema	Type[Identifiable]	The schema to add.	required
schemas	List[Type[Identifiable]]	The list of schemas.	required

Source code in aas_middleware\model\schema_util.py

def add_non_redundant_schema(schema: Type[Identifiable], schemas: List[Type[Identifiable]]):
    """
    Method to add a schema to a list of schemas if it is not already in the list.

    Args:
        schema (Type[Identifiable]): The schema to add.
        schemas (List[Type[Identifiable]]): The list of schemas.
    """
    if schema not in schemas:
        schemas.append(schema)

get_all_contained_schemas(schema)

Method to iterate over an Identifiable model and get all contained Identifiables.

Parameters:

Name	Type	Description	Default
schema	Type[Identifiable]	The referable data model.	required

Returns:

Type	Description
List[Type[Referable]]	The list of referables.

Source code in aas_middleware\model\schema_util.py

def get_all_contained_schemas(schema: Type[Identifiable]) -> List[Type[Identifiable]]:
    """
    Method to iterate over an Identifiable model and get all contained Identifiables.

    Args:
        schema (Type[Identifiable]): The referable data model.

    Returns:
        List[Type[Referable]]: The list of referables.
    """
    contained_schemas = []
    identifiable_schema_attributes = get_identifiable_attributes(schema)
    for identifiable_schema_attribute in identifiable_schema_attributes.values():
        in_attribute_contained_identifiable_schema = get_all_contained_schemas(
            identifiable_schema_attribute
        )
        for schema_attribute in in_attribute_contained_identifiable_schema:
            add_non_redundant_schema(schema_attribute, contained_schemas)
    if is_identifiable_type_container(schema):
        for item in typing.get_args(schema):
            in_attribute_contained_identifiable_schema = get_all_contained_schemas(item)
            for schema_attribute in in_attribute_contained_identifiable_schema:
                add_non_redundant_schema(schema_attribute, contained_schemas)
    elif is_identifiable_type(schema):
        add_non_redundant_schema(schema, contained_schemas)
    return contained_schemas

get_attribute_dict_of_schema(schema)

Method to get all attributes of a model.

Parameters:

Name	Type	Description	Default
model	Type[Identifiable]	The referable data model.	required

Returns:

Type	Description
Dict[str, Type[Identifiable]]	The dictionary of attributes.

Source code in aas_middleware\model\schema_util.py

def get_attribute_dict_of_schema(schema: Type[Identifiable]) -> Dict[str, Type[Identifiable]]:
    """
    Method to get all attributes of a model.

    Args:
        model (Type[Identifiable]): The referable data model.

    Returns:
        Dict[str, Type[Identifiable]]: The dictionary of attributes.
    """
    attribute_dict = {}
    if not isinstance(schema, type):
        return attribute_dict
    if issubclass(schema, BaseModel):
        for field_name, field in schema.model_fields.items():
            attribute_dict[field_name] = field.annotation
    else:
        annotations = typing.get_type_hints(schema.__init__)
        for parameter_name, parameter in annotations.items():
            attribute_dict[parameter_name] = parameter
    return attribute_dict

get_identifiable_attributes(schema)

Method to get all attributes of a model.

Parameters:

Name	Type	Description	Default
schema	Type[Identifiable]	The referable data model.	required

Returns:

Type	Description
List[str]	The list of attributes.

Source code in aas_middleware\model\schema_util.py

def get_identifiable_attributes(schema: Type[Identifiable]) -> Dict[str, Type[Identifiable]]:
    """
    Method to get all attributes of a model.

    Args:
        schema (Type[Identifiable]): The referable data model.

    Returns:
        List[str]: The list of attributes.
    """
    schema_attributes = get_attribute_dict_of_schema(schema)
    identifiable_attributes = {}
    for attribute_name, attribute_type in schema_attributes.items():
        if is_identifiable_type(attribute_type) or is_identifiable_type_container(attribute_type):
            identifiable_attributes[attribute_name] = attribute_type

    return identifiable_attributes

util

add_non_redundant_identifiable(model, identifiables)

Method to add an Identifiable to a list of Identifiables if it is not already in the list.

Parameters:

Name	Type	Description	Default
model	Identifiable	The Identifiable to add.	required
identifiables	List[Identifiable]	The list of contained Identifiables.	required

Returns:

Type	Description
List[Identifiable]	The list of Identifiables with the added model.

Source code in aas_middleware\model\util.py

def add_non_redundant_identifiable(
    model: Identifiable, identifiables: List[Identifiable]
) -> List[Identifiable]:
    """
    Method to add an Identifiable to a list of Identifiables if it is not already in the list.

    Args:
        model (Identifiable): The Identifiable to add.
        identifiables (List[Identifiable]): The list of contained Identifiables.

    Returns:
        List[Identifiable]: The list of Identifiables with the added model.
    """
    # TODO: maybe use directly a dict here to avoid iteration by using hashable ids
    if not any(
        get_id_with_patch(model) == get_id_with_patch(other_referable)
        for other_referable in identifiables
    ):
        identifiables.append(model)
    return identifiables

convert_camel_case_to_underscrore_str(came_case_string)

Convert a camel case string to an underscore seperated string.

Parameters:

Name	Type	Description	Default
class_name	str	The camel case string to convert.	required

Returns:

Type	Description
str	The underscore seperated string.

Source code in aas_middleware\model\util.py

def convert_camel_case_to_underscrore_str(came_case_string: str) -> str:
    """
    Convert a camel case string to an underscore seperated string.

    Args:
        class_name (str): The camel case string to convert.

    Returns:
        str: The underscore seperated string.
    """
    came_case_string = came_case_string[0].lower() + came_case_string[1:]
    new_class_name = re.sub(r"(?<!^)(?=[A-Z])", "_", came_case_string).lower()
    if all(len(el) == 1 for el in new_class_name.split("_")):
        new_class_name = new_class_name.replace("_", "")
    return new_class_name

convert_to_fitting_identifiable_container_type(list_container, container_type)

Function to convert a list of identifiables to a fitting container type.

Parameters:

Name	Type	Description	Default
list_container	List[Identifiable]	The list of identifiables.	required
container_type	Type[Any]	The container type.	required

Returns:

Type	Description
List[Identifiable] | Tuple[Identifiable] | Set[Identifiable]	The container type.

Source code in aas_middleware\model\util.py

def convert_to_fitting_identifiable_container_type(list_container: List[Identifiable], container_type: Type[Any]) -> List[Identifiable] | Tuple[Identifiable] | Set[Identifiable]:
    """
    Function to convert a list of identifiables to a fitting container type.

    Args:
        list_container (List[Identifiable]): The list of identifiables.
        container_type (Type[Any]): The container type.

    Returns:
        List[Identifiable] | Tuple[Identifiable] | Set[Identifiable]: The container type.
    """
    if container_type == list:
        return list_container
    elif container_type == tuple:
        return tuple(list_container)
    elif container_type == set:
        return set(list_container)
    else:
        raise ValueError("Container type not supported.")

convert_under_score_to_camel_case_str(underscore_str)

Convert a underscore seperated string to a camel case string.

Parameters:

Name	Type	Description	Default
class_name	str	The underscore seperated string to convert.	required

Returns:

Type	Description
str	The camel case string.

Source code in aas_middleware\model\util.py

def convert_under_score_to_camel_case_str(underscore_str: str) -> str:
    """
    Convert a underscore seperated string to a camel case string.

    Args:
        class_name (str): The underscore seperated string to convert.

    Returns:
        str: The camel case string.
    """
    words = underscore_str.split("_")
    camel_case_str = "".join(word[0].capitalize() + word[1:] for word in words)
    return camel_case_str

get_all_contained_identifiables(model)

Method to iterate over an Identifiable model and get all contained Identifiables.

Parameters:

Name	Type	Description	Default
model	REFERABLE_DATA_MODEL	The referable data model.	required

Returns:

Type	Description
List[Referable]	The list of referables.

Source code in aas_middleware\model\util.py

def get_all_contained_identifiables(model: Identifiable) -> List[Identifiable]:
    """
    Method to iterate over an Identifiable model and get all contained Identifiables.

    Args:
        model (REFERABLE_DATA_MODEL): The referable data model.

    Returns:
        List[Referable]: The list of referables.
    """
    contained_identifiables = []
    identifiable_attributes = get_identifiable_attributes_of_model(model)
    for identifiable_attribute in identifiable_attributes:
        in_attribute_contained_identifiables = get_all_contained_identifiables(
            identifiable_attribute
        )
        for identifiable in in_attribute_contained_identifiables:
            add_non_redundant_identifiable(identifiable, contained_identifiables)
    if is_identifiable_container(model):
        for item in model:
            in_attribute_contained_identifiables = get_all_contained_identifiables(item)
            for identifiable in in_attribute_contained_identifiables:
                add_non_redundant_identifiable(identifiable, contained_identifiables)
    elif is_identifiable(model):
        add_non_redundant_identifiable(model, contained_identifiables)
    return contained_identifiables

get_attribute_name_encoded_references(model)

Function to get the referenced ids of a model.

Parameters:

Name	Type	Description	Default
model	Referable	The model.	required

Returns:

Type	Description
List[str]	The referenced ids.

Source code in aas_middleware\model\util.py

def get_attribute_name_encoded_references(model: Identifiable) -> List[str]:
    """
    Function to get the referenced ids of a model.

    Args:
        model (Referable): The model.

    Returns:
        List[str]: The referenced ids.
    """
    referenced_ids = []
    for attribute_name, attribute_value in vars(model).items():
        if (
            attribute_name in STANDARD_AAS_FIELDS
            or attribute_name in REFERENCE_ATTRIBUTE_NAMES_SUFFIXES
        ):
            continue
        if not any(
            attribute_name.endswith(suffix)
            for suffix in REFERENCE_ATTRIBUTE_NAMES_SUFFIXES
        ):
            continue
        if isinstance(attribute_value, str | int | UUID):
            referenced_ids.append(str(attribute_value))
        else:
            referenced_ids += [str(item) for item in attribute_value if item]
    return referenced_ids

get_id(model)

Function to get the id attribute of an arbitrary model.

Parameters:

Name	Type	Description	Default
model	Any	The model.	required

Returns:

Type	Description
Optional[str | int | UUID]	The id attribute.

Exceptions:

Type	Description
ValueError	if the model is not an object, BaseModel or dict or if no id attribute is available

Source code in aas_middleware\model\util.py

def get_id(model: Any) -> str | int | UUID:
    """
    Function to get the id attribute of an arbitrary model.

    Args:
        model (Any): The model.

    Returns:
        Optional[str | int | UUID]: The id attribute.

    Raises:
        ValueError: if the model is not an object, BaseModel or dict or if no id attribute is available
    """
    if not is_identifiable(model):
        raise ValueError("Model is a basic type and has no id attribute.")

    if isinstance(model, BaseModel):
        identifiable_fields = get_identifier_type_fields(model.model_fields)
        if len(identifiable_fields) > 1:
            raise ValueError(f"Model has multiple Identifier attributes: {model}")
        if identifiable_fields:
            return getattr(model, identifiable_fields[0])
    elif hasattr(model, "__dict__"):
        # TODO: use typing.get_type_hints instead of inspect.signature
        sig = inspect.signature(type(model).__init__)
        potential_identifier = []
        for param in sig.parameters.values():
            if param.annotation == Identifier or param.annotation == "Identifier":
                potential_identifier.append(param.name)
        if len(potential_identifier) > 1:
            raise ValueError(f"Model {model} has multiple Identifier attributes.")
        if potential_identifier:
            return getattr(model, potential_identifier[0])

    if isinstance(model, BaseModel):
        data = model.model_dump()
    elif isinstance(model, dict):
        data = model
    else:
        data = vars(model)
    potential_id_attributes = [
        "id",
        "id_short",
        "Id",
        "ID",
        "Identifier",
        "identifier",
        "Identity",
        "identity",
    ]
    for id_attribute in potential_id_attributes:
        if id_attribute in data and isinstance(data[id_attribute], str | int | UUID):
            return data[id_attribute]

    raise ValueError(
        f"Model {model} has no attribute that can be used as id attribute."
    )

get_id_with_patch(model)

Function to get the id attribute of an arbitrary model.

Parameters:

Name	Type	Description	Default
model	Any	The model.	required

Returns:

Type	Description
Optional[str | int | UUID]	The id attribute.

Source code in aas_middleware\model\util.py

def get_id_with_patch(model: Any) -> str:
    """
    Function to get the id attribute of an arbitrary model.

    Args:
        model (Any): The model.

    Returns:
        Optional[str | int | UUID]: The id attribute.
    """
    if not is_identifiable(model):
        raise ValueError("Not identifiable object supplied.")
    try:
        return str(get_id(model))
    except ValueError:
        return "id_" + str(id(model))

get_identifier_type_fields(field_info_dict)

Function to get the fields of a model that are of type Identifier.

Parameters:

Name	Type	Description	Default
model	BaseModel	A Basemodel that is checked for identifier fields	required

Returns:

Type	Description
List[str]	The field names that are Identifiers

Source code in aas_middleware\model\util.py

def get_identifier_type_fields(field_info_dict: Dict[str, FieldInfo]) -> List[str]:
    """
    Function to get the fields of a model that are of type Identifier.

    Args:
        model (BaseModel): A Basemodel that is checked for identifier fields

    Returns:
        List[str]: The field names that are Identifiers
    """
    model_fields = []
    for field_name, field_info in field_info_dict.items():
        if field_info.annotation == Identifier:
            model_fields.append(field_name)
    return model_fields

get_reference_name(attribute_name, attribute_type)

Function to get the reference name of an attribute.

Parameters:

Name	Type	Description	Default
attribute_name	str	The attribute name.	required
attribute_type	Type[Any]	The type of the attribute.	required

Returns:

Type	Description
str	The name of the referenced type.

Source code in aas_middleware\model\util.py

def get_reference_name(attribute_name: str, attribute_type: Type[Any]) -> Optional[str]:
    """
    Function to get the reference name of an attribute.

    Args:
        attribute_name (str): The attribute name.
        attribute_type (Type[Any]): The type of the attribute.

    Returns:
        str: The name of the referenced type.
    """
    if attribute_name in REFERENCE_ATTRIBUTE_NAMES_SUFFIXES or attribute_name in STANDARD_AAS_FIELDS:
        return 

    if attribute_type == Reference or attribute_type == "Reference":
        return attribute_name
    elif typing.get_origin(attribute_type) in [List, Set, Tuple, Union] and Reference in typing.get_args(attribute_type):
        return attribute_name
    elif any (attribute_name.endswith(suffix) for suffix in REFERENCE_ATTRIBUTE_NAMES_SUFFIXES):
        suffix = next(suffix for suffix in REFERENCE_ATTRIBUTE_NAMES_SUFFIXES if attribute_name.endswith(suffix))
        underscore_consideration = False
        if attribute_name.endswith(f"_{suffix}"):
            underscore_consideration = True
        attribute_name_without_suffix = attribute_name[:-(len(suffix) + underscore_consideration)]
        if attribute_name_without_suffix.endswith("s") and not attribute_name_without_suffix.endswith("ss"):
            attribute_name_without_suffix = attribute_name_without_suffix[:-1]
        return convert_under_score_to_camel_case_str(attribute_name_without_suffix)

get_referenced_ids_of_model(model)

Function to get the referenced ids of a model by searching for type Reference and attribute names which suggest references.

Parameters:

Name	Type	Description	Default
model	Referable	The model to get the references from.	required

Returns:

Type	Description
List[str]	The referenced ids.

Source code in aas_middleware\model\util.py

def get_referenced_ids_of_model(model: Identifiable) -> Set[str]:
    """
    Function to get the referenced ids of a model by searching for type Reference and attribute names which suggest references.

    Args:
        model (Referable): The model to get the references from.

    Returns:
        List[str]: The referenced ids.
    """
    referenced_ids = []
    if isinstance(model, BaseModel):
        referenced_ids += get_references_of_reference_type_for_basemodel(model)
    elif hasattr(model, "__dict__"):
        referenced_ids += get_references_of_reference_type_for_object(model)
    referenced_ids += get_attribute_name_encoded_references(model)
    return set(referenced_ids)

get_references_of_reference_type_for_basemodel(model)

Function to get the references of a model that are of type Reference.

Parameters:

Name	Type	Description	Default
model	BaseModel	The model.	required

Returns:

Type	Description
List[str]	The reference fields.

Source code in aas_middleware\model\util.py

def get_references_of_reference_type_for_basemodel(model: BaseModel) -> List[str]:
    """
    Function to get the references of a model that are of type Reference.

    Args:
        model (BaseModel): The model.

    Returns:
        List[str]: The reference fields.
    """
    references = []
    for field_name, field_info in model.model_fields.items():
        if field_info.annotation == Reference or field_info.annotation == "Reference":
            references.append(getattr(model, field_name))
        if field_info.annotation == List[Reference]:
            references += getattr(model, field_name)
    return [str(ref) for ref in references if ref]

get_references_of_reference_type_for_object(model)

Function to get the references of a model that are of type Reference.

Parameters:

Name	Type	Description	Default
model	BaseModel	The model.	required

Returns:

Type	Description
List[str]	The reference fields.

Source code in aas_middleware\model\util.py

def get_references_of_reference_type_for_object(model: object) -> List[str]:
    """
    Function to get the references of a model that are of type Reference.

    Args:
        model (BaseModel): The model.

    Returns:
        List[str]: The reference fields.
    """
    references = []
    sig = inspect.signature(type(model).__init__)
    for param in sig.parameters.values():
        if param.annotation == Reference:
            references.append(getattr(model, param.name))
        if param.annotation == List[Reference]:
            references += getattr(model, param.name)
    return [str(ref) for ref in references if ref]

get_value_attributes(obj)

Function to get an dict of all attributes of an object without the private attributes and standard AAS attributes.

Parameters:

Name	Type	Description	Default
obj	object	The object.	required

Returns:

Type	Description
dict	The value attributes.

Source code in aas_middleware\model\util.py

def get_value_attributes(obj: object) -> Dict[str, Any]:
    """
    Function to get an dict of all attributes of an object without the private attributes and standard AAS attributes.

    Args:
        obj (object): The object.

    Returns:
        dict: The value attributes.
    """
    vars_dict = {}
    object_id = get_id_with_patch(obj)

    for attribute_name, attribute_value in vars(obj).items():
        if attribute_name.startswith("_"):
            continue
        if attribute_name in STANDARD_AAS_FIELDS:
            continue
        if attribute_value == object_id:
            continue
        if attribute_value is None:
            continue
        vars_dict[attribute_name] = attribute_value
    return vars_dict

is_identifiable(model)

Function to check if a model is identifiable.

Parameters:

Name	Type	Description	Default
model	Any	The model.	required

Returns:

Type	Description
bool	True if the model is identifiable, False otherwise.

Source code in aas_middleware\model\util.py

def is_identifiable(model: Any) -> bool:
    """
    Function to check if a model is identifiable.

    Args:
        model (Any): The model.

    Returns:
        bool: True if the model is identifiable, False otherwise.
    """
    if isinstance(model, UnIdentifiable):
        return False
    return True

is_identifiable_container(model)

Function to check if a model is an identifiable container.

Parameters:

Name	Type	Description	Default
model	Any	The model.	required

Returns:

Type	Description
bool	True if the model is an identifiable container, False otherwise.

Source code in aas_middleware\model\util.py

def is_identifiable_container(model: Any) -> bool:
    """
    Function to check if a model is an identifiable container.

    Args:
        model (Any): The model.

    Returns:
        bool: True if the model is an identifiable container, False otherwise.
    """
    if not isinstance(model, list | tuple | set | dict):
        return False
    if isinstance(model, list | tuple | set) and not all(
        is_identifiable(element) for element in model
    ):
        return False
    return True

is_identifiable_type(schema)

Function to check if a schema is identifiable.

Parameters:

Name	Type	Description	Default
schema	Type[Any]	The schema.	required

Returns:

Type	Description
bool	True if the schema is identifiable, False otherwise.

Source code in aas_middleware\model\util.py

def is_identifiable_type(schema: Type[Any]) -> bool:
    """
    Function to check if a schema is identifiable.

    Args:
        schema (Type[Any]): The schema.

    Returns:
        bool: True if the schema is identifiable, False otherwise.
    """
    # TODO: refactor to combine is_identifiable and is_identifiable_type
    # TODO: handle here also union types
    if not isinstance(schema, type):
        return False
    if issubclass(schema, UnIdentifiable):
        return False
    return True

is_identifiable_type_container(schema)

Method to check if a schema is a container of identifiables.

Parameters:

Name	Type	Description	Default
schema	Type[Any]	The schema.	required

Returns:

Type	Description
bool	True if the schema is a container of identifiables, False otherwise.

Source code in aas_middleware\model\util.py

def is_identifiable_type_container(schema: Type[Any]) -> bool:
    """
    Method to check if a schema is a container of identifiables.

    Args:
        schema (Type[Any]): The schema.

    Returns:
        bool: True if the schema is a container of identifiables, False otherwise.
    """
    # TODO: refactor to combine is_identifiable_container and is_identifiable_type_container
    if typing.get_origin(schema):
        outer_type = typing.get_origin(schema)
    else:
        outer_type = schema

    if not outer_type in [list,  tuple, set, dict, Union]:
        return False
    if outer_type == dict:
        raise NotImplementedError("Dicts are not supported yet. Try using classes instead.")
    type_arguments = get_identifiable_types(schema)
    if not type_arguments:
        return False
    type_arguments_with_none = [arg for arg in type_arguments if arg != NoneType]
    if not all(is_identifiable_type(element) for element in type_arguments_with_none):
        return False
    return True

models_are_equal(model1, model2)

Function to compare two models for equality.

Parameters:

Name	Type	Description	Default
model1	Identifiable	The first model.	required
model2	Identifiable	The second model.	required

Returns:

Type	Description
bool	True if the models are equal, False otherwise.

Source code in aas_middleware\model\util.py

def models_are_equal(model1: Identifiable, model2: Identifiable) -> bool:
    """
    Function to compare two models for equality.

    Args:
        model1 (Identifiable): The first model.
        model2 (Identifiable): The second model.

    Returns:
        bool: True if the models are equal, False otherwise.
    """
    model1_attributes = get_value_attributes(model1)
    model2_attributes = get_value_attributes(model2)
    if set(model1_attributes.keys()) != set(model2_attributes.keys()):
        return False
    for attribute_name1, attribute_value1 in model1_attributes.items():
        if is_identifiable(attribute_value1):
            if not models_are_equal(
                attribute_value1, model2_attributes[attribute_name1]
            ):
                return False
        elif is_identifiable_container(attribute_value1):
            if not is_identifiable_container(model2_attributes[attribute_name1]):
                return False
            if not len(attribute_value1) == len(model2_attributes[attribute_name1]):
                return False
            if not all(
                models_are_equal(item1, item2)
                for item1, item2 in zip(
                    attribute_value1, model2_attributes[attribute_name1]
                )
            ):
                return False
        elif attribute_value1 != model2_attributes[attribute_name1]:
            return False
    return True

replace_attribute_with_model(model, existing_model)

Function to replace an attribute with a model.

Parameters:

Name	Type	Description	Default
model	Identifiable	The model.	required
existing_model	Identifiable	The existing model.	required

Source code in aas_middleware\model\util.py

def replace_attribute_with_model(model: Identifiable, existing_model: Identifiable):
    """
    Function to replace an attribute with a model.

    Args:
        model (Identifiable): The model.
        existing_model (Identifiable): The existing model.
    """
    for attribute_name, attribute_value in vars(model).items():
        if is_identifiable(attribute_value):
            if attribute_value == existing_model:
                setattr(model, attribute_name, existing_model)
            else:
                replace_attribute_with_model(attribute_value, existing_model)
        elif is_identifiable_container(attribute_value):
            list_attribute_value = list(attribute_value)
            for i, item in enumerate(list_attribute_value):
                if item == existing_model:
                    list_attribute_value[i] = existing_model
                else:
                    replace_attribute_with_model(item, existing_model)
            setattr(model, attribute_name, convert_to_fitting_identifiable_container_type(list_attribute_value, type(attribute_value)))