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:**

```python
# 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:**
```python
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:**
```python
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`

```python
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:**

```python
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`

```python
@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`

```python
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`

```python
# 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

```python
# 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

```bash
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

```bash
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