giglez/docs/RTL433_INTEGRATION_PLAN.md

RTL_433 Integration Implementation Plan

Executive Summary

This document outlines the detailed implementation plan for integrating RTL_433 decoders into the GigLez platform to enable automatic device identification from RAW .sub files.

Current Status:

• ✅ RTL_433 protocol database loaded (data/rtl_433_protocols.json)
• ✅ RTL_433 source code available (signatures/rtl_433/)
• ✅ RTL_433 matcher for protocol name/timing matching (src/matcher/rtl433_matcher.py)
• ✅ .sub file parser with RAW data extraction (src/parser/sub_parser.py)
• ❌ RTL_433 binary not installed
• ❌ No bridge between RAW timing data and RTL_433 decoders

Goal: Enable automatic device identification by passing RAW timing data from .sub files to RTL_433's battle-tested decoders.

---

Phase 1: RTL_433 Installation & Environment Setup

1.1 Install RTL_433 Binary

Estimated Time: 30 minutes

Option A: Package Manager (Recommended for Ubuntu/Debian)

# Ubuntu 19.10+ / Debian sid
sudo apt-get update
sudo apt-get install -y rtl-433

# Verify installation
rtl_433 -V
rtl_433 -R help | head -20

Option B: Compile from Source (Required if package unavailable)

cd signatures/rtl_433

# Install build dependencies
sudo apt-get install -y \
    libtool \
    libusb-1.0-0-dev \
    librtlsdr-dev \
    rtl-sdr \
    build-essential \
    cmake \
    pkg-config

# Build
mkdir build
cd build
cmake ../ -DCMAKE_BUILD_TYPE=Release
make -j4
sudo make install
sudo ldconfig

# Verify
rtl_433 -V

Deliverable: Working rtl_433 binary accessible from command line

---

Phase 2: Understanding RTL_433 Input Formats

2.1 RTL_433 Input Format Options

RTL_433 accepts several input formats via the -r <filename> parameter:

Format	Extension	Description	Use Case
I/Q Samples	.cu8, .cs16	Raw SDR samples (2-channel, interleaved I/Q)	Direct from RTL-SDR hardware
Pulse Data	.ook, .fsk	PWM pulse width data	Converted from timing arrays
Test Data	-y {25}fb2dd58	Hex test strings	Protocol verification

For GigLez: We'll convert Flipper RAW_Data to Pulse Data format (.ook files)

2.2 RTL_433 Pulse Data Format

RTL_433's pulse data format encodes OOK/ASK signals as pulse width modulation:

Format: 16-bit signed integers (int16_t)
- Positive values = High pulse duration (microseconds)
- Negative values = Low pulse duration (microseconds)
- Binary little-endian encoding

Example Conversion:

Flipper RAW_Data: 1061 -13 59 -8 10 -24 18 -5 21 -5
RTL_433 Pulse:    [1061, -13, 59, -8, 10, -24, 18, -5, 21, -5] (as int16_t array)

2.3 File Format Specifications

Flipper .sub RAW Format:

Filetype: Flipper SubGhz RAW File
Version: 1
Frequency: 433920000
Preset: FuriHalSubGhzPresetOok650Async
Protocol: RAW
RAW_Data: 1061 -13 59 -8 10 -24 18 -5 21 -5 ...

RTL_433 Pulse File (.ook):

Binary file containing:
- Header: None (raw binary data)
- Content: int16_t array (little-endian)
- Each sample: 2 bytes signed integer

RTL_433 Command:

rtl_433 \
  -r pulse_data.ook \
  -F json \
  -M level \
  -M protocol \
  -M time \
  -R 0  # Disable all protocols
  -R 12 # Enable specific protocol (e.g., Acurite)

---

Phase 3: RAW Data Conversion Pipeline

3.1 Converter Implementation

File: src/parser/rtl433_converter.py

"""
RTL_433 Pulse Data Converter

Converts Flipper .sub RAW_Data to RTL_433 pulse data format
"""

import struct
from pathlib import Path
from typing import List, Optional
from loguru import logger

from .metadata import SignalMetadata


class RTL433Converter:
    """
    Convert Flipper RAW timing data to RTL_433 pulse format

    RTL_433 Pulse Format:
    - 16-bit signed integers (little-endian)
    - Positive = high pulse (μs)
    - Negative = low pulse (μs)
    """

    MAX_PULSE_VALUE = 32767  # int16_t max
    MIN_PULSE_VALUE = -32768  # int16_t min

    def convert_to_pulse_file(self,
                             raw_data: List[int],
                             output_path: str) -> bool:
        """
        Convert RAW_Data array to RTL_433 pulse data file

        Args:
            raw_data: Array of timing values from .sub file
            output_path: Path to output .ook file

        Returns:
            True if successful, False otherwise
        """
        try:
            # Validate input
            if not raw_data:
                logger.error("Empty RAW_Data array")
                return False

            # Clamp values to int16_t range
            clamped_data = []
            for value in raw_data:
                if value > self.MAX_PULSE_VALUE:
                    logger.warning(f"Pulse value {value} exceeds max, clamping")
                    clamped_data.append(self.MAX_PULSE_VALUE)
                elif value < self.MIN_PULSE_VALUE:
                    logger.warning(f"Pulse value {value} below min, clamping")
                    clamped_data.append(self.MIN_PULSE_VALUE)
                else:
                    clamped_data.append(value)

            # Pack as little-endian int16 array
            binary_data = struct.pack(f'<{len(clamped_data)}h', *clamped_data)

            # Write to file
            Path(output_path).parent.mkdir(parents=True, exist_ok=True)
            with open(output_path, 'wb') as f:
                f.write(binary_data)

            logger.info(f"Converted {len(raw_data)} pulses to {output_path}")
            return True

        except Exception as e:
            logger.error(f"Failed to convert pulse data: {e}")
            return False

    def convert_metadata(self, metadata: SignalMetadata) -> Optional[str]:
        """
        Convert SignalMetadata RAW_Data to temp pulse file

        Args:
            metadata: Parsed .sub file metadata

        Returns:
            Path to temporary .ook file, or None if failed
        """
        if not metadata.has_raw_data or not metadata.raw_data:
            logger.debug("No RAW data in metadata")
            return None

        # Create temp file path
        import tempfile
        temp_dir = Path(tempfile.gettempdir()) / "giglez_rtl433"
        temp_dir.mkdir(exist_ok=True)

        # Use frequency in filename for debugging
        freq_mhz = metadata.frequency / 1_000_000
        temp_file = temp_dir / f"pulse_{freq_mhz:.2f}MHz.ook"

        if self.convert_to_pulse_file(metadata.raw_data, str(temp_file)):
            return str(temp_file)

        return None

    def validate_pulse_data(self, raw_data: List[int]) -> tuple[bool, List[str]]:
        """
        Validate RAW_Data before conversion

        Args:
            raw_data: Array of timing values

        Returns:
            (is_valid, error_messages)
        """
        errors = []

        if not raw_data:
            errors.append("Empty pulse data")
            return (False, errors)

        # Check for minimum length
        if len(raw_data) < 10:
            errors.append(f"Too few pulses: {len(raw_data)} (minimum 10)")

        # Check for all zeros
        if all(p == 0 for p in raw_data):
            errors.append("All pulse values are zero")

        # Check for reasonable pulse widths
        max_pulse = max(abs(p) for p in raw_data)
        if max_pulse > 100_000:
            errors.append(f"Suspiciously large pulse: {max_pulse}μs")

        # Check for alternating sign pattern
        positive_count = sum(1 for p in raw_data if p > 0)
        negative_count = sum(1 for p in raw_data if p < 0)

        if positive_count == 0 or negative_count == 0:
            errors.append("Missing high or low pulses (should alternate)")

        return (len(errors) == 0, errors)


# Singleton instance
_converter = None

def get_converter() -> RTL433Converter:
    """Get global converter instance"""
    global _converter
    if _converter is None:
        _converter = RTL433Converter()
    return _converter

Test Script: tests/test_rtl433_converter.py

"""Test RTL_433 pulse converter"""

import struct
from pathlib import Path
from src.parser.rtl433_converter import RTL433Converter
from src.parser.sub_parser import parse_sub_file


def test_conversion():
    """Test conversion of real .sub file"""

    converter = RTL433Converter()

    # Parse test file
    test_file = "signatures/t-embed-rf/raw_7.sub"
    metadata = parse_sub_file(test_file)

    print(f"File: {test_file}")
    print(f"Frequency: {metadata.frequency / 1_000_000:.2f} MHz")
    print(f"Pulses: {len(metadata.raw_data)}")
    print(f"Duration: {metadata.duration_ms:.2f} ms")

    # Validate
    is_valid, errors = converter.validate_pulse_data(metadata.raw_data)
    if not is_valid:
        print("Validation errors:")
        for err in errors:
            print(f"  - {err}")
        return

    # Convert
    output_path = converter.convert_metadata(metadata)
    if output_path:
        print(f"Created: {output_path}")

        # Verify binary format
        with open(output_path, 'rb') as f:
            data = f.read()
            pulse_count = len(data) // 2
            pulses = struct.unpack(f'<{pulse_count}h', data)

            print(f"Verified: {pulse_count} pulses")
            print(f"First 10: {pulses[:10]}")
    else:
        print("Conversion failed")


if __name__ == '__main__':
    test_conversion()

---

Phase 4: RTL_433 Subprocess Wrapper

4.1 Decoder Implementation

File: src/matcher/rtl433_decoder.py

"""
RTL_433 Subprocess Decoder

Runs RTL_433 binary on pulse data and parses JSON output
"""

import json
import subprocess
import tempfile
from pathlib import Path
from typing import List, Dict, Optional
from dataclasses import dataclass
from loguru import logger

from ..parser.metadata import SignalMetadata
from ..parser.rtl433_converter import get_converter


@dataclass
class RTL433DecodedDevice:
    """Decoded device from RTL_433"""
    model: str
    manufacturer: Optional[str]
    device_id: Optional[str]
    protocol_id: int
    frequency: int
    raw_json: Dict
    confidence: float = 0.95  # RTL_433 decodes are high confidence

    @classmethod
    def from_json(cls, data: Dict, frequency: int):
        """Create from RTL_433 JSON output"""
        return cls(
            model=data.get('model', 'Unknown'),
            manufacturer=data.get('manufacturer'),
            device_id=data.get('id'),
            protocol_id=data.get('protocol', 0),
            frequency=frequency,
            raw_json=data
        )


class RTL433Decoder:
    """
    RTL_433 decoder wrapper

    Converts RAW pulse data and runs rtl_433 binary
    """

    def __init__(self,
                 rtl433_path: str = "rtl_433",
                 timeout: int = 10):
        """
        Initialize decoder

        Args:
            rtl433_path: Path to rtl_433 binary (default: system PATH)
            timeout: Max execution time in seconds
        """
        self.rtl433_path = rtl433_path
        self.timeout = timeout
        self.converter = get_converter()

        # Verify RTL_433 is available
        if not self._check_rtl433_available():
            logger.warning("rtl_433 binary not found in PATH")

    def _check_rtl433_available(self) -> bool:
        """Check if rtl_433 is installed"""
        try:
            result = subprocess.run(
                [self.rtl433_path, '-V'],
                capture_output=True,
                timeout=5
            )
            return result.returncode == 0
        except (FileNotFoundError, subprocess.TimeoutExpired):
            return False

    def decode(self,
               metadata: SignalMetadata,
               enable_all_protocols: bool = True,
               specific_protocols: Optional[List[int]] = None) -> List[RTL433DecodedDevice]:
        """
        Decode RAW signal using RTL_433

        Args:
            metadata: Parsed .sub file metadata with RAW_Data
            enable_all_protocols: Try all RTL_433 protocols (slower but comprehensive)
            specific_protocols: List of protocol IDs to enable (faster)

        Returns:
            List of decoded devices (may be empty if no match)
        """
        if not metadata.has_raw_data:
            logger.debug("No RAW data to decode")
            return []

        # Convert to pulse file
        pulse_file = self.converter.convert_metadata(metadata)
        if not pulse_file:
            logger.error("Failed to convert RAW data to pulse file")
            return []

        try:
            # Build RTL_433 command
            cmd = [
                self.rtl433_path,
                '-r', pulse_file,  # Read from file
                '-F', 'json',      # JSON output
                '-M', 'level',     # Add signal level
                '-M', 'protocol',  # Add protocol ID
                '-M', 'time',      # Add timestamp
                '-q',              # Quiet mode (no startup banner)
            ]

            # Protocol selection
            if specific_protocols:
                cmd.append('-R')
                cmd.append('0')  # Disable all
                for proto_id in specific_protocols:
                    cmd.append('-R')
                    cmd.append(str(proto_id))
            elif not enable_all_protocols:
                # Use frequency-based filtering
                freq_mhz = metadata.frequency / 1_000_000
                if 433.5 <= freq_mhz <= 434.5:
                    # Common 433MHz protocols
                    common_433 = [12, 19, 20, 40, 55]  # Acurite, Oregon, etc.
                    cmd.append('-R')
                    cmd.append('0')
                    for pid in common_433:
                        cmd.append('-R')
                        cmd.append(str(pid))

            # Execute
            logger.debug(f"Running: {' '.join(cmd)}")
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=self.timeout
            )

            # Parse output
            devices = self._parse_output(result.stdout, metadata.frequency)

            if devices:
                logger.info(f"RTL_433 decoded {len(devices)} device(s)")
            else:
                logger.debug("RTL_433 found no matches")

            return devices

        except subprocess.TimeoutExpired:
            logger.error(f"RTL_433 timed out after {self.timeout}s")
            return []

        except Exception as e:
            logger.error(f"RTL_433 decode error: {e}")
            return []

        finally:
            # Cleanup temp file
            try:
                Path(pulse_file).unlink()
            except:
                pass

    def _parse_output(self, output: str, frequency: int) -> List[RTL433DecodedDevice]:
        """
        Parse RTL_433 JSON output

        Args:
            output: JSON lines from RTL_433 stdout
            frequency: Original frequency from .sub file

        Returns:
            List of decoded devices
        """
        devices = []

        for line in output.strip().split('\n'):
            if not line or not line.startswith('{'):
                continue

            try:
                data = json.loads(line)

                # RTL_433 outputs various message types
                # We only want decoded device messages (have 'model' field)
                if 'model' in data:
                    device = RTL433DecodedDevice.from_json(data, frequency)
                    devices.append(device)

            except json.JSONDecodeError as e:
                logger.warning(f"Failed to parse JSON line: {e}")
                continue

        return devices

    def get_supported_protocols(self) -> Dict[int, str]:
        """
        Get list of supported RTL_433 protocols

        Returns:
            Dict mapping protocol ID to name
        """
        try:
            result = subprocess.run(
                [self.rtl433_path, '-R', 'help'],
                capture_output=True,
                text=True,
                timeout=5
            )

            protocols = {}
            for line in result.stdout.split('\n'):
                # Parse lines like: "  [12]  Acurite Tower Sensor"
                if line.strip().startswith('['):
                    parts = line.strip().split(']', 1)
                    if len(parts) == 2:
                        proto_id = int(parts[0].strip('[ '))
                        proto_name = parts[1].strip()
                        protocols[proto_id] = proto_name

            return protocols

        except Exception as e:
            logger.error(f"Failed to get RTL_433 protocols: {e}")
            return {}


# Global singleton
_decoder = None

def get_decoder() -> RTL433Decoder:
    """Get global decoder instance"""
    global _decoder
    if _decoder is None:
        _decoder = RTL433Decoder()
    return _decoder

Test Script: tests/test_rtl433_decoder.py

"""Test RTL_433 decoder"""

from src.parser.sub_parser import parse_sub_file
from src.matcher.rtl433_decoder import get_decoder


def test_decoder():
    """Test RTL_433 decoder on real file"""

    decoder = get_decoder()

    # Show supported protocols
    protocols = decoder.get_supported_protocols()
    print(f"RTL_433 supports {len(protocols)} protocols")
    print("Sample protocols:")
    for pid, name in list(protocols.items())[:10]:
        print(f"  [{pid:3d}] {name}")

    # Test decoding
    test_file = "signatures/t-embed-rf/raw_7.sub"
    metadata = parse_sub_file(test_file)

    print(f"\nDecoding: {test_file}")
    print(f"Frequency: {metadata.frequency / 1_000_000:.2f} MHz")

    # Try decoding
    devices = decoder.decode(metadata, enable_all_protocols=True)

    if devices:
        print(f"\nDecoded {len(devices)} device(s):")
        for device in devices:
            print(f"  Model: {device.model}")
            print(f"  Manufacturer: {device.manufacturer}")
            print(f"  Device ID: {device.device_id}")
            print(f"  Protocol: {device.protocol_id}")
            print(f"  Confidence: {device.confidence}")
            print(f"  Raw JSON: {device.raw_json}")
            print()
    else:
        print("No devices decoded")


if __name__ == '__main__':
    test_decoder()

---

Phase 5: Integration into Matcher Pipeline

5.1 Create RTL_433 Match Strategy

File: src/matcher/strategies.py (add new class)

class RTL433DecoderStrategy(MatchStrategy):
    """
    RTL_433 decoder strategy - highest accuracy for supported protocols

    Uses actual RTL_433 binary to decode RAW signals
    Confidence: 0.95 for successful decodes
    """

    def __init__(self):
        from ..matcher.rtl433_decoder import get_decoder
        self.decoder = get_decoder()

    def match(self, metadata: SignalMetadata, db) -> List[MatchResult]:
        """
        Decode using RTL_433 and convert to MatchResult

        Args:
            metadata: Signal metadata with RAW_Data
            db: Database connection (for storing decoded data)

        Returns:
            List of MatchResult objects
        """
        matches = []

        if not metadata.has_raw_data:
            return matches

        # Try RTL_433 decoding
        devices = self.decoder.decode(metadata, enable_all_protocols=True)

        for device in devices:
            # Query database for device
            # (or create new device entry if not exists)
            device_entry = self._find_or_create_device(
                db,
                device.model,
                device.manufacturer
            )

            matches.append(MatchResult(
                device_id=device_entry['id'],
                device_name=device.model,
                manufacturer=device.manufacturer or 'Unknown',
                confidence=device.confidence,
                match_method='rtl433_decode',
                match_details={
                    'protocol_id': device.protocol_id,
                    'device_id': device.device_id,
                    'rtl433_data': device.raw_json
                }
            ))

        logger.debug(f"RTL433DecoderStrategy found {len(matches)} matches")
        return matches

    def _find_or_create_device(self, db, model: str, manufacturer: Optional[str]) -> Dict:
        """
        Find device in database or create new entry

        Args:
            db: Database connection
            model: Device model name
            manufacturer: Manufacturer name

        Returns:
            Device record dict
        """
        # Search existing devices
        query = """
            SELECT id, manufacturer, model
            FROM devices
            WHERE model = %s
            AND (manufacturer = %s OR manufacturer IS NULL)
            LIMIT 1
        """

        results = db.execute(query, (model, manufacturer))

        if results:
            return {
                'id': results[0]['id'],
                'manufacturer': results[0]['manufacturer'],
                'model': results[0]['model']
            }

        # Create new device
        insert_query = """
            INSERT INTO devices (manufacturer, model, device_type, category)
            VALUES (%s, %s, 'rf_device', 'rtl433_decoded')
            RETURNING id
        """

        result = db.execute(insert_query, (manufacturer, model))
        new_id = result[0]['id']

        logger.info(f"Created new device: {manufacturer} {model} (ID={new_id})")

        return {
            'id': new_id,
            'manufacturer': manufacturer,
            'model': model
        }

5.2 Update Matcher Initialization

File: src/matcher/engine.py (update initialization)

def create_default_matcher(database):
    """
    Create matcher with all available strategies

    Strategies are tried in order of confidence/speed:
    1. RTL433 Decoder (0.95 confidence, slow)
    2. Exact Match (1.0 confidence, fast)
    3. Partial Match (0.8 confidence, fast)
    4. Pattern Match (0.7-0.9 confidence, medium)
    5. Timing Match (0.6-0.8 confidence, medium)
    6. Frequency Match (0.5-0.7 confidence, fast)
    """
    from .strategies import (
        RTL433DecoderStrategy,
        ExactMatcher,
        PartialMatcher,
        PatternMatcher,
        TimingMatcher,
        FrequencyMatcher
    )

    matcher = SignatureMatcher(database)

    # Add RTL_433 decoder first (highest accuracy)
    matcher.add_strategy(RTL433DecoderStrategy())

    # Add existing strategies
    matcher.add_strategy(ExactMatcher())
    matcher.add_strategy(PartialMatcher())
    matcher.add_strategy(PatternMatcher())
    matcher.add_strategy(TimingMatcher())
    matcher.add_strategy(FrequencyMatcher(tolerance_hz=10000))

    return matcher

---

Phase 6: Testing & Validation

6.1 Test Dataset

Use existing T-Embed captures:

signatures/t-embed-rf/
├── raw_7.sub
├── raw_6.sub
├── raw_5.sub
├── raw_4.sub
├── raw_8.sub
├── 34.0478N_118.2349W_1650_test_raw.sub
└── ...

6.2 Validation Script

File: scripts/test_rtl433_integration.py

"""
RTL_433 Integration Test

Tests complete pipeline:
1. Parse .sub file
2. Convert to pulse data
3. Run RTL_433 decoder
4. Store results
"""

from pathlib import Path
from src.parser.sub_parser import parse_sub_file
from src.matcher.rtl433_decoder import get_decoder
from src.matcher.engine import create_default_matcher
from src.database.connection import get_database


def test_full_pipeline():
    """Test complete RTL_433 integration"""

    test_files = list(Path("signatures/t-embed-rf").glob("*.sub"))

    print(f"Testing {len(test_files)} .sub files\n")
    print("="*70)

    decoder = get_decoder()
    results = []

    for test_file in test_files:
        print(f"\nFile: {test_file.name}")
        print("-"*70)

        try:
            # Parse
            metadata = parse_sub_file(str(test_file))
            print(f"Frequency: {metadata.frequency / 1_000_000:.2f} MHz")
            print(f"Format: {metadata.file_format}")

            if not metadata.has_raw_data:
                print("⚠ No RAW data, skipping")
                continue

            print(f"Pulses: {metadata.pulse_count}")
            print(f"Duration: {metadata.duration_ms:.2f} ms")

            # Decode
            devices = decoder.decode(metadata, enable_all_protocols=True)

            if devices:
                print(f"✓ Decoded {len(devices)} device(s):")
                for dev in devices:
                    print(f"  - {dev.manufacturer} {dev.model}")
                    print(f"    Protocol: {dev.protocol_id}")
                    print(f"    Confidence: {dev.confidence}")

                results.append({
                    'file': test_file.name,
                    'status': 'success',
                    'devices': len(devices)
                })
            else:
                print("✗ No devices decoded")
                results.append({
                    'file': test_file.name,
                    'status': 'no_match',
                    'devices': 0
                })

        except Exception as e:
            print(f"✗ Error: {e}")
            results.append({
                'file': test_file.name,
                'status': 'error',
                'error': str(e)
            })

    # Summary
    print("\n" + "="*70)
    print("SUMMARY")
    print("="*70)

    success = sum(1 for r in results if r['status'] == 'success')
    no_match = sum(1 for r in results if r['status'] == 'no_match')
    errors = sum(1 for r in results if r['status'] == 'error')

    print(f"Total files: {len(results)}")
    print(f"Successful decodes: {success}")
    print(f"No matches: {no_match}")
    print(f"Errors: {errors}")
    print(f"Success rate: {success / len(results) * 100:.1f}%")


if __name__ == '__main__':
    test_full_pipeline()

---

Phase 7: Performance Optimization

7.1 Caching Strategy

File: src/matcher/rtl433_cache.py

"""
RTL_433 Decode Cache

Caches RTL_433 decoding results to avoid re-processing
"""

import hashlib
import json
from pathlib import Path
from typing import List, Optional
from loguru import logger

from .rtl433_decoder import RTL433DecodedDevice


class RTL433Cache:
    """
    Cache for RTL_433 decode results

    Uses SHA256 of RAW_Data as cache key
    """

    def __init__(self, cache_dir: str = "/tmp/giglez_rtl433_cache"):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(parents=True, exist_ok=True)

    def _get_cache_key(self, raw_data: List[int]) -> str:
        """Generate cache key from RAW_Data"""
        data_str = ','.join(map(str, raw_data))
        return hashlib.sha256(data_str.encode()).hexdigest()[:16]

    def get(self, raw_data: List[int]) -> Optional[List[RTL433DecodedDevice]]:
        """Get cached decode result"""
        key = self._get_cache_key(raw_data)
        cache_file = self.cache_dir / f"{key}.json"

        if not cache_file.exists():
            return None

        try:
            with open(cache_file, 'r') as f:
                data = json.load(f)

            devices = [
                RTL433DecodedDevice.from_json(d, d['frequency'])
                for d in data
            ]

            logger.debug(f"Cache hit: {key}")
            return devices

        except Exception as e:
            logger.warning(f"Cache read error: {e}")
            return None

    def set(self, raw_data: List[int], devices: List[RTL433DecodedDevice]):
        """Store decode result in cache"""
        key = self._get_cache_key(raw_data)
        cache_file = self.cache_dir / f"{key}.json"

        try:
            data = [d.raw_json for d in devices]
            with open(cache_file, 'w') as f:
                json.dump(data, f)

            logger.debug(f"Cache write: {key}")

        except Exception as e:
            logger.warning(f"Cache write error: {e}")

7.2 Update Decoder to Use Cache

# In RTL433Decoder.decode()

from .rtl433_cache import RTL433Cache

class RTL433Decoder:
    def __init__(self, ...):
        # ...
        self.cache = RTL433Cache()

    def decode(self, metadata: SignalMetadata, ...) -> List[RTL433DecodedDevice]:
        if not metadata.has_raw_data:
            return []

        # Check cache first
        cached = self.cache.get(metadata.raw_data)
        if cached is not None:
            logger.info("Using cached RTL_433 decode result")
            return cached

        # ... existing decode logic ...

        devices = self._parse_output(result.stdout, metadata.frequency)

        # Cache result
        if devices:
            self.cache.set(metadata.raw_data, devices)

        return devices

---

Phase 8: API Integration

8.1 Update Capture Endpoint

File: src/api/routes/captures.py (update response)

@router.post("/submit")
async def submit_capture(...):
    # ... existing code ...

    # Run matching
    matches = matcher.match(metadata, max_results=10)

    # Separate RTL_433 matches
    rtl433_matches = [m for m in matches if m.match_method == 'rtl433_decode']
    other_matches = [m for m in matches if m.match_method != 'rtl433_decode']

    return {
        'status': 'success',
        'capture_id': capture_id,
        'rtl433_decoded': [m.to_dict() for m in rtl433_matches],
        'signature_matches': [m.to_dict() for m in other_matches],
        'best_match': matches[0].to_dict() if matches else None
    }

8.2 Add RTL_433 Info Endpoint

File: src/api/routes/hardware.py (new route)

from ..matcher.rtl433_decoder import get_decoder

@router.get("/rtl433/protocols")
async def get_rtl433_protocols():
    """Get list of supported RTL_433 protocols"""
    decoder = get_decoder()
    protocols = decoder.get_supported_protocols()

    return {
        'total': len(protocols),
        'protocols': [
            {'id': pid, 'name': name}
            for pid, name in protocols.items()
        ]
    }

@router.get("/rtl433/status")
async def get_rtl433_status():
    """Check RTL_433 availability"""
    decoder = get_decoder()
    available = decoder._check_rtl433_available()

    return {
        'available': available,
        'binary_path': decoder.rtl433_path,
        'timeout': decoder.timeout
    }

---

Implementation Timeline

Phase	Tasks	Duration	Dependencies
Phase 1	Install RTL_433 binary	30 min	None
Phase 2	Research input formats	1 hour	Phase 1
Phase 3	Implement converter	2-3 hours	Phase 2
Phase 4	Implement decoder wrapper	2-3 hours	Phase 3
Phase 5	Integrate into matcher	1-2 hours	Phase 4
Phase 6	Testing & validation	2-3 hours	Phase 5
Phase 7	Performance optimization	1-2 hours	Phase 6
Phase 8	API updates	1-2 hours	Phase 5

Total Estimated Time: 12-16 hours (1.5-2 work days)

---

Success Criteria

1. ✅ RTL_433 binary installed and accessible
2. ✅ Conversion from Flipper RAW_Data to RTL_433 pulse format working
3. ✅ RTL_433 decoder successfully processes pulse files
4. ✅ At least 50% of test .sub files decode successfully
5. ✅ Results stored in database with proper confidence scores
6. ✅ API returns RTL_433 decoded devices
7. ✅ Cache reduces redundant processing
8. ✅ Average decode time < 2 seconds per file

---

Deployment Checklist

Development Environment

• Install RTL_433 via package manager or compile from source
• Run converter tests
• Run decoder tests
• Run integration tests
• Verify cache functionality

Production Server

• Install RTL_433 binary: sudo apt-get install rtl-433
• Verify binary location: which rtl_433
• Test RTL_433 execution permissions
• Configure cache directory with proper permissions
• Set up log rotation for RTL_433 output
• Add RTL_433 version to system monitoring

Configuration

# config/settings.py

RTL433_CONFIG = {
    'binary_path': 'rtl_433',  # or '/usr/bin/rtl_433'
    'timeout': 10,  # seconds
    'cache_enabled': True,
    'cache_dir': '/tmp/giglez_rtl433_cache',
    'enable_all_protocols': True,
    'max_concurrent_decodes': 4
}

---

Troubleshooting

RTL_433 Not Found

# Install from package manager
sudo apt-get install rtl-433

# Or compile from source
cd signatures/rtl_433
mkdir build && cd build
cmake ..
make && sudo make install

Pulse Conversion Errors

• Check RAW_Data format (should alternate positive/negative)
• Verify pulse values are reasonable (< 100,000 μs)
• Ensure minimum 10 pulses

No Decodes

• Try enable_all_protocols=True
• Check frequency range (RTL_433 supports 300-928 MHz)
• Manually test with: rtl_433 -r pulse.ook -A

Performance Issues

• Enable caching
• Limit protocols to frequency-appropriate set
• Use specific_protocols parameter
• Consider async processing queue

---

Next Steps After Implementation

1. Expand Protocol Coverage: Add frequency-specific protocol hints
2. ML Enhancement: Use RTL_433 matches as training labels
3. Community Verification: Allow users to confirm/correct RTL_433 IDs
4. Signal Quality Metrics: Extract SNR/RSSI from RTL_433 output
5. Hybrid Matching: Combine RTL_433 + signature matching confidence

---

References

• RTL_433 GitHub: https://github.com/merbanan/rtl_433
• RTL_433 Documentation: https://triq.org/
• Pulse Data Format: include/pulse_data.h
• Decoder API: include/decoder.h
• Protocol List: conf/rtl_433.example.conf