Mirror/flock-you
1
0
Fork 0
mirror of https://github.com/colonelpanichacks/flock-you.git synced 2026-06-15 00:03:33 -07:00
Code Issues Packages Projects Releases Wiki Activity

Add files via upload

Browse Source
This commit is contained in:
Colonel Panic
2025-08-29 16:23:04 -04:00
committed by GitHub
parent 8497469d2e
commit 04d76e1274
2 changed files with 1001 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/flockyou.py
+181 -32
View File
@@ -10,6 +10,8 @@ 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')
@@ -17,6 +19,8 @@ socketio = SocketIO(app, cors_allowed_origins="*", async_mode='threading', logge
# Global variables
detections = []
cumulative_detections = []
session_start_time = datetime.now()
gps_data = None
serial_connection = None
gps_enabled = False
@@ -31,6 +35,59 @@ 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():
@@ -78,7 +135,7 @@ class GPSData:
self.satellites = 0
def parse_nmea_sentence(sentence):
"""Parse NMEA GPS sentence"""
"""Parse NMEA GPS sentence with improved accuracy"""
if not sentence.startswith('$'):
return None
@@ -88,33 +145,49 @@ def parse_nmea_sentence(sentence):
sentence_type = parts[0]
if sentence_type == '$GPGGA': # Global Positioning System Fix Data
if sentence_type in ['$GPGGA', '$GNGGA']: # Global Positioning System Fix Data (GPS + GLONASS)
if len(parts) >= 15:
try:
time_str = parts[1]
lat = float(parts[2]) / 100
lat_raw = parts[2]
lat_dir = parts[3]
lon = float(parts[4]) / 100
lon_raw = parts[4]
lon_dir = parts[5]
fix_quality = int(parts[6])
satellites = int(parts[7])
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
# Convert to decimal degrees
# 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': lat,
'longitude': lon,
'altitude': altitude,
'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):
except (ValueError, IndexError) as e:
print(f"GPS parsing error: {e}")
return None
return None
@@ -202,7 +275,7 @@ def flock_reader():
def add_detection_from_serial(data):
"""Add detection from serial data - counts detections per MAC address"""
global detections, gps_data, next_detection_id
global detections, cumulative_detections, gps_data, next_detection_id
# Add GPS data if available
if gps_data and gps_data.get('fix_quality') > 0:
@@ -250,6 +323,13 @@ def add_detection_from_serial(data):
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')}")
@@ -264,6 +344,10 @@ def add_detection_from_serial(data):
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}")
@@ -416,11 +500,19 @@ def index():
def get_detections():
"""Get all detections with optional filtering"""
filter_type = request.args.get('filter', 'all')
data_type = request.args.get('type', 'session')
if filter_type == 'all':
return jsonify(detections)
# Choose data source
if data_type == 'cumulative':
source_data = cumulative_detections
else:
filtered = [d for d in detections if d.get('detection_method') == filter_type]
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'])
@@ -573,11 +665,20 @@ def get_flock_ports():
@app.route('/api/export/csv', methods=['GET'])
def export_csv():
"""Export detections as CSV"""
if not detections:
"""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"flockyou_detections_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
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)
@@ -586,12 +687,12 @@ def export_csv():
fieldnames = [
'timestamp', 'detection_time', 'protocol', 'detection_method',
'ssid', 'mac_address', 'manufacturer', 'alias', 'rssi', 'signal_strength', 'channel',
'latitude', 'longitude', 'altitude', 'gps_timestamp', 'satellites'
'latitude', 'longitude', 'altitude', 'gps_timestamp', 'satellites', 'detection_count'
]
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
writer.writeheader()
for detection in detections:
for detection in data_to_export:
row = {
'timestamp': detection.get('timestamp'),
'detection_time': detection.get('detection_time'),
@@ -601,14 +702,15 @@ def export_csv():
'mac_address': detection.get('mac_address'),
'manufacturer': detection.get('manufacturer', 'Unknown'),
'alias': detection.get('alias', ''),
'rssi': detection.get('rssi'),
'rssi': detection.get('last_rssi') or detection.get('rssi'),
'signal_strength': detection.get('signal_strength'),
'channel': detection.get('channel'),
'channel': detection.get('last_channel') or detection.get('channel'),
'latitude': detection.get('gps', {}).get('latitude'),
'longitude': detection.get('gps', {}).get('longitude'),
'altitude': detection.get('gps', {}).get('altitude'),
'gps_timestamp': detection.get('gps', {}).get('timestamp'),
'satellites': detection.get('gps', {}).get('satellites')
'satellites': detection.get('gps', {}).get('satellites'),
'detection_count': detection.get('detection_count', 1)
}
writer.writerow(row)
@@ -617,10 +719,21 @@ def export_csv():
@app.route('/api/export/kml', methods=['GET'])
def export_kml():
"""Export detections as KML"""
if not detections:
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"flockyou_detections_{datetime.now().strftime('%Y%m%d_%H%M%S')}.kml"
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)
@@ -628,11 +741,11 @@ def export_kml():
kml_content = f"""<?xml version="1.0" encoding="UTF-8"?>
<kml xmlns="http://www.opengis.net/kml/2.2">
<Document>
<name>Flock You Detections</name>
<description>Surveillance device detections with GPS coordinates</description>
<name>{document_name}</name>
<description>Surveillance device detections with GPS coordinates ({len(data_to_export)} detections)</description>
"""
for i, detection in enumerate(detections):
for i, detection in enumerate(data_to_export):
gps = detection.get('gps', {})
if gps.get('latitude') and gps.get('longitude'):
kml_content += f"""
@@ -670,12 +783,13 @@ def export_kml():
@app.route('/api/clear', methods=['POST'])
def clear_detections():
"""Clear all detections"""
global detections, next_detection_id
"""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': 'All detections cleared'})
return jsonify({'status': 'success', 'message': 'Session detections cleared'})
@app.route('/api/test/detection', methods=['POST'])
def test_detection():
@@ -733,6 +847,39 @@ def update_detection_alias():
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"""
@@ -924,8 +1071,10 @@ def handle_serial_terminal_request(data):
emit('serial_error', {'message': f'Failed to start terminal: {str(e)}'})
if __name__ == '__main__':
# Load OUI database on startup
# 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)
api/templates/index.html
+820 -78
View File
File diff suppressed because it is too large Load Diff