7 Edge Data: Semantic Compression and Pipelines

7.1 Start With the Situation

This section: Start With the Situation — A single rule for cutting samples may still send data the app does not need.

A single rule for cutting samples may still send data the app does not need. The team must say which events matter. It must measure what semantic compression removes, then check that the edge pipeline still supports the same decisions.

7.2 Overview

This section: Overview — This route moves from event extraction into algorithm selection, factory validation, and complete edge pipelines.

This route moves from event extraction into algorithm selection, factory validation, and complete edge pipelines.

This is part 2 of 2. Review Edge Data: Sampling and Statistical Compression when you need the first route.

7.3 Learning Objectives

This section: Learning Objectives

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

• design semantic event extraction
• select compression from evidence needs
• validate a multi-sensor edge pipeline

from dataclasses import dataclass
from enum import Enum
from typing import Optional

class EventType(Enum):
    THRESHOLD_EXCEEDED = "threshold_exceeded"
    ANOMALY_DETECTED = "anomaly_detected"
    STATE_CHANGE = "state_change"
    PERIODIC_SUMMARY = "periodic_summary"

@dataclass
class SemanticEvent:
    timestamp: int
    device_id: str
    event_type: EventType
    value: float
    context: dict  # Additional info (threshold, previous state, etc.)

class SemanticCompressor:
    def __init__(self, device_id: str, threshold_high: float,
                 threshold_low: float, anomaly_std_factor: float = 3.0):
        self.device_id = device_id
        self.threshold_high = threshold_high
        self.threshold_low = threshold_low
        self.anomaly_std_factor = anomaly_std_factor
        self.history: list[float] = []
        self.last_state: Optional[str] = None
        self.summary_count = 0
        self.summary_sum = 0.0

    def process_sample(self, timestamp: int, value: float) -> list[SemanticEvent]:
        """
        Process a sample and return events (if any).
        Most samples produce NO events - that's the compression.
        """
        events = []

        # Update history for anomaly detection
        self.history.append(value)
        if len(self.history) > 100:
            self.history.pop(0)

        # Track for periodic summary
        self.summary_count += 1
        self.summary_sum += value

        # Check threshold crossing
        current_state = "normal"
        if value > self.threshold_high:
            current_state = "high"
        elif value < self.threshold_low:
            current_state = "low"

        if current_state != self.last_state and self.last_state is not None:
            events.append(SemanticEvent(
                timestamp=timestamp,
                device_id=self.device_id,
                event_type=EventType.STATE_CHANGE,
                value=value,
                context={
                    "previous_state": self.last_state,
                    "new_state": current_state
                }
            ))
        self.last_state = current_state

        # Check for statistical anomaly
        if len(self.history) >= 20:
            mean = sum(self.history) / len(self.history)
            std = (sum((x - mean)**2 for x in self.history) / len(self.history)) ** 0.5
            if std > 0 and abs(value - mean) > self.anomaly_std_factor * std:
                events.append(SemanticEvent(
                    timestamp=timestamp,
                    device_id=self.device_id,
                    event_type=EventType.ANOMALY_DETECTED,
                    value=value,
                    context={
                        "mean": mean,
                        "std": std,
                        "z_score": (value - mean) / std
                    }
                ))

        return events

    def get_periodic_summary(self, timestamp: int) -> SemanticEvent:
        """Call every N minutes to send a heartbeat/summary."""
        avg = self.summary_sum / self.summary_count if self.summary_count > 0 else 0
        event = SemanticEvent(
            timestamp=timestamp,
            device_id=self.device_id,
            event_type=EventType.PERIODIC_SUMMARY,
            value=avg,
            context={
                "sample_count": self.summary_count,
                "period_seconds": 300  # 5 minutes
            }
        )
        self.summary_count = 0
        self.summary_sum = 0.0
        return event

# Example: Temperature sensor, 1 sample/second
# Normal operation: 0 events per sample
# State change: 1 event (~100 bytes)
# 5-minute summary: 1 event (~80 bytes)
#
# Input: 300 samples x 8 bytes = 2400 bytes per 5 minutes
# Output: 1 summary + maybe 0-2 events = 80-280 bytes
# Compression ratio: 10:1 to 30:1 (varies by activity)

7.4 Algorithm Selection Decision Tree

This section: Algorithm Selection Decision Tree — The decision tree in Figure 7.1 turns the preceding methods into a selection sequence, beginning with what the signal and downstream decision are allowed to lose.

The decision tree in Figure 7.1 turns the preceding methods into a selection sequence, beginning with what the signal and downstream decision are allowed to lose.

+

Read Figure 7.1 from signal type to fidelity requirement and only then to compute limits. Event-like data can often become sparse semantic records; smooth numeric streams may support delta or window summaries; spectral decisions may justify selected frequency features; audit-grade reconstruction keeps the route lossless. The final resource branch checks whether memory, latency, and energy fit the edge device. This connects the tree to the chapter’s running narrative: compression is an evidence contract chosen from the downstream claim, then tested against device constraints and reconstruction needs.

7.5 Benchmark Results: ESP32 Edge Device

This section: Benchmark Results: ESP32 Edge Device

Real measurements on ESP32-WROOM-32 (240 MHz, 520KB RAM):

7.6 Raw JSON

This section: Raw JSON

• 1000 samples: -
• Compress time: 15 ms (serialize)
• Output size: 28,000 bytes
• Power: 2.4 mJ

7.7 GZIP-6

This section: GZIP-6 — GZIP keeps the data lossless, so the smaller output can support exact reconstruction when decompression reproduces the source bytes.

GZIP keeps the data lossless, so the smaller output can support exact reconstruction when decompression reproduces the source bytes. The size reduction must be weighed against compression time, memory use, and total energy for computation plus transmission. A smaller payload alone does not establish an energy advantage; measure the same workload on the target device before choosing the algorithm.

• 1000 samples: 28,000 bytes
• Compress time: 85 ms
• Output size: 8,200 bytes
• Power: Measure on the target ESP32 for the same 1,000-sample workload

7.8 Window aggregation

This section: Window aggregation — Window aggregation replaces individual samples with summary values, preserving trends while discarding waveform and frequency detail.

Window aggregation replaces individual samples with summary values, preserving trends while discarding waveform and frequency detail. Its short compression time does not include the wait for a complete summary window. Choose the retained statistics from the downstream decision, and include sample count, window duration, and timestamp precision so the result can be interpreted. Measure computation and transmission energy together before comparing battery costs.

• 1000 samples: 8 bytes/sample
• Compress time: 2 ms
• Output size: 48 bytes
• Power: Measure on the target ESP32 for the same 1,000-sample workload

7.9 FFT Top-10

This section: FFT Top-10 — Selected FFT bins reduce the payload by retaining chosen frequency features rather than the full waveform.

Selected FFT bins reduce the payload by retaining chosen frequency features rather than the full waveform. This can support frequency-based decisions, but discarded bins and time detail cannot be recovered. Validate the selected features on representative traces before relying on them for fault detection. Include sample rate, window size, and transmitted bin identifiers so the receiver can interpret the output. Compare computation and transmission energy together when assessing the operating cost.

• 1000 samples: 4 bytes/sample
• Compress time: 45 ms
• Output size: 140 bytes
• Power: 5.0 mJ

7.10 Semantic

This section: Semantic — Key insight: For battery-powered edge devices, window aggregation offers the best power efficiency.

• 1000 samples: 8 bytes/sample
• Compress time: 3 ms
• Output size: 0-100 bytes
• Power: 0.5 mJ

Key insight: For battery-powered edge devices, window aggregation offers the best power efficiency. FFT is valuable when frequency content matters, but the CPU cost is significant. Semantic compression is ideal for sparse event streams.

7.11 Memory Constraints on Edge Devices

This section: Memory Constraints on Edge Devices — Compression algorithms have memory overhead.

Compression algorithms have memory overhead. Consider carefully on constrained devices:

ESP32 recommendation: Use window aggregation or semantic compression as primary strategy. Reserve FFT for specific signals where frequency analysis is required.

7.12 Common Compression Pitfalls

This section: Common Compression Pitfalls

7.13 Industrial Edge Pipeline Check

This section: Industrial Edge Pipeline Check — The factory example turns the algorithm comparison into an operating choice: keep the frequency evidence, discard the raw waveform volume, and check latency against the maintenance requirement.

# Pseudocode for an edge processing pipeline (runs on ESP32 every 10 seconds)
def vibration_pipeline():
    # 1. Collect 10 seconds of data
    samples = collect_samples(rate=500, duration=10)  # 5000 samples

    # 2. Apply FFT (frequency-domain analysis)
    fft_result = numpy.fft.rfft(samples)  # -> 2501 nonnegative-frequency bins

    # 3. Extract critical frequency bins
    #    Bin width = 500 Hz / 5000 samples = 0.1 Hz per bin
    #    Bin index = frequency / bin_width
    bin_width = 500.0 / 5000  # 0.1 Hz per bin
    target_freqs = [50, 65, 80, 95, 110, 125, 140, 155, 170, 200]  # Hz
    bins = [fft_result[int(f / bin_width)] for f in target_freqs]
    # Ten complex FFT coefficients; select frequencies against fault traces.

    # 4. Quantize calibrated magnitudes to uint16 before transmission.
    # Pseudocode: calibrated_range defines magnitude units; saturate out-of-range values
    # and send scale/units metadata so the receiver can interpret each magnitude.
    magnitudes_u16 = quantize_to_uint16([abs(v) for v in bins], calibrated_range)
    transmit_to_cloud(magnitudes_u16)  # 10 x 2 bytes = 20 bytes vs 10,000 bytes

    return magnitudes_u16

# Data reduction: 5000 int16 samples -> ten uint16 magnitudes = 500x by payload bytes

The factory example turns the algorithm comparison into an operating choice: keep the frequency evidence, discard the raw waveform volume, and check latency against the maintenance requirement.

+Checkpoint: Factory Vibration Choice

You now know:

• The naive 50-sensor approach streams 500 Hz raw data into 129.6 GB per 30-day month, exceeding the 10 GB/month cap.
• Downsampling to 100 Hz cannot preserve components above 50 Hz; time-domain statistics discard waveform and frequency detail.
• FFT plus compression sends ten selected 16-bit magnitudes every 10 seconds: 500× payload reduction, or 250× including the example’s 20 B metadata, with total latency under the 1-minute requirement. Validate fault detection on representative traces.

f_sample_min = 2 × 210 Hz = 420 Hz

f_sample_recommended = 2.5 × 420 Hz = 1,050 Hz
f_sample_practical = 3 × 420 Hz = 1,260 Hz

# Pseudocode for edge processing every 10 seconds
import numpy

sample_rate_hz = 1000
window_s = 10
samples_per_window = sample_rate_hz * window_s  # 10000

# Perform FFT
fft_result = numpy.fft.rfft(samples)  # 5001 complex bins

# Extract illustrative shaft-order bins
illustrative_freqs_hz = [30, 60, 90, 150, 210]
bin_width = sample_rate_hz / samples_per_window  # 0.1 Hz
selected_bins = [int(f / bin_width) for f in illustrative_freqs_hz]
# Result: bins [300, 600, 900, 1500, 2100]

# Pseudocode: quantize five calibrated magnitudes to uint16, with declared scale,
# saturation handling, and receiver metadata for the selected frequencies.
calibrated_range = ...
magnitudes_u16 = quantize_to_uint16(
    [abs(fft_result[b]) for b in selected_bins], calibrated_range
)
compressed_data = serialize_uint16_with_metadata(
    magnitudes_u16, illustrative_freqs_hz, calibrated_range
)
payload_bytes = len(magnitudes_u16) * 2  # 10 bytes, excluding framing/metadata
compression_ratio = (samples_per_window * 2) / payload_bytes  # 2000x by payload bytes

class EdgeCompressionPipeline:
    def __init__(self):
        self.strategies = {
            'temperature': StatisticalAggregator(window_sec=300),    # 5-min windows
            'vibration': FFTCompressor(top_n=10, window_sec=10),     # Top 10 freq bins
            'motion': EventLogger(),                                 # State changes only
            'pressure': DeltaEncoder() + GZIPCompressor(),           # Lossless delta
            'door': EventLogger(),                                   # State changes only
        }

    def compress_sensor_data(self, sensor_id, raw_samples):
        sensor_type = self.get_sensor_type(sensor_id)
        strategy = self.strategies[sensor_type]
        return strategy.compress(raw_samples)

# Usage example:
pipeline = EdgeCompressionPipeline()

# Temperature: 300 samples → 4 summary values (min/max/mean/std)
temp_compressed = pipeline.compress_sensor_data("temp_01", temp_samples)
# Compression: 300 samples × 2 bytes = 600 bytes → 16 bytes (4 floats)
# Ratio: 37.5x

# Vibration: 10,000 samples → 10 FFT bins
vib_compressed = pipeline.compress_sensor_data("vib_01", vib_samples)
# Compression: 10,000 × 2 bytes = 20 KB → 40 bytes (10 complex floats)
# Ratio: 500x

# Motion: 1000 samples (mostly "no motion") → 3 events ("motion detected" at 3 timestamps)
motion_compressed = pipeline.compress_sensor_data("motion_01", motion_samples)
# Compression: 1000 × 1 byte = 1 KB → 24 bytes (3 events × 8 bytes each)
# Ratio: 42x

Shaft: 30 Hz
Illustrative 5× shaft-order component: 5 × 30 = 150 Hz

At 100 Hz sampling:
- Nyquist = 50 Hz
- 150 Hz aliases to |150 - round(150/100) × 100| = |150 - 200| = 50 Hz
- At exactly the Nyquist frequency, the signal is severely distorted

The illustrative component appears at the Nyquist boundary and cannot be reliably resolved.
Use this as a sampling exercise, not a bearing-failure prediction.

def calculate_illustrative_shaft_order_rate(motor_rpm, highest_shaft_order=7):
    """
    Calculate a sampling example for an explicitly chosen shaft order.
    This does not predict bearing-defect frequencies.
    """
    motor_hz = motor_rpm / 60
    highest_frequency = motor_hz * highest_shaft_order

    # Safety factor: 3x Nyquist for anti-aliasing filter
    recommended_rate = 3 * 2 * highest_frequency

    return {
        'motor_hz': motor_hz,
        'highest_illustrated_component_hz': highest_frequency,
        'nyquist_min': 2 * highest_frequency,
        'recommended': recommended_rate,
    }

# Example usage:
rate_info = calculate_illustrative_shaft_order_rate(motor_rpm=1800, highest_shaft_order=7)
# Shaft: 30 Hz | Highest illustrative component: 210 Hz

7.20 Knowledge Check

This section: Knowledge Check

7.21 Practice Exercises

This section: Practice Exercises

7.22 Time-Series Compression Audit

This section: Time-Series Compression Audit — The body above covers Nyquist sampling, aggregation, FFT compression, semantic extraction, and edge algorithm selection.

The body above covers Nyquist sampling, aggregation, FFT compression, semantic extraction, and edge algorithm selection. Continue to Time-Series Compression and Audit Boundaries when you need the deeper L2 material: delta and delta-of-delta encoders, Gorilla-style XOR compression, columnar layout, lossless-vs-lossy evidence boundaries, and compression metadata required for audit-grade reconstruction.

See how a pasture-monitoring question changes where bytes are processed in the diagram Figure 7.2.

+

In the diagram Figure 7.2, cow-collar motion moves from Measure the raw stream through Fog extracts motion evidence to Cloud compares the herd. The Keep raw windows around anomalies note makes the bargain explicit: compact events travel, while selected waveform evidence remains available for audit.

Before choosing a reducer, price the unreduced six-axis feed with the diagram Figure 7.3.

+

Work across Figure 7.3 one unit at a time. Six axes at 2 bytes per axis make a 12-byte frame. At 100 frames per second, that is 12 B/frame x 100 frames/s = 1,200 B/s. Multiplying by 60 s/min gives 72,000 B/min; extending through 60 min/h and 24 h/day gives 103,680,000 B/day, or 103.68 MB/day before timestamps and identifiers. Now read the evidence warning. Downsampling may miss short hoof-strike peaks, while event extraction keeps counts but discards waveform shape. Those losses support different claims, so byte savings alone cannot choose the reducer.

Use the diagram Figure 7.4 when the design review must decide which motion claim is worth transmitting.

+

In the diagram Figure 7.4, the RAW / 5 HZ lane retains Waveform evidence at a measurable radio cost; the TILT / CHANGE lane retains State evidence and reports on change only. The shared Same ID and clock premise prevents a misleading comparison between differently described records.

Scale the sampling policy across the estate before assigning storage tiers in the diagram Figure 7.5.

+

In the diagram Figure 7.5, the sequence One sensor → gateway, Five sensors × 100 fog nodes, and Downsample at fog exposes the jump from 17.28 MB/day to 8.64 GB/day and the reduction to 14.4 MB/day. Its retention panel separates a short replay buffer from ninety-day trends and an explicitly authorized long hold.

7.23 Summary

This section: Summary

Edge data acquisition requires careful balance between data fidelity and resource constraints:

• Nyquist compliance: Sample at 2x or higher than your highest frequency of interest to avoid aliasing
• Reduction techniques: Aggregation (10-50x), FFT compression (50-500x), and semantic extraction (100-1000x) each serve different use cases
• Algorithm selection: Match compression to downstream analytics needs - lossless for audit, statistical for trends, FFT for vibration, semantic for events
• Resource awareness: Consider CPU time and memory on constrained edge devices, not just compression ratio

7.24 Concept Relationships

This section: Concept Relationships — Key Insight: Compression algorithm selection depends on analytics requirements, not just compression ratio.

Sampling and compression determine data fidelity, bandwidth, and power consumption trade-offs:

Sampling Theory (This chapter):

• Nyquist theorem: sample at 2x highest frequency to avoid aliasing (vibration monitoring: sample at 2.5-3x Nyquist for anti-aliasing filter)
• Under-sampling can cause phantom patterns (an illustrative 150 Hz shaft-order component aliases when sampled below its Nyquist minimum)

Compression Strategies (This chapter):

• Lossless (GZIP): 2-4x reduction, preserves all data (audit trails, compliance)
• Lossy statistical (aggregation): 10-100x reduction, preserves trends (environmental monitoring)
• In the illustrative workload comparison, FFT-based: 50-500x reduction, preserves selected frequency components (vibration analysis).
• Semantic (event extraction): 100-1000x reduction, preserves state changes (threshold monitoring)

Power Impact:

• Edge Data Acquisition: Power and Gateways - Aggregation reduces transmission frequency and can extend battery life (illustrative factory case: 14,400x record-count reduction makes LoRa feasible versus the assumed cellular schedule).

Architecture Context:

• Edge Data Acquisition: Architecture - Device category determines compression need (cameras need heavy compression; temperature sensors need aggregation)
• Edge Compute Patterns - Edge ML requires compressed features (FFT bins, not raw waveforms)

Data Quality:

• data Quality and Preprocessing - Validation must occur before compression (compressing invalid data wastes resources)

Key Insight: Compression algorithm selection depends on analytics requirements, not just compression ratio. Vibration monitoring can use FFT features selected from representative traces; validate their fault-detection performance because discarded bins and time detail cannot be recovered. Choosing a compression method without checking the required evidence can permanently discard signal details needed for analysis.

7.25 What’s Next

This section: What’s Next

• Understand the acquisition architecture that applies these strategies: Edge Acquisition Architecture
• Dig deeper into lossless time-series compression audits: Time-Series Compression and Audit Boundaries
• Learn about power management for low-duty-cycle sampling: Edge Acquisition Power and Gateways
• Study edge compute patterns built on efficient sampling: Edge Compute Patterns
• Apply to the broader edge data acquisition context: Edge Data Acquisition
• Return to the module overview: Big Data Overview