orchestrata_ultimate.cxp.binary_serializer

Reference › orchestrata_ultimate › cxp › binary_serializer

BinarySerializer is the compact binary form of the ISerializer contract (Spec 004 US-006). It uses MessagePack when the msgpack package is installed and otherwise falls back to gzip-compressed JSON. The output carries a one-byte prefix, 0x01 for MessagePack and 0x02 for gzip JSON, so from_bytes can detect the format; it raises DeserializationError for input that is not bytes, is too short to hold a prefix, or cannot be decoded. to_bytes converts the packet with to_dict, prepares values with the private _encode_value (and _decode_value on the way back) and raises SerializationError with the packet id on failure. The text form is json_serializer. See exceptions and cxp.

At a glance

Class diagram

Dependency diagram

API

Classes

BinarySerializer

Decorators: traces(spec='004', story='US-006', reqs=['REQ-004-US-006-017', 'REQ-004-US-006-021']) · Line 93

Binary serializer for CXP packets.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/cxp/binary_serializer.py

# mypy: ignore-errors
"""
CXP Binary Serializer.

Binary serialization and deserialization for CXP packets.
Uses MessagePack when available, falls back to JSON + gzip.
Implements ISerializer binary methods per Spec 004 Section US-006.

Performance targets:
    - to_bytes: < 5ms for 1KB payload
    - from_bytes: < 5ms

Author: Mohamed
"""

import base64
import gzip
import json
import logging
from datetime import datetime
from enum import Enum
from typing import Any, cast

import msgpack as _msgpack

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket
from orchestrata_ultimate.cxp.exceptions import DeserializationError, SerializationError

# Spec: 004 - CXP Protocol


__all__ = ["BinarySerializer"]

logger = logging.getLogger(__name__)

# ---------- msgpack availability ----------


_msgpack_available = True


# ---------- helpers ----------


@traces(spec="004", story="US-006", reqs=["REQ-004-US-006-018"])
def _encode_value(obj: Any) -> Any:
    """Recursively convert non-serializable types for msgpack/json.

    Handles datetime, Enum, bytes, set, frozenset.
    """
    if isinstance(obj, datetime):
        return {"__datetime__": obj.isoformat()}
    if isinstance(obj, Enum):
        return obj.value
    if isinstance(obj, bytes):
        return {"__bytes__": base64.b64encode(obj).decode("ascii")}
    if isinstance(obj, (set, frozenset)):
        return sorted(obj, key=str)
    if isinstance(obj, dict):
        return {k: _encode_value(v) for k, v in obj.items()}
    if isinstance(obj, (list, tuple)):
        return [_encode_value(item) for item in obj]
    return obj


@traces(spec="004", story="US-006", reqs=["REQ-004-US-006-022"])
def _decode_value(obj: Any) -> Any:
    """Recursively restore encoded types from msgpack/json output."""
    if isinstance(obj, dict):
        if "__datetime__" in obj and len(obj) == 1:
            iso = obj["__datetime__"]
            return datetime.fromisoformat(iso.replace("Z", "+00:00"))
        if "__bytes__" in obj and len(obj) == 1:
            return base64.b64decode(obj["__bytes__"])
        return {k: _decode_value(v) for k, v in obj.items()}
    if isinstance(obj, (list, tuple)):
        return [_decode_value(item) for item in obj]
    if isinstance(obj, bytes):
        # msgpack sometimes returns bytes for string keys
        try:
            return obj.decode("utf-8")
        except UnicodeDecodeError:
            return obj
    return obj


@traces(
    spec="004",
    story="US-006",
    reqs=["REQ-004-US-006-017", "REQ-004-US-006-021"],
)
class BinarySerializer:
    """Binary serializer for CXP packets.

    Uses MessagePack when the ``msgpack`` package is installed.
    Falls back to JSON-encoded bytes compressed with gzip otherwise.

    Conforms to ISerializer binary methods (Spec 004 Section US-006).

    Usage:
        serializer = BinarySerializer()
        raw = serializer.to_bytes(packet)
        restored = serializer.from_bytes(raw)
    """

    # 1-byte format prefix so from_bytes knows which codec produced the blob.
    _MSGPACK_PREFIX = b"\x01"
    _GZIP_JSON_PREFIX = b"\x02"

    def __init__(self) -> None:
        logger.debug(
            "BinarySerializer initialized (msgpack=%s)",
            "available" if _msgpack_available else "fallback:json+gzip",
        )

    # -- public properties --

    @property
    def _has_msgpack(self) -> bool:
        """True when the msgpack library is importable."""
        return _msgpack_available

    # -- public API --

    @traces(
        spec="004",
        story="US-006",
        reqs=[
            "REQ-004-PERF-003",
            "REQ-004-US-006-017",
            "REQ-004-US-006-018",
            "REQ-004-US-006-019",
        ],
    )
    def to_bytes(self, packet: CXPPacket) -> bytes:
        """Serialize a CXPPacket to bytes.

        If msgpack is available, the output is prefixed with 0x01
        followed by msgpack-encoded data. Otherwise, the output is
        prefixed with 0x02 followed by gzip-compressed JSON.

        Args:
            packet: The CXP packet to serialize.

        Returns:
            Binary representation of the packet.

        Raises:
            SerializationError: If serialization fails.
        """
        try:
            data = packet.to_dict()
            prepared = _encode_value(data)
            payload = _msgpack.packb(prepared, use_bin_type=True)
            # cast: msgpack's fallback.Packer.pack() stub-infers `bytes | None`
            # (None only on the non-default autoreset=False path, which we
            # never take), so packb() is bytes here.
            return self._MSGPACK_PREFIX + cast(bytes, payload)

        except (TypeError, ValueError, OverflowError, RecursionError) as exc:
            raise SerializationError(
                message=f"Binary serialization failed: {exc}",
                format="binary",
                details={"packet_id": getattr(packet, "message_id", "unknown")},
            ) from exc

    @traces(
        spec="004",
        story="US-006",
        reqs=[
            "REQ-004-PERF-004",
            "REQ-004-US-006-021",
            "REQ-004-US-006-022",
            "REQ-004-US-006-023",
            "REQ-004-US-006-024",
        ],
    )
    def from_bytes(self, data: bytes) -> CXPPacket:
        """Deserialize bytes to a CXPPacket.

        Detects the format prefix (0x01 = msgpack, 0x02 = gzip+json)
        and decodes accordingly.

        Args:
            data: Binary data previously produced by to_bytes().

        Returns:
            Reconstructed CXPPacket instance.

        Raises:
            DeserializationError: If the data is invalid or packet
                reconstruction fails.
        """
        if not isinstance(data, (bytes, bytearray)):
            raise DeserializationError(
                message=f"Expected bytes, got {type(data).__name__}",
                format="binary",
            )

        if len(data) < 2:
            raise DeserializationError(
                message="Binary data too short (missing format prefix)",
                format="binary",
            )

        prefix = data[:1]
        payload = data[1:]

        try:
            if prefix == self._MSGPACK_PREFIX:
                raw = _msgpack.unpackb(payload, raw=False)
            elif prefix == self._GZIP_JSON_PREFIX:
                json_bytes = gzip.decompress(payload)
                raw = json.loads(json_bytes.decode("utf-8"))
            else:
                raise DeserializationError(
                    message=f"Unknown binary format prefix: {prefix!r}",
                    format="binary",
                )
        except DeserializationError:
            raise
        except Exception as exc:
            raise DeserializationError(
                message=f"Binary deserialization failed: {exc}",
                format="binary",
            ) from exc

        decoded = _decode_value(raw)

        if not isinstance(decoded, dict):
            raise DeserializationError(
                message=f"Decoded data must be a dict, got {type(decoded).__name__}",
                format="binary",
            )

        try:
            return CXPPacket.from_dict(decoded)
        except (KeyError, TypeError, ValueError) as exc:
            raise DeserializationError(
                message=f"Packet reconstruction failed: {exc}",
                format="binary",
                details={"keys_present": list(decoded.keys())},
            ) from exc