mirror of
https://github.com/smittix/intercept.git
synced 2026-05-29 21:09:26 -07:00
Prevent stale monitor start requests from retuning audio
This commit is contained in:
Smittix
@@ -38,13 +38,15 @@ receiver_bp = Blueprint('receiver', __name__, url_prefix='/receiver')
|
||||
# ============================================
|
||||
|
||||
# Audio demodulation state
|
||||
audio_process = None
|
||||
audio_rtl_process = None
|
||||
audio_lock = threading.Lock()
|
||||
audio_running = False
|
||||
audio_frequency = 0.0
|
||||
audio_modulation = 'fm'
|
||||
audio_source = 'process'
|
||||
audio_process = None
|
||||
audio_rtl_process = None
|
||||
audio_lock = threading.Lock()
|
||||
audio_start_lock = threading.Lock()
|
||||
audio_running = False
|
||||
audio_frequency = 0.0
|
||||
audio_modulation = 'fm'
|
||||
audio_source = 'process'
|
||||
audio_start_token = 0
|
||||
|
||||
# Scanner state
|
||||
scanner_thread: Optional[threading.Thread] = None
|
||||
@@ -1269,184 +1271,202 @@ def get_presets() -> Response:
|
||||
# MANUAL AUDIO ENDPOINTS (for direct listening)
|
||||
# ============================================
|
||||
|
||||
@receiver_bp.route('/audio/start', methods=['POST'])
|
||||
def start_audio() -> Response:
|
||||
"""Start audio at specific frequency (manual mode)."""
|
||||
global scanner_running, scanner_active_device, receiver_active_device, scanner_power_process, scanner_thread
|
||||
global audio_running, audio_frequency, audio_modulation, audio_source
|
||||
|
||||
# Stop scanner if running
|
||||
if scanner_running:
|
||||
scanner_running = False
|
||||
if scanner_active_device is not None:
|
||||
app_module.release_sdr_device(scanner_active_device)
|
||||
scanner_active_device = None
|
||||
if scanner_thread and scanner_thread.is_alive():
|
||||
try:
|
||||
scanner_thread.join(timeout=2.0)
|
||||
except Exception:
|
||||
pass
|
||||
if scanner_power_process and scanner_power_process.poll() is None:
|
||||
try:
|
||||
scanner_power_process.terminate()
|
||||
scanner_power_process.wait(timeout=1)
|
||||
except Exception:
|
||||
try:
|
||||
scanner_power_process.kill()
|
||||
except Exception:
|
||||
pass
|
||||
scanner_power_process = None
|
||||
try:
|
||||
subprocess.run(['pkill', '-9', 'rtl_power'], capture_output=True, timeout=0.5)
|
||||
except Exception:
|
||||
pass
|
||||
time.sleep(0.5)
|
||||
|
||||
data = request.json or {}
|
||||
|
||||
try:
|
||||
frequency = float(data.get('frequency', 0))
|
||||
modulation = normalize_modulation(data.get('modulation', 'wfm'))
|
||||
squelch = int(data.get('squelch', 0))
|
||||
gain = int(data.get('gain', 40))
|
||||
device = int(data.get('device', 0))
|
||||
sdr_type = str(data.get('sdr_type', 'rtlsdr')).lower()
|
||||
bias_t_raw = data.get('bias_t', scanner_config.get('bias_t', False))
|
||||
if isinstance(bias_t_raw, str):
|
||||
bias_t = bias_t_raw.strip().lower() in {'1', 'true', 'yes', 'on'}
|
||||
else:
|
||||
bias_t = bool(bias_t_raw)
|
||||
except (ValueError, TypeError) as e:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': f'Invalid parameter: {e}'
|
||||
}), 400
|
||||
|
||||
if frequency <= 0:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'frequency is required'
|
||||
}), 400
|
||||
|
||||
valid_sdr_types = ['rtlsdr', 'hackrf', 'airspy', 'limesdr', 'sdrplay']
|
||||
if sdr_type not in valid_sdr_types:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': f'Invalid sdr_type. Use: {", ".join(valid_sdr_types)}'
|
||||
}), 400
|
||||
|
||||
# Update config for audio
|
||||
scanner_config['squelch'] = squelch
|
||||
scanner_config['gain'] = gain
|
||||
scanner_config['device'] = device
|
||||
scanner_config['sdr_type'] = sdr_type
|
||||
scanner_config['bias_t'] = bias_t
|
||||
|
||||
# Preferred path: when waterfall WebSocket is active on the same SDR,
|
||||
# derive monitor audio from that IQ stream instead of spawning rtl_fm.
|
||||
try:
|
||||
from routes.waterfall_websocket import (
|
||||
get_shared_capture_status,
|
||||
start_shared_monitor_from_capture,
|
||||
)
|
||||
|
||||
shared = get_shared_capture_status()
|
||||
if shared.get('running') and shared.get('device') == device:
|
||||
_stop_audio_stream()
|
||||
ok, msg = start_shared_monitor_from_capture(
|
||||
device=device,
|
||||
frequency_mhz=frequency,
|
||||
modulation=modulation,
|
||||
squelch=squelch,
|
||||
)
|
||||
if ok:
|
||||
audio_running = True
|
||||
audio_frequency = frequency
|
||||
audio_modulation = modulation
|
||||
audio_source = 'waterfall'
|
||||
# Shared monitor uses the waterfall's existing SDR claim.
|
||||
if receiver_active_device is not None:
|
||||
app_module.release_sdr_device(receiver_active_device)
|
||||
receiver_active_device = None
|
||||
return jsonify({
|
||||
'status': 'started',
|
||||
'frequency': frequency,
|
||||
'modulation': modulation,
|
||||
'source': 'waterfall',
|
||||
})
|
||||
logger.warning(f"Shared waterfall monitor unavailable: {msg}")
|
||||
except Exception as e:
|
||||
logger.debug(f"Shared waterfall monitor probe failed: {e}")
|
||||
|
||||
# Stop waterfall if it's using the same SDR (SSE path)
|
||||
if waterfall_running and waterfall_active_device == device:
|
||||
_stop_waterfall_internal()
|
||||
time.sleep(0.2)
|
||||
|
||||
# Claim device for listening audio. The WebSocket waterfall handler
|
||||
# may still be tearing down its IQ capture process (thread join +
|
||||
# safe_terminate can take several seconds), so we retry with back-off
|
||||
# to give the USB device time to be fully released.
|
||||
if receiver_active_device is None or receiver_active_device != device:
|
||||
if receiver_active_device is not None:
|
||||
app_module.release_sdr_device(receiver_active_device)
|
||||
receiver_active_device = None
|
||||
|
||||
error = None
|
||||
max_claim_attempts = 6
|
||||
for attempt in range(max_claim_attempts):
|
||||
error = app_module.claim_sdr_device(device, 'receiver')
|
||||
if not error:
|
||||
break
|
||||
if attempt < max_claim_attempts - 1:
|
||||
logger.debug(
|
||||
f"Device claim attempt {attempt + 1}/{max_claim_attempts} "
|
||||
f"failed, retrying in 0.5s: {error}"
|
||||
)
|
||||
time.sleep(0.5)
|
||||
|
||||
if error:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'error_type': 'DEVICE_BUSY',
|
||||
'message': error
|
||||
}), 409
|
||||
receiver_active_device = device
|
||||
|
||||
_start_audio_stream(frequency, modulation)
|
||||
|
||||
if audio_running:
|
||||
audio_source = 'process'
|
||||
return jsonify({
|
||||
'status': 'started',
|
||||
'frequency': frequency,
|
||||
'modulation': modulation,
|
||||
'source': 'process',
|
||||
})
|
||||
else:
|
||||
# Avoid leaving a stale device claim after startup failure.
|
||||
if receiver_active_device is not None:
|
||||
app_module.release_sdr_device(receiver_active_device)
|
||||
receiver_active_device = None
|
||||
|
||||
start_error = ''
|
||||
for log_path in ('/tmp/rtl_fm_stderr.log', '/tmp/ffmpeg_stderr.log'):
|
||||
try:
|
||||
with open(log_path, 'r') as handle:
|
||||
content = handle.read().strip()
|
||||
if content:
|
||||
start_error = content.splitlines()[-1]
|
||||
break
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
message = 'Failed to start audio. Check SDR device.'
|
||||
if start_error:
|
||||
message = f'Failed to start audio: {start_error}'
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': message
|
||||
}), 500
|
||||
@receiver_bp.route('/audio/start', methods=['POST'])
|
||||
def start_audio() -> Response:
|
||||
"""Start audio at specific frequency (manual mode)."""
|
||||
global scanner_running, scanner_active_device, receiver_active_device, scanner_power_process, scanner_thread
|
||||
global audio_running, audio_frequency, audio_modulation, audio_source, audio_start_token
|
||||
|
||||
data = request.json or {}
|
||||
|
||||
try:
|
||||
frequency = float(data.get('frequency', 0))
|
||||
modulation = normalize_modulation(data.get('modulation', 'wfm'))
|
||||
squelch = int(data.get('squelch', 0))
|
||||
gain = int(data.get('gain', 40))
|
||||
device = int(data.get('device', 0))
|
||||
sdr_type = str(data.get('sdr_type', 'rtlsdr')).lower()
|
||||
request_token_raw = data.get('request_token')
|
||||
request_token = int(request_token_raw) if request_token_raw is not None else None
|
||||
bias_t_raw = data.get('bias_t', scanner_config.get('bias_t', False))
|
||||
if isinstance(bias_t_raw, str):
|
||||
bias_t = bias_t_raw.strip().lower() in {'1', 'true', 'yes', 'on'}
|
||||
else:
|
||||
bias_t = bool(bias_t_raw)
|
||||
except (ValueError, TypeError) as e:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': f'Invalid parameter: {e}'
|
||||
}), 400
|
||||
|
||||
if frequency <= 0:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'frequency is required'
|
||||
}), 400
|
||||
|
||||
valid_sdr_types = ['rtlsdr', 'hackrf', 'airspy', 'limesdr', 'sdrplay']
|
||||
if sdr_type not in valid_sdr_types:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': f'Invalid sdr_type. Use: {", ".join(valid_sdr_types)}'
|
||||
}), 400
|
||||
|
||||
with audio_start_lock:
|
||||
if request_token is not None:
|
||||
if request_token < audio_start_token:
|
||||
return jsonify({
|
||||
'status': 'stale',
|
||||
'message': 'Superseded audio start request',
|
||||
'source': audio_source,
|
||||
'superseded': True,
|
||||
}), 409
|
||||
audio_start_token = request_token
|
||||
else:
|
||||
audio_start_token += 1
|
||||
request_token = audio_start_token
|
||||
|
||||
# Stop scanner if running
|
||||
if scanner_running:
|
||||
scanner_running = False
|
||||
if scanner_active_device is not None:
|
||||
app_module.release_sdr_device(scanner_active_device)
|
||||
scanner_active_device = None
|
||||
if scanner_thread and scanner_thread.is_alive():
|
||||
try:
|
||||
scanner_thread.join(timeout=2.0)
|
||||
except Exception:
|
||||
pass
|
||||
if scanner_power_process and scanner_power_process.poll() is None:
|
||||
try:
|
||||
scanner_power_process.terminate()
|
||||
scanner_power_process.wait(timeout=1)
|
||||
except Exception:
|
||||
try:
|
||||
scanner_power_process.kill()
|
||||
except Exception:
|
||||
pass
|
||||
scanner_power_process = None
|
||||
try:
|
||||
subprocess.run(['pkill', '-9', 'rtl_power'], capture_output=True, timeout=0.5)
|
||||
except Exception:
|
||||
pass
|
||||
time.sleep(0.5)
|
||||
|
||||
# Update config for audio
|
||||
scanner_config['squelch'] = squelch
|
||||
scanner_config['gain'] = gain
|
||||
scanner_config['device'] = device
|
||||
scanner_config['sdr_type'] = sdr_type
|
||||
scanner_config['bias_t'] = bias_t
|
||||
|
||||
# Preferred path: when waterfall WebSocket is active on the same SDR,
|
||||
# derive monitor audio from that IQ stream instead of spawning rtl_fm.
|
||||
try:
|
||||
from routes.waterfall_websocket import (
|
||||
get_shared_capture_status,
|
||||
start_shared_monitor_from_capture,
|
||||
)
|
||||
|
||||
shared = get_shared_capture_status()
|
||||
if shared.get('running') and shared.get('device') == device:
|
||||
_stop_audio_stream()
|
||||
ok, msg = start_shared_monitor_from_capture(
|
||||
device=device,
|
||||
frequency_mhz=frequency,
|
||||
modulation=modulation,
|
||||
squelch=squelch,
|
||||
)
|
||||
if ok:
|
||||
audio_running = True
|
||||
audio_frequency = frequency
|
||||
audio_modulation = modulation
|
||||
audio_source = 'waterfall'
|
||||
# Shared monitor uses the waterfall's existing SDR claim.
|
||||
if receiver_active_device is not None:
|
||||
app_module.release_sdr_device(receiver_active_device)
|
||||
receiver_active_device = None
|
||||
return jsonify({
|
||||
'status': 'started',
|
||||
'frequency': frequency,
|
||||
'modulation': modulation,
|
||||
'source': 'waterfall',
|
||||
'request_token': request_token,
|
||||
})
|
||||
logger.warning(f"Shared waterfall monitor unavailable: {msg}")
|
||||
except Exception as e:
|
||||
logger.debug(f"Shared waterfall monitor probe failed: {e}")
|
||||
|
||||
# Stop waterfall if it's using the same SDR (SSE path)
|
||||
if waterfall_running and waterfall_active_device == device:
|
||||
_stop_waterfall_internal()
|
||||
time.sleep(0.2)
|
||||
|
||||
# Claim device for listening audio. The WebSocket waterfall handler
|
||||
# may still be tearing down its IQ capture process (thread join +
|
||||
# safe_terminate can take several seconds), so we retry with back-off
|
||||
# to give the USB device time to be fully released.
|
||||
if receiver_active_device is None or receiver_active_device != device:
|
||||
if receiver_active_device is not None:
|
||||
app_module.release_sdr_device(receiver_active_device)
|
||||
receiver_active_device = None
|
||||
|
||||
error = None
|
||||
max_claim_attempts = 6
|
||||
for attempt in range(max_claim_attempts):
|
||||
error = app_module.claim_sdr_device(device, 'receiver')
|
||||
if not error:
|
||||
break
|
||||
if attempt < max_claim_attempts - 1:
|
||||
logger.debug(
|
||||
f"Device claim attempt {attempt + 1}/{max_claim_attempts} "
|
||||
f"failed, retrying in 0.5s: {error}"
|
||||
)
|
||||
time.sleep(0.5)
|
||||
|
||||
if error:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'error_type': 'DEVICE_BUSY',
|
||||
'message': error
|
||||
}), 409
|
||||
receiver_active_device = device
|
||||
|
||||
_start_audio_stream(frequency, modulation)
|
||||
|
||||
if audio_running:
|
||||
audio_source = 'process'
|
||||
return jsonify({
|
||||
'status': 'started',
|
||||
'frequency': audio_frequency,
|
||||
'modulation': audio_modulation,
|
||||
'source': 'process',
|
||||
'request_token': request_token,
|
||||
})
|
||||
|
||||
# Avoid leaving a stale device claim after startup failure.
|
||||
if receiver_active_device is not None:
|
||||
app_module.release_sdr_device(receiver_active_device)
|
||||
receiver_active_device = None
|
||||
|
||||
start_error = ''
|
||||
for log_path in ('/tmp/rtl_fm_stderr.log', '/tmp/ffmpeg_stderr.log'):
|
||||
try:
|
||||
with open(log_path, 'r') as handle:
|
||||
content = handle.read().strip()
|
||||
if content:
|
||||
start_error = content.splitlines()[-1]
|
||||
break
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
message = 'Failed to start audio. Check SDR device.'
|
||||
if start_error:
|
||||
message = f'Failed to start audio: {start_error}'
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': message
|
||||
}), 500
|
||||
|
||||
|
||||
@receiver_bp.route('/audio/stop', methods=['POST'])
|
||||
|
||||
@@ -2704,6 +2704,7 @@ const Waterfall = (function () {
|
||||
gain,
|
||||
device,
|
||||
biasT,
|
||||
requestToken,
|
||||
}) {
|
||||
const response = await fetch('/receiver/audio/start', {
|
||||
method: 'POST',
|
||||
@@ -2716,6 +2717,7 @@ const Waterfall = (function () {
|
||||
device: device.deviceIndex,
|
||||
sdr_type: device.sdrType,
|
||||
bias_t: biasT,
|
||||
request_token: requestToken,
|
||||
}),
|
||||
});
|
||||
|
||||
@@ -2812,10 +2814,13 @@ const Waterfall = (function () {
|
||||
gain,
|
||||
device: monitorDevice,
|
||||
biasT,
|
||||
requestToken: nonce,
|
||||
});
|
||||
if (nonce !== _audioConnectNonce) return;
|
||||
|
||||
const busy = payload?.error_type === 'DEVICE_BUSY' || response.status === 409;
|
||||
const staleStart = payload?.superseded === true || payload?.status === 'stale';
|
||||
if (staleStart) return;
|
||||
const busy = payload?.error_type === 'DEVICE_BUSY' || (response.status === 409 && !staleStart);
|
||||
if (
|
||||
busy
|
||||
&& _running
|
||||
@@ -2834,8 +2839,10 @@ const Waterfall = (function () {
|
||||
gain,
|
||||
device: monitorDevice,
|
||||
biasT,
|
||||
requestToken: nonce,
|
||||
}));
|
||||
if (nonce !== _audioConnectNonce) return;
|
||||
if (payload?.superseded === true || payload?.status === 'stale') return;
|
||||
}
|
||||
|
||||
if (!response.ok || payload.status !== 'started') {
|
||||
|
||||
Reference in New Issue
Block a user