Trilltechnician
48fcb00241
Major Achievements: - ✅ Full web interface (1,520+ lines of frontend code) - ✅ Interactive Leaflet.js map with marker clustering - ✅ Drag-and-drop upload system with GPS input - ✅ Search & filter UI with multi-criteria - ✅ Statistics dashboard with Chart.js - ✅ Responsive mobile-friendly design Backend: - ✅ FastAPI static file serving - ✅ Simplified server mode (main_simple.py) - ✅ Improved startup script with port auto-selection - ✅ PostgreSQL schema ready (requires setup) Database: - ✅ SQLite populated with 85 Flipper Zero signatures - ✅ Device matching system operational - ✅ Frequency-based search working Documentation: - ✅ PHASE_3_COMPLETE.md - Technical summary - ✅ WEB_INTERFACE_README.md - User guide - ✅ WEBAPP_STARTUP_GUIDE.md - Troubleshooting - ✅ POSTGRESQL_SETUP_EXPLANATION.md - DB setup guide - ✅ DATABASE_POPULATION_SUCCESS.md - Import report - ✅ DEVICE_IDENTIFICATION_REPORT.md - Matching analysis Files Created: - templates/index.html (260 lines) - static/css/main.css (500 lines) - static/js/*.js (760 lines total) - src/api/main_simple.py (simplified server) - start_web.sh (auto port selection) Status: Production MVP Ready Next: Phase 4 - API & Integration 🛰️ Generated with Claude Code https://claude.com/claude-code Co-Authored-By: Claude <noreply@anthropic.com>
443 lines
16 KiB
Python
Executable File
443 lines
16 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Deep RF Signal Analysis and Device Identification
|
|
|
|
Analyzes T-Embed captures and identifies likely devices based on:
|
|
- Frequency band
|
|
- Timing patterns
|
|
- Pulse characteristics
|
|
- Known device signatures in the 915 MHz ISM band
|
|
"""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import List, Dict, Any, Tuple
|
|
import statistics
|
|
|
|
# Add project root to path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from src.parser.sub_parser import SubFileParser
|
|
|
|
|
|
class RFSignalAnalyzer:
|
|
"""Deep analysis of RF signals to identify device types"""
|
|
|
|
# Known 915 MHz ISM band devices and their characteristics
|
|
KNOWN_915MHZ_DEVICES = {
|
|
'wireless_sensor': {
|
|
'name': 'Wireless Sensor (Temperature/Humidity)',
|
|
'timing_range': (50, 1500),
|
|
'avg_pulse_range': (200, 600),
|
|
'pulse_count_range': (40, 100),
|
|
'characteristics': ['Regular pulses', 'Short transmission bursts'],
|
|
'manufacturers': ['Acurite', 'La Crosse', 'Oregon Scientific', 'Generic'],
|
|
'confidence_multiplier': 0.9
|
|
},
|
|
'tpms': {
|
|
'name': 'Tire Pressure Monitoring System (TPMS)',
|
|
'timing_range': (30, 800),
|
|
'avg_pulse_range': (100, 400),
|
|
'pulse_count_range': (50, 150),
|
|
'characteristics': ['Periodic transmission', 'Short data packets'],
|
|
'manufacturers': ['Schrader', 'Continental', 'Sensata'],
|
|
'confidence_multiplier': 0.85
|
|
},
|
|
'door_window_sensor': {
|
|
'name': 'Door/Window Security Sensor',
|
|
'timing_range': (100, 2000),
|
|
'avg_pulse_range': (300, 800),
|
|
'pulse_count_range': (20, 80),
|
|
'characteristics': ['On-demand transmission', 'Low duty cycle'],
|
|
'manufacturers': ['SimpliSafe', 'Ring', 'ADT', 'Generic'],
|
|
'confidence_multiplier': 0.8
|
|
},
|
|
'utility_meter': {
|
|
'name': 'Smart Utility Meter',
|
|
'timing_range': (200, 3000),
|
|
'avg_pulse_range': (400, 1200),
|
|
'pulse_count_range': (100, 300),
|
|
'characteristics': ['Regular interval transmission', 'Long packets'],
|
|
'manufacturers': ['Itron', 'Landis+Gyr', 'Sensus'],
|
|
'confidence_multiplier': 0.75
|
|
},
|
|
'motion_sensor': {
|
|
'name': 'Motion Detector / PIR Sensor',
|
|
'timing_range': (50, 1000),
|
|
'avg_pulse_range': (150, 500),
|
|
'pulse_count_range': (30, 90),
|
|
'characteristics': ['Event-triggered', 'Quick bursts'],
|
|
'manufacturers': ['Generic', 'Smart Home Brands'],
|
|
'confidence_multiplier': 0.7
|
|
},
|
|
'remote_control': {
|
|
'name': '915MHz Remote Control',
|
|
'timing_range': (100, 2500),
|
|
'avg_pulse_range': (250, 900),
|
|
'pulse_count_range': (20, 70),
|
|
'characteristics': ['Manual trigger', 'Short commands'],
|
|
'manufacturers': ['Generic', 'Industrial'],
|
|
'confidence_multiplier': 0.65
|
|
},
|
|
'iot_generic': {
|
|
'name': 'Generic IoT Device',
|
|
'timing_range': (10, 5000),
|
|
'avg_pulse_range': (50, 2000),
|
|
'pulse_count_range': (10, 500),
|
|
'characteristics': ['Variable patterns'],
|
|
'manufacturers': ['Various'],
|
|
'confidence_multiplier': 0.5
|
|
}
|
|
}
|
|
|
|
def __init__(self):
|
|
self.parser = SubFileParser()
|
|
|
|
def analyze_file(self, file_path: Path) -> Dict[str, Any]:
|
|
"""Perform deep analysis on a .sub file"""
|
|
|
|
print(f"\n{'='*80}")
|
|
print(f"ANALYZING: {file_path.name}")
|
|
print(f"{'='*80}\n")
|
|
|
|
# Parse file
|
|
try:
|
|
metadata = self.parser.parse(str(file_path))
|
|
except Exception as e:
|
|
return {'error': str(e)}
|
|
|
|
# Check if valid
|
|
if metadata.frequency == 0 or (metadata.file_format == 'RAW' and not metadata.raw_data):
|
|
return {'skipped': True, 'reason': 'Empty capture'}
|
|
|
|
# Basic info
|
|
print("BASIC SIGNAL INFORMATION")
|
|
print("-" * 80)
|
|
print(f"File Type: {metadata.file_type}")
|
|
print(f"Frequency: {metadata.frequency/1e6:.3f} MHz ({metadata.frequency} Hz)")
|
|
print(f"Protocol: {metadata.protocol or 'RAW (undecoded)'}")
|
|
print(f"Format: {metadata.file_format}")
|
|
print(f"Modulation: {metadata.modulation or 'Unknown'}")
|
|
|
|
# Analyze RAW data
|
|
if not metadata.raw_data:
|
|
print("\nNo RAW data to analyze")
|
|
return {'error': 'No RAW data'}
|
|
|
|
analysis = self._analyze_timing(metadata.raw_data)
|
|
|
|
print(f"\nRAW TIMING ANALYSIS")
|
|
print("-" * 80)
|
|
print(f"Total Samples: {analysis['total_samples']}")
|
|
print(f"Pulse Count: {analysis['pulse_count']} (positive values)")
|
|
print(f"Gap Count: {analysis['gap_count']} (negative values)")
|
|
print(f"\nTiming Statistics (microseconds):")
|
|
print(f" Min: {analysis['timing_min']} μs")
|
|
print(f" Max: {analysis['timing_max']} μs")
|
|
print(f" Average: {analysis['timing_avg']:.2f} μs")
|
|
print(f" Median: {analysis['timing_median']:.2f} μs")
|
|
print(f" Std Dev: {analysis['timing_stddev']:.2f} μs")
|
|
print(f"\nPulse Width Analysis:")
|
|
print(f" Avg Pulse: {analysis['avg_pulse_width']:.2f} μs")
|
|
print(f" Avg Gap: {analysis['avg_gap_width']:.2f} μs")
|
|
print(f" Pulse/Gap: {analysis['pulse_gap_ratio']:.2f}")
|
|
|
|
# Pattern analysis
|
|
pattern_analysis = self._analyze_pattern(metadata.raw_data)
|
|
|
|
print(f"\nPATTERN CHARACTERISTICS")
|
|
print("-" * 80)
|
|
print(f"Repeating Patterns: {pattern_analysis['has_repetition']}")
|
|
print(f"Pattern Regularity: {pattern_analysis['regularity']}")
|
|
print(f"Transmission Type: {pattern_analysis['transmission_type']}")
|
|
|
|
# Device identification
|
|
print(f"\n{'='*80}")
|
|
print("DEVICE IDENTIFICATION")
|
|
print(f"{'='*80}\n")
|
|
|
|
# Match against known devices
|
|
matches = self._identify_device(metadata.frequency, analysis, pattern_analysis)
|
|
|
|
if matches:
|
|
print(f"Found {len(matches)} potential match(es):\n")
|
|
|
|
for i, match in enumerate(matches, 1):
|
|
print(f"{i}. {match['name']}")
|
|
print(f" Confidence: {match['confidence']:.1%}")
|
|
print(f" Match Score: {match['score']:.2f}/1.0")
|
|
print(f" Manufacturers: {', '.join(match['manufacturers'])}")
|
|
print(f" Characteristics: {', '.join(match['characteristics'])}")
|
|
print(f"\n Match Details:")
|
|
|
|
for detail_key, detail_val in match['match_details'].items():
|
|
print(f" {detail_key}: {detail_val}")
|
|
|
|
print()
|
|
|
|
# Best match
|
|
best = matches[0]
|
|
print(f"{'='*80}")
|
|
print(f"MOST LIKELY DEVICE: {best['name']}")
|
|
print(f"Confidence: {best['confidence']:.1%}")
|
|
print(f"{'='*80}")
|
|
|
|
else:
|
|
print("❌ No matches found in known device database")
|
|
print("\nThis could be:")
|
|
print(" - A custom/proprietary device")
|
|
print(" - A new/unknown protocol")
|
|
print(" - Interference or noise")
|
|
|
|
return {
|
|
'file': file_path.name,
|
|
'frequency': metadata.frequency,
|
|
'analysis': analysis,
|
|
'pattern': pattern_analysis,
|
|
'matches': matches
|
|
}
|
|
|
|
def _analyze_timing(self, raw_data: List[int]) -> Dict[str, Any]:
|
|
"""Analyze timing characteristics"""
|
|
|
|
abs_timings = [abs(t) for t in raw_data]
|
|
pulses = [t for t in raw_data if t > 0]
|
|
gaps = [abs(t) for t in raw_data if t < 0]
|
|
|
|
analysis = {
|
|
'total_samples': len(raw_data),
|
|
'pulse_count': len(pulses),
|
|
'gap_count': len(gaps),
|
|
'timing_min': min(abs_timings),
|
|
'timing_max': max(abs_timings),
|
|
'timing_avg': statistics.mean(abs_timings),
|
|
'timing_median': statistics.median(abs_timings),
|
|
'timing_stddev': statistics.stdev(abs_timings) if len(abs_timings) > 1 else 0,
|
|
}
|
|
|
|
if pulses:
|
|
analysis['avg_pulse_width'] = statistics.mean(pulses)
|
|
else:
|
|
analysis['avg_pulse_width'] = 0
|
|
|
|
if gaps:
|
|
analysis['avg_gap_width'] = statistics.mean(gaps)
|
|
else:
|
|
analysis['avg_gap_width'] = 0
|
|
|
|
if analysis['avg_gap_width'] > 0:
|
|
analysis['pulse_gap_ratio'] = analysis['avg_pulse_width'] / analysis['avg_gap_width']
|
|
else:
|
|
analysis['pulse_gap_ratio'] = 0
|
|
|
|
return analysis
|
|
|
|
def _analyze_pattern(self, raw_data: List[int]) -> Dict[str, Any]:
|
|
"""Analyze signal patterns"""
|
|
|
|
# Check for repetition
|
|
has_repetition = self._check_repetition(raw_data)
|
|
|
|
# Calculate regularity (coefficient of variation)
|
|
abs_timings = [abs(t) for t in raw_data]
|
|
avg = statistics.mean(abs_timings)
|
|
stddev = statistics.stdev(abs_timings) if len(abs_timings) > 1 else 0
|
|
cv = (stddev / avg) if avg > 0 else 0
|
|
|
|
if cv < 0.5:
|
|
regularity = "High (uniform timing)"
|
|
elif cv < 1.5:
|
|
regularity = "Moderate (some variation)"
|
|
else:
|
|
regularity = "Low (highly variable)"
|
|
|
|
# Determine transmission type
|
|
if cv < 0.7 and has_repetition:
|
|
transmission_type = "Periodic (sensor/beacon)"
|
|
elif cv > 2.0:
|
|
transmission_type = "Bursty (on-demand)"
|
|
else:
|
|
transmission_type = "Mixed (varies)"
|
|
|
|
return {
|
|
'has_repetition': has_repetition,
|
|
'regularity': regularity,
|
|
'coefficient_variation': cv,
|
|
'transmission_type': transmission_type
|
|
}
|
|
|
|
def _check_repetition(self, raw_data: List[int], window_size: int = 10) -> bool:
|
|
"""Check if pattern has repetition"""
|
|
|
|
if len(raw_data) < window_size * 2:
|
|
return False
|
|
|
|
# Simple check: see if first window repeats
|
|
window1 = raw_data[:window_size]
|
|
|
|
for i in range(window_size, len(raw_data) - window_size):
|
|
window2 = raw_data[i:i+window_size]
|
|
|
|
# Check similarity
|
|
matches = sum(1 for j in range(window_size)
|
|
if abs(window1[j] - window2[j]) < abs(window1[j]) * 0.2)
|
|
|
|
if matches >= window_size * 0.7: # 70% similarity
|
|
return True
|
|
|
|
return False
|
|
|
|
def _identify_device(self, frequency: int, timing_analysis: Dict,
|
|
pattern_analysis: Dict) -> List[Dict[str, Any]]:
|
|
"""Identify device based on RF characteristics"""
|
|
|
|
freq_mhz = frequency / 1e6
|
|
|
|
# Only process 915 MHz ISM band
|
|
if not (900 <= freq_mhz <= 930):
|
|
return []
|
|
|
|
matches = []
|
|
|
|
for device_key, device_info in self.KNOWN_915MHZ_DEVICES.items():
|
|
score = 0.0
|
|
match_details = {}
|
|
|
|
# Check timing range
|
|
timing_match = self._check_range_match(
|
|
timing_analysis['timing_avg'],
|
|
device_info['timing_range']
|
|
)
|
|
score += timing_match * 0.3
|
|
match_details['Timing Match'] = f"{timing_match:.1%}"
|
|
|
|
# Check average pulse
|
|
pulse_match = self._check_range_match(
|
|
timing_analysis['avg_pulse_width'],
|
|
device_info['avg_pulse_range']
|
|
)
|
|
score += pulse_match * 0.3
|
|
match_details['Pulse Match'] = f"{pulse_match:.1%}"
|
|
|
|
# Check pulse count
|
|
pulse_count_match = self._check_range_match(
|
|
timing_analysis['pulse_count'],
|
|
device_info['pulse_count_range']
|
|
)
|
|
score += pulse_count_match * 0.2
|
|
match_details['Count Match'] = f"{pulse_count_match:.1%}"
|
|
|
|
# Pattern characteristics bonus
|
|
if 'Periodic' in pattern_analysis['transmission_type'] and 'sensor' in device_key:
|
|
score += 0.1
|
|
match_details['Pattern Bonus'] = 'Periodic transmission (sensor-like)'
|
|
|
|
if 'Bursty' in pattern_analysis['transmission_type'] and 'remote' in device_key:
|
|
score += 0.1
|
|
match_details['Pattern Bonus'] = 'Bursty transmission (control-like)'
|
|
|
|
# Only include if reasonable match
|
|
if score > 0.3:
|
|
confidence = score * device_info['confidence_multiplier']
|
|
|
|
matches.append({
|
|
'device_key': device_key,
|
|
'name': device_info['name'],
|
|
'confidence': confidence,
|
|
'score': score,
|
|
'manufacturers': device_info['manufacturers'],
|
|
'characteristics': device_info['characteristics'],
|
|
'match_details': match_details
|
|
})
|
|
|
|
# Sort by confidence
|
|
matches.sort(key=lambda x: x['confidence'], reverse=True)
|
|
|
|
return matches
|
|
|
|
def _check_range_match(self, value: float, range_tuple: Tuple[float, float]) -> float:
|
|
"""
|
|
Check how well a value fits within a range
|
|
|
|
Returns: 0.0-1.0 score
|
|
"""
|
|
min_val, max_val = range_tuple
|
|
|
|
if min_val <= value <= max_val:
|
|
# Value is within range
|
|
center = (min_val + max_val) / 2
|
|
distance = abs(value - center)
|
|
range_size = (max_val - min_val) / 2
|
|
|
|
# Score decreases as we move from center
|
|
score = 1.0 - (distance / range_size) if range_size > 0 else 1.0
|
|
return max(0.5, score) # At least 0.5 if in range
|
|
|
|
elif value < min_val:
|
|
# Below range
|
|
distance = min_val - value
|
|
return max(0.0, 1.0 - (distance / min_val))
|
|
|
|
else:
|
|
# Above range
|
|
distance = value - max_val
|
|
return max(0.0, 1.0 - (distance / max_val))
|
|
|
|
|
|
def main():
|
|
"""Main entry point"""
|
|
|
|
print("="*80)
|
|
print("T-EMBED RF DEVICE IDENTIFICATION")
|
|
print("Deep Signal Analysis & Device Detection")
|
|
print("="*80)
|
|
|
|
# Find T-Embed files
|
|
tembed_dir = Path(__file__).parent.parent / 'signatures' / 't-embed-rf'
|
|
|
|
if not tembed_dir.exists():
|
|
print(f"❌ Directory not found: {tembed_dir}")
|
|
return 1
|
|
|
|
sub_files = sorted(tembed_dir.glob('*.sub'))
|
|
print(f"\nFound {len(sub_files)} .sub files to analyze\n")
|
|
|
|
analyzer = RFSignalAnalyzer()
|
|
results = []
|
|
|
|
# Analyze each file
|
|
for sub_file in sub_files:
|
|
result = analyzer.analyze_file(sub_file)
|
|
if 'error' not in result and 'skipped' not in result:
|
|
results.append(result)
|
|
|
|
# Final summary
|
|
print(f"\n{'='*80}")
|
|
print("SUMMARY: DEVICES DETECTED")
|
|
print(f"{'='*80}\n")
|
|
|
|
if results:
|
|
for i, result in enumerate(results, 1):
|
|
print(f"{i}. {result['file']}")
|
|
print(f" Frequency: {result['frequency']/1e6:.2f} MHz")
|
|
|
|
if result['matches']:
|
|
best_match = result['matches'][0]
|
|
print(f" Identified: {best_match['name']}")
|
|
print(f" Confidence: {best_match['confidence']:.1%}")
|
|
print(f" Likely Manufacturer: {best_match['manufacturers'][0]}")
|
|
else:
|
|
print(f" Identified: Unknown device")
|
|
|
|
print()
|
|
else:
|
|
print("No valid devices detected (all files were empty or parse errors)\n")
|
|
|
|
print("="*80)
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|