giglez/signatures/t-embed-rf/quick_capture.py

#!/data/data/com.termux/files/usr/bin/python3
"""
Quick GPS+RF Capture - Uses existing wardriving infrastructure
Captures current GPS and associates it with a Bruce .sub file
"""

import json
import sys
import subprocess
import hashlib
from datetime import datetime, timezone
from pathlib import Path

# Import existing infrastructure
from database import SubGHzWardrivingDB


def get_current_gps(timeout=15):
    """Get current GPS using termux-location."""
    try:
        cmd = ["termux-location", "-p", "gps"]
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=timeout
        )

        if result.returncode == 0 and result.stdout.strip():
            gps_data = json.loads(result.stdout.strip())
            if 'latitude' in gps_data and 'longitude' in gps_data:
                print(f"✓ GPS Location: {gps_data['latitude']:.6f}, {gps_data['longitude']:.6f}")
                print(f"  Accuracy: {gps_data.get('accuracy', 'N/A')}m")
                return gps_data

    except subprocess.TimeoutExpired:
        print(f"✗ GPS timeout after {timeout}s")
    except Exception as e:
        print(f"✗ GPS error: {e}")

    return None


def parse_sub_file(sub_file):
    """Parse Bruce .sub file to extract RF metadata."""
    metadata = {}

    try:
        with open(sub_file, 'r') as f:
            for line in f:
                line = line.strip()
                if not line or not ':' in line:
                    continue

                key, value = line.split(':', 1)
                key = key.strip()
                value = value.strip()

                if key == 'Frequency':
                    metadata['frequency_hz'] = int(value)
                elif key == 'Protocol':
                    metadata['protocol'] = value
                elif key == 'Preset':
                    metadata['preset'] = value

        return metadata if metadata else None

    except Exception as e:
        print(f"✗ Error parsing {sub_file}: {e}")
        return None


def calculate_file_hash(file_path):
    """Calculate MD5 hash of file."""
    md5 = hashlib.md5()
    with open(file_path, 'rb') as f:
        for chunk in iter(lambda: f.read(4096), b''):
            md5.update(chunk)
    return md5.hexdigest()


def main():
    if len(sys.argv) < 2:
        print("Usage: quick_capture.py <bruce_file.sub> [session_id]")
        print("\nExample:")
        print("  python3 quick_capture.py raw10.sub")
        print("  python3 quick_capture.py ~/projects/device-integration/bruce/data/captures/raw10.sub mysession")
        return 1

    sub_file = Path(sys.argv[1])
    session_id = sys.argv[2] if len(sys.argv) > 2 else f"quick_{datetime.now().strftime('%Y%m%d_%H%M%S')}"

    if not sub_file.exists():
        print(f"✗ File not found: {sub_file}")
        return 1

    print(f"\n📡 SubGHz GPS Capture")
    print(f"━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
    print(f"Bruce File: {sub_file.name}")
    print(f"Session: {session_id}")
    print()

    # Get GPS
    print("🛰️  Getting GPS location...")
    gps_data = get_current_gps()
    if not gps_data:
        print("\n✗ Failed to get GPS - trying to use latest from log...")
        gps_log = Path.home() / "logs" / "gps_quick.jsonl"
        if gps_log.exists():
            with open(gps_log, 'r') as f:
                last_line = f.readlines()[-1]
                gps_data = json.loads(last_line.strip())
                print(f"✓ Using latest GPS: {gps_data['latitude']:.6f}, {gps_data['longitude']:.6f}")
        else:
            print("✗ No GPS data available")
            return 1

    # Parse RF file
    print("\n📻 Parsing RF file...")
    rf_metadata = parse_sub_file(sub_file)
    if not rf_metadata:
        print("✗ Failed to parse RF file")
        return 1

    print(f"✓ Frequency: {rf_metadata.get('frequency_hz', 'N/A')} Hz ({rf_metadata.get('frequency_hz', 0) / 1e6:.3f} MHz)")
    print(f"  Protocol: {rf_metadata.get('protocol', 'N/A')}")
    print(f"  Preset: {rf_metadata.get('preset', 'N/A')}")

    # Initialize database
    print("\n💾 Storing to database...")
    db = SubGHzWardrivingDB()

    # Create session if it doesn't exist
    db.create_session(session_id, device_info="Bruce T-Embed CC1101", notes="Quick GPS capture")

    # Add GPS track point
    db.add_gps_point(
        session_id=session_id,
        latitude=gps_data['latitude'],
        longitude=gps_data['longitude'],
        altitude=gps_data.get('altitude'),
        accuracy=gps_data.get('accuracy'),
        speed=gps_data.get('speed'),
        provider=gps_data.get('provider', 'gps')
    )

    # Add signal capture
    timestamp = datetime.now(timezone.utc).isoformat()
    capture_id = f"{session_id}_{sub_file.stem}_{int(datetime.now().timestamp())}"

    db.add_signal(
        session_id=session_id,
        capture_id=capture_id,
        timestamp=timestamp,
        bruce_filename=sub_file.name,
        bruce_file_path=str(sub_file),
        file_size_bytes=sub_file.stat().st_size,
        frequency_hz=rf_metadata.get('frequency_hz'),
        protocol=rf_metadata.get('protocol'),
        gps_latitude=gps_data['latitude'],
        gps_longitude=gps_data['longitude'],
        gps_accuracy_meters=gps_data.get('accuracy'),
        interpolated=False,
        interpolation_confidence=1.0,
        file_hash_md5=calculate_file_hash(sub_file)
    )

    print(f"✓ Stored signal: {capture_id}")
    print(f"\n✅ Capture paired successfully!")
    print(f"   Database: ~/logs/subghz_wardriving.db")
    print(f"   Export with: python3 -m subghz_wardriving.exporter")
    print()

    return 0


if __name__ == '__main__':
    sys.exit(main())