Initial commit: Phase 1 & Phase 2 infrastructure complete
This commit is contained in:
@@ -0,0 +1,307 @@
|
||||
"""
|
||||
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)
|
||||
|
||||
# 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}")
|
||||
|
||||
|
||||
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)
|
||||
Reference in New Issue
Block a user