195 lines
7.0 KiB
Python
195 lines
7.0 KiB
Python
"""A stand-in for the AGIBOT A3's speech RPC, for testing the REAL adapter path.
|
|
|
|
The mock robot (ROBOT_MODE=mock) tests the app. This tests the *wire protocol* -
|
|
it speaks the interface AgiBot documents for the A3, so you can run the app in
|
|
ROBOT_MODE=real against 127.0.0.1 and exercise `aimdk_transport.py` itself:
|
|
the URL shape, the JSON body, the trace_id round trip, chunking, Stop, and the
|
|
error paths. When the real robot arrives, only ROBOT_IP changes.
|
|
|
|
It mirrors the documented contract, including its quirks:
|
|
* route POST /rpc/<service>/<method>, Content-Type: application/json
|
|
* success flag `is_sucess` (one 'c' - as printed in AgiBot's docs)
|
|
* trace_id the reply appends a random suffix to the one you sent
|
|
* size limit 1024 bytes of UTF-8 on `text`
|
|
* unknown route 404; RPC-level failure 500
|
|
|
|
Docs: https://open.agibot.com/docs/en/aimdk/a3/v3_2/dev_guide/07-02-audio_play
|
|
|
|
Run:
|
|
python scripts/fake_a3_server.py # listens on 127.0.0.1:59301
|
|
|
|
Then in .env:
|
|
ROBOT_MODE=real
|
|
ROBOT_IP=127.0.0.1
|
|
ROBOT_PORT=59301
|
|
|
|
NOTE: this is a TEST DOUBLE written from public documentation. It is not the
|
|
robot, and passing against it proves the client is well-formed - not that the
|
|
robot's firmware behaves identically. Verify against the real unit.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import secrets
|
|
import sys
|
|
import time
|
|
from typing import Any, Dict, Optional
|
|
|
|
try:
|
|
import uvicorn
|
|
from fastapi import FastAPI, Request, Response
|
|
from fastapi.responses import JSONResponse
|
|
except ImportError: # pragma: no cover
|
|
print("Install dependencies first: pip install -r requirements.txt")
|
|
raise SystemExit(2)
|
|
|
|
def _use_utf8_console() -> None:
|
|
"""Windows consoles default to a legacy code page; the robot speaks Chinese."""
|
|
for stream in (sys.stdout, sys.stderr):
|
|
try:
|
|
if stream is not None and hasattr(stream, "reconfigure"):
|
|
stream.reconfigure(encoding="utf-8", errors="replace")
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
_use_utf8_console()
|
|
|
|
|
|
SERVICE = "aimdk.protocol.TTSService"
|
|
MAX_BYTES = 1024
|
|
|
|
app = FastAPI(title="Fake AGIBOT A3 speech RPC", docs_url=None, openapi_url=None)
|
|
|
|
STATE: Dict[str, Any] = {"utterances": {}, "active": None, "count": 0}
|
|
|
|
|
|
def _log(kind: str, message: str) -> None:
|
|
print(" {0} {1:<9} {2}".format(time.strftime("%H:%M:%S"), kind, message), flush=True)
|
|
|
|
|
|
@app.post("/rpc/{service}/{method}")
|
|
async def rpc(service: str, method: str, request: Request) -> Response:
|
|
if service != SERVICE:
|
|
_log("404", "unknown service {0}".format(service))
|
|
return JSONResponse({"error": "no such service"}, status_code=404)
|
|
|
|
# AimRT requires this header and rejects anything else.
|
|
content_type = (request.headers.get("content-type") or "").split(";")[0].strip()
|
|
if content_type != "application/json":
|
|
_log("500", "bad Content-Type: {0!r}".format(content_type))
|
|
return JSONResponse({"error": "unsupported content type"}, status_code=500)
|
|
|
|
try:
|
|
body = json.loads(await request.body() or b"{}")
|
|
except json.JSONDecodeError:
|
|
return JSONResponse({"error": "malformed json"}, status_code=500)
|
|
|
|
handler = {
|
|
"PlayTTS": _play_tts,
|
|
"StopTTSTraceId": _stop_tts,
|
|
"GetAudioStatus": _get_status,
|
|
}.get(method)
|
|
|
|
if handler is None:
|
|
_log("404", "unknown method {0}".format(method))
|
|
return JSONResponse({"error": "no such method"}, status_code=404)
|
|
|
|
return handler(body)
|
|
|
|
|
|
def _play_tts(body: Dict[str, Any]) -> Response:
|
|
text = body.get("text") or ""
|
|
size = len(text.encode("utf-8"))
|
|
|
|
if size > MAX_BYTES:
|
|
_log("REJECT", "text is {0} bytes (limit {1})".format(size, MAX_BYTES))
|
|
return JSONResponse(_reply(body, False, "text exceeds 1024 bytes"), status_code=200)
|
|
if not text.strip():
|
|
_log("REJECT", "empty text")
|
|
return JSONResponse(_reply(body, False, "empty text"), status_code=200)
|
|
|
|
trace = "{0}_{1}".format(body.get("trace_id") or "trace", secrets.token_urlsafe(16))
|
|
STATE["utterances"][trace] = {"text": text, "started": time.time(), "stopped": False}
|
|
STATE["active"] = trace
|
|
STATE["count"] += 1
|
|
|
|
_log("SPEAK", '"{0}" [{1} bytes, priority={2}, interrupt={3}]'.format(
|
|
text if len(text) <= 70 else text[:67] + "...",
|
|
size, body.get("priority_level"), body.get("is_interrupted"),
|
|
))
|
|
return JSONResponse(_reply(body, True, "", trace), status_code=200)
|
|
|
|
|
|
def _stop_tts(body: Dict[str, Any]) -> Response:
|
|
trace = body.get("trace_id")
|
|
entry = STATE["utterances"].get(trace)
|
|
if entry is None:
|
|
_log("STOP", "unknown trace_id {0!r}".format(trace))
|
|
return JSONResponse({"is_sucess": False, "error_message": "unknown trace_id"}, status_code=200)
|
|
entry["stopped"] = True
|
|
if STATE["active"] == trace:
|
|
STATE["active"] = None
|
|
_log("STOP", "stopped {0}".format(trace))
|
|
return JSONResponse({"is_sucess": True, "error_message": "", "trace_id": trace}, status_code=200)
|
|
|
|
|
|
def _get_status(body: Dict[str, Any]) -> Response:
|
|
trace = body.get("trace_id")
|
|
entry = STATE["utterances"].get(trace)
|
|
if entry is None:
|
|
status = "TTSStatusType_NOTInQue"
|
|
elif entry["stopped"]:
|
|
status = "TTSStatusType_Stop"
|
|
elif time.time() - entry["started"] < _estimate(entry["text"]):
|
|
status = "TTSStatusType_Playing"
|
|
else:
|
|
status = "TTSStatusType_End"
|
|
return JSONResponse({"trace_id": trace, "tts_status": status, "is_sucess": True}, status_code=200)
|
|
|
|
|
|
def _reply(body: Dict[str, Any], ok: bool, error: str, trace: Optional[str] = None) -> Dict[str, Any]:
|
|
return {
|
|
"text": body.get("text", ""),
|
|
"priority_level": body.get("priority_level", ""),
|
|
"priority_weight": 0,
|
|
"domain": body.get("domain", ""),
|
|
"trace_id": trace or body.get("trace_id", ""),
|
|
"is_sucess": ok, # the documented spelling - do not "fix" it
|
|
"error_message": error,
|
|
"estimated_duration": 0,
|
|
}
|
|
|
|
|
|
def _estimate(text: str) -> float:
|
|
words = max(1, len(text.split()))
|
|
return max(1.0, words / 150 * 60)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Fake AGIBOT A3 speech RPC endpoint.")
|
|
parser.add_argument("--host", default="127.0.0.1")
|
|
parser.add_argument("--port", type=int, default=59301)
|
|
args = parser.parse_args()
|
|
|
|
print("")
|
|
print(" Fake AGIBOT A3 speech RPC")
|
|
print(" " + "-" * 52)
|
|
print(" Listening : http://{0}:{1}".format(args.host, args.port))
|
|
print(" Endpoint : POST /rpc/{0}/PlayTTS".format(SERVICE))
|
|
print("")
|
|
print(" Point the app at it with:")
|
|
print(" ROBOT_MODE=real")
|
|
print(" ROBOT_IP={0}".format(args.host))
|
|
print(" ROBOT_PORT={0}".format(args.port))
|
|
print("")
|
|
print(" This is a TEST DOUBLE built from public docs - not the robot.")
|
|
print("")
|
|
uvicorn.run(app, host=args.host, port=args.port, log_level="warning", access_log=False)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|