Phase 1 & 2: Cleanup redundant code and integrate pattern decoder
## Phase 1: Code Cleanup (~1,859 lines removed) **Deleted Redundant Matchers:** - ❌ strategies_orm.py (356 lines) - Old ORM-based strategies - ❌ simple_matcher.py (351 lines) - Replaced by strategies.py - ❌ rtl433_matcher.py (352 lines) - Replaced by strategies.py **Archived Old Scripts:** - Moved 11 one-time analysis/import scripts to scripts/archive/ - Scripts: analyze_flipper_signatures, analyze_tembed_files, identify_tembed_devices, import_flipper_sqlite, import_tembed_signatures, match_tembed_with_db, match_with_flipper_db, rematch_captures, test_gps_extraction, test_tembed_matching, test_wardriving_import **Consolidated API:** - Renamed main.py → main_orm_legacy.py (archived old ORM-based API) - main_simple.py is now the primary production API ## Phase 2: Pattern Decoder Integration ✅ **CRITICAL FIX: Pattern decoder now integrated into production API!** **Changes:** 1. Updated main_simple.py to use unified SignatureMatcher 2. Added 6 strategies to matcher pipeline: - ExactMatcher (protocol + frequency) - FrequencyMatcher (frequency-based) - BitPatternMatcher (data patterns) - TimingMatcher (timing-based) - RTL433DecoderStrategy (RTL_433 decoder) - PatternBasedStrategy (NEW! Pattern decoder for short captures) 3. Created MockDB class for simplified mode (no real database) 4. Replaced old get_matcher() with get_matcher_engine() 5. Updated upload handler to use MatchResult format 6. All matches now include confidence scores and match methods **Result:** - Pattern decoder is NOW ACTIVE in production 🎉 - Unified matching pipeline with 6 strategies - Cleaner codebase (-1,859 lines) - Single source of truth for matching logic 🎉 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Executable
+150
@@ -0,0 +1,150 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test Wardriving Data Import
|
||||
|
||||
Tests importing GPS-tagged SubGhz captures from various formats
|
||||
"""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# Add project root to path
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||
|
||||
from src.parser.wardriving_importer import (
|
||||
WigleCSVImporter,
|
||||
GPSJSONImporter,
|
||||
BatchImporter
|
||||
)
|
||||
|
||||
|
||||
def test_wigle_csv_import():
|
||||
"""Test Wigle CSV format import"""
|
||||
print("=" * 80)
|
||||
print("TEST 1: Wigle CSV Import")
|
||||
print("=" * 80)
|
||||
|
||||
csv_path = "signatures/t-embed-rf/test_export_wigle.csv"
|
||||
|
||||
if not Path(csv_path).exists():
|
||||
print(f"❌ CSV file not found: {csv_path}")
|
||||
return
|
||||
|
||||
importer = WigleCSVImporter()
|
||||
captures = importer.parse_csv(csv_path)
|
||||
|
||||
print(f"\n✅ Imported {len(captures)} captures from CSV")
|
||||
|
||||
for i, capture in enumerate(captures, 1):
|
||||
print(f"\nCapture {i}:")
|
||||
print(f" Filename: {capture.filename}")
|
||||
print(f" Frequency: {capture.frequency / 1e6:.2f} MHz")
|
||||
print(f" Protocol: {capture.protocol}")
|
||||
print(f" GPS: {capture.latitude}, {capture.longitude}")
|
||||
print(f" Accuracy: {capture.accuracy}m")
|
||||
print(f" Timestamp: {capture.timestamp}")
|
||||
print(f" GPS Source: {capture.gps_source}")
|
||||
|
||||
|
||||
def test_gps_json_import():
|
||||
"""Test GPS JSON format import"""
|
||||
print("\n" + "=" * 80)
|
||||
print("TEST 2: GPS JSON Import")
|
||||
print("=" * 80)
|
||||
|
||||
json_files = list(Path("signatures/t-embed-rf").glob("gps_coordinates_*.json"))
|
||||
|
||||
if not json_files:
|
||||
print("❌ No GPS JSON files found")
|
||||
return
|
||||
|
||||
print(f"\nFound {len(json_files)} GPS JSON files")
|
||||
|
||||
for json_file in json_files[:3]: # Test first 3
|
||||
coords = GPSJSONImporter.parse_json(str(json_file))
|
||||
|
||||
if coords:
|
||||
print(f"\n✅ {json_file.name}")
|
||||
print(f" Latitude: {coords.latitude}")
|
||||
print(f" Longitude: {coords.longitude}")
|
||||
print(f" Accuracy: {coords.accuracy}m")
|
||||
print(f" Source: {coords.source}")
|
||||
else:
|
||||
print(f"\n❌ {json_file.name} - No GPS data")
|
||||
|
||||
|
||||
def test_batch_import():
|
||||
"""Test batch directory import"""
|
||||
print("\n" + "=" * 80)
|
||||
print("TEST 3: Batch Directory Import")
|
||||
print("=" * 80)
|
||||
|
||||
importer = BatchImporter()
|
||||
captures = importer.import_from_directory("signatures/t-embed-rf")
|
||||
|
||||
print(f"\n✅ Imported {len(captures)} total captures")
|
||||
|
||||
# Group by GPS source
|
||||
print("\nGPS Sources:")
|
||||
sources = {}
|
||||
for capture in captures:
|
||||
sources[capture.gps_source] = sources.get(capture.gps_source, 0) + 1
|
||||
|
||||
for source, count in sorted(sources.items()):
|
||||
print(f" {source}: {count}")
|
||||
|
||||
# Show sample captures
|
||||
print("\nSample Captures:")
|
||||
for i, capture in enumerate(captures[:5], 1):
|
||||
print(f"\n {i}. {capture.filename}")
|
||||
print(f" Frequency: {capture.frequency / 1e6:.2f} MHz")
|
||||
print(f" GPS: {capture.latitude}, {capture.longitude}")
|
||||
print(f" Source: {capture.gps_source}")
|
||||
|
||||
|
||||
def test_upload_manifest():
|
||||
"""Test generating upload manifest"""
|
||||
print("\n" + "=" * 80)
|
||||
print("TEST 4: Upload Manifest Generation")
|
||||
print("=" * 80)
|
||||
|
||||
importer = BatchImporter()
|
||||
captures = importer.import_from_directory("signatures/t-embed-rf")
|
||||
|
||||
manifest = importer.to_upload_manifest()
|
||||
|
||||
print(f"\n✅ Generated manifest with {len(manifest['captures'])} captures")
|
||||
print(f"Session UUID: {manifest['session_uuid']}")
|
||||
|
||||
print("\nSample capture data:")
|
||||
if manifest['captures']:
|
||||
import json
|
||||
print(json.dumps(manifest['captures'][0], indent=2))
|
||||
|
||||
|
||||
def main():
|
||||
"""Run all tests"""
|
||||
print("\n🧪 WARDRIVING DATA IMPORT TESTS")
|
||||
print("=" * 80)
|
||||
|
||||
try:
|
||||
test_wigle_csv_import()
|
||||
test_gps_json_import()
|
||||
test_batch_import()
|
||||
test_upload_manifest()
|
||||
|
||||
print("\n" + "=" * 80)
|
||||
print("✅ ALL TESTS COMPLETE")
|
||||
print("=" * 80)
|
||||
|
||||
except Exception as e:
|
||||
print(f"\n❌ ERROR: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return 1
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user