6 Edge Data: Sampling and Statistical Compression

6.11 Compression Algorithm Categories

6.12 Lossless Compression: DEFLATE/GZIP

Standard lossless compression works well for structured IoT data:

import gzip
import json

def compress_batch(readings: list[dict]) -> bytes:
    """
    Compress a batch of sensor readings losslessly.
    Typical compression: 3-5x for JSON sensor data.
    """
    json_str = json.dumps(readings)
    compressed = gzip.compress(json_str.encode('utf-8'), compresslevel=6)
    return compressed

# Example: 100 temperature readings
readings = [{"ts": 1704067200 + i, "v": 22.5 + (i % 10) * 0.1} for i in range(100)]
raw_size = len(json.dumps(readings).encode())       # ~4,500 bytes
compressed_size = len(compress_batch(readings))     # ~1,200 bytes
# Compression ratio: 3.75:1

Performance characteristics:

When to use: Audit trails, compliance data, any data that may be queried in original form. Do not use for real-time streams on constrained MCUs.

6.13 Lossy Statistical: Aggregation Windows

Compute statistics over time windows, discard raw samples:

import statistics
from dataclasses import dataclass
from typing import Optional

@dataclass
class AggregatedWindow:
    timestamp: int      # Window start
    count: int          # Number of samples
    mean: float
    min_val: float
    max_val: float
    std_dev: float
    p95: Optional[float] = None  # Optional percentile

class WindowAggregator:
    def __init__(self, window_seconds: int = 60):
        self.window_seconds = window_seconds
        self.buffer: list[float] = []
        self.window_start: Optional[int] = None

    def add_sample(self, timestamp: int, value: float) -> Optional[AggregatedWindow]:
        if self.window_start is None:
            self.window_start = timestamp

        # Check if window complete
        if timestamp - self.window_start >= self.window_seconds:
            result = self._compute_aggregate()
            self.buffer = [value]
            self.window_start = timestamp
            return result

        self.buffer.append(value)
        return None

    def _compute_aggregate(self) -> AggregatedWindow:
        sorted_buffer = sorted(self.buffer)
        p95_idx = int(len(sorted_buffer) * 0.95)

        return AggregatedWindow(
            timestamp=self.window_start,
            count=len(self.buffer),
            mean=statistics.mean(self.buffer),
            min_val=min(self.buffer),
            max_val=max(self.buffer),
            std_dev=statistics.stdev(self.buffer) if len(self.buffer) > 1 else 0,
            p95=sorted_buffer[p95_idx] if p95_idx < len(sorted_buffer) else None
        )

# Example: 1 Hz sensor, 60-second windows
# Input: 60 samples x 8 bytes = 480 bytes
# Output: 1 aggregate x 48 bytes = 48 bytes
# Compression ratio: 10:1
# Preserved: Trend (mean), anomaly detection (min/max/std), health (count)

Information loss analysis:

When to use: Temperature, humidity, air quality - any slowly changing signal where trends matter more than exact samples.

6.14 Lossy Transform: FFT-Based Compression

Transform to frequency domain, keep only significant components:

import numpy as np
from dataclasses import dataclass

@dataclass
class FFTCompressed:
    timestamp: int
    sample_rate: float
    duration: float
    frequencies: list[float]    # Top N frequency components
    magnitudes: list[float]     # Corresponding magnitudes
    phases: list[float]         # Phase angles for reconstruction

def fft_compress(samples: np.ndarray, sample_rate: float,
                 timestamp: int, top_n: int = 10) -> FFTCompressed:
    """
    Compress time-series data using FFT, keeping top N frequency components.

    Typical compression: 100:1 to 500:1 depending on signal complexity.
    Best for: Vibration, audio, periodic signals.
    """
    # Compute FFT
    fft_result = np.fft.rfft(samples)
    freqs = np.fft.rfftfreq(len(samples), 1/sample_rate)

    # Get magnitudes and find top N (excluding DC component)
    magnitudes = np.abs(fft_result[1:])  # Skip DC
    phases = np.angle(fft_result[1:])
    freqs = freqs[1:]

    # Select top N by magnitude
    top_indices = np.argsort(magnitudes)[-top_n:]

    return FFTCompressed(
        timestamp=timestamp,
        sample_rate=sample_rate,
        duration=len(samples) / sample_rate,
        frequencies=freqs[top_indices].tolist(),
        magnitudes=magnitudes[top_indices].tolist(),
        phases=phases[top_indices].tolist()
    )

def fft_decompress(compressed: FFTCompressed, num_samples: int) -> np.ndarray:
    """
    Reconstruct signal from FFT components (lossy reconstruction).
    """
    t = np.linspace(0, compressed.duration, num_samples)
    signal = np.zeros(num_samples)

    for freq, mag, phase in zip(compressed.frequencies,
                                 compressed.magnitudes,
                                 compressed.phases):
        signal += mag * np.cos(2 * np.pi * freq * t + phase)

    return signal

# Example: Vibration sensor, 1 second at 1000 Hz
samples = np.sin(2*np.pi*50*np.linspace(0, 1, 1000))  # 50 Hz signal
samples += 0.3 * np.sin(2*np.pi*150*np.linspace(0, 1, 1000))  # 150 Hz harmonic

# Input: 1000 samples x 4 bytes = 4000 bytes
# Output: 10 freq-mag-phase tuples x 12 bytes = 120 bytes + 20 bytes metadata
# Compression ratio: ~30:1

compressed = fft_compress(samples, 1000.0, 1704067200, top_n=10)
reconstructed = fft_decompress(compressed, 1000)

# Reconstruction error for this example: ~5% RMS
# Bearing fault detection: Still works (frequency peaks preserved)

When to use: Vibration analysis, acoustic monitoring, any signal where frequency content matters more than exact waveform.