pamoja.codec

  1"""Idiomatic codec facade.
  2
  3Python already has :mod:`json`, so this facade takes and returns ordinary values
  4and does the encoding itself, leaving callers to think in documents rather than
  5buffers. The conversion and the packing happen in the Rust core.
  6"""
  7
  8from __future__ import annotations
  9
 10import json
 11from typing import Any, Sequence
 12
 13from pamoja._native import Quantizer as _NativeQuantizer
 14from pamoja._native import cbor_to_json_bytes as _cbor_to_json_bytes
 15from pamoja._native import decode_delta_samples as _decode_delta_samples
 16from pamoja._native import encode_delta_samples as _encode_delta_samples
 17from pamoja._native import json_to_cbor_bytes as _json_to_cbor_bytes
 18
 19__all__ = ["Quantizer", "from_cbor", "pack_samples", "to_cbor", "unpack_samples"]
 20
 21
 22def to_cbor(value: Any) -> bytes:
 23    """Encode a value as CBOR, typically much smaller than its JSON form.
 24
 25    :param value: Any JSON-serializable value, or the raw bytes of a JSON
 26        document.
 27    :returns: The CBOR encoding.
 28    :raises PamojaError: If the value cannot be encoded.
 29    """
 30    if isinstance(value, (bytes, bytearray, memoryview)):
 31        document = bytes(value)
 32    else:
 33        document = json.dumps(value).encode("utf-8")
 34    return _json_to_cbor_bytes(document)
 35
 36
 37def from_cbor(cbor: bytes) -> Any:
 38    """Decode a CBOR document back into an ordinary Python value.
 39
 40    :param cbor: The CBOR document to decode.
 41    :returns: The decoded value.
 42    :raises PamojaError: If the document is malformed, or holds a construct with
 43        no JSON equivalent such as a non-string map key.
 44    """
 45    return json.loads(_cbor_to_json_bytes(bytes(cbor)).decode("utf-8"))
 46
 47
 48def pack_samples(samples: Sequence[int]) -> bytes:
 49    """Delta-encode a series of integer samples into a compact buffer.
 50
 51    :param samples: The samples, in order.
 52    :returns: The packed encoding, far smaller than the samples for a
 53        slow-moving series.
 54    """
 55    return _encode_delta_samples(list(samples))
 56
 57
 58def unpack_samples(data: bytes) -> list[int]:
 59    """Unpack a buffer produced by :func:`pack_samples`.
 60
 61    :param data: The packed encoding.
 62    :returns: The samples, in order.
 63    :raises PamojaError: If the buffer is malformed.
 64    """
 65    return _decode_delta_samples(bytes(data))
 66
 67
 68class Quantizer:
 69    """Packs float readings to a fixed precision, for a link charging per byte.
 70
 71    Example::
 72
 73        quantizer = Quantizer(100)  # keep two decimal places
 74        packed = quantizer.encode([20.0, 20.1, 20.2])
 75        quantizer.decode(packed)  # [20.0, 20.1, 20.2], to within 0.01
 76    """
 77
 78    __slots__ = ("_native",)
 79
 80    def __init__(self, scale: float) -> None:
 81        """Create a quantizer at the given precision.
 82
 83        :param scale: The multiplier applied before rounding; ``100`` keeps two
 84            decimal places. Must be positive and finite.
 85        :raises ValueError: If the scale is not positive and finite.
 86        """
 87        self._native = _NativeQuantizer(scale)
 88
 89    def encode(self, readings: Sequence[float]) -> bytes:
 90        """Quantize and pack a batch of readings.
 91
 92        :param readings: The readings, in order.
 93        :returns: The packed encoding.
 94        """
 95        return self._native.encode(list(readings))
 96
 97    def decode(self, data: bytes) -> list[float]:
 98        """Unpack a batch, to within this quantizer's precision.
 99
100        :param data: The encoding produced by :meth:`encode` at the same scale.
101        :returns: The readings, in order.
102        :raises PamojaError: If the buffer is malformed.
103        """
104        return self._native.decode(bytes(data))