giglez/src/parser/raw_parser.py

"""
RAW Signal Data Parser

Parses RAW_Data from Flipper Zero .sub files to extract timing signatures
for protocol identification.

Author: GigLez Team
Date: January 2026
"""

import re
import logging
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass
import statistics

logger = logging.getLogger(__name__)


@dataclass
class TimingSignature:
    """Timing signature extracted from RAW data"""
    pulses: List[int]  # All pulse durations (absolute values)
    high_pulses: List[int]  # RF-on pulse durations
    low_pulses: List[int]  # RF-off pulse durations

    mean_high: float
    mean_low: float
    std_high: float
    std_low: float

    short_pulse: int  # Likely short pulse width
    long_pulse: int  # Likely long pulse width
    gap: Optional[int]  # Gap between transmissions

    pulse_ratio: float  # long / short ratio
    encoding_type: str  # PWM, PPM, Manchester, etc.

    total_pulses: int
    duration_ms: float


class RAWParser:
    """
    Parser for Flipper Zero RAW_Data format

    RAW_Data format: space-separated integers
    - Positive: RF on duration (microseconds)
    - Negative: RF off duration (microseconds)

    Example:
        "2980 -240 520 -980 520 -980 980 -520 ..."
    """

    # Encoding detection thresholds
    PWM_RATIO_MIN = 1.5  # Long/Short ratio > 1.5 suggests PWM
    PPM_RATIO_MIN = 2.5  # Long/Short ratio > 2.5 suggests PPM
    MANCHESTER_RATIO_MAX = 1.3  # Long/Short ratio < 1.3 suggests Manchester

    def __init__(self):
        """Initialize RAW parser"""
        pass

    def parse(self, raw_data: str) -> Optional[TimingSignature]:
        """
        Parse RAW_Data string to extract timing signature

        Args:
            raw_data: RAW_Data string from .sub file

        Returns:
            TimingSignature or None if parsing fails
        """
        try:
            # Parse integers from string
            values = self._parse_raw_string(raw_data)
            if not values or len(values) < 10:
                logger.warning(f"Insufficient RAW data: {len(values) if values else 0} values")
                return None

            # Separate high and low pulses
            high_pulses = [v for v in values if v > 0]
            low_pulses = [abs(v) for v in values if v < 0]

            if not high_pulses or not low_pulses:
                logger.warning("No high or low pulses found")
                return None

            # Calculate statistics
            mean_high = statistics.mean(high_pulses)
            mean_low = statistics.mean(low_pulses)
            std_high = statistics.stdev(high_pulses) if len(high_pulses) > 1 else 0
            std_low = statistics.stdev(low_pulses) if len(low_pulses) > 1 else 0

            # Identify short and long pulses (cluster analysis)
            short_pulse, long_pulse = self._identify_pulse_widths(high_pulses)

            # Identify gap (longest low pulse, if significantly longer)
            gap = self._identify_gap(low_pulses)

            # Calculate pulse ratio
            pulse_ratio = long_pulse / short_pulse if short_pulse > 0 else 1.0

            # Detect encoding type
            encoding_type = self._detect_encoding(pulse_ratio, high_pulses, low_pulses)

            # Calculate total duration
            duration_ms = sum(abs(v) for v in values) / 1000.0

            signature = TimingSignature(
                pulses=[abs(v) for v in values],
                high_pulses=high_pulses,
                low_pulses=low_pulses,
                mean_high=mean_high,
                mean_low=mean_low,
                std_high=std_high,
                std_low=std_low,
                short_pulse=short_pulse,
                long_pulse=long_pulse,
                gap=gap,
                pulse_ratio=pulse_ratio,
                encoding_type=encoding_type,
                total_pulses=len(values),
                duration_ms=duration_ms
            )

            logger.debug(f"Parsed RAW signal: {short_pulse}µs/{long_pulse}µs, "
                        f"ratio={pulse_ratio:.2f}, encoding={encoding_type}")

            return signature

        except Exception as e:
            logger.error(f"Failed to parse RAW data: {e}")
            return None

    def _parse_raw_string(self, raw_data: str) -> List[int]:
        """Parse RAW_Data string to list of integers"""
        # Remove any non-numeric characters except spaces and minus signs
        cleaned = re.sub(r'[^0-9\s\-]', '', raw_data)

        # Split and convert to integers
        values = []
        for token in cleaned.split():
            try:
                values.append(int(token))
            except ValueError:
                continue

        return values

    def _identify_pulse_widths(self, pulses: List[int]) -> Tuple[int, int]:
        """
        Identify short and long pulse widths using clustering

        Uses simple percentile-based clustering:
        - Short pulse: 25th percentile
        - Long pulse: 75th percentile

        Args:
            pulses: List of pulse durations

        Returns:
            (short_pulse, long_pulse) in microseconds
        """
        if not pulses:
            return (0, 0)

        sorted_pulses = sorted(pulses)

        # Use percentiles for robust estimation
        short_idx = len(sorted_pulses) // 4  # 25th percentile
        long_idx = (3 * len(sorted_pulses)) // 4  # 75th percentile

        short_pulse = sorted_pulses[short_idx]
        long_pulse = sorted_pulses[long_idx]

        # Ensure they're different
        if short_pulse == long_pulse:
            # Fall back to min/max
            short_pulse = min(pulses)
            long_pulse = max(pulses)

        return (short_pulse, long_pulse)

    def _identify_gap(self, low_pulses: List[int]) -> Optional[int]:
        """
        Identify gap between transmissions

        Gap is typically the longest low pulse, if significantly longer
        than median low pulse.

        Args:
            low_pulses: List of RF-off durations

        Returns:
            Gap duration in microseconds or None
        """
        if not low_pulses or len(low_pulses) < 3:
            return None

        median = statistics.median(low_pulses)
        max_pulse = max(low_pulses)

        # Gap is significantly longer than median (3x threshold)
        if max_pulse > median * 3:
            return max_pulse

        return None

    def _detect_encoding(self, pulse_ratio: float,
                        high_pulses: List[int],
                        low_pulses: List[int]) -> str:
        """
        Detect encoding type from timing characteristics

        Encoding types:
        - PWM (Pulse Width Modulation): Different pulse widths (1.5 < ratio < 2.5)
        - PPM (Pulse Position Modulation): Very different pulse widths (ratio > 2.5)
        - Manchester: Similar pulse widths (ratio < 1.3)
        - OOK (On-Off Keying): Generic fallback

        Args:
            pulse_ratio: Long pulse / Short pulse ratio
            high_pulses: RF-on pulses
            low_pulses: RF-off pulses

        Returns:
            Encoding type string
        """
        # Manchester encoding: transitions at every bit
        if pulse_ratio < self.MANCHESTER_RATIO_MAX:
            # Check for consistent timing
            if statistics.stdev(high_pulses) < statistics.mean(high_pulses) * 0.3:
                return "Manchester"

        # PPM: position encoding
        if pulse_ratio > self.PPM_RATIO_MIN:
            return "PPM"

        # PWM: width encoding
        if pulse_ratio > self.PWM_RATIO_MIN:
            return "PWM"

        # Default to OOK
        return "OOK"

    def extract_te(self, signature: TimingSignature) -> int:
        """
        Extract TE (Timing Element) - the base timing unit

        For many protocols, TE is the GCD of pulse widths.
        Simpler approach: use short pulse as TE.

        Args:
            signature: Timing signature

        Returns:
            TE in microseconds
        """
        return signature.short_pulse


def parse_raw_data(raw_data: str) -> Optional[TimingSignature]:
    """
    Convenience function to parse RAW data

    Args:
        raw_data: RAW_Data string from .sub file

    Returns:
        TimingSignature or None
    """
    parser = RAWParser()
    return parser.parse(raw_data)


if __name__ == '__main__':
    # Test the parser
    logging.basicConfig(level=logging.DEBUG)

    print("\n" + "="*60)
    print("RAW DATA PARSER TEST")
    print("="*60)

    # Test 1: PWM signal
    test_pwm = "2980 -240 520 -980 520 -980 980 -520 520 -980 980 -520 520 -3000"
    print("\n1. PWM Signal Test")
    print(f"   Input: {test_pwm[:50]}...")

    signature = parse_raw_data(test_pwm)
    if signature:
        print(f"   Short pulse: {signature.short_pulse}µs")
        print(f"   Long pulse: {signature.long_pulse}µs")
        print(f"   Pulse ratio: {signature.pulse_ratio:.2f}")
        print(f"   Encoding: {signature.encoding_type}")
        print(f"   Gap: {signature.gap}µs" if signature.gap else "   Gap: None")
        print(f"   Total pulses: {signature.total_pulses}")
        print(f"   Duration: {signature.duration_ms:.1f}ms")

    # Test 2: Manchester-like signal
    test_manchester = "500 -500 500 -500 1000 -500 500 -1000 500 -500"
    print("\n2. Manchester-like Signal Test")
    print(f"   Input: {test_manchester}")

    signature = parse_raw_data(test_manchester)
    if signature:
        print(f"   Short pulse: {signature.short_pulse}µs")
        print(f"   Long pulse: {signature.long_pulse}µs")
        print(f"   Pulse ratio: {signature.pulse_ratio:.2f}")
        print(f"   Encoding: {signature.encoding_type}")

    # Test 3: Real Flipper Zero capture
    test_real = "7960 -3980 396 -796 400 -392 400 -792 400 -396 796 -396 400 -792 796 -396"
    print("\n3. Real Flipper Capture Test")
    print(f"   Input: {test_real}")

    signature = parse_raw_data(test_real)
    if signature:
        print(f"   Short pulse: {signature.short_pulse}µs")
        print(f"   Long pulse: {signature.long_pulse}µs")
        print(f"   Pulse ratio: {signature.pulse_ratio:.2f}")
        print(f"   Encoding: {signature.encoding_type}")
        print(f"   Mean high: {signature.mean_high:.0f}µs")
        print(f"   Mean low: {signature.mean_low:.0f}µs")

    print("\n" + "="*60)