mirror of
https://github.com/smittix/intercept.git
synced 2026-07-27 10:08:10 -07:00
Add GPS dongle support and fix Python 3.7/3.8 compatibility
- Add GPS dongle support with NMEA parsing (utils/gps.py, routes/gps.py) - Add GPS device selector to ADS-B and Satellite observer location sections - Add GPS dongle option to ADS-B dashboard - Fix Python 3.7/3.8 compatibility by adding 'from __future__ import annotations' to all SDR module files (fixes TypeError: 'type' object is not subscriptable) - Add pyserial to requirements.txt - Update README with GPS dongle documentation and troubleshooting 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -9,6 +9,7 @@ def register_blueprints(app):
|
||||
from .adsb import adsb_bp
|
||||
from .satellite import satellite_bp
|
||||
from .iridium import iridium_bp
|
||||
from .gps import gps_bp
|
||||
|
||||
app.register_blueprint(pager_bp)
|
||||
app.register_blueprint(sensor_bp)
|
||||
@@ -17,3 +18,4 @@ def register_blueprints(app):
|
||||
app.register_blueprint(adsb_bp)
|
||||
app.register_blueprint(satellite_bp)
|
||||
app.register_blueprint(iridium_bp)
|
||||
app.register_blueprint(gps_bp)
|
||||
|
||||
+218
@@ -0,0 +1,218 @@
|
||||
"""GPS dongle routes for USB GPS device support."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import queue
|
||||
import threading
|
||||
import time
|
||||
from typing import Generator
|
||||
|
||||
from flask import Blueprint, jsonify, request, Response
|
||||
|
||||
from utils.logging import get_logger
|
||||
from utils.sse import format_sse
|
||||
from utils.gps import (
|
||||
detect_gps_devices,
|
||||
is_serial_available,
|
||||
get_gps_reader,
|
||||
start_gps,
|
||||
stop_gps,
|
||||
get_current_position,
|
||||
GPSPosition,
|
||||
)
|
||||
|
||||
logger = get_logger('intercept.gps')
|
||||
|
||||
gps_bp = Blueprint('gps', __name__, url_prefix='/gps')
|
||||
|
||||
# Queue for SSE position updates
|
||||
_gps_queue: queue.Queue = queue.Queue(maxsize=100)
|
||||
|
||||
|
||||
def _position_callback(position: GPSPosition) -> None:
|
||||
"""Callback to queue position updates for SSE stream."""
|
||||
try:
|
||||
_gps_queue.put_nowait(position.to_dict())
|
||||
except queue.Full:
|
||||
# Discard oldest if queue is full
|
||||
try:
|
||||
_gps_queue.get_nowait()
|
||||
_gps_queue.put_nowait(position.to_dict())
|
||||
except queue.Empty:
|
||||
pass
|
||||
|
||||
|
||||
@gps_bp.route('/available')
|
||||
def check_gps_available():
|
||||
"""Check if GPS dongle support is available."""
|
||||
return jsonify({
|
||||
'available': is_serial_available(),
|
||||
'message': None if is_serial_available() else 'pyserial not installed - run: pip install pyserial'
|
||||
})
|
||||
|
||||
|
||||
@gps_bp.route('/devices')
|
||||
def list_gps_devices():
|
||||
"""List available GPS serial devices."""
|
||||
if not is_serial_available():
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'pyserial not installed'
|
||||
}), 503
|
||||
|
||||
devices = detect_gps_devices()
|
||||
return jsonify({
|
||||
'status': 'ok',
|
||||
'devices': devices
|
||||
})
|
||||
|
||||
|
||||
@gps_bp.route('/start', methods=['POST'])
|
||||
def start_gps_reader():
|
||||
"""Start GPS reader on specified device."""
|
||||
if not is_serial_available():
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'pyserial not installed'
|
||||
}), 503
|
||||
|
||||
# Check if already running
|
||||
reader = get_gps_reader()
|
||||
if reader and reader.is_running:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'GPS reader already running'
|
||||
}), 409
|
||||
|
||||
data = request.json or {}
|
||||
device_path = data.get('device')
|
||||
baudrate = data.get('baudrate', 9600)
|
||||
|
||||
if not device_path:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'Device path required'
|
||||
}), 400
|
||||
|
||||
# Validate baudrate
|
||||
valid_baudrates = [4800, 9600, 19200, 38400, 57600, 115200]
|
||||
if baudrate not in valid_baudrates:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': f'Invalid baudrate. Valid options: {valid_baudrates}'
|
||||
}), 400
|
||||
|
||||
# Clear the queue
|
||||
while not _gps_queue.empty():
|
||||
try:
|
||||
_gps_queue.get_nowait()
|
||||
except queue.Empty:
|
||||
break
|
||||
|
||||
# Start the GPS reader
|
||||
success = start_gps(device_path, baudrate)
|
||||
|
||||
if success:
|
||||
# Register callback for SSE streaming
|
||||
reader = get_gps_reader()
|
||||
if reader:
|
||||
reader.add_callback(_position_callback)
|
||||
|
||||
return jsonify({
|
||||
'status': 'started',
|
||||
'device': device_path,
|
||||
'baudrate': baudrate
|
||||
})
|
||||
else:
|
||||
reader = get_gps_reader()
|
||||
error = reader.error if reader else 'Unknown error'
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': f'Failed to start GPS reader: {error}'
|
||||
}), 500
|
||||
|
||||
|
||||
@gps_bp.route('/stop', methods=['POST'])
|
||||
def stop_gps_reader():
|
||||
"""Stop GPS reader."""
|
||||
reader = get_gps_reader()
|
||||
if reader:
|
||||
reader.remove_callback(_position_callback)
|
||||
|
||||
stop_gps()
|
||||
|
||||
return jsonify({'status': 'stopped'})
|
||||
|
||||
|
||||
@gps_bp.route('/status')
|
||||
def get_gps_status():
|
||||
"""Get current GPS reader status."""
|
||||
reader = get_gps_reader()
|
||||
|
||||
if not reader:
|
||||
return jsonify({
|
||||
'running': False,
|
||||
'device': None,
|
||||
'position': None,
|
||||
'error': None,
|
||||
'message': 'GPS reader not started'
|
||||
})
|
||||
|
||||
position = reader.position
|
||||
return jsonify({
|
||||
'running': reader.is_running,
|
||||
'device': reader.device_path,
|
||||
'position': position.to_dict() if position else None,
|
||||
'last_update': reader.last_update.isoformat() if reader.last_update else None,
|
||||
'error': reader.error,
|
||||
'message': 'Waiting for GPS fix - ensure GPS has clear view of sky' if reader.is_running and not position else None
|
||||
})
|
||||
|
||||
|
||||
@gps_bp.route('/position')
|
||||
def get_position():
|
||||
"""Get current GPS position."""
|
||||
position = get_current_position()
|
||||
|
||||
if position:
|
||||
return jsonify({
|
||||
'status': 'ok',
|
||||
'position': position.to_dict()
|
||||
})
|
||||
else:
|
||||
reader = get_gps_reader()
|
||||
if not reader or not reader.is_running:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'GPS reader not running'
|
||||
}), 400
|
||||
else:
|
||||
return jsonify({
|
||||
'status': 'waiting',
|
||||
'message': 'Waiting for GPS fix'
|
||||
})
|
||||
|
||||
|
||||
@gps_bp.route('/stream')
|
||||
def stream_gps():
|
||||
"""SSE stream of GPS position updates."""
|
||||
def generate() -> Generator[str, None, None]:
|
||||
last_keepalive = time.time()
|
||||
keepalive_interval = 30.0
|
||||
|
||||
while True:
|
||||
try:
|
||||
position = _gps_queue.get(timeout=1)
|
||||
last_keepalive = time.time()
|
||||
yield format_sse({'type': 'position', **position})
|
||||
except queue.Empty:
|
||||
now = time.time()
|
||||
if now - last_keepalive >= keepalive_interval:
|
||||
yield format_sse({'type': 'keepalive'})
|
||||
last_keepalive = now
|
||||
|
||||
response = Response(generate(), mimetype='text/event-stream')
|
||||
response.headers['Cache-Control'] = 'no-cache'
|
||||
response.headers['X-Accel-Buffering'] = 'no'
|
||||
response.headers['Connection'] = 'keep-alive'
|
||||
return response
|
||||
Reference in New Issue
Block a user