leetcrypt
4396c745a8
Reconstruct a signed pulse train from BinRAW Data_RAW demodulated bits (run-length encoded at TE-µs steps) so BinRAW files populate raw_data and flow through the existing RAW identification pipeline. Previously 0/120 BinRAW files yielded a predicted_category; now 118/120 categorize via the live SignatureMatcher engine. RAW code path is unchanged. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
348 lines
11 KiB
Python
348 lines
11 KiB
Python
"""
|
|
Flipper Zero .sub file parser
|
|
|
|
Parses both KEY and RAW format .sub files and extracts signal metadata
|
|
"""
|
|
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Optional, Dict, Any, List
|
|
from loguru import logger
|
|
|
|
from .metadata import SignalMetadata, PresetInfo
|
|
|
|
|
|
class SubFileParser:
|
|
"""Parser for Flipper Zero .sub files"""
|
|
|
|
def __init__(self):
|
|
self.reset()
|
|
|
|
def reset(self):
|
|
"""Reset parser state"""
|
|
self.fields = {}
|
|
self.errors = []
|
|
|
|
def parse(self, file_path: str) -> SignalMetadata:
|
|
"""
|
|
Parse a .sub file and extract metadata
|
|
|
|
Args:
|
|
file_path: Path to .sub file
|
|
|
|
Returns:
|
|
SignalMetadata object
|
|
|
|
Raises:
|
|
FileNotFoundError: If file doesn't exist
|
|
ValueError: If file format is invalid
|
|
"""
|
|
self.reset()
|
|
path = Path(file_path)
|
|
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"File not found: {file_path}")
|
|
|
|
# Read file
|
|
try:
|
|
with open(path, 'r', encoding='utf-8') as f:
|
|
content = f.read()
|
|
except Exception as e:
|
|
raise ValueError(f"Failed to read file: {e}")
|
|
|
|
# Parse fields
|
|
for line in content.split('\n'):
|
|
line = line.strip()
|
|
if not line or line.startswith('#'):
|
|
continue
|
|
|
|
if ':' in line:
|
|
key, value = line.split(':', 1)
|
|
self.fields[key.strip()] = value.strip()
|
|
|
|
# Validate required fields
|
|
if 'Filetype' not in self.fields:
|
|
raise ValueError("Missing required field: Filetype")
|
|
|
|
if 'Frequency' not in self.fields:
|
|
raise ValueError("Missing required field: Frequency")
|
|
|
|
# Extract metadata
|
|
return self._extract_metadata()
|
|
|
|
def _extract_metadata(self) -> SignalMetadata:
|
|
"""Extract metadata from parsed fields"""
|
|
|
|
# Basic fields
|
|
file_type = self.fields.get('Filetype', '')
|
|
version = int(self.fields.get('Version', 1))
|
|
frequency = int(self.fields.get('Frequency', 0))
|
|
preset = self.fields.get('Preset')
|
|
|
|
# Determine file format
|
|
protocol = self.fields.get('Protocol', 'Unknown')
|
|
if protocol == 'RAW':
|
|
file_format = 'RAW'
|
|
elif protocol == 'BinRAW':
|
|
file_format = 'BinRAW'
|
|
else:
|
|
file_format = 'KEY'
|
|
|
|
# Extract modulation from preset
|
|
modulation = None
|
|
if preset:
|
|
preset_info = PresetInfo.from_preset_name(preset)
|
|
modulation = preset_info.modulation
|
|
|
|
# Initialize metadata
|
|
metadata = SignalMetadata(
|
|
file_type=file_type,
|
|
version=version,
|
|
file_format=file_format,
|
|
frequency=frequency,
|
|
preset=preset,
|
|
modulation=modulation,
|
|
protocol=protocol if protocol not in ['RAW', 'BinRAW'] else None
|
|
)
|
|
|
|
# Extract format-specific fields
|
|
if file_format == 'KEY':
|
|
self._extract_key_fields(metadata)
|
|
elif file_format == 'RAW':
|
|
self._extract_raw_fields(metadata)
|
|
elif file_format == 'BinRAW':
|
|
self._extract_binraw_fields(metadata)
|
|
|
|
# Extract custom preset if present
|
|
if 'Custom_preset_module' in self.fields:
|
|
metadata.custom_preset_module = self.fields['Custom_preset_module']
|
|
if 'Custom_preset_data' in self.fields:
|
|
metadata.custom_preset_data = bytes.fromhex(
|
|
self.fields['Custom_preset_data'].replace(' ', '')
|
|
)
|
|
|
|
# Add any parsing errors
|
|
metadata.parse_errors = self.errors
|
|
|
|
return metadata
|
|
|
|
def _extract_key_fields(self, metadata: SignalMetadata):
|
|
"""Extract fields from KEY format file"""
|
|
try:
|
|
if 'Bit' in self.fields:
|
|
metadata.bit_length = int(self.fields['Bit'])
|
|
|
|
if 'Key' in self.fields:
|
|
# Parse hex key data
|
|
key_hex = self.fields['Key'].replace(' ', '')
|
|
metadata.key_data = bytes.fromhex(key_hex)
|
|
|
|
if 'TE' in self.fields:
|
|
metadata.timing_element = int(self.fields['TE'])
|
|
|
|
except ValueError as e:
|
|
self.errors.append(f"Error parsing KEY fields: {e}")
|
|
logger.warning(f"Parse error: {e}")
|
|
|
|
def _extract_raw_fields(self, metadata: SignalMetadata):
|
|
"""Extract fields from RAW format file"""
|
|
try:
|
|
if 'RAW_Data' in self.fields:
|
|
# Parse timing array
|
|
raw_str = self.fields['RAW_Data']
|
|
timings = [int(x) for x in raw_str.split()]
|
|
metadata.raw_data = timings
|
|
|
|
# Calculate statistics
|
|
metadata.pulse_count = len(timings)
|
|
metadata.duration_ms = sum(abs(t) for t in timings) / 1000.0
|
|
|
|
# Average pulse width (positive values only)
|
|
positive_pulses = [t for t in timings if t > 0]
|
|
if positive_pulses:
|
|
metadata.avg_pulse_width = sum(positive_pulses) / len(positive_pulses)
|
|
|
|
except ValueError as e:
|
|
self.errors.append(f"Error parsing RAW fields: {e}")
|
|
logger.warning(f"Parse error: {e}")
|
|
|
|
def _extract_binraw_fields(self, metadata: SignalMetadata):
|
|
"""Extract fields from BinRAW format file"""
|
|
try:
|
|
if 'Bit' in self.fields:
|
|
metadata.bit_length = int(self.fields['Bit'])
|
|
|
|
if 'TE' in self.fields:
|
|
metadata.bin_raw_te = int(self.fields['TE'])
|
|
|
|
if 'Bit_RAW' in self.fields:
|
|
metadata.bin_raw_bit = int(self.fields['Bit_RAW'])
|
|
|
|
if 'Data_RAW' in self.fields:
|
|
data_hex = self.fields['Data_RAW'].replace(' ', '')
|
|
metadata.bin_raw_data = bytes.fromhex(data_hex)
|
|
|
|
# Reconstruct a signed pulse train (µs) from the demodulated bit
|
|
# stream so BinRAW captures flow through the same RAW identification
|
|
# path as RAW_Data files. Data_RAW is the signal sampled at TE-µs
|
|
# steps: a run of N identical bits is one pulse of N*TE µs (+high on
|
|
# 1s, -low on 0s). This populates raw_data, which makes the
|
|
# has_raw_data property true.
|
|
n_bits = metadata.bin_raw_bit or metadata.bit_length
|
|
if metadata.bin_raw_data and metadata.bin_raw_te and n_bits:
|
|
pulses = self._binraw_to_pulses(
|
|
metadata.bin_raw_data, n_bits, metadata.bin_raw_te)
|
|
if pulses:
|
|
metadata.raw_data = pulses
|
|
metadata.pulse_count = len(pulses)
|
|
|
|
# Calculate duration
|
|
if metadata.bit_length and metadata.bin_raw_te:
|
|
metadata.duration_ms = (metadata.bit_length * metadata.bin_raw_te) / 1000.0
|
|
|
|
except ValueError as e:
|
|
self.errors.append(f"Error parsing BinRAW fields: {e}")
|
|
logger.warning(f"Parse error: {e}")
|
|
|
|
@staticmethod
|
|
def _binraw_to_pulses(data: bytes, n_bits: int, te: int) -> List[int]:
|
|
"""Expand a BinRAW bit stream into a signed timing array (µs).
|
|
|
|
Bits are read MSB-first, truncated to ``n_bits``, then run-length
|
|
encoded: each run of ``k`` equal bits becomes ``k*te`` µs, signed
|
|
positive for 1s (pulse) and negative for 0s (gap).
|
|
"""
|
|
bits = []
|
|
for byte in data:
|
|
for i in range(7, -1, -1):
|
|
bits.append((byte >> i) & 1)
|
|
bits = bits[:n_bits]
|
|
if not bits:
|
|
return []
|
|
pulses: List[int] = []
|
|
current, run = bits[0], 1
|
|
for b in bits[1:]:
|
|
if b == current:
|
|
run += 1
|
|
else:
|
|
pulses.append(run * te * (1 if current else -1))
|
|
current, run = b, 1
|
|
pulses.append(run * te * (1 if current else -1))
|
|
return pulses
|
|
|
|
|
|
def parse_sub_file(file_path: str) -> SignalMetadata:
|
|
"""
|
|
Convenience function to parse a .sub file
|
|
|
|
Args:
|
|
file_path: Path to .sub file
|
|
|
|
Returns:
|
|
SignalMetadata object
|
|
"""
|
|
parser = SubFileParser()
|
|
return parser.parse(file_path)
|
|
|
|
|
|
def extract_protocol_features(metadata: SignalMetadata) -> Dict[str, Any]:
|
|
"""
|
|
Extract features for protocol matching
|
|
|
|
Args:
|
|
metadata: Parsed signal metadata
|
|
|
|
Returns:
|
|
Dictionary of matchable features
|
|
"""
|
|
features = {
|
|
'frequency': metadata.frequency,
|
|
'modulation': metadata.modulation,
|
|
'protocol': metadata.protocol,
|
|
}
|
|
|
|
if metadata.is_decoded:
|
|
# For decoded signals
|
|
features.update({
|
|
'bit_length': metadata.bit_length,
|
|
'timing_element': metadata.timing_element,
|
|
'key_pattern': metadata.key_data[:4] if metadata.key_data else None # First 4 bytes
|
|
})
|
|
|
|
elif metadata.has_raw_data:
|
|
# For RAW signals, extract timing characteristics
|
|
features.update({
|
|
'pulse_count': metadata.pulse_count,
|
|
'avg_pulse_width': metadata.avg_pulse_width,
|
|
'duration_ms': metadata.duration_ms,
|
|
'timing_pattern': _extract_timing_pattern(metadata.raw_data)
|
|
})
|
|
|
|
return features
|
|
|
|
|
|
def _extract_timing_pattern(raw_data: List[int], max_samples: int = 20) -> List[int]:
|
|
"""
|
|
Extract representative timing pattern from RAW data
|
|
|
|
Args:
|
|
raw_data: Array of timing values
|
|
max_samples: Maximum number of samples to return
|
|
|
|
Returns:
|
|
Representative timing pattern
|
|
"""
|
|
if not raw_data:
|
|
return []
|
|
|
|
# Take first few pulses as pattern
|
|
pattern = raw_data[:max_samples]
|
|
|
|
# Normalize to absolute values
|
|
return [abs(t) for t in pattern]
|
|
|
|
|
|
def validate_sub_file(file_path: str) -> tuple[bool, List[str]]:
|
|
"""
|
|
Validate a .sub file without full parsing
|
|
|
|
Args:
|
|
file_path: Path to .sub file
|
|
|
|
Returns:
|
|
Tuple of (is_valid, error_messages)
|
|
"""
|
|
errors = []
|
|
|
|
try:
|
|
with open(file_path, 'r', encoding='utf-8') as f:
|
|
content = f.read()
|
|
|
|
# Check for required header
|
|
if 'Filetype:' not in content:
|
|
errors.append("Missing Filetype field")
|
|
|
|
if 'Frequency:' not in content:
|
|
errors.append("Missing Frequency field")
|
|
|
|
if 'Version:' not in content:
|
|
errors.append("Missing Version field")
|
|
|
|
# Check file type
|
|
if 'Flipper SubGhz' not in content:
|
|
errors.append("Not a valid Flipper SubGhz file")
|
|
|
|
# Validate frequency value
|
|
freq_match = re.search(r'Frequency:\s*(\d+)', content)
|
|
if freq_match:
|
|
freq = int(freq_match.group(1))
|
|
if freq < 300_000_000 or freq > 928_000_000:
|
|
errors.append(f"Frequency out of range: {freq} Hz")
|
|
|
|
except FileNotFoundError:
|
|
errors.append("File not found")
|
|
except Exception as e:
|
|
errors.append(f"Validation error: {e}")
|
|
|
|
return (len(errors) == 0, errors)
|