pamoja_codec/
delta.rs

1//! Compact batch encoding for metered links: delta plus variable-length integers.
2//!
3//! On a long-range radio or a metered cellular link, every byte costs power or money,
4//! so it pays to send a batch of readings in as few bytes as possible rather than one
5//! full-width value at a time. The functions here encode a sequence of integers as a
6//! starting value followed by the differences between consecutive values, each
7//! written as a variable-length integer. A slowly changing signal - a temperature, a
8//! tank level, a battery voltage - then costs about one byte per sample instead of
9//! eight, with no loss. [`Quantizer`] extends this to `f32` readings by rounding each
10//! to a fixed precision first.
11
12use alloc::format;
13use alloc::vec::Vec;
14
15use pamoja_core::{Error, Result};
16
17/// The magnitude past which a quantized reading no longer fits the `i64` it is packed
18/// as, kept a little inside `i64::MAX` so the conversion never saturates.
19const QUANTIZED_LIMIT: f32 = 9.2e18;
20
21// Writes an unsigned integer as LEB128: seven bits per byte, high bit as a continue
22// flag.
23fn write_uvarint(mut value: u64, out: &mut Vec<u8>) {
24    loop {
25        let mut byte = (value & 0x7f) as u8;
26        value >>= 7;
27        if value != 0 {
28            byte |= 0x80;
29        }
30        out.push(byte);
31        if value == 0 {
32            break;
33        }
34    }
35}
36
37// Reads a LEB128 unsigned integer, advancing `pos`. The tenth byte holds bit 63 alone,
38// so anything more in it is a value past 64 bits.
39fn read_uvarint(bytes: &[u8], pos: &mut usize) -> Result<u64> {
40    let mut result = 0u64;
41    let mut shift = 0u32;
42    loop {
43        let byte = *bytes
44            .get(*pos)
45            .ok_or_else(|| Error::Codec("the batch ends part-way through a value".into()))?;
46        *pos += 1;
47        if shift == 63 && byte > 1 {
48            return Err(Error::Codec(
49                "a value in the batch does not fit in 64 bits".into(),
50            ));
51        }
52        result |= u64::from(byte & 0x7f) << shift;
53        if byte & 0x80 == 0 {
54            return Ok(result);
55        }
56        shift += 7;
57    }
58}
59
60// Rounds half away from zero, as `f32::round` does, with only `core`. Every `f32` of
61// magnitude 2^23 or more is already whole, and below that the fraction is exact.
62fn round(value: f32) -> f32 {
63    const WHOLE: f32 = 8_388_608.0;
64    if value.is_nan() || value <= -WHOLE || value >= WHOLE {
65        return value;
66    }
67    let truncated = value as i32 as f32;
68    let fraction = value - truncated;
69    if fraction >= 0.5 {
70        truncated + 1.0
71    } else if fraction <= -0.5 {
72        truncated - 1.0
73    } else {
74        truncated
75    }
76}
77
78// Maps a signed integer to an unsigned one whose size grows with magnitude, so small
79// negative deltas stay small (protobuf zigzag).
80fn zigzag(value: i64) -> u64 {
81    ((value << 1) ^ (value >> 63)) as u64
82}
83
84fn unzigzag(value: u64) -> i64 {
85    ((value >> 1) as i64) ^ -((value & 1) as i64)
86}
87
88/// Encodes a batch of integer samples as a starting value plus variable-length deltas.
89///
90/// # Arguments
91///
92/// * `samples` - the integers to encode, in order.
93///
94/// # Returns
95///
96/// The compact encoding. A slowly changing series is far smaller than the eight bytes
97/// per sample a raw encoding would use.
98///
99/// # Examples
100///
101/// ```
102/// use pamoja_codec::{decode_deltas, encode_deltas};
103///
104/// let samples = [1000, 1001, 1003, 1002];
105/// let bytes = encode_deltas(&samples);
106/// assert!(bytes.len() < samples.len() * 8); // far smaller than eight bytes each
107/// assert_eq!(decode_deltas(&bytes).unwrap(), samples);
108/// ```
109pub fn encode_deltas(samples: &[i64]) -> Vec<u8> {
110    let mut out = Vec::new();
111    write_uvarint(samples.len() as u64, &mut out);
112    let mut previous = 0i64;
113    for &sample in samples {
114        let delta = sample.wrapping_sub(previous);
115        write_uvarint(zigzag(delta), &mut out);
116        previous = sample;
117    }
118    out
119}
120
121/// Decodes a batch encoded by [`encode_deltas`].
122///
123/// # Arguments
124///
125/// * `bytes` - the encoded batch.
126///
127/// # Returns
128///
129/// The decoded samples, in order.
130///
131/// # Errors
132///
133/// Returns [`Error::Codec`](pamoja_core::Error::Codec) if `bytes` ends part-way
134/// through a value, holds a value past 64 bits, or carries bytes after the last
135/// sample, which a batch cut short or run into another would.
136pub fn decode_deltas(bytes: &[u8]) -> Result<Vec<i64>> {
137    let mut pos = 0;
138    let count = read_uvarint(bytes, &mut pos)?;
139    // The count is untrusted, so the vector grows with the data actually present
140    // rather than pre-allocating from a claimed length.
141    let mut samples = Vec::new();
142    let mut previous = 0i64;
143    for _ in 0..count {
144        let delta = unzigzag(read_uvarint(bytes, &mut pos)?);
145        let sample = previous.wrapping_add(delta);
146        samples.push(sample);
147        previous = sample;
148    }
149    match bytes.len() - pos {
150        0 => Ok(samples),
151        extra => Err(Error::Codec(format!(
152            "{extra} bytes follow the batch's last sample"
153        ))),
154    }
155}
156
157/// Packs a batch of `f32` readings into a compact byte form for a metered link.
158///
159/// A quantizer rounds each reading to a fixed precision - set by the `scale`, where
160/// `100.0` keeps two decimal places - turns it into an integer, and delta-encodes the
161/// batch with [`encode_deltas`]. This is lossy by exactly the rounding step, which is
162/// the right trade for a cheap sensor on an expensive link: a temperature to the
163/// nearest hundredth of a degree costs a byte or two per sample instead of four.
164/// The same `scale` must be used to encode and decode, and nothing in the bytes
165/// records it.
166///
167/// A reading that is not a number, or is infinite, is refused rather than packed:
168/// the format has no way to say "no reading", and a dead sensor's `NaN` sent as a
169/// real value would be worse than an error.
170///
171/// # Examples
172///
173/// ```
174/// use pamoja_codec::Quantizer;
175///
176/// // Quantize to 0.1 precision and pack a slowly-rising series.
177/// let quantizer = Quantizer::new(10.0);
178/// let readings = [20.0, 20.1, 20.2, 20.3];
179/// let packed = quantizer.encode(&readings).unwrap();
180/// assert!(packed.len() < readings.len() * 4); // smaller than four bytes per reading
181///
182/// let restored = quantizer.decode(&packed).unwrap();
183/// assert!((restored[2] - 20.2).abs() < 0.05);
184/// ```
185#[derive(Clone, Copy, Debug)]
186pub struct Quantizer {
187    scale: f32,
188}
189
190impl Quantizer {
191    /// Creates a quantizer with the given precision scale.
192    ///
193    /// # Arguments
194    ///
195    /// * `scale` - the multiplier applied before rounding; `100.0` keeps two decimal
196    ///   places. It must be a positive, finite number, which
197    ///   [`encode`](Quantizer::encode) and [`decode`](Quantizer::decode) check.
198    ///
199    /// # Returns
200    ///
201    /// The quantizer.
202    pub fn new(scale: f32) -> Self {
203        Self { scale }
204    }
205
206    /// Quantizes and delta-encodes a batch of readings.
207    ///
208    /// # Arguments
209    ///
210    /// * `readings` - the readings to pack, in order.
211    ///
212    /// # Returns
213    ///
214    /// The compact encoding of the batch.
215    ///
216    /// # Errors
217    ///
218    /// Returns [`Error::Codec`](pamoja_core::Error::Codec) if the scale is not a
219    /// positive, finite number, if a reading is not a number or is infinite, or if a
220    /// reading at this scale is past what the batch can carry, about 9.2e18.
221    pub fn encode(&self, readings: &[f32]) -> Result<Vec<u8>> {
222        self.check_scale()?;
223        let mut samples = Vec::with_capacity(readings.len());
224        for (index, &reading) in readings.iter().enumerate() {
225            if !reading.is_finite() {
226                return Err(Error::Codec(format!(
227                    "reading {index} is {reading}, which cannot be quantized"
228                )));
229            }
230            let scaled = round(reading * self.scale);
231            if !scaled.is_finite() || scaled <= -QUANTIZED_LIMIT || scaled >= QUANTIZED_LIMIT {
232                return Err(Error::Codec(format!(
233                    "reading {index} is {reading}, too large for a scale of {}",
234                    self.scale
235                )));
236            }
237            samples.push(scaled as i64);
238        }
239        Ok(encode_deltas(&samples))
240    }
241
242    /// Decodes a batch back into readings, to within the quantizer's precision.
243    ///
244    /// # Arguments
245    ///
246    /// * `bytes` - the encoding produced by [`encode`](Quantizer::encode) with the
247    ///   same scale.
248    ///
249    /// # Returns
250    ///
251    /// The decoded readings, in order.
252    ///
253    /// # Errors
254    ///
255    /// Returns [`Error::Codec`](pamoja_core::Error::Codec) if the scale is not a
256    /// positive, finite number, or `bytes` is malformed.
257    pub fn decode(&self, bytes: &[u8]) -> Result<Vec<f32>> {
258        self.check_scale()?;
259        let samples = decode_deltas(bytes)?;
260        Ok(samples
261            .iter()
262            .map(|&sample| sample as f32 / self.scale)
263            .collect())
264    }
265
266    /// Refuses a scale that cannot round a reading to anything meaningful.
267    fn check_scale(&self) -> Result<()> {
268        if self.scale.is_finite() && self.scale > 0.0 {
269            Ok(())
270        } else {
271            Err(Error::Codec(format!(
272                "a quantizer's scale must be a positive, finite number, not {}",
273                self.scale
274            )))
275        }
276    }
277}
278
279#[cfg(test)]
280mod tests {
281    use super::*;
282
283    #[test]
284    fn an_integer_batch_round_trips() {
285        for samples in [
286            vec![],
287            vec![0],
288            vec![42],
289            vec![-5, -4, -3, -2],
290            vec![1000, 1001, 1003, 1002, 999],
291            vec![i64::MIN, 0, i64::MAX],
292        ] {
293            let bytes = encode_deltas(&samples);
294            assert_eq!(decode_deltas(&bytes).expect("decode"), samples);
295        }
296    }
297
298    #[test]
299    fn a_slow_series_is_far_smaller_than_raw() {
300        let samples: Vec<i64> = (0..100).map(|i| 5000 + i).collect();
301        let bytes = encode_deltas(&samples);
302        // Each delta is one, so a sample costs about a byte instead of eight.
303        assert!(bytes.len() < samples.len() * 2);
304    }
305
306    #[test]
307    fn truncated_bytes_are_a_codec_error() {
308        // Claims three samples but supplies none.
309        match decode_deltas(&[3]) {
310            Err(Error::Codec(reason)) => {
311                assert_eq!(reason, "the batch ends part-way through a value");
312            }
313            other => panic!("a batch cut short decoded: {other:?}"),
314        }
315    }
316
317    #[test]
318    fn a_value_past_64_bits_is_refused_rather_than_truncated() {
319        let mut largest = Vec::new();
320        write_uvarint(u64::MAX, &mut largest);
321        assert_eq!(largest.len(), 10);
322        assert_eq!(largest[9], 0x01);
323        let mut pos = 0;
324        assert_eq!(
325            read_uvarint(&largest, &mut pos).expect("u64::MAX"),
326            u64::MAX
327        );
328
329        for tenth in [0x02, 0x7f, 0x81] {
330            let mut bytes = largest.clone();
331            bytes[9] = tenth;
332            bytes.push(0x00);
333            let mut pos = 0;
334            match read_uvarint(&bytes, &mut pos) {
335                Err(Error::Codec(reason)) => {
336                    assert_eq!(reason, "a value in the batch does not fit in 64 bits");
337                }
338                other => panic!("a tenth byte of {tenth:#04x} read as {other:?}"),
339            }
340        }
341    }
342
343    #[test]
344    fn rounding_matches_the_standard_library_at_its_edges() {
345        for value in [
346            0.0,
347            -0.0,
348            0.5,
349            -0.5,
350            1.5,
351            -1.5,
352            2.5,
353            -2.5,
354            0.499_999_97,
355            -0.499_999_97,
356            8_388_607.5,
357            -8_388_607.5,
358            8_388_608.0,
359            16_777_217.0,
360            f32::MAX,
361            f32::MIN,
362            f32::MIN_POSITIVE,
363            f32::INFINITY,
364            f32::NEG_INFINITY,
365        ] {
366            assert_eq!(round(value), value.round(), "rounding {value}");
367        }
368        assert!(round(f32::NAN).is_nan());
369    }
370
371    proptest::proptest! {
372        #[test]
373        fn rounding_matches_the_standard_library(value in proptest::num::f32::ANY) {
374            let ours = round(value);
375            let theirs = value.round();
376            proptest::prop_assert!(ours == theirs || (ours.is_nan() && theirs.is_nan()));
377        }
378    }
379
380    #[test]
381    fn varints_cover_the_boundaries() {
382        for value in [0u64, 1, 127, 128, 16_383, 16_384, u64::MAX] {
383            let mut out = Vec::new();
384            write_uvarint(value, &mut out);
385            let mut pos = 0;
386            assert_eq!(read_uvarint(&out, &mut pos).expect("read"), value);
387            assert_eq!(pos, out.len());
388        }
389    }
390
391    #[test]
392    fn varints_match_the_canonical_leb128_encodings() {
393        let cases: [(u64, &[u8]); 4] = [
394            (0, &[0x00]),
395            (1, &[0x01]),
396            (128, &[0x80, 0x01]),
397            (300, &[0xAC, 0x02]),
398        ];
399        for (value, encoded) in cases {
400            let mut out = Vec::new();
401            write_uvarint(value, &mut out);
402            assert_eq!(out, encoded, "varint of {value}");
403        }
404    }
405
406    #[test]
407    fn zigzag_matches_the_protobuf_mapping() {
408        let cases: [(i64, u64); 6] = [
409            (0, 0),
410            (-1, 1),
411            (1, 2),
412            (-2, 3),
413            (2, 4),
414            (2_147_483_647, 4_294_967_294),
415        ];
416        for (signed, unsigned) in cases {
417            assert_eq!(zigzag(signed), unsigned, "zigzag of {signed}");
418            assert_eq!(unzigzag(unsigned), signed, "unzigzag of {unsigned}");
419        }
420    }
421
422    #[test]
423    fn a_quantizer_round_trips_within_its_precision() {
424        let quantizer = Quantizer::new(100.0);
425        let readings = [4.0, 4.62, 5.13, 4.77, 3.98];
426        let packed = quantizer.encode(&readings).expect("encode");
427        let restored = quantizer.decode(&packed).expect("decode");
428        for (original, decoded) in readings.iter().zip(&restored) {
429            assert!((original - decoded).abs() <= 0.005 + f32::EPSILON);
430        }
431    }
432
433    #[test]
434    fn a_reading_that_is_not_a_number_is_refused_rather_than_packed_as_zero() {
435        let quantizer = Quantizer::new(100.0);
436        for (readings, index) in [
437            (vec![20.0, f32::NAN], 1),
438            (vec![f32::INFINITY], 0),
439            (vec![20.0, 20.1, f32::NEG_INFINITY], 2),
440        ] {
441            match quantizer.encode(&readings) {
442                Err(Error::Codec(reason)) => {
443                    assert!(
444                        reason.starts_with(&format!("reading {index} is")),
445                        "{reason}"
446                    );
447                }
448                other => panic!("{readings:?} was packed: {other:?}"),
449            }
450        }
451    }
452
453    #[test]
454    fn a_reading_too_large_for_its_scale_is_refused() {
455        let quantizer = Quantizer::new(1e10);
456        match quantizer.encode(&[1e10]) {
457            Err(Error::Codec(reason)) => assert!(reason.contains("too large"), "{reason}"),
458            other => panic!("an overflowing reading was packed: {other:?}"),
459        }
460        assert!(Quantizer::new(1e10).encode(&[1.0]).is_ok());
461    }
462
463    #[test]
464    fn a_scale_that_is_not_a_positive_number_is_refused() {
465        for scale in [0.0, -100.0, f32::NAN, f32::INFINITY] {
466            let quantizer = Quantizer::new(scale);
467            assert!(matches!(quantizer.encode(&[1.0]), Err(Error::Codec(_))));
468            assert!(matches!(quantizer.decode(&[0]), Err(Error::Codec(_))));
469        }
470    }
471
472    #[test]
473    fn bytes_after_the_last_sample_are_refused() {
474        let mut bytes = encode_deltas(&[1, 2, 3]);
475        bytes.extend_from_slice(&encode_deltas(&[4]));
476        match decode_deltas(&bytes) {
477            Err(Error::Codec(reason)) => {
478                assert_eq!(reason, "2 bytes follow the batch's last sample");
479            }
480            other => panic!("two batches run together decoded: {other:?}"),
481        }
482    }
483}
pamoja_codec/delta.rs

pamoja_codec/
delta.rs
