from flask import Flask, render_template, request, jsonify, send_file import json import csv import os from datetime import datetime import time from flask_socketio import SocketIO, emit, join_room, leave_room import threading import serial import serial.tools.list_ports import queue import uuid import pickle from pathlib import Path app = Flask(__name__) app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY', 'flockyou_dev_key_2024') socketio = SocketIO(app, cors_allowed_origins="*", async_mode='threading', logger=True, engineio_logger=True) # Global variables detections = [] cumulative_detections = [] session_start_time = datetime.now() gps_data = None gps_history = [] # Buffer of recent GPS readings for temporal matching MAX_GPS_HISTORY = 100 # Keep last 100 GPS readings GPS_MATCH_THRESHOLD = 30 # Max seconds between detection and GPS reading serial_connection = None gps_enabled = False flock_device_connected = False flock_device_port = None flock_serial_connection = None oui_database = {} serial_data_buffer = [] reconnect_attempts = {'flock': 0, 'gps': 0} max_reconnect_attempts = 5 reconnect_delay = 3 # seconds connection_lock = threading.Lock() serial_queue = queue.Queue() next_detection_id = 1 # Unique ID counter settings = {'gps_port': '', 'flock_port': '', 'filter': 'all'} # Data storage paths DATA_DIR = Path('data') CUMULATIVE_DATA_FILE = DATA_DIR / 'cumulative_detections.pkl' SETTINGS_FILE = DATA_DIR / 'settings.json' # Ensure data directory exists DATA_DIR.mkdir(exist_ok=True) # Persistent storage functions def load_cumulative_detections(): """Load cumulative detections from disk""" global cumulative_detections try: if CUMULATIVE_DATA_FILE.exists(): with open(CUMULATIVE_DATA_FILE, 'rb') as f: cumulative_detections = pickle.load(f) print(f"Loaded {len(cumulative_detections)} cumulative detections") else: cumulative_detections = [] except Exception as e: print(f"Error loading cumulative detections: {e}") cumulative_detections = [] def save_cumulative_detections(): """Save cumulative detections to disk""" try: with open(CUMULATIVE_DATA_FILE, 'wb') as f: pickle.dump(cumulative_detections, f) print(f"Saved {len(cumulative_detections)} cumulative detections") except Exception as e: print(f"Error saving cumulative detections: {e}") def load_settings(): """Load settings from disk""" global settings try: if SETTINGS_FILE.exists(): with open(SETTINGS_FILE, 'r') as f: settings.update(json.load(f)) print(f"Loaded settings: {settings}") except Exception as e: print(f"Error loading settings: {e}") def save_settings(): """Save settings to disk""" try: with open(SETTINGS_FILE, 'w') as f: json.dump(settings, f, indent=2) print(f"Saved settings: {settings}") except Exception as e: print(f"Error saving settings: {e}") # Load OUI database def load_oui_database(): """Load the IEEE OUI database for manufacturer lookups""" global oui_database try: with open('oui.txt', 'r', encoding='utf-8', errors='ignore') as f: for line in f: line = line.strip() if line and not line.startswith('#') and '(hex)' in line: # Parse OUI line format: "28-6F-B9 (hex) Nokia Shanghai Bell Co., Ltd." parts = line.split('(hex)') if len(parts) == 2: mac_prefix = parts[0].strip().replace('-', '').replace(' ', '').upper() manufacturer = parts[1].strip() if mac_prefix and manufacturer and len(mac_prefix) == 6: oui_database[mac_prefix] = manufacturer print(f"Loaded {len(oui_database)} OUI entries") except Exception as e: print(f"Error loading OUI database: {e}") def lookup_manufacturer(mac_address): """Look up manufacturer information for a MAC address""" if not mac_address: return None # Extract first 6 characters (3 bytes) of MAC address mac_clean = mac_address.replace(':', '').replace('-', '').upper() if len(mac_clean) >= 6: oui = mac_clean[:6] return oui_database.get(oui, "Unknown Manufacturer") return "Unknown Manufacturer" # GPS Dongle Configuration GPS_BAUDRATE = 9600 GPS_TIMEOUT = 1 class GPSData: def __init__(self): self.latitude = None self.longitude = None self.altitude = None self.timestamp = None self.fix_quality = 0 self.satellites = 0 def parse_nmea_sentence(sentence): """Parse NMEA GPS sentence with improved accuracy""" if not sentence.startswith('$'): return None parts = sentence.strip().split(',') if len(parts) < 1: return None sentence_type = parts[0] if sentence_type in ['$GPGGA', '$GNGGA']: # Global Positioning System Fix Data (GPS + GLONASS) if len(parts) >= 15: try: time_str = parts[1] lat_raw = parts[2] lat_dir = parts[3] lon_raw = parts[4] lon_dir = parts[5] fix_quality = int(parts[6]) if parts[6] else 0 satellites = int(parts[7]) if parts[7] else 0 hdop = float(parts[8]) if parts[8] else 0 # Horizontal Dilution of Precision altitude = float(parts[9]) if parts[9] else 0 # Skip if no fix if fix_quality == 0 or not lat_raw or not lon_raw: return None # Convert NMEA format (DDMM.MMMM) to decimal degrees with high precision lat_degrees = int(lat_raw[:2]) lat_minutes = float(lat_raw[2:]) lat = lat_degrees + (lat_minutes / 60.0) lon_degrees = int(lon_raw[:3]) lon_minutes = float(lon_raw[3:]) lon = lon_degrees + (lon_minutes / 60.0) # Apply direction if lat_dir == 'S': lat = -lat if lon_dir == 'W': lon = -lon return { 'latitude': round(lat, 8), # 8 decimal places for ~1.1mm accuracy 'longitude': round(lon, 8), 'altitude': round(altitude, 3), 'fix_quality': fix_quality, 'satellites': satellites, 'hdop': hdop, 'timestamp': time_str } except (ValueError, IndexError) as e: print(f"GPS parsing error: {e}") return None return None def safe_socket_emit(event, data, room=None): """Safely emit socket events with error handling""" try: if room: socketio.emit(event, data, room=room) else: socketio.emit(event, data) except Exception as e: print(f"Socket emit error for {event}: {e}") def gps_reader(): """Background thread for reading GPS data""" global gps_data, serial_connection, gps_enabled while gps_enabled: if serial_connection and serial_connection.is_open: try: line = serial_connection.readline().decode('utf-8', errors='ignore') if line: # Send raw GPS data to serial terminal safe_socket_emit('serial_data', f"GPS: {line.strip()}", room='serial_terminal') parsed = parse_nmea_sentence(line) if parsed: gps_data = parsed # Add to GPS history with timestamp for temporal matching if parsed.get('fix_quality') > 0: gps_entry = parsed.copy() gps_entry['system_timestamp'] = time.time() gps_history.append(gps_entry) # Keep only recent GPS readings if len(gps_history) > MAX_GPS_HISTORY: gps_history.pop(0) safe_socket_emit('gps_update', parsed) # Also send parsed GPS data to terminal if parsed.get('fix_quality') > 0: gps_info = f"GPS Fix: {parsed.get('latitude', 'N/A')}, {parsed.get('longitude', 'N/A')} - {parsed.get('satellites', 0)} satellites" safe_socket_emit('serial_data', gps_info, room='serial_terminal') except Exception as e: print(f"GPS read error: {e}") with connection_lock: gps_enabled = False safe_socket_emit('gps_disconnected', {}) break time.sleep(0.1) def flock_reader(): """Background thread for reading Flock device data""" global flock_serial_connection, flock_device_connected, serial_data_buffer while flock_device_connected: if flock_serial_connection and flock_serial_connection.is_open: try: line = flock_serial_connection.readline().decode('utf-8', errors='ignore') if line: line = line.strip() if line: # Store in buffer for terminal serial_data_buffer.append(line) if len(serial_data_buffer) > 1000: # Keep last 1000 lines serial_data_buffer.pop(0) # Forward to all serial terminal clients safe_socket_emit('serial_data', line, room='serial_terminal') print(f"Serial data sent to terminal: {line}") # Try to parse as detection data try: data = json.loads(line) if 'detection_method' in data: # This is a detection, add it add_detection_from_serial(data) else: print(f"JSON data without detection_method: {data}") except json.JSONDecodeError: # Not JSON, just log it print(f"Flock device (non-JSON): {line}") except Exception as e: print(f"Flock device read error: {e}") with connection_lock: flock_device_connected = False safe_socket_emit('flock_disconnected', {}) # Trigger reconnection immediately attempt_reconnect_flock() break time.sleep(0.1) def find_best_gps_match(detection_timestamp): """Find the GPS reading closest in time to the detection timestamp""" global gps_history if not gps_history: return None try: # Convert detection timestamp to epoch time if isinstance(detection_timestamp, str): # Try parsing ISO format first try: dt = datetime.fromisoformat(detection_timestamp.replace('Z', '+00:00')) except: # Try parsing the display format dt = datetime.strptime(detection_timestamp, '%Y-%m-%d %H:%M:%S') detection_time = dt.timestamp() else: detection_time = detection_timestamp best_match = None min_time_diff = float('inf') for gps_entry in gps_history: gps_time = gps_entry['system_timestamp'] time_diff = abs(detection_time - gps_time) if time_diff < min_time_diff and time_diff <= GPS_MATCH_THRESHOLD: min_time_diff = time_diff best_match = gps_entry return best_match except Exception as e: print(f"Error finding GPS match: {e}") return None def validate_gps_data(gps_data): """Validate GPS data integrity""" if not gps_data: return False, "No GPS data" lat = gps_data.get('latitude') lon = gps_data.get('longitude') if lat is None or lon is None: return False, "Missing coordinates" # Basic coordinate validation if not (-90 <= lat <= 90): return False, f"Invalid latitude: {lat}" if not (-180 <= lon <= 180): return False, f"Invalid longitude: {lon}" fix_quality = gps_data.get('fix_quality', 0) if fix_quality < 1: return False, f"Poor GPS fix quality: {fix_quality}" return True, "Valid GPS data" def add_detection_from_serial(data): """Add detection from serial data - counts detections per MAC address""" global detections, cumulative_detections, gps_data, next_detection_id # Add server timestamp first (system time when detection was processed) system_time = time.time() data['server_timestamp'] = datetime.fromtimestamp(system_time).isoformat() # Try to find the best GPS match for this detection's timestamp best_gps = find_best_gps_match(system_time) preferred_timestamp = None if best_gps: # Validate GPS data before using it is_valid, validation_msg = validate_gps_data(best_gps) if is_valid: time_diff = abs(system_time - best_gps['system_timestamp']) data['gps'] = { 'latitude': best_gps.get('latitude'), 'longitude': best_gps.get('longitude'), 'altitude': best_gps.get('altitude'), 'timestamp': best_gps.get('timestamp'), 'satellites': best_gps.get('satellites'), 'fix_quality': best_gps.get('fix_quality'), 'time_diff': time_diff, 'match_quality': 'temporal' } # Prefer GPS timestamp when available and accurate if time_diff < 5: # Very close temporal match preferred_timestamp = best_gps.get('timestamp') print(f"✓ Using GPS timestamp for MAC {data.get('mac_address', 'unknown')}: {time_diff:.2f}s difference") else: print(f"✓ GPS temporal match for MAC {data.get('mac_address', 'unknown')}: {time_diff:.2f}s difference") else: print(f"⚠ Invalid GPS data for temporal match: {validation_msg}") best_gps = None # Fallback to current GPS if no good temporal match if not best_gps and gps_data and gps_data.get('fix_quality') > 0: is_valid, validation_msg = validate_gps_data(gps_data) if is_valid: data['gps'] = { 'latitude': gps_data.get('latitude'), 'longitude': gps_data.get('longitude'), 'altitude': gps_data.get('altitude'), 'timestamp': gps_data.get('timestamp'), 'satellites': gps_data.get('satellites'), 'fix_quality': gps_data.get('fix_quality'), 'time_diff': None, # Unknown time difference 'match_quality': 'current' } # Use current GPS timestamp if available preferred_timestamp = gps_data.get('timestamp') print(f"○ Using current GPS timestamp for MAC {data.get('mac_address', 'unknown')} (no temporal match)") else: print(f"⚠ Current GPS data invalid: {validation_msg}") # Set timestamps - prefer GPS timestamp when available if preferred_timestamp: data['timestamp'] = preferred_timestamp data['detection_time'] = preferred_timestamp data['timestamp_source'] = 'gps' print(f"📍 Using GPS timestamp as primary timestamp for {data.get('mac_address', 'unknown')}") else: # Fallback to system timestamps system_dt = datetime.fromtimestamp(system_time) data['timestamp'] = system_dt.isoformat() data['detection_time'] = system_dt.strftime('%Y-%m-%d %H:%M:%S') data['timestamp_source'] = 'system' print(f"🕐 Using system timestamp for {data.get('mac_address', 'unknown')} (no GPS available)") # Log if no GPS could be assigned if not data.get('gps'): print(f"✗ No valid GPS data available for MAC {data.get('mac_address', 'unknown')}") # Add manufacturer information if 'mac_address' in data: data['manufacturer'] = lookup_manufacturer(data['mac_address']) # Check if we already have a detection for this MAC address mac_address = data.get('mac_address') existing_detection = None if mac_address: for detection in detections: if detection.get('mac_address') == mac_address: existing_detection = detection break if existing_detection: # Update existing detection with new data and increment count existing_detection['detection_count'] = existing_detection.get('detection_count', 1) + 1 existing_detection['last_seen'] = datetime.now().isoformat() existing_detection['last_rssi'] = data.get('rssi', existing_detection.get('last_rssi')) existing_detection['last_channel'] = data.get('channel', existing_detection.get('last_channel')) existing_detection['last_frequency'] = data.get('frequency', existing_detection.get('last_frequency')) existing_detection['last_ssid'] = data.get('ssid', existing_detection.get('last_ssid')) existing_detection['last_device_name'] = data.get('device_name', existing_detection.get('last_device_name')) # Preserve detection_method if not already set if not existing_detection.get('detection_method') and data.get('detection_method'): existing_detection['detection_method'] = data.get('detection_method') # Update GPS if new data is available if data.get('gps'): existing_detection['gps'] = data['gps'] # Update cumulative detections for cum_detection in cumulative_detections: if cum_detection.get('mac_address') == mac_address: cum_detection.update(existing_detection) break save_cumulative_detections() # Emit updated detection safe_socket_emit('detection_updated', existing_detection) print(f"Updated detection: MAC {mac_address}, Count: {existing_detection['detection_count']}, Method: {existing_detection.get('detection_method')}") else: # Create new detection data['id'] = next_detection_id next_detection_id += 1 data['alias'] = '' # Empty alias by default data['detection_count'] = 1 data['first_seen'] = datetime.now().isoformat() data['last_seen'] = datetime.now().isoformat() detections.append(data) # Add to cumulative detections cumulative_detections.append(data.copy()) save_cumulative_detections() # Emit to connected clients safe_socket_emit('new_detection', data) print(f"New detection added: ID {data['id']}, Method: {data.get('detection_method')}, MAC: {mac_address}") def connection_monitor(): """Background thread for monitoring device connections""" global gps_enabled, flock_device_connected, serial_connection, reconnect_attempts while True: # Check GPS connection if gps_enabled: try: if not serial_connection or not serial_connection.is_open: with connection_lock: gps_enabled = False safe_socket_emit('gps_disconnected', {}) print("GPS connection lost") # Start reconnection attempts attempt_reconnect_gps() else: # Test if the connection is still valid serial_connection.in_waiting except Exception as e: print(f"GPS connection test failed: {e}") with connection_lock: gps_enabled = False safe_socket_emit('gps_disconnected', {}) attempt_reconnect_gps() # Check Flock You device connection if flock_device_connected: try: # Test if the connection is still valid if not flock_serial_connection or not flock_serial_connection.is_open: with connection_lock: flock_device_connected = False safe_socket_emit('flock_disconnected', {}) print("Flock You device connection lost") # Start reconnection attempts attempt_reconnect_flock() else: # Try a simple read to test connection flock_serial_connection.in_waiting except Exception as e: print(f"Flock device connection test failed: {e}") with connection_lock: flock_device_connected = False safe_socket_emit('flock_disconnected', {}) # Start reconnection attempts attempt_reconnect_flock() time.sleep(2) # Check every 2 seconds def attempt_reconnect_flock(): """Attempt to reconnect to Flock device""" global flock_device_connected, reconnect_attempts, flock_serial_connection def reconnect_thread(): global flock_device_connected, reconnect_attempts, flock_serial_connection while not flock_device_connected and reconnect_attempts['flock'] < max_reconnect_attempts: try: print(f"Attempting to reconnect to Flock device (attempt {reconnect_attempts['flock'] + 1}/{max_reconnect_attempts})") # Try to reconnect if flock_serial_connection: try: flock_serial_connection.close() except: pass # Wait a moment for the device to be ready time.sleep(1) flock_serial_connection = serial.Serial(flock_device_port, 115200, timeout=1) # Test the connection test_data = flock_serial_connection.readline() # If successful, update status with connection_lock: flock_device_connected = True reconnect_attempts['flock'] = 0 print(f"Successfully reconnected to Flock device on {flock_device_port}") safe_socket_emit('flock_reconnected', {'port': flock_device_port}) # Restart the reading thread flock_thread = threading.Thread(target=flock_reader, daemon=True) flock_thread.start() return except Exception as e: print(f"Reconnection attempt failed: {e}") reconnect_attempts['flock'] += 1 time.sleep(reconnect_delay) if reconnect_attempts['flock'] >= max_reconnect_attempts: print("Max reconnection attempts reached for Flock device") safe_socket_emit('reconnect_failed', {'device': 'flock'}) reconnect_attempts['flock'] = 0 # Reset for future attempts thread = threading.Thread(target=reconnect_thread, daemon=True) thread.start() def attempt_reconnect_gps(): """Attempt to reconnect to GPS device""" global gps_enabled, reconnect_attempts def reconnect_thread(): global gps_enabled, reconnect_attempts while not gps_enabled and reconnect_attempts['gps'] < max_reconnect_attempts: try: print(f"Attempting to reconnect to GPS device (attempt {reconnect_attempts['gps'] + 1}/{max_reconnect_attempts})") # Try to reconnect test_ser = serial.Serial(serial_connection.port, GPS_BAUDRATE, timeout=1) test_ser.close() # If successful, update status with connection_lock: gps_enabled = True reconnect_attempts['gps'] = 0 print(f"Successfully reconnected to GPS device on {serial_connection.port}") safe_socket_emit('gps_reconnected', {'port': serial_connection.port}) # Restart the reading thread gps_thread = threading.Thread(target=gps_reader, daemon=True) gps_thread.start() return except Exception as e: print(f"GPS reconnection attempt failed: {e}") reconnect_attempts['gps'] += 1 time.sleep(reconnect_delay) if reconnect_attempts['gps'] >= max_reconnect_attempts: print("Max reconnection attempts reached for GPS device") safe_socket_emit('reconnect_failed', {'device': 'gps'}) reconnect_attempts['gps'] = 0 # Reset for future attempts thread = threading.Thread(target=reconnect_thread, daemon=True) thread.start() @app.route('/') def index(): return render_template('index.html') @app.route('/api/detections', methods=['GET']) def get_detections(): """Get all detections with optional filtering""" filter_type = request.args.get('filter', 'all') data_type = request.args.get('type', 'session') # Choose data source if data_type == 'cumulative': source_data = cumulative_detections else: source_data = detections # Apply filter if filter_type == 'all': return jsonify(source_data) else: filtered = [d for d in source_data if d.get('detection_method') == filter_type] return jsonify(filtered) @app.route('/api/detections', methods=['POST']) def add_detection(): """Add a new detection from serial data""" global detections, gps_data data = request.json # Add GPS data if available if gps_data and gps_data.get('fix_quality') > 0: data['gps'] = { 'latitude': gps_data.get('latitude'), 'longitude': gps_data.get('longitude'), 'altitude': gps_data.get('altitude'), 'timestamp': gps_data.get('timestamp'), 'satellites': gps_data.get('satellites') } # Add manufacturer information if 'mac_address' in data: data['manufacturer'] = lookup_manufacturer(data['mac_address']) # Add server timestamp data['server_timestamp'] = datetime.now().isoformat() detections.append(data) # Emit to connected clients socketio.emit('new_detection', data) return jsonify({'status': 'success', 'id': len(detections)}) @app.route('/api/gps/connect', methods=['POST']) def connect_gps(): """Connect to GPS dongle""" global serial_connection, gps_enabled data = request.json port = data.get('port') try: if serial_connection: serial_connection.close() serial_connection = serial.Serial(port, GPS_BAUDRATE, timeout=GPS_TIMEOUT) with connection_lock: gps_enabled = True # Start GPS reading thread gps_thread = threading.Thread(target=gps_reader, daemon=True) gps_thread.start() return jsonify({'status': 'success', 'message': f'Connected to {port}'}) except Exception as e: return jsonify({'status': 'error', 'message': str(e)}), 400 @app.route('/api/gps/disconnect', methods=['POST']) def disconnect_gps(): """Disconnect GPS dongle""" global serial_connection, gps_enabled with connection_lock: gps_enabled = False if serial_connection: serial_connection.close() serial_connection = None return jsonify({'status': 'success', 'message': 'GPS disconnected'}) @app.route('/api/flock/connect', methods=['POST']) def connect_flock(): """Connect to Flock You device""" global flock_device_connected, flock_device_port, flock_serial_connection data = request.json port = data.get('port') try: # Create persistent connection to the port flock_serial_connection = serial.Serial(port, 115200, timeout=1) with connection_lock: flock_device_connected = True flock_device_port = port # Start reading thread flock_thread = threading.Thread(target=flock_reader, daemon=True) flock_thread.start() return jsonify({'status': 'success', 'message': f'Connected to Flock You device on {port}'}) except Exception as e: return jsonify({'status': 'error', 'message': str(e)}), 400 @app.route('/api/flock/disconnect', methods=['POST']) def disconnect_flock(): """Disconnect Flock You device""" global flock_device_connected, flock_device_port, flock_serial_connection with connection_lock: flock_device_connected = False flock_device_port = None if flock_serial_connection and flock_serial_connection.is_open: flock_serial_connection.close() flock_serial_connection = None return jsonify({'status': 'success', 'message': 'Flock You device disconnected'}) @app.route('/api/status', methods=['GET']) def get_status(): """Get connection status of both devices""" return jsonify({ 'gps_connected': gps_enabled, 'gps_port': serial_connection.port if serial_connection else None, 'flock_connected': flock_device_connected, 'flock_port': flock_device_port }) @app.route('/api/gps/ports', methods=['GET']) def get_gps_ports(): """Get available serial ports for GPS""" ports = [] for port in serial.tools.list_ports.comports(): port_info = { 'device': port.device, 'description': port.description, 'manufacturer': port.manufacturer if port.manufacturer else 'Unknown', 'product': port.product if port.product else 'Unknown', 'vid': port.vid, 'pid': port.pid } ports.append(port_info) return jsonify(ports) @app.route('/api/flock/ports', methods=['GET']) def get_flock_ports(): """Get available serial ports for Flock You device""" ports = [] for port in serial.tools.list_ports.comports(): port_info = { 'device': port.device, 'description': port.description, 'manufacturer': port.manufacturer if port.manufacturer else 'Unknown', 'product': port.product if port.product else 'Unknown', 'vid': port.vid, 'pid': port.pid } ports.append(port_info) return jsonify(ports) @app.route('/api/export/csv', methods=['GET']) def export_csv(): """Export session detections as CSV""" export_type = request.args.get('type', 'session') if export_type == 'cumulative': data_to_export = cumulative_detections filename_prefix = "flockyou_cumulative" else: data_to_export = detections filename_prefix = f"flockyou_session_{session_start_time.strftime('%Y%m%d_%H%M%S')}" if not data_to_export: return jsonify({'status': 'error', 'message': 'No detections to export'}), 400 filename = f"{filename_prefix}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv" filepath = os.path.join('exports', filename) os.makedirs('exports', exist_ok=True) with open(filepath, 'w', newline='', encoding='utf-8') as csvfile: fieldnames = [ 'timestamp', 'detection_time', 'server_timestamp', 'protocol', 'detection_method', 'ssid', 'device_name', 'mac_address', 'manufacturer', 'alias', 'rssi', 'last_rssi', 'signal_strength', 'channel', 'last_channel', 'detection_count', 'latitude', 'longitude', 'altitude', 'gps_timestamp', 'satellites', 'fix_quality', 'gps_time_diff', 'gps_match_quality', 'timestamp_source' ] writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() for detection in data_to_export: gps_data = detection.get('gps', {}) row = { 'timestamp': detection.get('timestamp'), 'detection_time': detection.get('detection_time'), 'server_timestamp': detection.get('server_timestamp'), 'protocol': detection.get('protocol'), 'detection_method': detection.get('detection_method'), 'ssid': detection.get('ssid', ''), 'device_name': detection.get('device_name', ''), 'mac_address': detection.get('mac_address'), 'manufacturer': detection.get('manufacturer', 'Unknown'), 'alias': detection.get('alias', ''), 'rssi': detection.get('rssi'), 'last_rssi': detection.get('last_rssi'), 'signal_strength': detection.get('signal_strength'), 'channel': detection.get('channel'), 'last_channel': detection.get('last_channel'), 'detection_count': detection.get('detection_count', 1), 'latitude': gps_data.get('latitude'), 'longitude': gps_data.get('longitude'), 'altitude': gps_data.get('altitude'), 'gps_timestamp': gps_data.get('timestamp'), 'satellites': gps_data.get('satellites'), 'fix_quality': gps_data.get('fix_quality'), 'gps_time_diff': gps_data.get('time_diff'), 'gps_match_quality': gps_data.get('match_quality'), 'timestamp_source': detection.get('timestamp_source', 'unknown') } writer.writerow(row) return send_file(filepath, as_attachment=True, download_name=filename) @app.route('/api/export/kml', methods=['GET']) def export_kml(): """Export detections as KML""" export_type = request.args.get('type', 'session') if export_type == 'cumulative': data_to_export = cumulative_detections filename_prefix = "flockyou_cumulative" document_name = "Flock You Cumulative Detections" else: data_to_export = detections filename_prefix = f"flockyou_session_{session_start_time.strftime('%Y%m%d_%H%M%S')}" document_name = f"Flock You Session Detections - {session_start_time.strftime('%Y-%m-%d %H:%M:%S')}" if not data_to_export: return jsonify({'status': 'error', 'message': 'No detections to export'}), 400 filename = f"{filename_prefix}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.kml" filepath = os.path.join('exports', filename) os.makedirs('exports', exist_ok=True) kml_content = f""" {document_name} Surveillance device detections with GPS coordinates ({len(data_to_export)} detections) """ for i, detection in enumerate(data_to_export): gps = detection.get('gps', {}) if gps.get('latitude') and gps.get('longitude'): # Use alias if available, otherwise use detection number placemark_name = detection.get('alias') or f"Detection {i+1}" # GPS accuracy indicator gps_accuracy = "" if gps.get('time_diff') is not None: time_diff = gps.get('time_diff') if time_diff < 5: gps_accuracy = f" (✓ Precise: {time_diff:.1f}s)" elif time_diff < 15: gps_accuracy = f" (~ Good: {time_diff:.1f}s)" else: gps_accuracy = f" (⚠ Approximate: {time_diff:.1f}s)" else: gps_accuracy = " (? Unknown accuracy)" # Build device info device_info = "" if detection.get('ssid'): device_info += f"SSID: {detection.get('ssid')}
" if detection.get('device_name'): device_info += f"Device Name: {detection.get('device_name')}
" # RSSI info rssi_info = detection.get('last_rssi') or detection.get('rssi', 'N/A') # Channel info channel_info = detection.get('last_channel') or detection.get('channel', 'N/A') kml_content += f""" {placemark_name} Protocol: {detection.get('protocol')}
Detection Method: {detection.get('detection_method')}
{device_info} MAC Address: {detection.get('mac_address')}
Manufacturer: {detection.get('manufacturer', 'Unknown')}
Alias: {detection.get('alias', 'None')}
RSSI: {rssi_info} dBm
Signal Strength: {detection.get('signal_strength', 'N/A')}
Channel: {channel_info}
Detection Count: {detection.get('detection_count', 1)}
Detection Time: {detection.get('detection_time', 'N/A')}
Server Timestamp: {detection.get('server_timestamp', 'N/A')}
GPS Coordinates: {gps.get('latitude'):.6f}, {gps.get('longitude'):.6f}{gps_accuracy}
GPS Altitude: {gps.get('altitude', 'N/A')} m
GPS Satellites: {gps.get('satellites', 'N/A')}
GPS Fix Quality: {gps.get('fix_quality', 'N/A')}
GPS Match Quality: {gps.get('match_quality', 'N/A')}
GPS Timestamp: {gps.get('timestamp', 'N/A')}
Timestamp Source: {detection.get('timestamp_source', 'Unknown').upper()} ]]> {gps.get('longitude')},{gps.get('latitude')},{gps.get('altitude', 0)} """ kml_content += """ """ with open(filepath, 'w', encoding='utf-8') as f: f.write(kml_content) return send_file(filepath, as_attachment=True, download_name=filename) @app.route('/api/clear', methods=['POST']) def clear_detections(): """Clear session detections""" global detections, next_detection_id, session_start_time detections.clear() next_detection_id = 1 # Reset ID counter session_start_time = datetime.now() # Reset session start time safe_socket_emit('detections_cleared', {}) return jsonify({'status': 'success', 'message': 'Session detections cleared'}) @app.route('/api/test/detection', methods=['POST']) def test_detection(): """Test endpoint to add a sample detection""" if request.is_json: # Use provided detection data sample_detection = request.json # Ensure required fields are present if 'detection_method' not in sample_detection: sample_detection['detection_method'] = 'probe_request' if 'protocol' not in sample_detection: sample_detection['protocol'] = 'wifi' if 'mac_address' not in sample_detection: sample_detection['mac_address'] = 'AA:BB:CC:DD:EE:FF' if 'detection_time' not in sample_detection: sample_detection['detection_time'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') if 'timestamp' not in sample_detection: sample_detection['timestamp'] = datetime.now().isoformat() else: # Use default sample detection sample_detection = { 'detection_method': 'probe_request', 'protocol': 'wifi', 'mac_address': 'AA:BB:CC:DD:EE:FF', 'ssid': 'TestNetwork', 'rssi': -45, 'signal_strength': 'Excellent', 'channel': 6, 'detection_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'timestamp': datetime.now().isoformat() } add_detection_from_serial(sample_detection) return jsonify({'status': 'success', 'message': 'Test detection added'}) @app.route('/api/detection/alias', methods=['POST']) def update_detection_alias(): """Update detection alias""" global detections data = request.json detection_id = data.get('id') alias = data.get('alias', '').strip() if detection_id is None: return jsonify({'status': 'error', 'message': 'Detection ID required'}), 400 # Find and update the detection for detection in detections: if detection.get('id') == detection_id: detection['alias'] = alias # Emit update to all clients safe_socket_emit('detection_updated', detection) return jsonify({'status': 'success', 'message': 'Alias updated'}) return jsonify({'status': 'error', 'message': 'Detection not found'}), 404 @app.route('/api/settings', methods=['GET']) def get_settings(): """Get current settings""" return jsonify(settings) @app.route('/api/settings', methods=['POST']) def update_settings(): """Update settings""" global settings data = request.json settings.update(data) save_settings() return jsonify({'status': 'success', 'settings': settings}) @app.route('/api/stats', methods=['GET']) def get_stats(): """Get detection statistics""" return jsonify({ 'session': { 'total': len(detections), 'wifi': len([d for d in detections if d.get('protocol') == 'wifi']), 'ble': len([d for d in detections if d.get('protocol') in ['bluetooth_le', 'bluetooth_classic']]), 'gps': len([d for d in detections if d.get('gps')]), 'start_time': session_start_time.isoformat() }, 'cumulative': { 'total': len(cumulative_detections), 'wifi': len([d for d in cumulative_detections if d.get('protocol') == 'wifi']), 'ble': len([d for d in cumulative_detections if d.get('protocol') in ['bluetooth_le', 'bluetooth_classic']]), 'gps': len([d for d in cumulative_detections if d.get('gps')]) } }) @app.route('/api/oui/search', methods=['POST']) def search_oui(): """Search OUI database""" global oui_database data = request.json query = data.get('query', '').strip() if not query: return jsonify({'status': 'error', 'message': 'Query required'}), 400 results = [] # Clean the query - remove colons and spaces, convert to uppercase clean_query = query.replace(':', '').replace(' ', '').upper() # Check if query looks like a MAC address (6 hex characters) if len(clean_query) >= 6 and all(c in '0123456789ABCDEF' for c in clean_query[:6]): # Search by MAC prefix mac_prefix = clean_query[:6] if mac_prefix in oui_database: results.append({ 'mac': mac_prefix, 'manufacturer': oui_database[mac_prefix] }) else: # Search by manufacturer name query_lower = query.lower() for mac, manufacturer in oui_database.items(): if query_lower in manufacturer.lower(): results.append({ 'mac': mac, 'manufacturer': manufacturer }) if len(results) >= 100: # Increased limit break print(f"Search query: '{query}' -> '{clean_query}', found {len(results)} results") return jsonify({ 'status': 'success', 'results': results, 'count': len(results) }) @app.route('/api/oui/all') def get_all_oui(): """Get all OUI entries""" global oui_database # Return all entries results = [] for mac, manufacturer in oui_database.items(): results.append({ 'mac': mac, 'manufacturer': manufacturer }) return jsonify({ 'status': 'success', 'results': results, 'count': len(results), 'total': len(oui_database) }) @app.route('/api/oui/refresh', methods=['POST']) def refresh_oui_database(): """Refresh OUI database from IEEE website""" global oui_database try: import urllib.request import tempfile import os # Download the latest OUI database url = "https://standards-oui.ieee.org/oui/oui.txt" print(f"Downloading OUI database from {url}...") # Create a temporary file with tempfile.NamedTemporaryFile(delete=False, suffix='.txt') as temp_file: temp_path = temp_file.name # Download the file urllib.request.urlretrieve(url, temp_path) # Parse the downloaded file new_oui_database = {} with open(temp_path, 'r', encoding='utf-8', errors='ignore') as f: for line in f: line = line.strip() if line and not line.startswith('#') and '\t' in line: parts = line.split('\t') if len(parts) >= 3: mac = parts[0].strip().replace('-', '').upper() manufacturer = parts[2].strip() if len(mac) == 6 and manufacturer: new_oui_database[mac] = manufacturer # Clean up temporary file os.unlink(temp_path) # Update the global database oui_database = new_oui_database # Save to local file with open('oui.txt', 'w', encoding='utf-8') as f: for mac, manufacturer in oui_database.items(): f.write(f"{mac}\t{manufacturer}\n") print(f"Successfully refreshed OUI database with {len(oui_database)} entries") return jsonify({ 'status': 'success', 'message': 'Database refreshed successfully', 'count': len(oui_database) }) except Exception as e: print(f"Error refreshing OUI database: {str(e)}") return jsonify({ 'status': 'error', 'message': f'Failed to refresh database: {str(e)}' }), 500 # Socket.IO event handlers @socketio.on('connect') def handle_connect(): print(f"Client connected: {request.sid}") @socketio.on('disconnect') def handle_disconnect(): print(f"Client disconnected: {request.sid}") # Clean up any room memberships try: leave_room('serial_terminal') except: pass @socketio.on('heartbeat') def handle_heartbeat(): """Handle client heartbeat to keep connection alive""" try: emit('heartbeat_ack') except Exception as e: print(f"Heartbeat response error: {e}") def send_heartbeat(): """Send periodic heartbeat to all clients""" while True: try: safe_socket_emit('heartbeat', {}) time.sleep(30) # Send heartbeat every 30 seconds except Exception as e: print(f"Heartbeat error: {e}") time.sleep(5) @socketio.on('request_serial_terminal') def handle_serial_terminal_request(data): """Handle serial terminal connection request""" global serial_data_buffer port = data.get('port') print(f"Serial terminal request from {request.sid} for port: {port}") if not port: emit('serial_error', {'message': 'No port specified'}) return if not flock_device_connected or flock_device_port != port: emit('serial_error', {'message': 'Device not connected. Please connect to the Sniffer device first.'}) return try: # Add to serial terminal room join_room('serial_terminal') emit('serial_connected') # Send recent buffer data buffer_count = len(serial_data_buffer) print(f"Sending {min(50, buffer_count)} recent lines to terminal") for line in serial_data_buffer[-50:]: # Send last 50 lines emit('serial_data', line) print(f"Serial terminal connected for client {request.sid}") except Exception as e: print(f"Serial terminal connection error: {e}") emit('serial_error', {'message': f'Failed to start terminal: {str(e)}'}) if __name__ == '__main__': # Load data on startup load_oui_database() load_cumulative_detections() load_settings() # Start connection monitor thread monitor_thread = threading.Thread(target=connection_monitor, daemon=True) monitor_thread.start() # Start heartbeat thread heartbeat_thread = threading.Thread(target=send_heartbeat, daemon=True) heartbeat_thread.start() print("Starting Flock You API server...") print("Server will be available at: http://localhost:5000") print("Press Ctrl+C to stop the server") try: socketio.run(app, debug=False, host='0.0.0.0', port=5000) except KeyboardInterrupt: print("\nShutting down server...") # Clean up connections if flock_serial_connection and flock_serial_connection.is_open: flock_serial_connection.close() if serial_connection and serial_connection.is_open: serial_connection.close() print("Server stopped.")