187 lines
6.8 KiB
Python
187 lines
6.8 KiB
Python
"""The desk API: conversation (with the command relay) and text-to-speech.
|
|
|
|
text -> POST /desk/converse_stream (NDJSON, sentence by sentence)
|
|
or /desk/converse
|
|
[server may relay a `command`; the answer goes to /desk/tool_result]
|
|
reply <- final text
|
|
text -> POST /desk/tts -> raw 16 kHz mono PCM16, streamed
|
|
|
|
Same protocol and conventions as Bolt Console's server_client and Bolt-Pet's
|
|
tts, which this is adapted from.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Callable, Iterable, Iterator, NamedTuple, Optional
|
|
|
|
import requests
|
|
|
|
from . import config
|
|
|
|
_MAX_RELAY_HOPS = 16
|
|
|
|
|
|
class ServerError(Exception):
|
|
pass
|
|
|
|
|
|
class Reply(NamedTuple):
|
|
text: str
|
|
streamed: bool = False # already handed over sentence by sentence
|
|
voice_id: str = "" # the server's speak_as pick, if any
|
|
|
|
|
|
def _headers() -> dict:
|
|
return {"X-Desk-Api-Key": config.API_KEY}
|
|
|
|
|
|
def check_health(timeout: float = 8.0) -> dict:
|
|
response = requests.get(f"{config.SERVER_URL}/desk/health", headers=_headers(), timeout=timeout)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
def _reply_from(payload: dict, streamed: bool = False) -> Reply:
|
|
if payload.get("type") == "reply":
|
|
return Reply(str(payload.get("text") or ""), streamed, str(payload.get("voice_id") or ""))
|
|
raise ServerError(str(payload.get("error") or "unknown server response"))
|
|
|
|
|
|
def _relay(payload: dict, on_command, on_say) -> dict:
|
|
for _ in range(_MAX_RELAY_HOPS):
|
|
if payload.get("type") != "command":
|
|
break
|
|
holding = str(payload.get("say") or "").strip()
|
|
if holding and on_say is not None:
|
|
try:
|
|
on_say(holding)
|
|
except Exception:
|
|
pass
|
|
output = on_command(str(payload.get("command") or ""))
|
|
try:
|
|
payload = requests.post(
|
|
f"{config.SERVER_URL}/desk/tool_result",
|
|
json={"session_id": config.SESSION_ID, "token": payload.get("token"), "output": output},
|
|
headers=_headers(), timeout=180,
|
|
).json()
|
|
except Exception as exc:
|
|
raise ServerError(f"couldn't reach the server during tool relay: {exc}") from exc
|
|
return payload
|
|
|
|
|
|
def converse(text: str, on_command: Callable[[str], str],
|
|
on_say: Optional[Callable[[str], None]] = None, timeout: float = 120.0) -> Reply:
|
|
try:
|
|
payload = requests.post(
|
|
f"{config.SERVER_URL}/desk/converse",
|
|
json={"session_id": config.SESSION_ID, "text": text},
|
|
headers=_headers(), timeout=timeout,
|
|
).json()
|
|
except Exception as exc:
|
|
raise ServerError(f"couldn't reach the server: {exc}") from exc
|
|
return _reply_from(_relay(payload, on_command, on_say))
|
|
|
|
|
|
def converse_stream(text: str, on_say: Callable[[str], None], on_command: Callable[[str], str],
|
|
timeout: float = 180.0) -> Reply:
|
|
"""Each sentence to *on_say* as it's generated, so speaking starts after
|
|
the first sentence, not the whole answer. Raises ServerError only if
|
|
nothing was said yet (the caller then uses converse())."""
|
|
said = []
|
|
try:
|
|
response = requests.post(
|
|
f"{config.SERVER_URL}/desk/converse_stream",
|
|
json={"session_id": config.SESSION_ID, "text": text},
|
|
headers=_headers(), timeout=timeout, stream=True,
|
|
)
|
|
response.raise_for_status()
|
|
for raw in response.iter_lines(decode_unicode=True):
|
|
if not raw:
|
|
continue
|
|
try:
|
|
event = json.loads(raw)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
kind = str(event.get("type") or "")
|
|
if kind == "say":
|
|
line = str(event.get("text") or "").strip()
|
|
if line:
|
|
said.append(line)
|
|
on_say(line)
|
|
elif kind == "command":
|
|
response.close()
|
|
return _reply_from(_relay(event, on_command, on_say))
|
|
elif kind == "reply":
|
|
if event.get("already_spoken"):
|
|
return _reply_from(event, streamed=True)
|
|
return unsaid(_reply_from(event), said)
|
|
elif kind == "error":
|
|
raise ServerError(str(event.get("error") or "stream failed"))
|
|
except ServerError:
|
|
raise
|
|
except Exception as exc:
|
|
if said:
|
|
return Reply("", streamed=True) # never repeat half an answer out loud
|
|
raise ServerError(f"streaming failed: {exc}") from exc
|
|
raise ServerError("stream ended without a reply")
|
|
|
|
|
|
def unsaid(reply: Reply, said: list) -> Reply:
|
|
"""The part of *reply* not already spoken as streamed sentences — the
|
|
server can stream the sentences and then send the whole text again
|
|
without `already_spoken` (compared ignoring whitespace)."""
|
|
if not said:
|
|
return reply
|
|
target = len("".join("".join(said).split()))
|
|
seen, cut = 0, None
|
|
for index, char in enumerate(reply.text):
|
|
if not char.isspace():
|
|
seen += 1
|
|
if seen == target:
|
|
cut = index + 1
|
|
break
|
|
if cut is None or "".join(reply.text[:cut].split()) != "".join("".join(said).split()):
|
|
return reply
|
|
rest = reply.text[cut:].strip()
|
|
return Reply(rest, streamed=not rest, voice_id=reply.voice_id)
|
|
|
|
|
|
# ── text to speech ──────────────────────────────────────────────────────────
|
|
|
|
class TtsError(Exception):
|
|
pass
|
|
|
|
|
|
def tts_stream(text: str, voice_id: str = "", chunk_bytes: int = 4096) -> Iterator[bytes]:
|
|
"""PCM16 bytes as the server synthesises them (whole samples only).
|
|
Raises TtsError before yielding anything if the request fails."""
|
|
voice = (voice_id or "").strip() or config.VOICE_ID
|
|
if not voice:
|
|
raise TtsError("ELEVENLABS_VOICE_ID isn't set, so the server doesn't know which voice to use")
|
|
try:
|
|
response = requests.post(
|
|
f"{config.SERVER_URL}/desk/tts", headers=_headers(),
|
|
json={"session_id": config.SESSION_ID, "text": text, "voice_id": voice},
|
|
timeout=60, stream=True,
|
|
)
|
|
response.raise_for_status()
|
|
except Exception as exc:
|
|
raise TtsError(f"text-to-speech request failed: {exc}") from exc
|
|
return whole_samples(response.iter_content(chunk_size=chunk_bytes))
|
|
|
|
|
|
def whole_samples(chunks: Iterable[bytes]) -> Iterator[bytes]:
|
|
"""HTTP chunks split wherever they like, including inside a 16-bit
|
|
sample. Carry the odd byte forward, or everything after the first split
|
|
plays as static (a Bolt-Pet lesson)."""
|
|
carry = b""
|
|
for chunk in chunks:
|
|
if not chunk:
|
|
continue
|
|
data = carry + chunk
|
|
usable = len(data) - len(data) % 2
|
|
carry = data[usable:]
|
|
if usable:
|
|
yield data[:usable]
|