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 ValueError: If the value holds a NaN or an infinity, which JSON
 29        has no way to write.
 30    :raises TypeError: If the value holds something else JSON cannot write,
 31        such as a set.
 32    :raises PamojaError: If the bytes given are not a valid JSON document.
 33    """
 34    if isinstance(value, (bytes, bytearray, memoryview)):
 35        document = bytes(value)
 36    else:
 37        document = json.dumps(value, allow_nan=False).encode("utf-8")
 38    return _json_to_cbor_bytes(document)
 39
 40
 41def from_cbor(cbor: bytes) -> Any:
 42    """Decode a CBOR document back into an ordinary Python value.
 43
 44    :param cbor: The CBOR document to decode.
 45    :returns: The decoded value.
 46    :raises PamojaError: If the document is malformed, or holds a construct with
 47        no JSON equivalent such as a non-string map key.
 48    """
 49    return json.loads(_cbor_to_json_bytes(bytes(cbor)).decode("utf-8"))
 50
 51
 52def pack_samples(samples: Sequence[int]) -> bytes:
 53    """Delta-encode a series of integer samples into a compact buffer.
 54
 55    :param samples: The samples, in order.
 56    :returns: The packed encoding, far smaller than the samples for a
 57        slow-moving series.
 58    :raises TypeError: If a sample is not an ``int``.
 59    :raises OverflowError: If a sample does not fit in 64 bits.
 60    """
 61    return _encode_delta_samples(list(samples))
 62
 63
 64def unpack_samples(data: bytes) -> list[int]:
 65    """Unpack a buffer produced by :func:`pack_samples`.
 66
 67    :param data: The packed encoding.
 68    :returns: The samples, in order.
 69    :raises PamojaError: If the buffer is malformed or carries bytes after its
 70        last sample.
 71    """
 72    return _decode_delta_samples(bytes(data))
 73
 74
 75class Quantizer:
 76    """Packs float readings to a fixed precision, for a link charging per byte.
 77
 78    Example::
 79
 80        quantizer = Quantizer(100)  # keep two decimal places
 81        packed = quantizer.encode([20.0, 20.1, 20.2])
 82        quantizer.decode(packed)  # [20.0, 20.1, 20.2], to within 0.01
 83    """
 84
 85    __slots__ = ("_native",)
 86
 87    def __init__(self, scale: float) -> None:
 88        """Create a quantizer at the given precision.
 89
 90        :param scale: The multiplier applied before rounding; ``100`` keeps two
 91            decimal places. Must be positive and finite.
 92        :raises ValueError: If the scale is not positive and finite.
 93        """
 94        self._native = _NativeQuantizer(scale)
 95
 96    def encode(self, readings: Sequence[float]) -> bytes:
 97        """Quantize and pack a batch of readings.
 98
 99        :param readings: The readings, in order.
100        :returns: The packed encoding.
101        :raises PamojaError: If a reading is not a number, is infinite, or is too
102            large for the scale; the format has no way to carry a missing reading.
103        """
104        return self._native.encode(list(readings))
105
106    def decode(self, data: bytes) -> list[float]:
107        """Unpack a batch, to within this quantizer's precision.
108
109        :param data: The encoding produced by :meth:`encode` at the same scale.
110        :returns: The readings, in order.
111        :raises PamojaError: If the buffer is malformed or carries bytes after its
112            last reading.
113        """
114        return self._native.decode(bytes(data))