serializer.py

Go to the documentation of this file.

from __future__ import annotations
 
import json
from typing import TextIO, cast, Any, Union, Dict, List
 
from ids_peak_common.serialization.iarchive import IArchive
from ids_peak_icv.exceptions import NotSupportedException
from ids_peak_icv.serialization.serialization_type import SerializationType
from ids_peak_icv.serialization.archive import Archive
 
 
class Serializer:
    """!
    @brief
    Serializer for writing `IArchive` objects to an output stream.
 
    This class converts an `IArchive` object
    into a specific serialization format
    and writes the result to a stream.
 
    **Example**
 
        from ids_peak_icv.serialization import Archive
        from ids_peak_icv.serialization import Serializer, SerializationType
 
        archive = Archive()
        archive.set("id", 42)
        archive.set("name", "example")
 
        serializer = Serializer(SerializationType.JSON)
        with open("archive.json", "w") as f:
            serializer.write(archive, f)
 
    @since ids_peak_icv 1.0
 
    @ingroup ids_peak_icv_python_serialization
    """
 
    def __init__(self, serialization_type: SerializationType = SerializationType.JSON) -> None:
        """!
        @brief
        Initializes the serializer with the specified serialization type.
 
        The chosen format determines how the archive is written
        when `write()` is called.
        If no type is provided,
        JSON is used by default.
 
        @param serialization_type: The serialization format to use
                                   (defaults to :class:`SerializationType.JSON`).
 
        @since ids_peak_icv 1.0
        """
        self._type_type = serialization_type
 
    def write(self, archive: IArchive, stream: TextIO) -> None:
        """!
        @brief
        Serializes an archive and writes it to a stream.
 
        Converts the given `IArchive` object
        into the configured format
        and writes the serialized data to the provided stream.
 
        @param archive: The archive to serialize.
        @param stream:  The text stream that receives the serialized data
                        (e.g., file or StringIO).
 
        @exception NotSupportedException: The serialization type is not supported.
 
        @since ids_peak_icv 1.0
        """
        if self._type_type == SerializationType.JSON:
            data = self._to_dict(archive)
            json.dump(data, cast(Any, stream), ensure_ascii=False, indent=4)
        else:
            raise NotSupportedException(f"Serialization type {self._type} is not supported.")
 
    def _to_dict(self, archive: IArchive) -> dict:
        """!
        @brief
        Recursively converts an `IArchive` into a plain Python dict.
 
        @param archive: The archive to convert.
 
        @return A dictionary representation of the archive,
                 suitable for serialization.
 
        @since ids_peak_icv 1.0
        """
        result: Dict[Any, Union[Dict[Any, Any], List[Dict[Any, Any]]]] = {}
 
        if not isinstance(archive, Archive):
            raise NotSupportedException("Serialization is only supported for the Archive implementation.")
 
        archive = cast(Archive, archive)
        for key in archive.keys():
            value = archive.get(key, None)
            if isinstance(value, IArchive):
                result[key] = self._to_dict(value)
            elif isinstance(value, list) and value and isinstance(value[0], IArchive):
                result[key] = [self._to_dict(v) for v in value]
            else:
                result[key] = value
        return result