7 Edge Data: Semantic Compression and Pipelines

Highest compression, but requires domain knowledge:

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)

When to use: Monitoring systems where “nothing happening” is the common case. Alarm systems, threshold monitoring, sparse event streams.

Choose the appropriate compression strategy based on signal characteristics, edge compute capabilities, and analytical requirements:

Signal Type	Recommended Compression	Typical Ratio	Edge CPU	Bandwidth	Information Loss	Best For
Slowly changing continuous (temperature, humidity)	Statistical aggregation (min/max/mean/std)	100-1000x	Very Low (1% CPU)	99%+ reduction	Loses individual samples, keeps trends	Environmental monitoring, agriculture
Periodic vibration (motors, pumps)	FFT + top-N frequency bins	100-5000x	High (50% CPU)	99%+ reduction	Loses waveform, keeps frequency spectrum	Predictive maintenance, bearing analysis
Event-driven sparse (motion, door switches)	Event logging (timestamp + state change only)	1000-10000x	Very Low	99.9%+ reduction	Loses “no event” periods (acceptable)	Security, occupancy, access control
High-frequency transient (acoustics, ultrasound)	Triggered capture + FFT	50-500x	High	98%+ reduction	Loses non-trigger periods	Leak detection, acoustic monitoring
Bounded range analog (pressure, flow)	Delta encoding + GZIP	3-10x	Medium (10% CPU)	70-90% reduction	None (lossless)	Critical measurements requiring full fidelity
Audit trail / compliance (access logs, alarms)	GZIP compression only	2-5x	Low (5% CPU)	50-80% reduction	None (lossless)	Regulatory compliance, security logs

Decision Tree:

Read the tree from information requirements toward implementation cost. Start by asking whether silence itself carries meaning or whether only state changes matter; event logging is appropriate only when the application can interpret the gaps correctly. Next decide whether audit or diagnostic work requires the exact waveform. If it does, keep the transformation lossless and prove that decompression reproduces the source bytes. If a lossy representation is acceptable, inspect the signal: stable frequency components can support FFT features, slow trends can support window statistics, and bounded low-variance values can support delta encoding. At each branch, measure edge CPU and memory cost, reconstructed or retained information, output rate, and end-to-end latency on representative data. Keep a raw review window and a versioned compression rule so that a model or threshold change can be audited. The ratios in the comparison table are illustrative workload ranges, not guarantees for a new sensor stream.

• is the signal event-driven (state changes only)?

  • YES → Use event logging (transmit only state changes)
  • NO → Continue to step 2

• do you need to preserve the exact waveform for audit/compliance?

  • YES → Use lossless compression only (GZIP, DEFLATE)
  • NO → Continue to step 3

• does the signal have strong frequency-domain features?

  • YES (vibration, acoustics) → Use FFT + top-N bins
  • NO → Continue to step 4

• is the signal slowly changing (< 1% per sample)?

  • YES → Use statistical aggregation over time windows
  • NO → Continue to step 5

• is the signal bounded with low variance?

  • YES → Use delta encoding + lossless compression
  • NO → Use adaptive sampling rate based on rate-of-change

Example: Multi-Sensor System with Different Compression Strategies:

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

Buffer Sizing for Each Strategy:

Strategy	RAM Required (per sensor)	Latency Added	Notes
Statistical Aggregation	1-2 kB (circular buffer)	5-60 seconds (window duration)	Minimal memory, acceptable latency
FFT Compression	16-64 kB (FFT working memory)	1-10 seconds (FFT window)	High memory, fast processing
Event Logging	< 1 kB (state machine)	None (immediate)	Minimal resources, real-time
Delta Encoding	4-8 kB (recent history)	< 1 second	Low memory, minimal latency

Verification Checklist:

• Compression ratio measured on representative data (not just ideal cases)
• CPU usage stays under 60% during peak sensor activity
• RAM usage leaves 30%+ margin for bursts
• Decompression/reconstruction tested to verify information preservation
• Bandwidth reduction measured end-to-end (including protocol overhead)

Objective: Determine optimal sampling rates for different sensor types.

Tasks:

1. Identify signal characteristics for 4 sensors: temperature (max 0.1 Hz), vibration (max 500 Hz), audio (max 20 kHz), motion IMU (max 50 Hz)
2. Apply Nyquist theorem: calculate minimum sampling rates
3. Implement with margin and measure data rate impact

Expected Outcome: Understand the relationship between signal bandwidth and sampling requirements.

Objective: Implement multiple data reduction techniques and compare bandwidth savings.

Tasks:

1. Collect 1-minute of high-rate sensor data (100 Hz = 6,000 samples)
2. Apply 4 reduction strategies: downsampling, statistical aggregation, delta encoding, event-based
3. Transmit reduced data and compare bandwidth
4. Validate: can you detect a 1C temperature spike with each method?

Expected Outcome: Understand trade-offs between compression ratio and information preservation.