giglez/tests/benchmark/test_data_generator.py

#!/usr/bin/env python3
"""
Synthetic Test Signal Generator for Benchmarking

Generates .sub files with known protocols for accuracy testing.
"""

import random
from pathlib import Path
from typing import List, Dict, Tuple


class SyntheticSignalGenerator:
    """Generate synthetic RF signals for known protocols"""

    def __init__(self, output_dir: Path):
        self.output_dir = output_dir
        self.output_dir.mkdir(parents=True, exist_ok=True)

    def generate_pwm_signal(
        self,
        protocol_name: str,
        frequency: int,
        short_pulse: int,
        long_pulse: int,
        short_gap: int,
        bit_pattern: str,
        preamble: str = "",
        noise_level: float = 0.05
    ) -> Path:
        """
        Generate PWM-encoded signal (SHORT=0, LONG=1)

        Args:
            protocol_name: Protocol name for filename
            frequency: Center frequency in Hz
            short_pulse: Short pulse duration (μs)
            long_pulse: Long pulse duration (μs)
            short_gap: Gap duration (μs)
            bit_pattern: Binary pattern to encode
            preamble: Preamble pattern (optional)
            noise_level: Jitter as percentage (0.0-1.0)

        Returns:
            Path to generated .sub file
        """
        pulses = []

        # Add preamble if specified
        if preamble:
            for bit in preamble:
                if bit == '1':
                    pulses.extend([self._jitter(long_pulse, noise_level), -short_gap])
                else:
                    pulses.extend([self._jitter(short_pulse, noise_level), -short_gap])

        # Encode bit pattern
        for bit in bit_pattern:
            if bit == '1':
                pulses.extend([self._jitter(long_pulse, noise_level), -short_gap])
            else:
                pulses.extend([self._jitter(short_pulse, noise_level), -short_gap])

        # Write .sub file
        filename = f"{protocol_name.lower().replace(' ', '_')}_synthetic.sub"
        filepath = self.output_dir / filename

        content = f"""Filetype: Flipper SubGhz RAW File
Version: 1
Frequency: {frequency}
Preset: FuriHalSubGhzPresetOok650Async
Protocol: RAW
RAW_Data: {' '.join(map(str, pulses))}
"""

        filepath.write_text(content)
        return filepath

    def generate_long_burst_signal(
        self,
        protocol_name: str,
        frequency: int,
        burst_duration: int,
        short_pulse: int,
        long_pulse: int,
        bit_pattern: str,
        noise_level: float = 0.05
    ) -> Path:
        """Generate signal with long burst preamble (Princeton style)"""
        pulses = []

        # Long burst preamble
        pulses.append(self._jitter(burst_duration, noise_level))
        pulses.append(-short_pulse)

        # Data
        for bit in bit_pattern:
            if bit == '1':
                pulses.extend([self._jitter(long_pulse, noise_level), -short_pulse])
            else:
                pulses.extend([self._jitter(short_pulse, noise_level), -short_pulse])

        filename = f"{protocol_name.lower().replace(' ', '_')}_synthetic.sub"
        filepath = self.output_dir / filename

        content = f"""Filetype: Flipper SubGhz RAW File
Version: 1
Frequency: {frequency}
Preset: FuriHalSubGhzPresetOok650Async
Protocol: RAW
RAW_Data: {' '.join(map(str, pulses))}
"""

        filepath.write_text(content)
        return filepath

    def _jitter(self, value: int, noise_level: float) -> int:
        """Add random jitter to pulse duration"""
        if noise_level == 0:
            return value
        jitter = random.gauss(0, noise_level)
        return int(value * (1 + jitter))

    def generate_test_suite(self) -> List[Tuple[Path, str, Dict]]:
        """
        Generate comprehensive test suite with known protocols

        Returns:
            List of (filepath, expected_protocol, metadata)
        """
        test_cases = []

        # === Weather Sensors ===

        # 1. LaCrosse TX141-BV2 (433.92 MHz, PWM 500/1000μs)
        filepath = self.generate_pwm_signal(
            protocol_name="LaCrosse TX141-BV2",
            frequency=433920000,
            short_pulse=500,
            long_pulse=1000,
            short_gap=500,
            bit_pattern="10101010" + "11001100" * 4,  # 40 bits
            preamble="10101010",  # Alternating preamble
            noise_level=0.05
        )
        test_cases.append((filepath, "LaCrosse TX141-BV2", {
            'category': 'weather_sensor',
            'frequency': 433920000,
            'encoding': 'PWM',
            'timing': '500/1000μs'
        }))

        # 2. Acurite 609TXC (433.92 MHz, PWM 1000/2000μs - CORRECTED)
        filepath = self.generate_pwm_signal(
            protocol_name="Acurite 609TXC",
            frequency=433920000,
            short_pulse=1000,  # FIXED: Was 500
            long_pulse=2000,   # FIXED: Was 1000
            short_gap=1000,
            bit_pattern="1100" * 10,  # 40 bits
            preamble="1010",
            noise_level=0.05
        )
        test_cases.append((filepath, "Acurite 609TXC", {
            'category': 'weather_sensor',
            'frequency': 433920000
        }))

        # 3. Oregon Scientific v2.1 (433.92 MHz, Manchester 488/976μs - CORRECTED)
        filepath = self.generate_pwm_signal(
            protocol_name="Oregon Scientific v2.1",
            frequency=433920000,
            short_pulse=488,   # FIXED: Was 500
            long_pulse=976,    # FIXED: Was 1000
            short_gap=488,
            bit_pattern="1000" + "11001010" * 6,  # Sync word + data (56 bits total)
            preamble="10101010" * 4,  # 32-bit preamble
            noise_level=0.05
        )
        test_cases.append((filepath, "Oregon Scientific v2.1", {
            'category': 'weather_sensor',
            'frequency': 433920000,
            'encoding': 'Manchester'
        }))

        # 4. Nexus-TH (433.92 MHz)
        filepath = self.generate_pwm_signal(
            protocol_name="Nexus Temperature-Humidity",
            frequency=433920000,
            short_pulse=500,
            long_pulse=1000,
            short_gap=500,
            bit_pattern="11110000" * 5,
            preamble="1111",
            noise_level=0.05
        )
        test_cases.append((filepath, "Nexus Temperature-Humidity", {
            'category': 'weather_sensor',
            'frequency': 433920000
        }))

        # === Garage Door Openers ===

        # 5. Princeton (315 MHz, long burst preamble)
        filepath = self.generate_long_burst_signal(
            protocol_name="Princeton",
            frequency=315000000,
            burst_duration=4000,  # 4ms burst
            short_pulse=400,
            long_pulse=1200,
            bit_pattern="110101101001" * 2,  # 24 bits
            noise_level=0.05
        )
        test_cases.append((filepath, "Princeton", {
            'category': 'garage_door',
            'frequency': 315000000,
            'preamble_type': 'long_burst'
        }))

        # 6. PT2262 (433.92 MHz, long burst)
        filepath = self.generate_long_burst_signal(
            protocol_name="PT2262",
            frequency=433920000,
            burst_duration=3200,
            short_pulse=350,
            long_pulse=1050,
            bit_pattern="1111000011110000" + "10101010",
            noise_level=0.05
        )
        test_cases.append((filepath, "PT2262", {
            'category': 'remote_control',
            'frequency': 433920000
        }))

        # === Tire Pressure Sensors ===

        # 7. Schrader TPMS (315 MHz)
        filepath = self.generate_pwm_signal(
            protocol_name="Schrader TPMS",
            frequency=315000000,
            short_pulse=100,
            long_pulse=200,
            short_gap=100,
            bit_pattern="10101100" * 8,  # 64 bits
            preamble="10101010" * 2,
            noise_level=0.03
        )
        test_cases.append((filepath, "Schrader TPMS", {
            'category': 'tire_pressure',
            'frequency': 315000000
        }))

        # 8. Toyota TPMS (315 MHz)
        filepath = self.generate_pwm_signal(
            protocol_name="Toyota TPMS",
            frequency=315000000,
            short_pulse=100,
            long_pulse=200,
            short_gap=100,
            bit_pattern="11001100" * 9,  # 72 bits
            preamble="1111000011110000",
            noise_level=0.03
        )
        test_cases.append((filepath, "Toyota TPMS", {
            'category': 'tire_pressure',
            'frequency': 315000000
        }))

        # === Security Sensors (868 MHz) ===

        # 9. Honeywell Security (868 MHz)
        filepath = self.generate_pwm_signal(
            protocol_name="Honeywell Security",
            frequency=868000000,
            short_pulse=250,
            long_pulse=750,
            short_gap=250,
            bit_pattern="10110011" * 6,
            preamble="1010" * 4,
            noise_level=0.04
        )
        test_cases.append((filepath, "Honeywell Security", {
            'category': 'security',
            'frequency': 868000000
        }))

        # === Doorbells ===

        # 10. Generic Doorbell (433.92 MHz)
        filepath = self.generate_pwm_signal(
            protocol_name="Generic Doorbell",
            frequency=433920000,
            short_pulse=300,
            long_pulse=900,
            short_gap=300,
            bit_pattern="110011001100" * 2,
            preamble="1111",
            noise_level=0.06
        )
        test_cases.append((filepath, "Generic Doorbell", {
            'category': 'doorbell',
            'frequency': 433920000
        }))

        # === Noisy Signals (High Jitter) ===

        # 11. Noisy LaCrosse (test noise tolerance)
        filepath = self.generate_pwm_signal(
            protocol_name="LaCrosse TX141-BV2 Noisy",
            frequency=433920000,
            short_pulse=500,
            long_pulse=1000,
            short_gap=500,
            bit_pattern="10101010" + "11001100" * 4,
            preamble="10101010",
            noise_level=0.15  # 15% jitter
        )
        test_cases.append((filepath, "LaCrosse TX141-BV2", {
            'category': 'weather_sensor',
            'frequency': 433920000,
            'noise': 'high'
        }))

        # 12. Noisy Princeton
        filepath = self.generate_long_burst_signal(
            protocol_name="Princeton Noisy",
            frequency=315000000,
            burst_duration=4000,
            short_pulse=400,
            long_pulse=1200,
            bit_pattern="110101101001" * 2,
            noise_level=0.12
        )
        test_cases.append((filepath, "Princeton", {
            'category': 'garage_door',
            'frequency': 315000000,
            'noise': 'high'
        }))

        return test_cases


if __name__ == '__main__':
    # Generate test suite
    output_dir = Path(__file__).parent / "synthetic_signals"
    generator = SyntheticSignalGenerator(output_dir)

    print("=== Generating Synthetic Test Signals ===")
    print()

    test_cases = generator.generate_test_suite()

    print(f"Generated {len(test_cases)} test signals:")
    for filepath, expected_protocol, metadata in test_cases:
        print(f"  ✓ {filepath.name} → {expected_protocol}")

    print()
    print(f"Output directory: {output_dir}")
    print(f"Total files: {len(list(output_dir.glob('*.sub')))}")