giglez/docs/PATTERN_BASED_DECODING_PLAN.md

Pattern-Based Device Identification for Short Captures

Problem Statement

Current Situation:

• LilyGo T-Embed CC1101 + Bruce firmware captures single transmissions
• Flipper Zero captures are similar (100-500 pulses)
• RTL_433 requires multiple repetitions (1000+ pulses)
• 0% decode success rate with RTL_433

Goal: Create a pattern-based decoder that can identify devices from single-transmission captures by analyzing:

1. Pulse timing patterns
2. Protocol characteristics
3. Frequency/modulation metadata
4. Statistical fingerprints

---

Approach: Multi-Strategy Device Identification

Strategy 1: Timing Pattern Analysis ⭐ Primary

Concept: Different protocols have unique timing signatures

Example Patterns:

# Oregon Scientific v2.1 (Weather Sensor)
SHORT_PULSE = 488  # μs
LONG_PULSE = 976   # μs
SYNC_PATTERN = [SHORT, LONG, SHORT, SHORT]  # Preamble

# PT2262 (Generic Remote)
SHORT_PULSE = 350
LONG_PULSE = 1050
SYNC_PATTERN = [SHORT, LONG*31]  # 31x long pulse

# Princeton (Garage Opener)
SHORT_PULSE = 400
LONG_PULSE = 1200
SYNC_PATTERN = [SHORT*4, LONG*4]

Implementation:

1. Extract pulse widths from RAW_Data
2. Identify SHORT/LONG pulse durations (clustering)
3. Find repeating patterns (preamble, sync words)
4. Match against known protocol signatures

---

Strategy 2: Statistical Fingerprinting

Concept: Protocols have distinct statistical properties

Metrics to Extract:

class SignalFingerprint:
    # Timing statistics
    mean_pulse_width: float
    std_pulse_width: float
    pulse_count: int

    # Pattern characteristics
    unique_pulse_widths: int  # Usually 2-4 for OOK
    pulse_width_ratio: float  # LONG/SHORT ratio
    duty_cycle: float  # HIGH/total time

    # Frequency analysis
    dominant_frequency: float  # From FFT of timing
    repetition_rate: float  # Bits per second

    # Protocol hints
    has_manchester: bool  # Manchester encoding detected
    has_pwm: bool  # Pulse width modulation
    has_ppm: bool  # Pulse position modulation

Matching:

• Compare fingerprint against database of known protocols
• Calculate similarity score (0-1)
• Return top matches with confidence

---

Strategy 3: Protocol Library Integration

Leverage Flipper's Built-in Decoders:

Flipper Zero firmware has decoders for:

• Princeton (garage doors)
• PT2262/PT2264 (generic remotes)
• KeeLoq (rolling code)
• Star Line (car alarms)
• GateTX (gate openers)
• Came (access control)
• Nice (gate openers)
• And 40+ more

Approach:

1. Extract protocol definitions from Flipper firmware
2. Port decoding logic to Python
3. Run each decoder against capture
4. Return successful decodes

---

Strategy 4: Machine Learning Classification

Concept: Train classifier on labeled captures

Features:

features = [
    pulse_count,
    mean_pulse_width,
    std_pulse_width,
    max_pulse_width,
    min_pulse_width,
    pulse_width_ratio,
    duty_cycle,
    unique_pulse_count,
    # Histogram bins
    *pulse_width_histogram(10_bins),
    # Frequency domain
    *fft_features(5_components),
]

Model:

• Random Forest or XGBoost
• Trained on 10,000+ labeled captures
• Outputs: Device category, confidence

Categories:

• Garage Door Opener
• Gate Remote
• Car Key Fob
• Weather Sensor
• Doorbell
• Security Sensor
• Generic Remote
• Unknown

---

Implementation Plan

Phase 1: Timing Pattern Decoder (2-3 hours)

Create: src/matcher/pattern_decoder.py

class PatternDecoder:
    """Decode devices from single-transmission captures"""

    def decode(self, metadata: SignalMetadata) -> List[DeviceMatch]:
        """
        Multi-strategy decoder for short captures

        Strategies:
        1. Timing pattern matching
        2. Statistical fingerprinting
        3. Protocol library matching
        """
        matches = []

        # Extract pulse timings
        pulses = metadata.raw_data

        # Strategy 1: Timing patterns
        timing_matches = self._decode_timing_patterns(pulses)
        matches.extend(timing_matches)

        # Strategy 2: Statistical fingerprint
        fingerprint = self._extract_fingerprint(pulses)
        fingerprint_matches = self._match_fingerprint(fingerprint)
        matches.extend(fingerprint_matches)

        # Strategy 3: Protocol library
        protocol_matches = self._decode_protocols(pulses)
        matches.extend(protocol_matches)

        # Deduplicate and rank by confidence
        return self._rank_matches(matches)

Key Functions:

def _decode_timing_patterns(self, pulses: List[int]) -> List[DeviceMatch]:
    """Match timing patterns against known protocols"""

    # 1. Identify SHORT and LONG pulses
    short_pulse, long_pulse = self._identify_pulse_widths(pulses)

    # 2. Extract pattern
    pattern = self._normalize_pattern(pulses, short_pulse, long_pulse)
    # Example: [S, L, S, S, L, L, L, S] → "10011110"

    # 3. Match against database
    for protocol in KNOWN_PROTOCOLS:
        if self._pattern_matches(pattern, protocol.signature):
            yield DeviceMatch(
                device_name=protocol.name,
                category=protocol.category,
                confidence=self._calculate_confidence(pattern, protocol),
                method='timing_pattern'
            )


def _identify_pulse_widths(self, pulses: List[int]) -> Tuple[int, int]:
    """
    Identify SHORT and LONG pulse durations using clustering

    Most OOK protocols have 2 pulse widths:
    - SHORT: 300-600 μs
    - LONG: 900-1500 μs (usually 3x SHORT)
    """
    from sklearn.cluster import KMeans

    # Get absolute pulse widths
    abs_pulses = [abs(p) for p in pulses]

    # Cluster into 2 groups
    kmeans = KMeans(n_clusters=2)
    kmeans.fit([[p] for p in abs_pulses])

    centers = sorted(kmeans.cluster_centers_.flatten())
    short_pulse = centers[0]
    long_pulse = centers[1]

    return int(short_pulse), int(long_pulse)


def _extract_fingerprint(self, pulses: List[int]) -> SignalFingerprint:
    """Extract statistical fingerprint from signal"""
    import numpy as np

    abs_pulses = [abs(p) for p in pulses]

    return SignalFingerprint(
        mean_pulse_width=np.mean(abs_pulses),
        std_pulse_width=np.std(abs_pulses),
        pulse_count=len(pulses),
        unique_pulse_widths=len(set(abs_pulses)),
        pulse_width_ratio=max(abs_pulses) / min(abs_pulses),
        duty_cycle=self._calculate_duty_cycle(pulses),
    )

---

Phase 2: Protocol Database (1-2 hours)

Create: src/matcher/protocol_database.py

@dataclass
class ProtocolSignature:
    """Known protocol signature"""
    name: str
    category: str
    manufacturer: Optional[str]

    # Timing characteristics
    short_pulse_us: int  # Expected SHORT pulse width
    long_pulse_us: int   # Expected LONG pulse width
    tolerance: float = 0.2  # 20% tolerance

    # Pattern signature
    preamble_pattern: str  # e.g., "10101010"
    sync_pattern: str      # e.g., "1000"
    min_bits: int
    max_bits: int

    # Statistical hints
    typical_pulse_count: int
    frequency: int = 433920000  # Hz


# Database of known protocols
KNOWN_PROTOCOLS = [
    # Weather Sensors
    ProtocolSignature(
        name="Oregon Scientific v2.1",
        category="Weather Sensor",
        manufacturer="Oregon Scientific",
        short_pulse_us=488,
        long_pulse_us=976,
        preamble_pattern="10" * 16,  # 16x alternating
        sync_pattern="1000",
        min_bits=64,
        max_bits=128,
        typical_pulse_count=200,
    ),

    ProtocolSignature(
        name="Acurite 592TXR",
        category="Weather Sensor",
        manufacturer="Acurite",
        short_pulse_us=500,
        long_pulse_us=1000,
        preamble_pattern="10" * 4,
        sync_pattern="1110",
        min_bits=56,
        max_bits=64,
        typical_pulse_count=150,
    ),

    # Generic Remotes
    ProtocolSignature(
        name="PT2262",
        category="Generic Remote",
        manufacturer="Princeton Tech",
        short_pulse_us=350,
        long_pulse_us=1050,
        preamble_pattern="",
        sync_pattern="1" + "0" * 31,  # 31x short gap
        min_bits=24,
        max_bits=24,
        typical_pulse_count=100,
    ),

    ProtocolSignature(
        name="Princeton",
        category="Garage Door",
        manufacturer="Various",
        short_pulse_us=400,
        long_pulse_us=1200,
        preamble_pattern="1111",
        sync_pattern="10000000",
        min_bits=24,
        max_bits=24,
        typical_pulse_count=120,
    ),

    # Car Remotes
    ProtocolSignature(
        name="KeeLoq",
        category="Car Key Fob",
        manufacturer="Microchip",
        short_pulse_us=400,
        long_pulse_us=800,
        preamble_pattern="10" * 12,
        sync_pattern="1111000",
        min_bits=66,
        max_bits=66,
        typical_pulse_count=180,
    ),

    # Add 40+ more from Flipper firmware
]

---

Phase 3: Integration (1 hour)

Update: src/matcher/strategies.py

class PatternBasedStrategy(MatchStrategy):
    """
    Pattern-based decoder for single-transmission captures

    Works with:
    - Flipper Zero .sub files
    - LilyGo T-Embed captures
    - Any short (<500 pulse) RAW captures
    """

    def __init__(self):
        self.decoder = PatternDecoder()

    def match(self, metadata: SignalMetadata, db) -> List[MatchResult]:
        """Decode using pattern analysis"""

        if not metadata.has_raw_data:
            return []

        # Decode using patterns
        devices = self.decoder.decode(metadata)

        # Convert to MatchResult
        results = []
        for device in devices:
            # Find or create device in DB
            device_entry = self._find_or_create_device(
                db,
                device.device_name,
                device.manufacturer
            )

            results.append(MatchResult(
                device_id=device_entry['id'],
                device_name=device.device_name,
                manufacturer=device.manufacturer or 'Unknown',
                confidence=device.confidence,
                match_method='pattern_decode',
                match_details={
                    'strategy': device.strategy,
                    'pattern': device.pattern,
                    'fingerprint': device.fingerprint,
                }
            ))

        return results

Update: src/matcher/simple_matcher.py

# Add PatternBasedStrategy as FIRST strategy
STRATEGIES = [
    PatternBasedStrategy(),      # NEW - for short captures
    RTL433DecoderStrategy(),     # Existing - for long captures
    FrequencyMatchStrategy(),
    ProtocolMatchStrategy(),
    TimingAnalysisStrategy(),
]

---

Expected Results

With Pattern-Based Decoder

Capture Type	RTL_433	Pattern Decoder	Improvement
Flipper unit tests	0%	40-60%	+40-60%
Weather sensors	0%	30-50%	+30-50%
Generic remotes	0%	60-80%	+60-80%
Car key fobs	0%	20-40%	+20-40%
Overall	0%	50-70%	+50-70%

Why Higher Success?

Pattern Decoder Advantages:

1. ✅ Works with single transmissions
2. ✅ No repetition required
3. ✅ Fast (<10ms per file)
4. ✅ Handles short captures (100-500 pulses)
5. ✅ Protocol-agnostic (learns patterns)

RTL_433 Advantages:

1. ✅ Very high confidence (statistical analysis)
2. ✅ 244 protocols supported
3. ✅ Battle-tested
4. ❌ Requires 1000+ pulses
5. ❌ Needs multiple repetitions

Best of Both:

• Use Pattern Decoder first (single transmission)
• Use RTL_433 if available (long captures)
• Combine confidence scores

---

Implementation Timeline

Week 1: Core Pattern Decoder

Day 1-2: Timing Pattern Extraction

• Pulse width clustering (SHORT/LONG identification)
• Pattern normalization
• Preamble detection
• Sync word detection

Day 3-4: Protocol Database

• Extract 50+ protocol signatures from Flipper firmware
• Create ProtocolSignature dataclass
• Implement pattern matching logic

Day 5: Statistical Fingerprinting

• Extract signal fingerprint
• Calculate similarity scores
• Match against database

Week 2: Integration & Testing

Day 1-2: Integration

• Create PatternBasedStrategy
• Update matcher pipeline
• Update API responses

Day 3-4: Testing

• Test with Flipper unit test files
• Test with weather sensor captures
• Test with generic remote captures
• Measure decode success rate

Day 5: Documentation

• API documentation updates
• User guide for capture methodology
• Performance benchmarks

---

Code Example: Complete Decoder

# src/matcher/pattern_decoder.py

from dataclasses import dataclass
from typing import List, Optional, Tuple
import numpy as np
from sklearn.cluster import KMeans

@dataclass
class DeviceMatch:
    device_name: str
    category: str
    manufacturer: Optional[str]
    confidence: float
    strategy: str  # 'timing', 'fingerprint', 'protocol'
    pattern: Optional[str] = None
    fingerprint: Optional[dict] = None


class PatternDecoder:
    """Decode devices from single-transmission captures"""

    def __init__(self):
        self.protocols = KNOWN_PROTOCOLS
        self.min_confidence = 0.3

    def decode(self, metadata: SignalMetadata) -> List[DeviceMatch]:
        """Multi-strategy decoder"""
        matches = []

        pulses = metadata.raw_data
        if not pulses or len(pulses) < 20:
            return []

        # Strategy 1: Timing patterns
        try:
            timing_matches = self._decode_timing_patterns(pulses, metadata.frequency)
            matches.extend(timing_matches)
        except Exception as e:
            print(f"Timing decode failed: {e}")

        # Strategy 2: Statistical fingerprint
        try:
            fingerprint = self._extract_fingerprint(pulses)
            fingerprint_matches = self._match_fingerprint(fingerprint, metadata.frequency)
            matches.extend(fingerprint_matches)
        except Exception as e:
            print(f"Fingerprint decode failed: {e}")

        # Deduplicate and rank
        return self._rank_matches(matches)

    def _decode_timing_patterns(self, pulses: List[int], frequency: int) -> List[DeviceMatch]:
        """Match timing patterns"""
        matches = []

        # Identify pulse widths
        try:
            short_pulse, long_pulse = self._identify_pulse_widths(pulses)
        except:
            return []

        # Check against each protocol
        for protocol in self.protocols:
            if protocol.frequency != frequency:
                continue

            # Check pulse width match
            short_match = abs(short_pulse - protocol.short_pulse_us) / protocol.short_pulse_us
            long_match = abs(long_pulse - protocol.long_pulse_us) / protocol.long_pulse_us

            if short_match <= protocol.tolerance and long_match <= protocol.tolerance:
                # Pulse widths match!
                confidence = 1.0 - max(short_match, long_match)

                # Check pulse count
                count_match = abs(len(pulses) - protocol.typical_pulse_count) / protocol.typical_pulse_count
                if count_match <= 0.5:  # Within 50%
                    confidence *= (1.0 - count_match * 0.5)
                else:
                    confidence *= 0.5

                matches.append(DeviceMatch(
                    device_name=protocol.name,
                    category=protocol.category,
                    manufacturer=protocol.manufacturer,
                    confidence=confidence,
                    strategy='timing',
                    pattern=f"SHORT={short_pulse}, LONG={long_pulse}"
                ))

        return matches

    def _identify_pulse_widths(self, pulses: List[int]) -> Tuple[int, int]:
        """Identify SHORT and LONG pulse durations"""
        abs_pulses = [abs(p) for p in pulses if abs(p) > 50]  # Filter noise

        if len(abs_pulses) < 10:
            raise ValueError("Too few pulses")

        # Cluster into 2-4 groups
        unique_widths = len(set(abs_pulses))
        n_clusters = min(unique_widths, 4)

        if n_clusters < 2:
            raise ValueError("Not enough pulse variation")

        kmeans = KMeans(n_clusters=n_clusters, random_state=42)
        kmeans.fit([[p] for p in abs_pulses])

        centers = sorted(kmeans.cluster_centers_.flatten())

        return int(centers[0]), int(centers[1])

    def _extract_fingerprint(self, pulses: List[int]) -> dict:
        """Extract statistical fingerprint"""
        abs_pulses = [abs(p) for p in pulses]

        return {
            'mean': np.mean(abs_pulses),
            'std': np.std(abs_pulses),
            'min': np.min(abs_pulses),
            'max': np.max(abs_pulses),
            'count': len(pulses),
            'unique': len(set(abs_pulses)),
            'ratio': np.max(abs_pulses) / np.min(abs_pulses) if np.min(abs_pulses) > 0 else 0,
        }

    def _match_fingerprint(self, fingerprint: dict, frequency: int) -> List[DeviceMatch]:
        """Match fingerprint against database"""
        matches = []

        for protocol in self.protocols:
            if protocol.frequency != frequency:
                continue

            # Calculate similarity score
            score = 0.0
            factors = 0

            # Pulse count similarity
            if fingerprint['count'] > 0:
                count_sim = 1.0 - abs(fingerprint['count'] - protocol.typical_pulse_count) / protocol.typical_pulse_count
                score += max(0, count_sim)
                factors += 1

            # Pulse width ratio similarity
            expected_ratio = protocol.long_pulse_us / protocol.short_pulse_us
            if fingerprint['ratio'] > 0:
                ratio_sim = 1.0 - abs(fingerprint['ratio'] - expected_ratio) / expected_ratio
                score += max(0, ratio_sim)
                factors += 1

            if factors > 0:
                confidence = score / factors

                if confidence >= self.min_confidence:
                    matches.append(DeviceMatch(
                        device_name=protocol.name,
                        category=protocol.category,
                        manufacturer=protocol.manufacturer,
                        confidence=confidence * 0.8,  # Lower confidence for fingerprint
                        strategy='fingerprint',
                        fingerprint=fingerprint
                    ))

        return matches

    def _rank_matches(self, matches: List[DeviceMatch]) -> List[DeviceMatch]:
        """Deduplicate and rank by confidence"""
        # Group by device name
        grouped = {}
        for match in matches:
            if match.device_name not in grouped:
                grouped[match.device_name] = match
            else:
                # Keep highest confidence
                if match.confidence > grouped[match.device_name].confidence:
                    grouped[match.device_name] = match

        # Sort by confidence
        return sorted(grouped.values(), key=lambda m: m.confidence, reverse=True)

---

Testing Plan

Test 1: Flipper Unit Tests

Expected: 40-60% success rate

PYTHONPATH=. python3 -c "
from src.matcher.pattern_decoder import PatternDecoder
from src.parser.sub_parser import parse_sub_file

decoder = PatternDecoder()

files = [
    'data/test_known_devices/GateTX_Gate_Opener.sub',
    'data/test_known_devices/Holtek_Remote.sub',
]

for file in files:
    metadata = parse_sub_file(file)
    matches = decoder.decode(metadata)
    print(f'{file}: {len(matches)} matches')
    for m in matches:
        print(f'  → {m.device_name} ({m.confidence:.2%})')
"

Test 2: Real Weather Sensors

Expected: 30-50% success rate

python3 scripts/test_pattern_decoder.py data/test_rtl433_real/

---

Summary

Current: 0% decode rate with RTL_433 (requires long captures)

Proposed: 50-70% decode rate with Pattern Decoder (works with short captures)

Method:

1. Timing pattern matching (SHORT/LONG pulse detection)
2. Statistical fingerprinting (signal characteristics)
3. Protocol library (50+ known signatures)

Timeline: 2 weeks for complete implementation

Next Step: Implement PatternDecoder class with timing analysis