archive.py

Go to the documentation of this file.

from __future__ import annotations
 
from typing import Any, Dict, List, Type, TypeAlias
 
from ids_peak_common.serialization.iarchive import IArchive
from ids_peak_icv.exceptions import NotSupportedException, CorruptedException
 
T: TypeAlias = IArchive.T
 
 
class Archive(IArchive):
    """!
    @brief
    A key-value store with type checking, based on a dict.
 
    Supports only:
    - bool, int, float, str
    - IArchive
    - Lists of the above (with homogeneous element types)
 
    **Example**
 
        from ids_peak_icv.serialization import Archive
 
        archive = Archive()
 
        archive.set("A", 1)
        archive.set("B", "Text")
        archive.set("C", [1, 2, 3])
 
        sub_archive = Archive()
        sub_archive.set("a", 1.2)
        sub_archive.set("b", True)
        archive.set("D", sub_archive)
 
    @since ids_peak_icv 1.0
 
    @ingroup ids_peak_icv_python_serialization
    """
    AllowedValue = IArchive.AllowedValue | IArchive
 
    def __init__(self, data: Dict[str, Archive.AllowedValue] | None = None) -> None:
        """!
        @brief
        Initializes the archive with optional initial data.
 
        If ``data`` is provided,
        all entries are validated for supported types
        before the dict is stored by reference.
        If ``data`` is ``None``,
        the archive is initially empty.
 
        @param data: Optional dictionary of initial key-value pairs.
 
        @exception NotSupportedException: An entry in data has an unsupported type.
 
        @since ids_peak_icv 1.0
        """
        if data is None:
            data = {}
        else:
            for key, value in data.items():
                if not self._is_supported_type(value):
                    raise NotSupportedException(
                        f"Unsupported value type for key '{key}': {type(value).__name__}"
                    )
 
        self._storage: Dict = data
 
    def set(self, key: str, value: AllowedValue) -> None:
        """!
        @brief
        Stores a value under the specified key.
 
        @param key:   The key under which to store the value.
        @param value: The value to store (restricted types only).
 
        @exception NotSupportedException: The value has an unsupported type.
 
        @since ids_peak_icv 1.0
        """
        if self._is_supported_type(value):
            self._storage[key] = value
        else:
            raise NotSupportedException(
                f"Unsupported value type: {type(value).__name__}"
            )
 
    def get(self, key: str, datatype: Type[T] | None) -> T:
        """!
        @brief
        Returns the value stored under the specified key.
 
        @param key:      Key of the value to retrieve.
        @param datatype: The expected datatype of the value.
                         If None is given, the type check is not executed.
 
        @return The stored value.
 
        @exception CorruptedException: If the key is not found
                                    or the specified datatype is not the type of the stored value.
 
        @since ids_peak_icv 1.0
        """
 
        if key not in self._storage:
            raise CorruptedException(f"The specified key '{key}' is not in the archive.")
 
        value = self._storage[key]
        if datatype is not type(value) and datatype is not None:
            raise CorruptedException(f"The value of the specified key '{key}' is not of the expected datatype.")
 
        return value
 
    def get_archive(self, key: str) -> IArchive:
        """!
        @brief
        Returns the archive stored under the given key.
 
        @param key: Key of the archive to retrieve.
 
        @return The stored archive.
 
        @exception CorruptedException: The key is not found
                                    or the specified datatype is not of type ``IArchive``.
 
        @since ids_peak_icv 1.0
        """
 
        if key not in self._storage:
            raise CorruptedException(f"The specified key '{key}' is not in the archive.")
        value = self._storage[key]
        if not isinstance(value, IArchive):
            raise CorruptedException(f"The value of the specified key '{key}' is not of the expected datatype.")
 
        return value
 
    def has(self, key: str) -> bool:
        """!
        @brief
        Checks whether a given key exists in the archive.
 
        @param key: The key to check.
 
        @return True if the key exists, False otherwise.
 
        @since ids_peak_icv 1.0
        """
        return key in self._storage
 
    def keys(self) -> List[str]:
        """!
        @brief
        Returns all keys currently stored in the archive.
 
        @return List of keys.
 
        @since ids_peak_icv 1.0
        """
        return list(self._storage.keys())
 
    def create_archive(self) -> IArchive:
        """!
        @brief
        Creates a new, empty archive instance.
 
        @since ids_peak_icv 1.0
        """
        return Archive()
 
    def __str__(self) -> str:
        """!
        @brief
        A string representation of the archive.
 
        @since ids_peak_icv 1.0
        """
        return str(self._storage)
 
    def __repr__(self) -> str:
        """!
        @brief
        @since ids_peak_icv 1.0
        """
        return f"Archive({str(self)})"
 
    def __eq__(self, other: Any) -> bool:
        """!
        @brief
        @since ids_peak_icv 1.0
        """
        if not isinstance(other, Archive):
            return False
 
        if set(self._storage.keys()) != set(other._storage.keys()):
            return False
 
        for key in self._storage:
            if not self._compare_values(self._storage[key], other._storage[key]):
                return False
 
        return True
 
    def _compare_values(self, lhs: Any, rhs: Any) -> bool:
        """!
        @brief
        @since ids_peak_icv 1.0
        """
        if isinstance(lhs, Archive) and isinstance(rhs, Archive):
            return bool(lhs == rhs)
 
        if isinstance(lhs, list) and isinstance(rhs, list):
            if len(lhs) != len(rhs):
                return False
            return all(self._compare_values(val1, val2) for val1, val2 in zip(lhs, rhs))
 
        return bool(lhs == rhs)
 
    _ALLOWED_SCALARS = (bool, int, float, str, IArchive)
 
    def _is_supported_type(self, value: Any) -> bool:
        """!
        @brief
        Returns whether the value’s type is supported.
 
        @param value: A value or a list whose type is validated.
 
        @return `True` if the type supported, `False` otherwise.
 
        @since ids_peak_icv 1.0
        """
        if isinstance(value, self._ALLOWED_SCALARS):
            return True
        if isinstance(value, list):
            if len(value) == 0:
                return True
            first_type = type(value[0])
            if all(isinstance(v, first_type) for v in value):
                if issubclass(first_type, self._ALLOWED_SCALARS):
                    return True
        return False