"""WebSocket handshake helpers. Keep this module free of ASR/TTS imports."""
|
|
import json
|
|
|
def is_websocket_upgrade(headers) -> bool:
|
"""True when the HTTP request is a WebSocket handshake, not a health check."""
|
if headers is None:
|
return False
|
try:
|
connection_hdr = str(headers.get("connection", "") or "")
|
upgrade_hdr = str(headers.get("upgrade", "") or "")
|
except Exception:
|
return False
|
return "upgrade" in connection_hdr.lower() or "websocket" in upgrade_hdr.lower()
|
|
|
def is_client_hello(message) -> bool:
|
"""Device hello must be answered even while private config is still loading."""
|
if not isinstance(message, str):
|
return False
|
try:
|
data = json.loads(message)
|
except Exception:
|
return False
|
return isinstance(data, dict) and data.get("type") == "hello"
|
|
|
def merge_hello_audio_params(welcome_params, client_params) -> dict:
|
"""Keep server TTS output rate. Device hello sample_rate is capture-only."""
|
merged = dict(welcome_params or {})
|
if not isinstance(client_params, dict):
|
return merged
|
fmt = client_params.get("format")
|
if fmt:
|
merged["format"] = fmt
|
channels = client_params.get("channels")
|
if channels:
|
merged["channels"] = channels
|
return merged
|
|
|
def is_priority_listen_detect(message) -> bool:
|
"""Boot/app-task detect must not be dropped while bind config is still loading."""
|
if not isinstance(message, str):
|
return False
|
try:
|
data = json.loads(message)
|
except Exception:
|
return False
|
if not isinstance(data, dict) or data.get("type") != "listen":
|
return False
|
if data.get("state") != "detect":
|
return False
|
text = str(data.get("text") or "")
|
return (
|
text.startswith("[app_task]")
|
or text.startswith("[voice_entry]")
|
or text.startswith("[skill_event]")
|
or text.startswith("[lesson_sync]")
|
)
|
