307 lines
11 KiB
Python
307 lines
11 KiB
Python
"""Voice-session control — the on/off switch behind the Interaction page.
|
|
|
|
The conversation loop is a systemd *user* unit (sanad_agibot.service) that the
|
|
dashboard already has the rights to drive, because both run as `agi` in the same
|
|
user session. So "turn speaking on" is: persist the operator's selection, then
|
|
start/restart that unit; "off" is stopping it.
|
|
|
|
The (gender, language, model) -> (persona, voice) mapping deliberately lives in
|
|
Sanad's own voice/session_profile.py and is imported from there by path rather
|
|
than copied here. A copy would drift, and a drifted copy means the dashboard
|
|
shows one character while the robot speaks as another.
|
|
|
|
Everything in here degrades to a readable status instead of raising: if Sanad
|
|
is not deployed, or systemd is unavailable, the card still renders and explains
|
|
what is missing.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
SERVICE = os.environ.get("SANAD_SERVICE_NAME", "sanad_agibot.service")
|
|
|
|
SANAD_DIR = Path(os.environ.get(
|
|
"SANAD_DIR",
|
|
Path.home() / "sanad_deploy" / "sanad_package_1" / "vendor" / "Sanad",
|
|
))
|
|
_PROFILE_MODULE_PATH = SANAD_DIR / "voice" / "session_profile.py"
|
|
|
|
# Choices the UI renders. Kept server-side so the page and the robot can never
|
|
# disagree about what is selectable.
|
|
OPTIONS: dict[str, list[dict[str, str]]] = {
|
|
"gender": [
|
|
{"value": "female", "label": "Female — Muza / موزة"},
|
|
{"value": "male", "label": "Male — Lumi / لومي"},
|
|
],
|
|
"language": [
|
|
{"value": "arabic", "label": "Arabic only (Emirati)"},
|
|
{"value": "multi", "label": "Multi-language"},
|
|
],
|
|
# The second value is still keyed "linksoul" so an already-saved selection
|
|
# keeps working; the label describes what it actually does now.
|
|
"model": [
|
|
{"value": "gemini", "label": "Gemini Live"},
|
|
{"value": "linksoul", "label": "Pipeline — Fatima / Hamdan"},
|
|
],
|
|
}
|
|
|
|
_profile_mod = None
|
|
_profile_err: str | None = None
|
|
|
|
|
|
def _profile():
|
|
"""Sanad's session_profile module, loaded from disk on first use."""
|
|
global _profile_mod, _profile_err
|
|
if _profile_mod is not None or _profile_err is not None:
|
|
return _profile_mod
|
|
|
|
try:
|
|
spec = importlib.util.spec_from_file_location(
|
|
"sanad_session_profile", _PROFILE_MODULE_PATH)
|
|
if spec is None or spec.loader is None:
|
|
raise ImportError(f"cannot load {_PROFILE_MODULE_PATH}")
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
_profile_mod = module
|
|
except Exception as exc: # noqa: BLE001 - reported, never fatal
|
|
_profile_err = f"{type(exc).__name__}: {exc}"
|
|
return _profile_mod
|
|
|
|
|
|
def _systemctl(*args: str, timeout: float = 20.0) -> tuple[int, str]:
|
|
if not shutil.which("systemctl"):
|
|
return 127, "systemctl not available on this host"
|
|
try:
|
|
proc = subprocess.run(
|
|
["systemctl", "--user", *args],
|
|
capture_output=True, text=True, timeout=timeout,
|
|
)
|
|
return proc.returncode, (proc.stdout + proc.stderr).strip()
|
|
except subprocess.TimeoutExpired:
|
|
return 124, f"systemctl --user {' '.join(args)} timed out"
|
|
except OSError as exc:
|
|
return 1, str(exc)
|
|
|
|
|
|
_VOICE_PATTERNS = (
|
|
r"vendor/Sanad/voice/sanad_voice\.py",
|
|
r"vendor/Sanad/pipeline/runner\.py",
|
|
)
|
|
|
|
|
|
def _kill_voice_processes(grace: float = 6.0) -> int:
|
|
"""Stop the running voice process whoever started it.
|
|
|
|
Needed because the process is launched by keepalive_daemon.sh rather than
|
|
systemd, so stopping the unit is not enough to silence the robot. SIGINT
|
|
first so it releases ROS audio focus and closes the Gemini session cleanly.
|
|
"""
|
|
import signal
|
|
import subprocess as sp
|
|
|
|
pids: list[str] = []
|
|
for pat in _VOICE_PATTERNS:
|
|
try:
|
|
out = sp.run(["pgrep", "-f", pat], capture_output=True, text=True, timeout=5)
|
|
pids += [p for p in out.stdout.split() if p.strip()]
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
if not pids:
|
|
return 0
|
|
for p in pids:
|
|
try:
|
|
os.kill(int(p), signal.SIGINT)
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
time.sleep(grace)
|
|
for p in pids:
|
|
try:
|
|
os.kill(int(p), signal.SIGKILL)
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
return len(pids)
|
|
|
|
|
|
def service_status() -> dict[str, Any]:
|
|
code, out = _systemctl("is-active", SERVICE, timeout=10.0)
|
|
active = out.strip() or ("unknown" if code else "inactive")
|
|
return {
|
|
"unit": SERVICE,
|
|
"state": active,
|
|
"running": active == "active",
|
|
}
|
|
|
|
|
|
ENV_FILE = Path(os.environ.get("SANAD_ENV_FILE", Path.home() / ".sanad_agibot_env"))
|
|
|
|
_CRED_KEYS = ("LINKSOUL_APP_ID", "LINKSOUL_APP_KEY", "LINKSOUL_APP_SECRET")
|
|
|
|
|
|
def _env_file_keys() -> set[str]:
|
|
"""Names assigned in ~/.sanad_agibot_env.
|
|
|
|
Checked here rather than in os.environ because that file is sourced by the
|
|
launcher, not by the dashboard — the credentials are never visible in this
|
|
process's own environment, so reading os.environ would always report them
|
|
missing. Only key names are collected; the values are never read.
|
|
"""
|
|
found: set[str] = set()
|
|
try:
|
|
for line in ENV_FILE.read_text(encoding="utf-8").splitlines():
|
|
line = line.strip()
|
|
if line.startswith("export "):
|
|
line = line[len("export "):].lstrip()
|
|
name, sep, value = line.partition("=")
|
|
if sep and value.strip().strip("\"'"):
|
|
found.add(name.strip())
|
|
except OSError:
|
|
pass
|
|
return found
|
|
|
|
|
|
def linksoul_ready() -> dict[str, Any]:
|
|
"""Whether the pipeline brain can actually run right now.
|
|
|
|
It needs edge-tts for the Emirati voice, miniaudio to decode that MP3 (this
|
|
robot has no ffmpeg and the `agi` account has no sudo), requests for the STT
|
|
call, and a Gemini key. It does NOT need the LinkSoul cloud SDK or its app
|
|
credentials - that was the earlier cloud design, which required AgiBot to
|
|
bind this robot to an application. The pipeline talks to the mic and speaker
|
|
directly, so it works with no platform registration at all.
|
|
"""
|
|
missing_pkgs = [p for p in ("edge_tts", "miniaudio", "requests")
|
|
if importlib.util.find_spec(p) is None]
|
|
has_key = ("SANAD_GEMINI_API_KEY" in _env_file_keys()
|
|
or bool(os.environ.get("SANAD_GEMINI_API_KEY"))
|
|
or bool(os.environ.get("GEMINI_API_KEY")))
|
|
|
|
reasons = []
|
|
if missing_pkgs:
|
|
reasons.append("missing Python packages: " + ", ".join(missing_pkgs)
|
|
+ " (pip3 install --user " + " ".join(missing_pkgs) + ")")
|
|
if not has_key:
|
|
reasons.append(f"SANAD_GEMINI_API_KEY is not set in {ENV_FILE}")
|
|
return {"ready": not reasons, "reasons": reasons}
|
|
|
|
|
|
def snapshot() -> dict[str, Any]:
|
|
"""Everything the Interaction card needs to render itself."""
|
|
mod = _profile()
|
|
if mod is None:
|
|
return {
|
|
"available": False,
|
|
"error": _profile_err or "session_profile.py not found",
|
|
"sanad_dir": str(SANAD_DIR),
|
|
"options": OPTIONS,
|
|
"service": service_status(),
|
|
"linksoul": linksoul_ready(),
|
|
}
|
|
|
|
resolved = mod.resolve()
|
|
return {
|
|
"available": True,
|
|
"error": "",
|
|
"options": OPTIONS,
|
|
**mod.catalog(),
|
|
"profile": {k: resolved[k] for k in ("enabled", "gender", "language", "model")},
|
|
"persona": {
|
|
"file": resolved["persona_file"],
|
|
"exists": resolved["persona_exists"],
|
|
"name_en": resolved["name_en"],
|
|
"name_ar": resolved["name_ar"],
|
|
},
|
|
"voice": {
|
|
"gemini": resolved["gemini_voice"],
|
|
"edge": resolved["edge_voice"],
|
|
},
|
|
"greeting": resolved["greeting"],
|
|
"service": service_status(),
|
|
"linksoul": linksoul_ready(),
|
|
}
|
|
|
|
|
|
def apply(enabled: bool, gender: str | None, language: str | None,
|
|
model: str | None) -> dict[str, Any]:
|
|
"""Persist the selection and bring the conversation loop to that state.
|
|
|
|
Returns the new snapshot plus a `greeting` the caller should speak when the
|
|
session has just been switched on. The greeting is spoken by the dashboard
|
|
rather than by the persona so it fires exactly once, at the moment Apply is
|
|
pressed, on whichever model is selected — the LinkSoul path is a passive
|
|
callback and cannot start talking on its own.
|
|
"""
|
|
mod = _profile()
|
|
if mod is None:
|
|
return {"ok": False, "error": _profile_err or "session_profile.py not found",
|
|
"snapshot": snapshot()}
|
|
|
|
# Validate a CANDIDATE before persisting anything. Saving first and
|
|
# rejecting afterwards left an unlaunchable selection on disk: the dashboard
|
|
# refused to start it, but start_agibot.sh would happily replay it at the
|
|
# next boot, exec a runner that exits immediately, and Restart=on-failure
|
|
# would turn that into a crash loop with the robot mute.
|
|
current = mod.load()
|
|
candidate = mod.normalize({
|
|
"enabled": enabled,
|
|
"gender": gender or current["gender"],
|
|
"language": language or current["language"],
|
|
"model": model or current["model"],
|
|
})
|
|
resolved = mod.resolve(candidate)
|
|
|
|
if not resolved["persona_exists"]:
|
|
return {
|
|
"ok": False,
|
|
"error": f"Persona file missing: {resolved['persona_file']}",
|
|
"snapshot": snapshot(),
|
|
}
|
|
|
|
if enabled and resolved["model"] == "linksoul":
|
|
ready = linksoul_ready()
|
|
if not ready["ready"]:
|
|
return {"ok": False,
|
|
"error": "LinkSoul is not usable yet — " + "; ".join(ready["reasons"]),
|
|
"snapshot": snapshot()}
|
|
|
|
saved = mod.save(**{k: candidate[k] for k in ("enabled", "gender", "language", "model")})
|
|
resolved = mod.resolve(saved)
|
|
|
|
# The voice process is owned by keepalive_daemon.sh, not by systemd - the
|
|
# user manager dies with the last login session (no linger, needs root), so
|
|
# the daemon had to take over. That means `systemctl stop` no longer stops
|
|
# anything: pressing "Speaking off" left the robot still listening. So act
|
|
# on the processes directly here, and let the daemon keep enforcing it.
|
|
_kill_voice_processes()
|
|
|
|
if enabled:
|
|
_systemctl("reset-failed", SERVICE, timeout=10.0)
|
|
code, out = _systemctl("restart", SERVICE, timeout=45.0)
|
|
action = "restarted"
|
|
else:
|
|
code, out = _systemctl("stop", SERVICE, timeout=30.0)
|
|
action = "stopped"
|
|
# The voice process does not exit on SIGINT within TimeoutStopSec, so
|
|
# systemd SIGKILLs it and books the unit as failed. That is an artifact
|
|
# of an operator-requested stop, not a fault, and showing "failed" on
|
|
# the card after pressing Off would be plainly wrong.
|
|
if code == 0:
|
|
_systemctl("reset-failed", SERVICE, timeout=10.0)
|
|
|
|
if code != 0:
|
|
return {"ok": False, "error": f"systemctl {action} failed: {out}",
|
|
"snapshot": snapshot()}
|
|
|
|
return {
|
|
"ok": True,
|
|
"action": action,
|
|
"greeting": resolved["greeting"] if enabled else "",
|
|
"snapshot": snapshot(),
|
|
}
|