deserializer.py

Go to the documentation of this file.

from __future__ import annotations
import json
from typing import TextIO, cast
 
from ids_peak_common.serialization.iarchive import IArchive
from ids_peak_icv.exceptions import IOException
from ids_peak_icv.serialization.archive import Archive
from ids_peak_icv.exceptions import NotSupportedException, CorruptedException
from ids_peak_icv.serialization.serializer import SerializationType
 
 
class Deserializer:
    """!
    @brief
    Deserializer for reading serialized data
    and converting it into an `IArchive` object.
 
    This class provides functionality to read data
    from a supported serialization format
    (currently, only JSON)
    and reconstruct it as an `IArchive` object.
 
    **Example**
 
        from ids_peak_icv.serialization import Deserializer, SerializationType
 
        deserializer = Deserializer(SerializationType.JSON)
        with open("archive.json", "r") as f:
            archive = deserializer.read(f)
 
        id = archive.get("id", int)
        name = archive.get("name", str)
 
    @since ids_peak_icv 1.0
 
    @ingroup ids_peak_icv_python_serialization
    """
 
    def __init__(self, serialization_type: SerializationType = SerializationType.JSON) -> None:
        """!
        @brief
        Initializes a deserializer instance.
 
        @param serialization_type: The format to use for deserialization.
                                   Defaults to `SerializationType.JSON`.
 
        @since ids_peak_icv 1.0
        """
        self._type = serialization_type
 
    def read(self, stream: TextIO) -> IArchive:
        """!
        @brief
        Reads serialized data from the specified stream
        and deserializes it into an archive.
 
        @param stream: The stream to read serialized data from.
 
        @return An `IArchive` object containing the deserialized data.
 
        @exception NotSupportedException: The serialization type is not supported.
        @exception IOException:           The parsing failed.
        @exception CorruptedException:    The format of the given data is invalid.
 
        @since ids_peak_icv 1.0
        """
        if self._type != SerializationType.JSON:
            raise NotSupportedException(f"Serialization type {self._type} is not supported.")
 
        data = stream.read()
        try:
            parsed = json.loads(data)
        except json.JSONDecodeError as e:
            raise IOException(f"Failed to parse JSON: {e}")
 
        archive = Archive()
        self._populate_archive(archive, parsed)
        return archive
 
    def _populate_archive(self, archive: Archive, data: list | dict) -> None:
        """!
        @brief
        Recursively populates an Archive with parsed JSON data.
 
        @param archive: The archive to populate.
        @param data: Parsed JSON data (dict, list, scalar).
 
        @exception CorruptedException: The format of the given data is invalid.
 
        @since ids_peak_icv 1.0
        """
        if not isinstance(data, dict):
            raise CorruptedException("Root of serialized data must be a JSON object.")
 
        for key, value in data.items():
            if isinstance(value, dict):
                nested = Archive()
                self._populate_archive(nested, value)
                archive.set(key, nested)
            elif isinstance(value, list):
                if value and isinstance(value[0], dict):
                    nested_list = []
                    for item in value:
                        nested = Archive()
                        self._populate_archive(nested, item)
                        nested_list.append(nested)
                    archive.set(key, cast(list[IArchive], nested_list))
                else:
                    archive.set(key, value)
            else:
                archive.set(key, value)