6 Edge Data: Sampling and Statistical Compression

6.1 Overview

This section: Overview — This first route protects signal evidence through sampling and compares lossless, statistical, and transform reduction.

This first route protects signal evidence through sampling and compares lossless, statistical, and transform reduction.

This is part 1 of 2. Continue with Edge Data: Semantic Compression and Pipelines for the second focused route.

6.2 Start With the Story

This section: Start With the Story — Picture a motor whose vibration changes faster than a slow sensor check can see.

Keep the Shape Before Shrinking the Data

Picture a motor whose vibration changes faster than a slow sensor check can see. If the device measures too rarely, no later data trick can rebuild the missing motion.

Sampling means taking readings at chosen times. The sampling rate is how many readings are taken in one second. Start with the fastest change that matters. Measure often enough to show that change before you remove or combine any values.

Compression means using fewer bits to carry useful information. First decide what the receiver must still know. Exact records may need a lossless method, which can rebuild every bit. Trends may allow summaries. Alarms may allow the device to send only a named event plus enough context to check it.

Compare the smaller result with the original. Test normal work, sharp changes, noise, and a lost link. Record the method, settings, time window, and error.

Start with one short raw record. Mark the fastest useful change. Mark the slow trend. Mark known noise. Mark any gap. Choose the rate. Take the sample. Check the shape. Keep the time. Keep the unit. Keep the test state.

Now make a smaller copy. Count its bits. Check its shape. Check its peak. Check its mean. Check its alarm. Rebuild it when the method allows that. Measure the error. Repeat with a sharp event. Repeat with a quiet event. Repeat after a restart.

Test the full field path. Fill the local store. Remove the link. Keep taking samples. Restore the link. Send old data in order. Mark repeats. Mark lost blocks. Check the far copy. Check the power used. Check the time needed.

Write one clear rule for use. State which facts may change. State which facts must not change. State the largest allowed error. State who owns the setting. State what change forces a new test. Keep the raw test set so a later method can be judged on the same evidence.

The simple rule has limits. Less data saves power and link use, but it can also hide faults.

Use Practitioner to choose rates and reduction steps. Use Under the Hood for signal, transform, battery, and error limits.

Choose a sample rate and reduction step by testing whether faults remain visible after data is discarded.

6.3 Learning Objectives

This section: Learning Objectives

By the end of this chapter, you will be able to:

• Apply Nyquist Theorem: Calculate appropriate sampling rates for different sensor types
• Implement Data Reduction Techniques: Use aggregation, compression, event-based reporting, and delta encoding
• Select Compression Algorithms: Choose optimal algorithms based on data type and edge device constraints
• Avoid Common Pitfalls: Prevent sampling aliasing, buffer overflow, and rate mismatch errors

6.4 Prerequisites

This section: Prerequisites

Before diving into this chapter, you should be familiar with:

• Edge Data Acquisition: Architecture: Understanding device categories and data generation patterns
• Basic signal processing concepts: Familiarity with frequency and time-domain representations
• Python programming: Code examples use Python for data processing

6.5 Nyquist Sampling Rate

This section: Nyquist Sampling Rate

6.6 Nyquist Rate Calculator

This section: Nyquist Rate Calculator — Use this calculator to explore how motor speed, harmonic order, and safety factor affect the required sampling rate and resulting data volume.

Use this calculator to explore how motor speed, harmonic order, and safety factor affect the required sampling rate and resulting data volume.

6.7 Interactive: Nyquist Sampling Animation

This section: Interactive: Nyquist Sampling Animation

6.8 Edge Data Reduction Techniques

This section: Edge Data Reduction Techniques

Before transmitting data to the cloud, edge devices can apply several reduction strategies:

1. Aggregation: Compute statistics over time windows (mean, min, max, variance)
2. Compression: Apply lossless (ZIP) or lossy (threshold-based) compression
3. Event-based reporting: Only transmit when values exceed thresholds
4. Delta encoding: Send only changes from previous values

# Example: Edge aggregation for temperature sensor
class EdgeAggregator:
    def __init__(self, window_size=60):  # 60 samples = 1 minute at 1 Hz
        self.window_size = window_size
        self.buffer = []

    def add_sample(self, value):
        self.buffer.append(value)
        if len(self.buffer) >= self.window_size:
            return self.compute_summary()
        return None

    def compute_summary(self):
        summary = {
            "min": min(self.buffer),
            "max": max(self.buffer),
            "mean": sum(self.buffer) / len(self.buffer),
            "samples": len(self.buffer)
        }
        self.buffer = []
        return summary

# Usage: Send 1 summary per minute instead of 60 raw samples
aggregator = EdgeAggregator(window_size=60)
for temp_reading in sensor_stream:
    summary = aggregator.add_sample(temp_reading)
    if summary:
        transmit_to_cloud(summary)  # 60x bandwidth reduction

6.9 Battery vs Aggregation Calculator

This section: Battery vs Aggregation Calculator — Adjust the parameters below to see how edge aggregation affects battery life and bandwidth for a LoRa-connected sensor.

Adjust the parameters below to see how edge aggregation affects battery life and bandwidth for a LoRa-connected sensor.

# Resample high-frequency data to match low-frequency
vibration_1hz = vibration_100hz.resample('1S').mean()
# Or upsample low-frequency with interpolation
temp_100hz = temp_1hz.resample('10ms').interpolate(method='linear')
# Then merge on aligned timestamps
merged = pd.merge_asof(vibration_1hz, temp_1hz, on='timestamp', tolerance=pd.Timedelta('500ms'))

# Measured peak accepted arrival rate during the outage
samples_per_second = 10
max_outage_duration_seconds = 3600  # 1 hour
safety_margin = 1.5

min_buffer_size = samples_per_second * max_outage_duration_seconds * safety_margin
# = 10 * 3600 * 1.5 = 54,000 samples

# If memory-constrained, implement tiered retention:
# - Last 5 minutes: Full resolution
# - 5-60 minutes: 10x downsampled
# - Beyond 60 minutes: Statistical summary only

+Checkpoint: Reduce Before Transmit

You now know:

• Aggregation can replace 60 raw samples with 1 summary when min, max, mean, and count preserve the decision evidence.
• Transmission dominates many edge power budgets: the worked LoRa case moves from 35 days to 496 days by sending 4 summaries per hour.
• Buffer plans need peak-load math, including the 54,000-sample outage calculation, before teams trust compressed or tiered retention data.

6.10 Compression Algorithms Deep Dive

This section: Compression Algorithms Deep Dive

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

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)

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
# Selected spectral features remain available; validate fault detection on representative traces

6.15 Continue to Part 2

This section: Continue to Part 2 — Continue with Edge Data: Semantic Compression and Pipelines.

Continue with Edge Data: Semantic Compression and Pipelines.