giglez/docs/HEURISTIC_ALGORITHM_PROPOSAL.md

Improved Pattern-Matching Algorithm Using FOSS Resources

Focus: No ML/DL - Pure algorithmic improvements using open-source RF databases Goal: 3x accuracy improvement on unknown devices (5% → 40%)

---

Core Strategy: Multi-Pass Heuristic Pipeline

┌─────────────────────────────────────────────────────────────────┐
│ Stage 1: Signal Preprocessing & Feature Extraction             │
│  ┌───────────────────────────────────────────────────────────┐ │
│  │ 1. Outlier Removal (IQR method)                           │ │
│  │ 2. Noise Floor Estimation                                 │ │
│  │ 3. Pulse Normalization                                    │ │
│  │ 4. Preamble Detection (autocorrelation)                   │ │
│  └───────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
                           ↓
┌─────────────────────────────────────────────────────────────────┐
│ Stage 2: Protocol Database Filtering                           │
│  ┌───────────────────────────────────────────────────────────┐ │
│  │ • Frequency-based filtering (±100kHz)                     │ │
│  │ • Modulation type matching (OOK/FSK from preset)          │ │
│  │ • Candidate pool: 286 → 20-30 protocols                   │ │
│  └───────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
                           ↓
┌─────────────────────────────────────────────────────────────────┐
│ Stage 3: Timing Analysis (Multi-Method)                        │
│  ┌───────────────────────────────────────────────────────────┐ │
│  │ Method A: K-Means (k=2,3,4) + Silhouette scoring         │ │
│  │ Method B: Histogram Peak Detection                        │ │
│  │ Method C: DBSCAN (density-based clustering)               │ │
│  │ → Ensemble vote: Best method selected by confidence       │ │
│  └───────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
                           ↓
┌─────────────────────────────────────────────────────────────────┐
│ Stage 4: Pattern Matching & Scoring                            │
│  ┌───────────────────────────────────────────────────────────┐ │
│  │ Score = Σ weighted features:                              │ │
│  │  • Timing accuracy (30%)                                  │ │
│  │  • Frequency match (25%)                                  │ │
│  │  • Bit count match (20%)                                  │ │
│  │  • Preamble pattern (15%)                                 │ │
│  │  • Statistical fingerprint (10%)                          │ │
│  └───────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
                           ↓
┌─────────────────────────────────────────────────────────────────┐
│ Stage 5: Disambiguation & Ranking                              │
│  ┌───────────────────────────────────────────────────────────┐ │
│  │ • Geographic priors (common devices in region)            │ │
│  │ • Category likelihood (residential vs industrial)         │ │
│  │ • Time-of-day patterns (TPMS = daytime, security = 24/7)  │ │
│  │ • Return top-3 matches with confidence intervals          │ │
│  └───────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘

---

Implementation Details

Component 1: Robust Timing Extraction

File: src/matcher/timing_analyzer.py (new)

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

@dataclass
class TimingSignature:
    """Extracted timing characteristics"""
    short_pulse_us: int
    long_pulse_us: int
    mid_pulse_us: Optional[int]  # For tri-bit encoding
    short_gap_us: int
    long_gap_us: int
    num_levels: int  # 2, 3, or 4
    confidence: float  # How clean the clustering is
    method: str  # 'kmeans', 'histogram', 'dbscan'


class RobustTimingAnalyzer:
    """Multi-method timing extraction with ensemble voting"""

    def __init__(self):
        self.methods = {
            'kmeans': self._kmeans_method,
            'histogram': self._histogram_method,
            'dbscan': self._dbscan_method
        }

    def extract_timing(self, pulses: List[int]) -> TimingSignature:
        """
        Try multiple methods, return best by confidence
        """
        # Preprocess: remove outliers
        pulses_clean = self._remove_outliers(pulses)

        # Separate HIGH (positive) and LOW (negative) pulses
        high_pulses = [p for p in pulses_clean if p > 0]
        low_pulses = [abs(p) for p in pulses_clean if p < 0]

        # Try each method, score by quality
        results = []
        for name, method_func in self.methods.items():
            try:
                timing = method_func(high_pulses, low_pulses)
                timing.method = name
                results.append(timing)
            except Exception as e:
                continue

        # Return best result
        if results:
            return max(results, key=lambda t: t.confidence)
        else:
            # Fallback: simple mean
            return self._fallback_method(high_pulses, low_pulses)

    def _remove_outliers(self, pulses: List[int], method='iqr') -> List[int]:
        """
        Remove statistical outliers (noise spikes, glitches)

        Methods:
          - IQR: Interquartile range (robust to outliers)
          - Z-score: Standard deviation based
          - MAD: Median Absolute Deviation
        """
        if len(pulses) < 10:
            return pulses

        abs_pulses = [abs(p) for p in pulses]
        signs = [1 if p >= 0 else -1 for p in pulses]

        if method == 'iqr':
            q1, q3 = np.percentile(abs_pulses, [25, 75])
            iqr = q3 - q1
            lower = q1 - 1.5 * iqr
            upper = q3 + 1.5 * iqr

            cleaned = [
                p for p in pulses
                if lower <= abs(p) <= upper
            ]
            return cleaned

        elif method == 'zscore':
            mean = np.mean(abs_pulses)
            std = np.std(abs_pulses)
            threshold = 3  # 3 standard deviations

            cleaned = [
                p for p in pulses
                if abs(abs(p) - mean) <= threshold * std
            ]
            return cleaned

        return pulses

    def _kmeans_method(self, high_pulses: List[int], low_pulses: List[int]) -> TimingSignature:
        """
        K-means clustering for k=2,3,4 with silhouette scoring
        """
        best_clustering = None
        best_score = -1

        for k in [2, 3, 4]:
            if len(high_pulses) < k * 5:  # Need enough samples
                continue

            X = np.array(high_pulses).reshape(-1, 1)
            kmeans = KMeans(n_clusters=k, random_state=42, n_init=10)
            labels = kmeans.fit_predict(X)

            # Silhouette score: how well-separated are clusters?
            from sklearn.metrics import silhouette_score
            score = silhouette_score(X, labels)

            if score > best_score:
                best_score = score
                centers = sorted(kmeans.cluster_centers_.flatten())
                best_clustering = (k, centers, score)

        if not best_clustering:
            raise ValueError("K-means failed")

        k, centers, score = best_clustering

        # Extract timing based on number of levels
        if k == 2:
            short_pulse = int(centers[0])
            long_pulse = int(centers[1])
            mid_pulse = None
        elif k == 3:
            short_pulse = int(centers[0])
            mid_pulse = int(centers[1])
            long_pulse = int(centers[2])
        else:  # k == 4
            # Rare: some protocols have 4 distinct widths
            short_pulse = int(centers[0])
            mid_pulse = int(centers[1])
            long_pulse = int(centers[2])

        # Repeat for gaps
        if low_pulses:
            Y = np.array(low_pulses).reshape(-1, 1)
            kmeans_gaps = KMeans(n_clusters=min(2, len(set(low_pulses))), random_state=42, n_init=10)
            gap_centers = sorted(kmeans_gaps.fit_predict(Y))
            short_gap = int(gap_centers[0]) if len(gap_centers) > 0 else 0
            long_gap = int(gap_centers[1]) if len(gap_centers) > 1 else short_gap
        else:
            short_gap = 0
            long_gap = 0

        return TimingSignature(
            short_pulse_us=short_pulse,
            long_pulse_us=long_pulse,
            mid_pulse_us=mid_pulse,
            short_gap_us=short_gap,
            long_gap_us=long_gap,
            num_levels=k,
            confidence=score,  # Silhouette score
            method='kmeans'
        )

    def _histogram_method(self, high_pulses: List[int], low_pulses: List[int]) -> TimingSignature:
        """
        Histogram peak detection - find modes in distribution
        """
        from scipy.signal import find_peaks

        # Create histogram
        hist, bin_edges = np.histogram(high_pulses, bins=50)

        # Find peaks (local maxima)
        peaks, properties = find_peaks(hist, height=len(high_pulses) * 0.05)  # At least 5% of pulses

        if len(peaks) < 2:
            raise ValueError("Not enough peaks detected")

        # Get pulse widths at peak locations
        peak_widths = [bin_edges[p] for p in peaks]
        peak_widths_sorted = sorted(peak_widths)

        short_pulse = int(peak_widths_sorted[0])
        long_pulse = int(peak_widths_sorted[1]) if len(peak_widths_sorted) > 1 else short_pulse * 2
        mid_pulse = int(peak_widths_sorted[1]) if len(peak_widths_sorted) > 2 else None

        # Confidence: ratio of peak heights to noise
        peak_heights = [hist[p] for p in peaks]
        noise_level = np.median(hist)
        snr = np.mean(peak_heights) / (noise_level + 1)
        confidence = min(1.0, snr / 10)  # Normalize to 0-1

        # Gaps (simplified)
        short_gap = int(np.percentile(low_pulses, 25)) if low_pulses else 0
        long_gap = int(np.percentile(low_pulses, 75)) if low_pulses else 0

        return TimingSignature(
            short_pulse_us=short_pulse,
            long_pulse_us=long_pulse,
            mid_pulse_us=mid_pulse,
            short_gap_us=short_gap,
            long_gap_us=long_gap,
            num_levels=len(peak_widths_sorted),
            confidence=confidence,
            method='histogram'
        )

    def _dbscan_method(self, high_pulses: List[int], low_pulses: List[int]) -> TimingSignature:
        """
        DBSCAN density-based clustering - handles noise naturally
        """
        X = np.array(high_pulses).reshape(-1, 1)

        # DBSCAN parameters
        eps = np.std(high_pulses) * 0.3  # Cluster radius
        min_samples = max(3, len(high_pulses) // 20)  # At least 5% of data

        dbscan = DBSCAN(eps=eps, min_samples=min_samples)
        labels = dbscan.fit_predict(X)

        # Extract cluster centers
        unique_labels = set(labels)
        if -1 in unique_labels:
            unique_labels.remove(-1)  # Remove noise label

        if len(unique_labels) < 2:
            raise ValueError("DBSCAN found < 2 clusters")

        centers = []
        for label in unique_labels:
            cluster_points = X[labels == label]
            center = np.mean(cluster_points)
            centers.append(center)

        centers_sorted = sorted(centers)

        short_pulse = int(centers_sorted[0])
        long_pulse = int(centers_sorted[1]) if len(centers_sorted) > 1 else short_pulse * 2
        mid_pulse = int(centers_sorted[1]) if len(centers_sorted) > 2 else None

        # Confidence: fraction of non-noise points
        noise_count = sum(labels == -1)
        confidence = 1.0 - (noise_count / len(labels))

        # Gaps
        short_gap = int(np.percentile(low_pulses, 25)) if low_pulses else 0
        long_gap = int(np.percentile(low_pulses, 75)) if low_pulses else 0

        return TimingSignature(
            short_pulse_us=short_pulse,
            long_pulse_us=long_pulse,
            mid_pulse_us=mid_pulse,
            short_gap_us=short_gap,
            long_gap_us=long_gap,
            num_levels=len(centers_sorted),
            confidence=confidence,
            method='dbscan'
        )

    def _fallback_method(self, high_pulses: List[int], low_pulses: List[int]) -> TimingSignature:
        """Simple percentile-based fallback"""
        short_pulse = int(np.percentile(high_pulses, 25))
        long_pulse = int(np.percentile(high_pulses, 75))
        short_gap = int(np.percentile(low_pulses, 25)) if low_pulses else 0
        long_gap = int(np.percentile(low_pulses, 75)) if low_pulses else 0

        return TimingSignature(
            short_pulse_us=short_pulse,
            long_pulse_us=long_pulse,
            mid_pulse_us=None,
            short_gap_us=short_gap,
            long_gap_us=long_gap,
            num_levels=2,
            confidence=0.5,
            method='fallback'
        )

---

Component 2: Preamble Detection

File: src/matcher/preamble_detector.py (new)

import numpy as np
from typing import Optional, Dict

class PreambleDetector:
    """Detect preamble/sync patterns at start of transmission"""

    def detect(self, pulses: List[int]) -> Optional[Dict]:
        """
        Detect common preamble types:
          1. Alternating pattern (101010...) - used by Oregon, Manchester
          2. Long burst (11111...) - used by Princeton, PT2262
          3. Sync word (specific pattern) - used by Acurite, LaCrosse
        """
        if len(pulses) < 20:
            return None

        # Check first 50 pulses
        preamble_window = pulses[:50]

        # Method 1: Autocorrelation for periodic patterns
        periodic = self._detect_periodic(preamble_window)
        if periodic:
            return periodic

        # Method 2: Long burst detection
        burst = self._detect_long_burst(preamble_window)
        if burst:
            return burst

        # Method 3: Known sync patterns
        sync = self._detect_sync_pattern(preamble_window)
        if sync:
            return sync

        return None

    def _detect_periodic(self, pulses: List[int]) -> Optional[Dict]:
        """
        Detect repeating patterns via autocorrelation

        Example: Oregon Scientific sends "1010101010..." (16 times)
        """
        # Autocorrelation
        autocorr = np.correlate(pulses, pulses, mode='full')
        autocorr = autocorr[len(autocorr)//2:]  # Keep positive lags

        # Find first peak after lag=1
        peaks = []
        for i in range(2, min(20, len(autocorr))):
            if autocorr[i] > autocorr[i-1] and autocorr[i] > autocorr[i+1]:
                peaks.append((i, autocorr[i]))

        if not peaks:
            return None

        # Strongest peak = period
        period, strength = max(peaks, key=lambda x: x[1])

        # Verify: pattern repeats at least 4 times
        pattern = pulses[:period]
        repetitions = 0
        for i in range(0, len(pulses) - period, period):
            chunk = pulses[i:i+period]
            if self._pattern_matches(pattern, chunk, tolerance=0.2):
                repetitions += 1
            else:
                break

        if repetitions >= 4:
            return {
                'type': 'periodic',
                'period': period,
                'repetitions': repetitions,
                'pattern': pattern,
                'length': period * repetitions
            }

        return None

    def _pattern_matches(self, pattern1: List[int], pattern2: List[int], tolerance: float) -> bool:
        """Check if two patterns match within tolerance"""
        if len(pattern1) != len(pattern2):
            return False

        for p1, p2 in zip(pattern1, pattern2):
            if abs(p1 - p2) > abs(p1) * tolerance:
                return False

        return True

    def _detect_long_burst(self, pulses: List[int]) -> Optional[Dict]:
        """
        Detect long HIGH pulse at start

        Example: Princeton sends 4x long HIGH as preamble
        """
        if pulses[0] <= 0:
            return None  # First pulse must be HIGH

        mean_pulse = np.mean([abs(p) for p in pulses])

        # First pulse significantly longer than average?
        if pulses[0] > mean_pulse * 2:
            return {
                'type': 'long_burst',
                'duration_us': pulses[0],
                'length': 1
            }

        return None

    def _detect_sync_pattern(self, pulses: List[int]) -> Optional[Dict]:
        """
        Detect known sync patterns from protocol database

        Example: Acurite uses "10" (short HIGH, short LOW) as sync
        """
        # Convert pulses to binary (simplified)
        mean_pulse = np.median([abs(p) for p in pulses if p > 0])
        bits = []
        for p in pulses[:20]:
            if p > 0:
                bits.append('1' if p > mean_pulse else '0')
            else:
                bits.append('0')

        bit_string = ''.join(bits)

        # Check against known sync patterns
        known_syncs = {
            '10': 'Acurite',
            '1000': 'Oregon Scientific',
            '1111': 'PT2262',
            '0110': 'LaCrosse'
        }

        for pattern, protocol in known_syncs.items():
            if bit_string.startswith(pattern):
                return {
                    'type': 'sync_pattern',
                    'pattern': pattern,
                    'protocol_hint': protocol,
                    'length': len(pattern)
                }

        return None

---

Component 3: Protocol Database Loader (RTL_433 Import)

File: scripts/import_rtl433_database.py (new)

#!/usr/bin/env python3
"""
Import RTL_433 protocol database to expand GigLez signatures

Reads: data/rtl_433_protocols.json (286 devices)
Writes: Updated src/matcher/protocol_database.py
"""

import json
from pathlib import Path

def load_rtl433_protocols():
    """Load RTL_433 JSON database"""
    json_path = Path("data/rtl_433_protocols.json")

    with open(json_path) as f:
        data = json.load(f)

    return data['devices']


def convert_to_protocol_signature(device):
    """
    Convert RTL_433 format to ProtocolSignature

    RTL_433 format:
      {
        "name": "Acurite Tower Sensor",
        "modulation": "OOK",
        "short_width": 1000,
        "long_width": 2000,
        "gap_limit": 3500,
        "frequency": null,  # Often missing
        "manufacturer": "Acurite",
        "category": "weather"
      }
    """
    # Map modulation
    modulation_map = {
        'OOK': 'Modulation.OOK',
        'FSK': 'Modulation.FSK',
        'ASK': 'Modulation.ASK'
    }

    # Estimate frequency from category if missing
    freq = device.get('frequency')
    if not freq:
        # Defaults by category
        if device['category'] == 'weather':
            freq = 433920000  # Most weather sensors
        elif device['category'] == 'automotive':
            freq = 315000000  # TPMS (North America)
        else:
            freq = 433920000  # Default ISM band

    # Calculate typical pulse count
    # Estimate: typical_pulses = 2 * (short + long) * 24 bits
    short = device.get('short_width', 500)
    long = device.get('long_width', 1000)
    typical_pulse_count = int(2.5 * (short + long) / 100)  # Heuristic

    return {
        'name': device['name'],
        'manufacturer': device.get('manufacturer'),
        'category': device.get('category', 'Unknown'),
        'modulation': modulation_map.get(device.get('modulation'), 'Modulation.OOK'),
        'short_pulse_us': short,
        'long_pulse_us': long,
        'frequency': freq,
        'frequency_tolerance': 100000,  # ±100kHz
        'gap_limit': device.get('gap_limit', short * 4),
        'reset_limit': device.get('reset_limit', long * 5),
        'typical_pulse_count': typical_pulse_count,
        'source': 'RTL_433'
    }


def generate_python_code(protocols):
    """Generate Python code for protocol_database.py"""

    code = '''# Auto-generated from RTL_433 database
# Total protocols: {}
# Source: scripts/import_rtl433_database.py

RTL433_PROTOCOLS = [
'''.format(len(protocols))

    for proto in protocols:
        code += f'''    ProtocolSignature(
        name="{proto['name']}",
        category="{proto['category']}",
        manufacturer="{proto.get('manufacturer', 'Unknown')}",
        modulation={proto['modulation']},
        short_pulse_us={proto['short_pulse_us']},
        long_pulse_us={proto['long_pulse_us']},
        frequency={proto['frequency']},
        frequency_tolerance={proto['frequency_tolerance']},
        typical_pulse_count={proto['typical_pulse_count']},
    ),
'''

    code += ''']

# Combine with hand-curated protocols
ALL_PROTOCOLS = (
    WEATHER_SENSORS +
    GARAGE_DOOR_OPENERS +
    DOORBELLS +
    TIRE_PRESSURE +
    SECURITY_SENSORS +
    REMOTE_CONTROLS +
    RTL433_PROTOCOLS  # Add imported protocols
)
'''

    return code


if __name__ == '__main__':
    print("Loading RTL_433 database...")
    devices = load_rtl433_protocols()
    print(f"Found {len(devices)} devices")

    print("Converting to ProtocolSignature format...")
    protocols = [convert_to_protocol_signature(d) for d in devices]

    print("Generating Python code...")
    code = generate_python_code(protocols)

    output_file = "src/matcher/rtl433_protocols_generated.py"
    with open(output_file, 'w') as f:
        f.write(code)

    print(f"✅ Written to {output_file}")
    print(f"   Total protocols: {len(protocols)}")
    print(f"   Categories: {set(p['category'] for p in protocols)}")

---

Performance Targets

Feature	Current	Improved	Method
Protocol Database	25	311 (25 + 286 RTL_433)	Auto-import script
Timing Accuracy	65%	80%	Multi-method ensemble
Single-Tx Success	20%	55%	Outlier removal + preamble
Processing Speed	150ms	90ms	Frequency filtering
Unknown Accuracy	5%	40%	Expanded DB + scoring

---

FOSS Resources Utilized

1. RTL_433 Protocol Database: 286 device signatures
2. Flipper Zero Protocols: 40+ protocol decoders
3. Scikit-learn: K-means, DBSCAN clustering
4. SciPy: Statistical analysis, peak detection
5. NumPy: Signal processing, autocorrelation
6. FCC Database: Device frequency validation

---

Next Steps

1. Implement Component 1: Robust timing analyzer (timing_analyzer.py)
2. Implement Component 2: Preamble detector (preamble_detector.py)
3. Run Component 3: Import RTL_433 database (expand to 311 protocols)
4. Integrate: Update pattern_decoder.py to use new components
5. Benchmark: Test on labeled dataset, measure accuracy improvement

---

Implementation Time: 1-2 weeks Dependencies: None (FOSS only) Risk: Low (incremental improvements to existing system)