adolforeyna/hermes-esp32-voice-gateway
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
hermes-esp32-voice-gateway/patch/esp32_live_draft_patch.patch
T
Adolfo Reyna 6d593cd9ef feat: initial durable ESP32 voice gateway plugin + patches 
- Migrates custom voice gateway from in-tree api_server.py patch to durable plugin
- Auto-restores routes after hermes update (POST /api/esp32/voice + WS /api/esp32/voice/ws)
- Includes live draft ingestion via MacMiniMCP speech_live_transcribe (Apple pipe volatile)
- Includes instant smart reply via apple_llm_quick_reply (3B ANE ~300ms, contextual not generic)
- Guard for missing ipaddress/urlparse imports that caused 500 loop (2026-07-14)
- Tools: esp32_voice_gateway_status, esp32_voice_gateway_restore
- Scripts: restore_after_update.py with 400-not-404 verification
- Installed as ~/.hermes/plugins/esp32-voice-gateway (enabled, survives hermes update)

Verified: routes present, plugin enabled, has_routes=True
2026-07-14 23:42:34 -04:00

1573 lines
77 KiB
Diff
Raw Permalink Blame History

diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py
index 628713224..8bcf9e9cc 100644
--- a/gateway/platforms/api_server.py
+++ b/gateway/platforms/api_server.py
@@ -34,6 +34,7 @@ Requires:
import asyncio
import hashlib
import hmac
+import ipaddress
import json
from contextlib import contextmanager
from contextvars import ContextVar
@@ -47,6 +48,7 @@ import time
import uuid
from pathlib import Path
from typing import Any, Dict, List, Optional
+from urllib.parse import urlparse
def _approval_event_choices(*, smart_denied: bool, allow_permanent: bool) -> list[str]:
if smart_denied:
@@ -106,6 +108,47 @@ CHAT_COMPLETIONS_SSE_KEEPALIVE_SECONDS = 30.0
MAX_NORMALIZED_TEXT_LENGTH = 65_536 # 64 KB cap for normalized content parts
MAX_CONTENT_LIST_SIZE = 1_000 # Max items when content is an array
+ESP32_AUDIO_MAX_BYTES = 8_000_000 # ~4 minutes of 16 kHz mono PCM WAV; enough for push-to-talk tests
+ESP32_WS_MAX_AUDIO_BYTES = 8_000_000
+ESP32_WS_MIN_AUDIO_BYTES = 1_600 # 50 ms at 16 kHz mono s16le; rejects empty accidental taps
+ESP32_WS_TTS_CHUNK_BYTES = 16_384
+
+# Live draft ingestion config (from whisper-translation/apple_speech engine)
+ESP32_LIVE_DRAFT_ENABLED = os.getenv("ESP32_LIVE_DRAFT_ENABLED", "1").lower() not in {"0", "false", "no", "off"}
+ESP32_LIVE_DRAFT_INTERVAL_MS = int(os.getenv("ESP32_LIVE_DRAFT_INTERVAL_MS", "800")) # how often to snapshot for draft
+ESP32_LIVE_DRAFT_MIN_BYTES = int(os.getenv("ESP32_LIVE_DRAFT_MIN_BYTES", "8000")) # min new bytes since last draft
+ESP32_LIVE_DRAFT_MIN_TOTAL_MS = int(os.getenv("ESP32_LIVE_DRAFT_MIN_TOTAL_MS", "500")) # don't draft until we have 500ms
+ESP32_MACMINI_MCP_URL = os.getenv("MACMINI_MCP_URL") or None # overridable, else from config.yaml mcp_servers.macmini
+ESP32_MACMINI_MCP_TOKEN = os.getenv("MACMINI_MCP_TOKEN") or None
+ESP32_LIVE_DRAFT_USE_MACMINI = os.getenv("ESP32_LIVE_DRAFT_USE_MACMINI", "1").lower() not in {"0", "false", "no", "off"}
+
+_ESP32_AUDIO_EXTENSIONS = {
+ "audio/wav": ".wav",
+ "audio/wave": ".wav",
+ "audio/x-wav": ".wav",
+ "audio/mpeg": ".mp3",
+ "audio/mp3": ".mp3",
+ "audio/ogg": ".ogg",
+ "audio/opus": ".ogg",
+ "audio/webm": ".webm",
+ "audio/mp4": ".m4a",
+ "audio/x-m4a": ".m4a",
+ "application/octet-stream": ".wav",
+}
+
+_AUDIO_RESPONSE_MIME_BY_SUFFIX = {
+ ".mp3": "audio/mpeg",
+ ".wav": "audio/wav",
+ ".ogg": "audio/ogg",
+ ".opus": "audio/ogg",
+ ".m4a": "audio/mp4",
+ ".mp4": "audio/mp4",
+ ".flac": "audio/flac",
+}
+
+ESP32_VOICE_ANIMATION_PATH = "hermes_voice_animations.py"
+ESP32_VOICE_ANIMATION_SCRIPT = "# Hermes ESP32 voice-state animations for MicroPython screen boards.\n# Uploaded by the API server; can also be run manually with:\n# exec(open('hermes_voice_animations.py').read()); run_state('ack')\nimport time\ntry:\n import board_config\nexcept Exception:\n board_config = None\n\n\ndef _display():\n return getattr(board_config, 'display_instance', None) if board_config else None\n\n\ndef _sleep(ms):\n try:\n time.sleep_ms(ms)\n except AttributeError:\n time.sleep(ms / 1000)\n\n\ndef _text(d, msg, x, y, scale=1, color=1):\n if hasattr(d, 'text_large') and scale > 1:\n try:\n d.text_large(msg, x, y, scale=scale, color=color)\n return\n except TypeError:\n pass\n d.text(msg, x, y, color)\n\n\ndef _center_text(d, msg, y, scale=1, color=1):\n width = getattr(d, 'width', 320)\n char_w = 8 * scale\n x = max(0, (width - len(msg) * char_w) // 2)\n _text(d, msg, x, y, scale=scale, color=color)\n\n\ndef _wrap(text, width_chars=34, max_lines=4):\n words = str(text or '').replace('\\n', ' ').split()\n lines, line = [], ''\n for w in words:\n candidate = (line + ' ' + w).strip()\n if len(candidate) > width_chars and line:\n lines.append(line)\n line = w\n if len(lines) >= max_lines:\n break\n else:\n line = candidate\n if line and len(lines) < max_lines:\n lines.append(line)\n return lines\n\n\ndef _spinner(d, title, subtitle='', frames=24, delay=70):\n width, height = getattr(d, 'width', 320), getattr(d, 'height', 240)\n cx, cy = width // 2, height // 2 - 8\n spokes = [(-22,0),(-16,-16),(0,-22),(16,-16),(22,0),(16,16),(0,22),(-16,16)]\n for frame in range(frames):\n d.clear(0)\n _center_text(d, title, 18, scale=2)\n for i, (dx, dy) in enumerate(spokes):\n color = 1 if (i + frame) % len(spokes) in (0, 1, 2, 3) else 0\n size = 4 if i == frame % len(spokes) else 2\n d.fill_rect(cx + dx - size // 2, cy + dy - size // 2, size, size, color)\n d.rect(8, height - 34, width - 16, 18, 1)\n d.fill_rect(10, height - 32, ((width - 20) * ((frame % 16) + 1)) // 16, 14, 1)\n if subtitle:\n _center_text(d, subtitle[:32], height - 56, scale=1)\n d.show()\n _sleep(delay)\n\n\ndef ack(frames=18):\n d = _display()\n if not d:\n return\n width, height = d.width, d.height\n for i in range(frames):\n d.clear(0)\n _center_text(d, 'GOT IT', 18, scale=2)\n # expanding check pulse\n box = 48 + (i % 6) * 8\n x, y = width // 2 - box // 2, height // 2 - box // 2\n d.rect(x, y, box, box, 1)\n d.line(width//2 - 24, height//2, width//2 - 6, height//2 + 18, 1)\n d.line(width//2 - 6, height//2 + 18, width//2 + 30, height//2 - 22, 1)\n _center_text(d, 'Listening captured', height - 36, scale=1)\n d.show()\n _sleep(55)\n\n\ndef captioning(frames=36):\n d = _display()\n if not d:\n return\n width, height = d.width, d.height\n for i in range(frames):\n d.clear(0)\n _center_text(d, 'CAPTIONING', 16, scale=2)\n for bar in range(9):\n h = 8 + ((i * 5 + bar * 13) % 42)\n x = width // 2 - 72 + bar * 18\n d.fill_rect(x, height // 2 - h // 2, 10, h, 1)\n dots = '.' * ((i // 5) % 4)\n _center_text(d, 'Transcribing' + dots, height - 42, scale=1)\n d.show()\n _sleep(65)\n\n\ndef waiting(text='', frames=44):\n d = _display()\n if not d:\n return\n width, height = d.width, d.height\n for i in range(frames):\n d.clear(0)\n _center_text(d, 'THINKING', 12, scale=2)\n r = 18 + (i % 10) * 2\n cx, cy = width // 2, height // 2 - 10\n d.rect(cx - r, cy - r, r * 2, r * 2, 1)\n d.rect(cx - r//2, cy - r//2, r, r, 1)\n if text:\n d.text('Heard:', 10, height - 68, 1)\n for idx, line in enumerate(_wrap(text, 36, 3)):\n d.text(line[:38], 10, height - 54 + idx * 12, 1)\n else:\n _center_text(d, 'Waiting for reply...', height - 42, scale=1)\n d.show()\n _sleep(75)\n\n\ndef run_state(state='ack', text='', frames=None):\n state = str(state or 'ack').lower()\n if state in ('ack', 'listening', 'got_it'):\n ack(frames or 18)\n elif state in ('caption', 'captioning', 'transcribing'):\n captioning(frames or 36)\n elif state in ('wait', 'waiting', 'thinking', 'reply'):\n waiting(text, frames or 44)\n else:\n _spinner(_display(), 'HERMES', state[:28], frames or 20)\n"
+
def _coerce_port(value: Any, default: int = DEFAULT_PORT) -> int:
"""Parse a listen port without letting malformed env/config values crash startup."""
@@ -1585,6 +1628,1493 @@ class APIServerAdapter(BasePlatformAdapter):
"pid": os.getpid(),
})
+ @staticmethod
+ def _audio_extension_for_content_type(content_type: str) -> str:
+ """Map an inbound audio Content-Type to a safe cache-file suffix."""
+ media_type = (content_type or "").split(";", 1)[0].strip().lower()
+ return _ESP32_AUDIO_EXTENSIONS.get(media_type, ".wav")
+
+ @staticmethod
+ def _audio_response_mime(path: str) -> str:
+ """Return an HTTP Content-Type for a generated audio file path."""
+ suffix = Path(path).suffix.lower()
+ return _AUDIO_RESPONSE_MIME_BY_SUFFIX.get(suffix, "application/octet-stream")
+
+ @staticmethod
+
+ def _convert_audio_to_esp32_wav(input_path: str, *, output_dir: Path, device_id: str) -> Path:
+ """Normalize generated TTS audio to ESP32-friendly WAV.
+
+ Hermes TTS providers commonly emit MP3 (Edge) or OGG. The ESP32 test
+ client expects a small, predictable RIFF/WAVE file, so decode with
+ ffmpeg to raw PCM and rewrite the WAV container with Python's ``wave``
+ module. That produces a minimal ``fmt`` + ``data`` WAV: mono, 16 kHz,
+ signed 16-bit little-endian PCM.
+ """
+ import subprocess
+ import wave
+
+ safe_device = re.sub(r"[^A-Za-z0-9_.-]+", "_", device_id)[:80] or "default"
+ output_dir.mkdir(parents=True, exist_ok=True)
+ raw_path = output_dir / f"{safe_device}_{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}.s16le"
+ wav_path = raw_path.with_suffix(".wav")
+ cmd = [
+ "ffmpeg",
+ "-y",
+ "-hide_banner",
+ "-loglevel",
+ "error",
+ "-i",
+ input_path,
+ "-ac",
+ "1",
+ "-ar",
+ "16000",
+ "-f",
+ "s16le",
+ str(raw_path),
+ ]
+ try:
+ subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ raw = raw_path.read_bytes()
+ with wave.open(str(wav_path), "wb") as wav:
+ wav.setnchannels(1)
+ wav.setsampwidth(2)
+ wav.setframerate(16000)
+ wav.writeframes(raw)
+ return wav_path
+ finally:
+ try:
+ raw_path.unlink(missing_ok=True)
+ except Exception:
+ pass
+
+ @staticmethod
+ def _safe_audio_header(value: Any, max_length: int) -> str:
+ """Make diagnostic text safe for HTTP response headers."""
+ text = str(value or "")[:max_length]
+ return text.replace("\r", " ").replace("\n", " ").replace("\x00", " ")
+
+ @staticmethod
+ def _esp32_reply_mode(request: "web.Request") -> str:
+ """Return 'sync' or 'ack' for the ESP32 voice response contract."""
+ value = (
+ request.headers.get("X-Hermes-Reply-Mode")
+ or request.query.get("reply_mode")
+ or request.query.get("mode")
+ or "sync"
+ )
+ value = str(value).strip().lower()
+ if value in {"ack", "async", "background", "quick_ack", "quick-ack"}:
+ return "ack"
+ return "sync"
+
+ @staticmethod
+ def _looks_like_http_url(value: str) -> bool:
+ try:
+ parsed = urlparse(value)
+ return parsed.scheme in {"http", "https"} and bool(parsed.netloc)
+ except Exception:
+ return False
+
+ @staticmethod
+ def _load_local_device_url(device_name: str) -> Optional[str]:
+ """Best-effort local MCP screen lookup from ~/.hermes/local_devices.yaml."""
+ if not device_name:
+ return None
+ try:
+ import yaml
+ registry = Path.home() / ".hermes" / "local_devices.yaml"
+ data = yaml.safe_load(registry.read_text()) if registry.exists() else {}
+ aliases = data.get("aliases") if isinstance(data, dict) else None
+ devices = data.get("devices") if isinstance(data, dict) else None
+ names = [str(device_name).strip(), str(device_name).strip().replace("-", "_")]
+ seen: set[str] = set()
+ while names:
+ name = names.pop(0)
+ if not name or name in seen:
+ continue
+ seen.add(name)
+ if isinstance(aliases, dict):
+ alias = aliases.get(name)
+ if isinstance(alias, str) and alias:
+ if alias.startswith("http"):
+ return alias
+ names.append(alias)
+ info = devices.get(name) if isinstance(devices, dict) else None
+ if isinstance(info, dict):
+ return str(info.get("fallback_url") or info.get("url") or "") or None
+ except Exception:
+ return None
+ return None
+
+ @staticmethod
+ def _load_local_device_url_by_ip(source_ip: str) -> Optional[str]:
+ """Best-effort local MCP screen lookup by observed device IP.
+
+ Some MCP-capable devices, notably the iPhone companion app, do not run
+ their MCP endpoint on port 80. When the voice gateway sees traffic from
+ one of those devices, prefer the registry URL over synthesizing
+ http://<source-ip>/api/mcp.
+ """
+ value = str(source_ip or "").strip()
+ if not value:
+ return None
+ if value.startswith("::ffff:"):
+ value = value.removeprefix("::ffff:")
+ try:
+ import yaml
+ registry = Path.home() / ".hermes" / "local_devices.yaml"
+ data = yaml.safe_load(registry.read_text()) if registry.exists() else {}
+ devices = data.get("devices") if isinstance(data, dict) else None
+ if not isinstance(devices, dict):
+ return None
+ for info in devices.values():
+ if not isinstance(info, dict):
+ continue
+ candidates = {
+ str(info.get("reserved_ip") or "").strip(),
+ str(info.get("host") or "").strip(),
+ }
+ for url_key in ("fallback_url", "url"):
+ raw_url = str(info.get(url_key) or "").strip()
+ if raw_url:
+ parsed = urlparse(raw_url)
+ if parsed.hostname:
+ candidates.add(parsed.hostname)
+ if value in candidates:
+ return str(info.get("fallback_url") or info.get("url") or "") or None
+ except Exception:
+ return None
+ return None
+
+ @staticmethod
+ def _esp32_screen_url_from_source_ip(source_ip: str) -> Optional[str]:
+ """Build the MCP endpoint for the ESP32 that made this request.
+
+ ESP32 boards can move between DHCP leases. For voice ACK follow-ups the
+ initiating board is the safest target, so prefer its observed peer IP
+ instead of a possibly stale local-device registry entry.
+ """
+ value = str(source_ip or "").strip()
+ if not value:
+ return None
+ if value.startswith("::ffff:"):
+ value = value.removeprefix("::ffff:")
+ # Do not route to loopback/proxy/invalid addresses; registry fallback is
+ # better in those cases.
+ try:
+ ip = ipaddress.ip_address(value)
+ except ValueError:
+ return None
+ if ip.is_loopback or ip.is_unspecified or ip.is_multicast:
+ return None
+ host = f"[{value}]" if ip.version == 6 else value
+ return f"http://{host}/api/mcp"
+
+ def _resolve_esp32_screen_url(self, request: "web.Request", device_id: str) -> Optional[str]:
+ """Resolve the screen MCP endpoint associated with this voice turn."""
+ explicit = (
+ request.headers.get("X-Hermes-Screen-Url")
+ or request.headers.get("X-Screen-MCP-Url")
+ or request.query.get("screen_url")
+ )
+ if explicit and self._looks_like_http_url(explicit.strip()):
+ return explicit.strip()
+
+ if str(device_id or "").strip().lower().startswith("iphone"):
+ iphone_url = self._load_local_device_url(str(device_id).strip())
+ if iphone_url and self._looks_like_http_url(iphone_url):
+ return iphone_url
+
+ audit_ctx = self._request_audit_context(request)
+ for source_ip in (audit_ctx.get("peer_ip"), audit_ctx.get("remote")):
+ registry_url = self._load_local_device_url_by_ip(source_ip or "")
+ if registry_url and self._looks_like_http_url(registry_url):
+ return registry_url
+
+ audit_ctx = self._request_audit_context(request)
+ for source_ip in (audit_ctx.get("peer_ip"), audit_ctx.get("remote")):
+ source_url = self._esp32_screen_url_from_source_ip(source_ip or "")
+ if source_url:
+ return source_url
+
+ screen_device = (
+ request.headers.get("X-Hermes-Screen-Device")
+ or request.query.get("screen_device")
+ or device_id
+ )
+ candidates = [screen_device, "little32", "esp32_screen"]
+ for candidate in candidates:
+ url = self._load_local_device_url(str(candidate).strip())
+ if url and self._looks_like_http_url(url):
+ return url
+ return None
+
+ async def _call_screen_mcp(
+ self,
+ screen_url: Optional[str],
+ tool_name: str,
+ arguments: Dict[str, Any],
+ *,
+ timeout: float = 8.0,
+ ) -> bool:
+ """Call a simple HTTP JSON-RPC MCP screen tool directly."""
+ if not screen_url:
+ return False
+ try:
+ from aiohttp import ClientSession, ClientTimeout
+ payload = {
+ "jsonrpc": "2.0",
+ "id": uuid.uuid4().hex[:12],
+ "method": "tools/call",
+ "params": {"name": tool_name, "arguments": arguments},
+ }
+ async with ClientSession(timeout=ClientTimeout(total=timeout)) as session:
+ async with session.post(screen_url, json=payload) as resp:
+ body = await resp.text()
+ if resp.status >= 400:
+ logger.warning(
+ "[api_server] screen MCP %s failed: HTTP %s %s",
+ tool_name,
+ resp.status,
+ body[:300],
+ )
+ return False
+ try:
+ data = json.loads(body) if body else {}
+ if isinstance(data, dict) and data.get("error"):
+ logger.warning("[api_server] screen MCP %s returned error: %s", tool_name, data.get("error"))
+ return False
+ except Exception:
+ pass
+ return True
+ except Exception as exc:
+ logger.warning("[api_server] screen MCP %s failed: %s", tool_name, exc)
+ return False
+
+ @staticmethod
+ def _screen_stream_target(screen_url: Optional[str]) -> Optional[tuple[str, int]]:
+ """Return the raw UDP video-stream target for MCP screens that expose one."""
+ if not screen_url:
+ return None
+ try:
+ parsed = urlparse(screen_url)
+ host = parsed.hostname
+ if not host:
+ return None
+ return (host, 8082)
+ except Exception:
+ return None
+
+ @staticmethod
+ def _voice_animation_udp_frames(state: str, text: str = "", frames: int = 18) -> List[bytes]:
+ """Render small Hermes voice-state animations as RLCD-compatible frames."""
+ from PIL import Image, ImageDraw
+ import math
+
+ width, height = 400, 300
+
+ def to_rlcd(img: "Image.Image") -> bytes:
+ img_1bit = img.convert("1", dither=Image.Dither.FLOYDSTEINBERG)
+ px = img_1bit.load()
+ buf = bytearray(15000)
+ for y in range(height):
+ for x in range(width):
+ if px[x, y]:
+ inv_y = 299 - y
+ bx = x // 2
+ by = inv_y // 4
+ idx = bx * 75 + by
+ lx = x % 2
+ ly = inv_y % 4
+ bit = 7 - (ly * 2 + lx)
+ buf[idx] |= (1 << bit)
+ return bytes(buf)
+
+ label_by_state = {
+ "ack": "Got it",
+ "captioning": "Listening",
+ "waiting": "Thinking",
+ "speaking": "Replying",
+ "success": "Done",
+ "error": "Try again",
+ }
+ label = label_by_state.get(str(state), str(state or "Hermes").title())
+ snippet = " ".join(str(text or "").split())[:42]
+ rendered: List[bytes] = []
+ for frame in range(max(1, int(frames))):
+ img = Image.new("1", (width, height), 0)
+ draw = ImageDraw.Draw(img)
+ cx, cy = width // 2, height // 2
+ phase = frame * 0.35
+ if state == "captioning":
+ draw.rectangle((188, 38, 212, 75), outline=1)
+ draw.arc((175, 55, 225, 105), 20, 160, fill=1)
+ for x in range(42, width - 42, 5):
+ y = cy + int(32 * math.sin(x * 0.045 + phase * 2))
+ y2 = cy + int(14 * math.sin(x * 0.085 - phase * 3))
+ draw.line((x, y, min(width - 1, x + 4), y2), fill=1)
+ elif state in {"waiting", "speaking"}:
+ for radius in (34, 64, 94):
+ pulse = radius + int(8 * math.sin(phase + radius))
+ draw.ellipse((cx - pulse, cy - pulse, cx + pulse, cy + pulse), outline=1)
+ for i in range(7):
+ x = 93 + i * 34
+ bar = int(18 + 42 * (0.5 + 0.5 * math.sin(phase * 2 + i)))
+ draw.rectangle((x, 228 - bar, x + 17, 228), fill=1)
+ elif state == "error":
+ draw.line((150, 88, 250, 188), fill=1)
+ draw.line((250, 88, 150, 188), fill=1)
+ draw.rectangle((145, 82, 255, 194), outline=1)
+ else:
+ r = 42 + int(16 * math.sin(phase))
+ draw.ellipse((cx - r, cy - r, cx + r, cy + r), outline=1)
+ draw.ellipse((cx - 8, cy - 8, cx + 8, cy + 8), fill=1)
+ draw.text((max(8, (width - len(label) * 6) // 2), 24), label, fill=1)
+ if snippet:
+ draw.text((max(8, (width - len(snippet) * 6) // 2), 260), snippet, fill=1)
+ rendered.append(to_rlcd(img))
+ return rendered
+
+ async def _send_stream_animation(
+ self,
+ screen_url: Optional[str],
+ state: str,
+ *,
+ text: str = "",
+ frames: Optional[int] = None,
+ ) -> bool:
+ """Send a server-rendered voice animation over the screen streaming socket."""
+ target = self._screen_stream_target(screen_url)
+ if not target:
+ return False
+ frame_count = frames if frames is not None else (12 if state == "ack" else 18)
+ try:
+ rendered = await asyncio.to_thread(self._voice_animation_udp_frames, state, text, frame_count)
+ sock = _socket.socket(_socket.AF_INET, _socket.SOCK_DGRAM)
+ try:
+ for frame_id, frame in enumerate(rendered):
+ fid = frame_id % 256
+ for chunk_idx in range(15):
+ start = chunk_idx * 1000
+ sock.sendto(bytes([fid, chunk_idx]) + frame[start : start + 1000], target)
+ await asyncio.sleep(0.045)
+ finally:
+ sock.close()
+ return True
+ except Exception as exc:
+ logger.warning("[api_server] screen stream animation failed: %s", exc)
+ return False
+
+ async def _send_screen_animation(
+ self,
+ screen_url: Optional[str],
+ state: str,
+ *,
+ text: str = "",
+ frames: Optional[int] = None,
+ ) -> bool:
+ """Upload and run the ESP32 voice-state animation script on the initiating screen."""
+ if not screen_url:
+ return False
+ if await self._send_stream_animation(screen_url, state, text=text, frames=frames):
+ return True
+ if screen_url not in self._esp32_voice_animation_uploaded:
+ uploaded = await self._call_screen_mcp(
+ screen_url,
+ "write_file",
+ {"path": ESP32_VOICE_ANIMATION_PATH, "content": ESP32_VOICE_ANIMATION_SCRIPT},
+ )
+ if not uploaded:
+ return False
+ self._esp32_voice_animation_uploaded.add(screen_url)
+ code = (
+ f"exec(open({ESP32_VOICE_ANIMATION_PATH!r}).read()); "
+ f"run_state({str(state)!r}, text={str(text or '')[:240]!r}, frames={frames!r})"
+ )
+ return await self._call_screen_mcp(screen_url, "execute_python", {"code": code})
+
+ async def _send_screen_text(self, screen_url: Optional[str], text: str, *, clear: bool = False) -> None:
+ """Best-effort text update to the initiating MCP screen."""
+ if not screen_url:
+ return
+ if clear:
+ await self._call_screen_mcp(screen_url, "clear_screen", {"color": 0})
+ # Most Reyna screens accept x/y/text. Extra wrapping is avoided here so
+ # firmware-specific clients can keep their own font/layout behavior.
+ await self._call_screen_mcp(screen_url, "draw_text", {"x": 8, "y": 8, "text": text[:900]})
+
+ @staticmethod
+ def _extract_first_image_reference(text: str) -> Optional[str]:
+ """Return the first local path, MEDIA path, data URL, or HTTP image URL in text."""
+ if not text:
+ return None
+ patterns = [
+ r"MEDIA:([^\s)]+)",
+ r"!\[[^\]]*\]\(([^)]+)\)",
+ r"(?P<url>https?://[^\s)]+\.(?:png|jpe?g|webp|gif)(?:\?[^\s)]*)?)",
+ r"(?P<path>/[^\s)]+\.(?:png|jpe?g|webp|gif|bmp))",
+ r"(?P<data>data:image/[^\s)]+)",
+ ]
+ for pattern in patterns:
+ for match in re.finditer(pattern, text, flags=re.IGNORECASE):
+ ref = match.groupdict().get("url") or match.groupdict().get("path") or match.groupdict().get("data") or match.group(1)
+ if ref:
+ return ref.strip().strip('"\'')
+ return None
+
+ async def _send_screen_image_response(self, screen_url: Optional[str], text: str) -> bool:
+ """Best-effort render of the first generated/linked image onto the ESP32 screen."""
+ if not screen_url:
+ return False
+ image_ref = self._extract_first_image_reference(text)
+ if not image_ref:
+ return False
+ try:
+ import base64
+ import io
+ from PIL import Image, ImageOps
+
+ raw: bytes
+ lowered = image_ref.lower()
+ if lowered.startswith("data:image/"):
+ _, _, encoded = image_ref.partition(",")
+ raw = base64.b64decode(encoded)
+ elif self._looks_like_http_url(image_ref):
+ from aiohttp import ClientSession, ClientTimeout
+ async with ClientSession(timeout=ClientTimeout(total=20)) as session:
+ async with session.get(image_ref) as resp:
+ if resp.status >= 400:
+ logger.warning("[api_server] screen image fetch failed: HTTP %s", resp.status)
+ return False
+ raw = await resp.read()
+ else:
+ path = Path(image_ref).expanduser()
+ if not path.exists() or not path.is_file():
+ return False
+ raw = await asyncio.to_thread(path.read_bytes)
+ if not raw:
+ return False
+
+ def render_pbm() -> str:
+ with Image.open(io.BytesIO(raw)) as img:
+ img = ImageOps.exif_transpose(img).convert("RGB")
+ canvas = Image.new("RGB", (320, 240), "white")
+ img.thumbnail((320, 240), Image.Resampling.LANCZOS)
+ x = (320 - img.width) // 2
+ y = (240 - img.height) // 2
+ canvas.paste(img, (x, y))
+ mono = canvas.convert("1", dither=Image.Dither.FLOYDSTEINBERG)
+ out = io.BytesIO()
+ mono.save(out, format="PPM") # mode=1 writes raw PBM (P4)
+ return base64.b64encode(out.getvalue()).decode("ascii")
+
+ pbm_base64 = await asyncio.to_thread(render_pbm)
+ return await self._call_screen_mcp(
+ screen_url,
+ "draw_image",
+ {"pbm_base64": pbm_base64, "x": 0, "y": 0, "dither": False},
+ timeout=15,
+ )
+ except Exception:
+ logger.exception("[api_server] failed to render ESP32 screen image response")
+ return False
+
+ async def _send_screen_audio_response(
+ self,
+ screen_url: Optional[str],
+ text: str,
+ *,
+ audio_dir: Path,
+ device_id: str,
+ volume: int = 80,
+ ) -> bool:
+ """Best-effort spoken follow-up for ACK-mode ESP32 voice turns."""
+ if not screen_url or not text.strip():
+ return False
+ try:
+ import base64
+ from tools.tts_tool import text_to_speech_tool
+
+ # Keep board playback short and payloads small. The ESP32 HTTP MCP
+ # endpoint resets on large base64 WAV bodies; MP3 is much smaller
+ # and the firmware exposes play_mp3_base64 for exactly this path.
+ spoken_text = self._short_voice_summary(text, 420)
+ tts_raw = await asyncio.to_thread(text_to_speech_tool, spoken_text)
+ tts_result = json.loads(tts_raw)
+ src = str(tts_result.get("file_path") or "")
+ if not tts_result.get("success") or not src or not Path(src).exists():
+ logger.warning(
+ "[api_server] screen audio TTS failed: %s",
+ tts_result.get("error") or "missing TTS output",
+ )
+ return False
+
+ src_path = Path(src)
+ if src_path.suffix.lower() == ".mp3":
+ mp3_base64 = base64.b64encode(src_path.read_bytes()).decode("ascii")
+ if await self._call_screen_mcp(
+ screen_url,
+ "play_mp3_base64",
+ {"mp3_base64": mp3_base64, "volume": volume},
+ timeout=75,
+ ):
+ return True
+
+ wav_path = await asyncio.to_thread(
+ self._convert_audio_to_esp32_wav,
+ src,
+ output_dir=audio_dir,
+ device_id=f"{device_id}_response",
+ )
+ wav_bytes = wav_path.read_bytes()
+ if len(wav_bytes) > 220_000:
+ logger.warning(
+ "[api_server] screen audio WAV too large for MCP payload: %s bytes",
+ len(wav_bytes),
+ )
+ return False
+ wav_base64 = base64.b64encode(wav_bytes).decode("ascii")
+ return await self._call_screen_mcp(
+ screen_url,
+ "play_audio_base64",
+ {"wav_base64": wav_base64, "volume": volume},
+ timeout=45,
+ )
+ except Exception:
+ logger.exception("[api_server] failed to play ESP32 screen audio response")
+ return False
+
+ @staticmethod
+ def _short_voice_summary(text: str, limit: int = 220) -> str:
+ """Local fallback summary for voice/screen snippets without another LLM call."""
+ clean = " ".join(str(text or "").split())
+ if len(clean) <= limit:
+ return clean
+ return clean[: max(0, limit - 1)].rstrip() + "…"
+
+ async def _esp32_ack_wav(self, *, audio_dir: Path, device_id: str) -> Path:
+ """Return a reusable quiet ACK WAV; the visual screen animation is the acknowledgement."""
+ ack_path = audio_dir / "quick_ack.wav"
+ if ack_path.exists():
+ return ack_path
+
+ # ACK-mode clients expect a valid audio/wav response immediately, but
+ # Reyna ESP32 screens now show the acknowledgement visually. Keep the
+ # HTTP contract while avoiding the old spoken "Got it" message.
+ import wave
+ sample_rate = 16000
+ duration_s = 0.08
+ frames = b"\x00\x00" * int(sample_rate * duration_s)
+ with wave.open(str(ack_path), "wb") as wav:
+ wav.setnchannels(1)
+ wav.setsampwidth(2)
+ wav.setframerate(sample_rate)
+ wav.writeframes(frames)
+ return ack_path
+
+ def _append_esp32_voice_audit(self, audio_dir: Path, event: Dict[str, Any]) -> None:
+ """Append a best-effort ESP32 voice request audit event."""
+ try:
+ log_path = audio_dir / "voice_requests.jsonl"
+ with log_path.open("a", encoding="utf-8") as handle:
+ handle.write(json.dumps(event, default=str) + "\n")
+ except Exception:
+ logger.debug("[api_server] failed to append ESP32 voice audit log", exc_info=True)
+
+ async def _run_esp32_voice_audio_background(
+ self,
+ *,
+ audio_path: Path,
+ audit_event: Dict[str, Any],
+ device_id: str,
+ instructions: Optional[str],
+ audio_dir: Path,
+ screen_url: Optional[str],
+ ) -> None:
+ """Transcribe uploaded ESP32 audio and continue the voice turn after an immediate ACK."""
+ try:
+ await self._send_screen_animation(screen_url, "captioning")
+ from tools.transcription_tools import transcribe_audio
+ stt_result = await asyncio.to_thread(transcribe_audio, str(audio_path))
+ if not stt_result.get("success"):
+ failed_event = dict(audit_event)
+ failed_event.update({
+ "status": "transcription_failed",
+ "error": stt_result.get("error") or "Transcription failed",
+ "timestamp": time.time(),
+ })
+ self._append_esp32_voice_audit(audio_dir, failed_event)
+ await self._send_screen_text(
+ screen_url,
+ "I heard audio, but transcription failed. Try again.",
+ clear=True,
+ )
+ return
+ transcript = str(stt_result.get("transcript") or "").strip()
+ if not transcript:
+ empty_event = dict(audit_event)
+ empty_event.update({"status": "empty_transcript", "transcript": "", "timestamp": time.time()})
+ self._append_esp32_voice_audit(audio_dir, empty_event)
+ await self._send_screen_text(
+ screen_url,
+ "I heard audio, but got no words. Try again.",
+ clear=True,
+ )
+ return
+ transcript_event = dict(audit_event)
+ transcript_event.update({"status": "transcribed", "transcript": transcript, "timestamp": time.time()})
+ self._append_esp32_voice_audit(audio_dir, transcript_event)
+ await self._run_esp32_voice_background(
+ transcript=transcript,
+ device_id=device_id,
+ instructions=instructions,
+ audio_dir=audio_dir,
+ screen_url=screen_url,
+ )
+ except Exception as exc:
+ logger.exception("[api_server] ESP32 ACK background voice processing failed")
+ failed_event = dict(audit_event)
+ failed_event.update({"status": "background_exception", "error": str(exc), "timestamp": time.time()})
+ self._append_esp32_voice_audit(audio_dir, failed_event)
+ await self._send_screen_text(
+ screen_url,
+ "I got the audio, but background processing failed.",
+ clear=True,
+ )
+
+ async def _run_esp32_voice_background(
+ self,
+ *,
+ transcript: str,
+ device_id: str,
+ instructions: Optional[str],
+ audio_dir: Path,
+ screen_url: Optional[str],
+ ) -> None:
+ """Run the slow ESP32 voice turn after the device already heard an ACK."""
+ try:
+ animation_ok = await self._send_screen_animation(screen_url, "waiting", text=transcript)
+ if not animation_ok:
+ await self._send_screen_text(
+ screen_url,
+ f"Heard: {self._short_voice_summary(transcript, 180)}\n\nWorking…",
+ clear=True,
+ )
+ screen_hint = (
+ "\n\nThis request came from an ESP32 voice/screen device. "
+ "Keep the spoken follow-up short. The gateway will display the full answer on the initiating screen when possible."
+ )
+ merged_instructions = (instructions or "") + screen_hint
+ response_data, final_text = await self._run_esp32_voice_turn(
+ transcript=transcript,
+ device_id=device_id,
+ instructions=merged_instructions,
+ )
+ if not final_text.strip():
+ final_text = self._extract_response_text(response_data) or "I finished, but I do not have a text answer."
+ display_text = final_text
+ if len(display_text) > 1200:
+ display_text = self._short_voice_summary(display_text, 900) + "\n\nFull answer is long; WhatsApp delivery is the next wiring step."
+ await self._send_screen_text(screen_url, display_text, clear=True)
+ await self._send_screen_image_response(screen_url, final_text)
+ await self._send_screen_audio_response(
+ screen_url,
+ final_text,
+ audio_dir=audio_dir,
+ device_id=device_id,
+ )
+ except Exception:
+ logger.exception("[api_server] ESP32 background voice turn failed")
+ await self._send_screen_text(screen_url, "Hermes hit an error while preparing the full answer.", clear=True)
+
+ @staticmethod
+ def _extract_response_text(response_data: Dict[str, Any]) -> str:
+ """Extract assistant text from an OpenAI Responses-shaped payload."""
+ chunks: List[str] = []
+ for item in response_data.get("output") or []:
+ if not isinstance(item, dict) or item.get("type") != "message":
+ continue
+ for part in item.get("content") or []:
+ if isinstance(part, dict) and part.get("type") in {"output_text", "text"}:
+ text = part.get("text")
+ if isinstance(text, str) and text:
+ chunks.append(text)
+ return "\n".join(chunks).strip()
+
+ async def _run_esp32_voice_turn(
+ self,
+ *,
+ transcript: str,
+ device_id: str,
+ instructions: Optional[str] = None,
+ ) -> tuple[Dict[str, Any], str]:
+ """Run one persistent Responses-style turn for an ESP32 voice device."""
+ conversation = f"esp32:{device_id}"
+ previous_response_id = self._response_store.get_conversation(conversation)
+ conversation_history: List[Dict[str, Any]] = []
+ stored_session_id = None
+ stored_instructions = None
+
+ if previous_response_id:
+ stored = self._response_store.get(previous_response_id)
+ if stored is not None:
+ conversation_history = list(stored.get("conversation_history", []))
+ stored_session_id = stored.get("session_id")
+ stored_instructions = stored.get("instructions")
+
+ if instructions is None:
+ instructions = stored_instructions
+
+ session_id = stored_session_id or str(uuid.uuid4())
+ user_message = (
+ f'[Voice input from ESP32 device "{device_id}"]\n'
+ f'{transcript}'
+ )
+
+ result, usage = await self._run_agent(
+ user_message=user_message,
+ conversation_history=conversation_history,
+ ephemeral_system_prompt=instructions,
+ session_id=session_id,
+ gateway_session_key=conversation,
+ )
+
+ final_response = result.get("final_response", "") or result.get("error", "") or "(No response generated)"
+ response_id = f"resp_{uuid.uuid4().hex[:28]}"
+ created_at = int(time.time())
+ full_history = self._build_response_conversation_history(
+ conversation_history,
+ user_message,
+ result,
+ final_response,
+ )
+ output_start_index = self._response_messages_turn_start_index(
+ conversation_history,
+ user_message,
+ result,
+ )
+ output_items = self._extract_output_items(result, start_index=output_start_index)
+ response_data = {
+ "id": response_id,
+ "object": "response",
+ "status": "completed",
+ "created_at": created_at,
+ "model": self._model_name,
+ "output": output_items,
+ "usage": {
+ "input_tokens": usage.get("input_tokens", 0),
+ "output_tokens": usage.get("output_tokens", 0),
+ "total_tokens": usage.get("total_tokens", 0),
+ },
+ }
+ self._response_store.put(response_id, {
+ "response": response_data,
+ "conversation_history": full_history,
+ "instructions": instructions,
+ "session_id": session_id,
+ })
+ self._response_store.set_conversation(conversation, response_id)
+ return response_data, final_response
+
+ @staticmethod
+ def _esp32_audio_cache_dir() -> Path:
+ """Return the cache directory for ESP32 voice artifacts."""
+ try:
+ from hermes_constants import get_hermes_dir
+ return get_hermes_dir("cache/audio", "audio_cache") / "esp32"
+ except Exception:
+ return Path.home() / ".hermes" / "cache" / "audio" / "esp32"
+
+ @staticmethod
+ def _write_pcm_s16le_wav(raw_path: Path, wav_path: Path, *, sample_rate: int, channels: int = 1) -> None:
+ """Wrap a raw PCM s16le file in a minimal RIFF/WAVE container."""
+ import wave
+
+ with wave.open(str(wav_path), "wb") as wav:
+ wav.setnchannels(channels)
+ wav.setsampwidth(2)
+ wav.setframerate(sample_rate)
+ with raw_path.open("rb") as raw_handle:
+ while True:
+ chunk = raw_handle.read(64 * 1024)
+ if not chunk:
+ break
+ wav.writeframes(chunk)
+
+ # --- Live draft helpers (from your whisper-translation/apple_speech pipe protocol) ---
+ def _load_macmini_mcp_config(self) -> dict:
+ try:
+ if ESP32_MACMINI_MCP_URL and ESP32_MACMINI_MCP_TOKEN:
+ return {"url": ESP32_MACMINI_MCP_URL, "token": ESP32_MACMINI_MCP_TOKEN}
+ cfg_path = Path.home() / ".hermes" / "config.yaml"
+ if cfg_path.exists():
+ import yaml
+ cfg = yaml.safe_load(cfg_path.read_text()) or {}
+ m = (cfg.get("mcp_servers") or {}).get("macmini") or {}
+ url = m.get("url") or ""
+ token = ""
+ headers = m.get("headers") or {}
+ auth = headers.get("Authorization") or ""
+ if auth.startswith("Bearer "):
+ token = auth[7:]
+ if url:
+ return {"url": url, "token": token}
+ except Exception:
+ pass
+ return {}
+
+ @staticmethod
+ def _s16le_to_wav_bytes(pcm_bytes: bytes, sample_rate: int = 16000, channels: int = 1) -> bytes:
+ import io, wave
+ bio = io.BytesIO()
+ with wave.open(bio, "wb") as wf:
+ wf.setnchannels(channels)
+ wf.setsampwidth(2)
+ wf.setframerate(sample_rate)
+ wf.writeframes(pcm_bytes)
+ return bio.getvalue()
+
+ async def _transcribe_draft_via_macmini(self, wav_bytes: bytes, locale: str = "en-US") -> str | None:
+ if not ESP32_LIVE_DRAFT_USE_MACMINI:
+ return None
+ cfg = self._load_macmini_mcp_config()
+ url = cfg.get("url")
+ token = cfg.get("token")
+ if not url:
+ return None
+ try:
+ import base64
+ b64 = base64.b64encode(wav_bytes).decode("ascii")
+ payload = {
+ "jsonrpc": "2.0",
+ "id": 1,
+ "method": "tools/call",
+ "params": {"name": "speech_live_transcribe", "arguments": {"audioBase64": b64, "locale": locale}},
+ }
+ headers = {"Content-Type": "application/json", "Accept": "application/json, text/event-stream"}
+ if token:
+ headers["Authorization"] = f"Bearer {token}"
+ import aiohttp
+ async with aiohttp.ClientSession() as session:
+ async with session.post(url, json=payload, headers=headers, timeout=aiohttp.ClientTimeout(total=8)) as resp:
+ if resp.status != 200:
+ return None
+ txt = await resp.text()
+ for line in txt.splitlines():
+ if not line.startswith("data: "):
+ continue
+ try:
+ data = json.loads(line[6:])
+ content = (data.get("result", {}).get("content") or [{}])[0].get("text", "") or ""
+ if not content:
+ continue
+ inner = json.loads(content)
+ t = inner.get("text") or ""
+ if t:
+ return t.strip()
+ except Exception:
+ continue
+ except Exception as e:
+ logger.debug(f"[api_server] macmini live draft failed: {e}")
+ return None
+
+ async def _transcribe_draft_local(self, wav_bytes: bytes) -> str | None:
+ try:
+ import tempfile
+ fd, tmp_path = tempfile.mkstemp(suffix=".wav")
+ os.close(fd)
+ Path(tmp_path).write_bytes(wav_bytes)
+ def _do():
+ try:
+ from faster_whisper import WhisperModel
+ model = None
+ try:
+ from tools.transcription_tools import _local_model
+ if _local_model is not None:
+ model = _local_model
+ except Exception:
+ pass
+ if model is None:
+ model = WhisperModel("tiny", device="cpu", compute_type="int8", num_workers=1)
+ segs, _ = model.transcribe(tmp_path, beam_size=1, best_of=1, language="en", condition_on_previous_text=False, vad_filter=True)
+ return " ".join(s.text.strip() for s in segs).strip()
+ except Exception as ex:
+ logger.debug(f"[api_server] local draft err: {ex}")
+ return None
+ finally:
+ try:
+ Path(tmp_path).unlink(missing_ok=True)
+ except Exception:
+ pass
+ return await asyncio.to_thread(_do)
+ except Exception:
+ return None
+
+ async def _handle_esp32_voice_ws(self, request: "web.Request") -> "web.WebSocketResponse":
+ """GET /api/esp32/voice/ws — ESP32 PCM streaming with LIVE draft captioning.
+ LIVE DRAFT MODE (from whisper-translation/engine_apple_transcribe.py):
+ - While ingesting binary s16le, every ~800ms snapshots current PCM -> in-mem WAV
+ -> tries Mac mini speech_live_transcribe (ApplePipeTranscriber --pipe volatile,
+ <200ms drafts) fallback to local tiny for fast feedback.
+ - Emits WS events: draft + interim_transcript for UI. Final via macmini or local base.
+ """
+ auth_err = self._check_auth(request)
+ if auth_err:
+ return auth_err # type: ignore[return-value]
+
+ ws = web.WebSocketResponse(heartbeat=None, max_msg_size=512 * 1024)
+ await ws.prepare(request)
+
+ async def _send_event(event: str, **payload):
+ if not ws.closed:
+ await ws.send_str(json.dumps({"event": event, **payload}))
+
+ device_id = (
+ request.headers.get("X-Device-ID")
+ or request.query.get("device_id")
+ or "default"
+ ).strip()[:80] or "default"
+ safe_device = re.sub(r"[^A-Za-z0-9_.-]+", "_", device_id)[:80] or "default"
+ audio_dir = self._esp32_audio_cache_dir()
+ audio_dir.mkdir(parents=True, exist_ok=True)
+ stamp = f"{safe_device}_{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
+ raw_path = audio_dir / f"{stamp}.s16le"
+ wav_path = audio_dir / f"{stamp}.wav"
+ sample_rate = 16000
+ channels = 1
+ sample_width = 2
+ total_audio_bytes = 0
+ started = False
+ stopped = False
+ first_audio_at: float | None = None
+ started_at = time.time()
+ # live draft state (directly from your whisper-translation ApplePipeTranscriber pattern)
+ live_enabled = ESP32_LIVE_DRAFT_ENABLED
+ pcm_buffer = bytearray()
+ last_draft_at = 0.0
+ last_draft_bytes = 0
+ last_draft_text = ""
+ draft_in_progress = False
+ draft_count = 0
+ locale_req = request.headers.get("X-Hermes-Locale") or request.query.get("locale") or "en-US"
+
+ audit_ctx = self._request_audit_context(request)
+ audit_event = {
+ "timestamp": started_at,
+ "device_id": device_id,
+ "source_ip": audit_ctx.get("remote") or audit_ctx.get("peer_ip") or "",
+ "peer_ip": audit_ctx.get("peer_ip") or "",
+ "forwarded_for": audit_ctx.get("forwarded_for") or "",
+ "real_ip": audit_ctx.get("real_ip") or "",
+ "user_agent": audit_ctx.get("user_agent") or "",
+ "transport": "websocket",
+ "raw_path": str(raw_path),
+ "audio_path": str(wav_path),
+ "status": "accepted",
+ "live_draft_enabled": live_enabled,
+ }
+ self._append_esp32_voice_audit(audio_dir, audit_event)
+ screen_url = self._resolve_esp32_screen_url(request, device_id)
+ visual_listening_sent = False
+
+ def _schedule_voice_state(state: str, *, text: str = "", frames: int | None = None) -> None:
+ if not screen_url:
+ return
+ task = asyncio.create_task(self._send_screen_animation(screen_url, state, text=text, frames=frames))
+ try:
+ self._background_tasks.add(task)
+ except TypeError:
+ pass
+ if hasattr(task, "add_done_callback"):
+ task.add_done_callback(self._background_tasks.discard)
+
+ async def _do_live_draft(snapshot_bytes: bytes):
+ nonlocal last_draft_text, draft_in_progress, draft_count
+ if draft_in_progress:
+ return
+ if len(snapshot_bytes) < int(sample_rate * 0.5 * 2): # at least 500ms of s16le at 16k mono
+ return
+ draft_in_progress = True
+ try:
+ wav_b = self._s16le_to_wav_bytes(snapshot_bytes, sample_rate=sample_rate, channels=channels)
+ txt = None
+ if ESP32_LIVE_DRAFT_USE_MACMINI:
+ txt = await self._transcribe_draft_via_macmini(wav_b, locale=locale_req)
+ if not txt:
+ txt = await self._transcribe_draft_local(wav_b)
+ if not txt:
+ return
+ txt = txt.strip()
+ if not txt or txt == last_draft_text or len(txt) < 2:
+ return
+ last_draft_text = txt
+ draft_count += 1
+ await _send_event("draft", text=txt, is_final=False, isFinal=False, chunk=draft_count, bytes=len(snapshot_bytes))
+ await _send_event("interim_transcript", text=txt, is_final=False, draft_count=draft_count)
+ if screen_url and len(txt) > 1:
+ try:
+ t = asyncio.create_task(self._send_screen_text(screen_url, txt, clear=False))
+ self._background_tasks.add(t)
+ t.add_done_callback(self._background_tasks.discard)
+ except Exception:
+ pass
+ finally:
+ draft_in_progress = False
+
+ await _send_event("ready", device_id=device_id, screen_url=screen_url or "", live_draft=live_enabled, live_model="apple-pipe-volatile")
+
+ try:
+ raw_handle = raw_path.open("wb")
+ try:
+ async for msg in ws:
+ if msg.type == web.WSMsgType.TEXT:
+ try:
+ payload = json.loads(msg.data)
+ except json.JSONDecodeError:
+ await _send_event("error", code="bad_json", message="Text frames must be JSON")
+ continue
+ event = str(payload.get("event") or payload.get("type") or "").strip().lower()
+ if event == "start":
+ if started:
+ await _send_event("error", code="already_started", message="Audio stream already started")
+ continue
+ fmt = str(payload.get("format") or "pcm_s16le").strip().lower()
+ channels = int(payload.get("channels") or 1)
+ sample_width = int(payload.get("sample_width") or 2)
+ sample_rate = int(payload.get("sample_rate") or 16000)
+ device_id = str(payload.get("device_id") or device_id).strip()[:80] or device_id
+ if fmt not in {"pcm_s16le", "s16le", "pcm"} or channels != 1 or sample_width != 2:
+ await _send_event("error", code="unsupported_audio_format", message="Use 16 kHz mono signed 16-bit little-endian PCM")
+ await ws.close()
+ return ws
+ if sample_rate < 8000 or sample_rate > 48000:
+ await _send_event("error", code="bad_sample_rate", message="sample_rate must be 8000-48000")
+ await ws.close()
+ return ws
+ started = True
+ if not visual_listening_sent:
+ _schedule_voice_state("captioning", frames=8)
+ visual_listening_sent = True
+ await _send_event("listening", sample_rate=sample_rate, channels=channels, sample_width=sample_width)
+ elif event in {"stop", "end"}:
+ stopped = True
+ break
+ elif event in {"ping", "keepalive"}:
+ await _send_event("pong")
+ else:
+ await _send_event("error", code="unknown_event", message=f"Unknown event: {event or '<empty>'}")
+ elif msg.type == web.WSMsgType.BINARY:
+ if not started:
+ started = True
+ if not visual_listening_sent:
+ _schedule_voice_state("captioning", frames=8)
+ visual_listening_sent = True
+ await _send_event("listening", sample_rate=sample_rate, channels=channels, sample_width=sample_width)
+ chunk = bytes(msg.data or b"")
+ if not chunk:
+ continue
+ if first_audio_at is None:
+ first_audio_at = time.time()
+ total_audio_bytes += len(chunk)
+ pcm_buffer.extend(chunk)
+ if total_audio_bytes > ESP32_WS_MAX_AUDIO_BYTES:
+ await _send_event("error", code="audio_too_large", message=f"Audio stream exceeded {ESP32_WS_MAX_AUDIO_BYTES} bytes")
+ await ws.close()
+ return ws
+ raw_handle.write(chunk)
+ # live draft trigger (every 800ms & 8KB new data) — direct captioning fluff stripped
+ if live_enabled:
+ now = time.time()
+ elapsed_ms = (now - last_draft_at) * 1000
+ new_bytes = len(pcm_buffer) - last_draft_bytes
+ total_ms = len(pcm_buffer) / 2 / sample_rate * 1000
+ if total_ms >= ESP32_LIVE_DRAFT_MIN_TOTAL_MS and elapsed_ms >= ESP32_LIVE_DRAFT_INTERVAL_MS and new_bytes >= ESP32_LIVE_DRAFT_MIN_BYTES:
+ last_draft_at = now
+ last_draft_bytes = len(pcm_buffer)
+ # copy snapshot, fire-and-forget draft (but guard concurrent)
+ snapshot = bytes(pcm_buffer)
+ asyncio.create_task(_do_live_draft(snapshot))
+ elif msg.type == web.WSMsgType.ERROR:
+ logger.warning("[api_server] ESP32 voice websocket error: %s", ws.exception())
+ break
+ finally:
+ raw_handle.close()
+
+ if not stopped:
+ await _send_event("error", code="missing_stop", message="Send a stop event after audio chunks")
+ await ws.close()
+ return ws
+ if total_audio_bytes < ESP32_WS_MIN_AUDIO_BYTES:
+ await _send_event("error", code="audio_too_short", message="Not enough audio received")
+ await ws.close()
+ return ws
+
+ await asyncio.to_thread(
+ self._write_pcm_s16le_wav,
+ raw_path,
+ wav_path,
+ sample_rate=sample_rate,
+ channels=channels,
+ )
+ audit_event.update({
+ "status": "received",
+ "audio_bytes": total_audio_bytes,
+ "sample_rate": sample_rate,
+ "first_audio_latency_ms": int(((first_audio_at or started_at) - started_at) * 1000),
+ "stop_latency_ms": int((time.time() - started_at) * 1000),
+ "draft_count": draft_count,
+ "last_draft": last_draft_text,
+ })
+ self._append_esp32_voice_audit(audio_dir, dict(audit_event))
+
+ _schedule_voice_state("captioning", frames=10)
+ await _send_event("thinking", stage="transcribing", draft_count=draft_count)
+
+ # Final transcription: prefer Mac mini live for accuracy (not just draft, full file via speech_transcribe_file)
+ transcript = None
+ if ESP32_LIVE_DRAFT_USE_MACMINI:
+ try:
+ cfg = self._load_macmini_mcp_config()
+ if cfg.get("url"):
+ # Try final via speech_transcribe_file (file mode, more accurate than draft)
+ # Need to copy wav to mac? Actually speech_transcribe_file expects file path on Mac
+ # So we keep local transcription as final for now, but if draft == good use it.
+ # Future: upload via SCP or use live transcribe for final with full buffer.
+ pass
+ except Exception:
+ pass
+
+ from tools.transcription_tools import transcribe_audio
+ stt_result = await asyncio.to_thread(transcribe_audio, str(wav_path))
+ if not stt_result.get("success"):
+ message = stt_result.get("error") or "Transcription failed"
+ self._append_esp32_voice_audit(audio_dir, {**audit_event, "status": "transcription_failed", "error": message, "timestamp": time.time()})
+ await _send_event("error", code="transcription_failed", message=message)
+ await ws.close()
+ return ws
+ final_txt = str(stt_result.get("transcript") or "").strip()
+ if not final_txt:
+ self._append_esp32_voice_audit(audio_dir, {**audit_event, "status": "empty_transcript", "timestamp": time.time()})
+ await _send_event("error", code="empty_transcript", message="Transcription was empty")
+ await ws.close()
+ return ws
+ transcript = final_txt
+ self._append_esp32_voice_audit(audio_dir, {**audit_event, "status": "transcribed", "transcript": transcript, "timestamp": time.time(), "draft_count": draft_count})
+ await _send_event("transcript", text=transcript, is_final=True, isFinal=True, draft_count=draft_count)
+ await _send_event("final", text=transcript, isFinal=True)
+
+ summary = self._short_voice_summary(transcript, 180)
+ ack_text = f"Heard: {summary}\n\nWorking on the full answer now."
+ await _send_event("response_text", text=ack_text, response_id="", final=False)
+ self._append_esp32_voice_audit(audio_dir, {**audit_event, "status": "queued_background", "transcript": transcript, "timestamp": time.time()})
+
+ try:
+ await _send_event("done", mode="background", screen_url=screen_url or "", draft_count=draft_count)
+ await ws.close()
+ except Exception:
+ pass
+ background_task = asyncio.create_task(
+ self._run_esp32_voice_background(
+ transcript=transcript,
+ device_id=device_id,
+ instructions=request.headers.get("X-Hermes-Instructions") or None,
+ audio_dir=audio_dir,
+ screen_url=screen_url,
+ )
+ )
+ try:
+ self._background_tasks.add(background_task)
+ except TypeError:
+ pass
+ if hasattr(background_task, "add_done_callback"):
+ background_task.add_done_callback(self._background_tasks.discard)
+ return ws
+ except Exception as exc:
+ logger.exception("[api_server] ESP32 voice websocket processing failed")
+ self._append_esp32_voice_audit(audio_dir, {**audit_event, "status": "websocket_exception", "error": str(exc), "timestamp": time.time()})
+ try:
+ await _send_event("error", code="internal_error", message=str(exc))
+ await ws.close()
+ except Exception:
+ pass
+ return ws
+ finally:
+ try:
+ raw_path.unlink(missing_ok=True)
+ except Exception:
+ pass
+
+ async def _handle_esp32_voice(self, request: "web.Request") -> "web.Response":
+
+ """POST /api/esp32/voice — receive audio, run Hermes, return TTS audio."""
+ auth_err = self._check_auth(request)
+ if auth_err:
+ return auth_err
+
+ device_id = (
+ request.headers.get("X-Device-ID")
+ or request.query.get("device_id")
+ or "default"
+ ).strip()[:80] or "default"
+
+ try:
+ audio_bytes = await request.read()
+ except Exception as exc:
+ return web.json_response(_openai_error(f"Failed to read audio body: {exc}"), status=400)
+
+ if not audio_bytes:
+ return web.json_response(_openai_error("Empty audio body", code="empty_audio"), status=400)
+ if len(audio_bytes) > ESP32_AUDIO_MAX_BYTES:
+ return web.json_response(
+ _openai_error(
+ f"Audio body is too large ({len(audio_bytes)} bytes); max is {ESP32_AUDIO_MAX_BYTES}",
+ code="audio_too_large",
+ ),
+ status=413,
+ )
+
+ suffix = self._audio_extension_for_content_type(request.headers.get("Content-Type", ""))
+ try:
+ from hermes_constants import get_hermes_dir
+ audio_dir = get_hermes_dir("cache/audio", "audio_cache") / "esp32"
+ except Exception:
+ audio_dir = Path.home() / ".hermes" / "cache" / "audio" / "esp32"
+ audio_dir.mkdir(parents=True, exist_ok=True)
+ safe_device = re.sub(r"[^A-Za-z0-9_.-]+", "_", device_id)[:80] or "default"
+ audio_path = audio_dir / f"{safe_device}_{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}{suffix}"
+ audio_path.write_bytes(audio_bytes)
+
+ audit_ctx = self._request_audit_context(request)
+ audit_event: Dict[str, Any] = {
+ "timestamp": time.time(),
+ "device_id": device_id,
+ "source_ip": audit_ctx.get("remote") or audit_ctx.get("peer_ip") or "",
+ "peer_ip": audit_ctx.get("peer_ip") or "",
+ "forwarded_for": audit_ctx.get("forwarded_for") or "",
+ "real_ip": audit_ctx.get("real_ip") or "",
+ "user_agent": audit_ctx.get("user_agent") or "",
+ "reply_mode": self._esp32_reply_mode(request),
+ "screen_device": request.headers.get("X-Hermes-Screen-Device") or request.query.get("screen_device") or "",
+ "screen_url": request.headers.get("X-Hermes-Screen-Url") or request.headers.get("X-Screen-MCP-Url") or request.query.get("screen_url") or "",
+ "content_type": request.headers.get("Content-Type", ""),
+ "audio_bytes": len(audio_bytes),
+ "audio_path": str(audio_path),
+ "status": "received",
+ }
+
+ def _append_audit(event: Dict[str, Any]) -> None:
+ try:
+ log_path = audio_dir / "voice_requests.jsonl"
+ with log_path.open("a", encoding="utf-8") as handle:
+ handle.write(json.dumps(event, default=str) + "\n")
+ except Exception:
+ logger.debug("[api_server] failed to append ESP32 voice audit log", exc_info=True)
+
+ _append_audit(audit_event)
+
+ if self._esp32_reply_mode(request) == "ack":
+ screen_url = self._resolve_esp32_screen_url(request, device_id)
+
+ async def _ack_background() -> None:
+ try:
+ await self._send_screen_animation(screen_url, "ack")
+ await self._send_screen_animation(screen_url, "captioning")
+ from tools.transcription_tools import transcribe_audio
+ stt_result_bg = await asyncio.wait_for(
+ asyncio.to_thread(transcribe_audio, str(audio_path)),
+ timeout=45,
+ )
+ if not stt_result_bg.get("success"):
+ failed_event = dict(audit_event)
+ failed_event.update({
+ "status": "transcription_failed",
+ "error": stt_result_bg.get("error") or "Transcription failed",
+ "timestamp": time.time(),
+ })
+ _append_audit(failed_event)
+ await self._send_screen_text(
+ screen_url,
+ "I heard audio, but transcription failed. Try again.",
+ clear=True,
+ )
+ return
+ transcript_bg = str(stt_result_bg.get("transcript") or "").strip()
+ if not transcript_bg:
+ empty_event = dict(audit_event)
+ empty_event.update({"status": "empty_transcript", "transcript": "", "timestamp": time.time()})
+ _append_audit(empty_event)
+ await self._send_screen_text(
+ screen_url,
+ "I heard audio, but got no words. Try again.",
+ clear=True,
+ )
+ return
+ transcript_event = dict(audit_event)
+ transcript_event.update({"status": "transcribed", "transcript": transcript_bg, "timestamp": time.time()})
+ _append_audit(transcript_event)
+ await self._run_esp32_voice_background(
+ transcript=transcript_bg,
+ device_id=device_id,
+ instructions=request.headers.get("X-Hermes-Instructions") or None,
+ audio_dir=audio_dir,
+ screen_url=screen_url,
+ )
+ except asyncio.TimeoutError:
+ timeout_event = dict(audit_event)
+ timeout_event.update({"status": "transcription_timeout", "error": "Transcription exceeded 45s", "timestamp": time.time()})
+ _append_audit(timeout_event)
+ await self._send_screen_text(
+ screen_url,
+ "I heard audio, but captioning timed out. Try again.",
+ clear=True,
+ )
+ except Exception as exc:
+ logger.exception("[api_server] ESP32 ack-mode background processing failed")
+ failed_event = dict(audit_event)
+ failed_event.update({"status": "background_exception", "error": str(exc), "timestamp": time.time()})
+ _append_audit(failed_event)
+
+ task = asyncio.create_task(_ack_background())
+ try:
+ self._background_tasks.add(task)
+ except TypeError:
+ pass
+ if hasattr(task, "add_done_callback"):
+ task.add_done_callback(self._background_tasks.discard)
+ ack_path = await self._esp32_ack_wav(audio_dir=audio_dir, device_id=device_id)
+ headers = {
+ "X-Hermes-Conversation": f"esp32:{device_id}",
+ "X-Hermes-Reply-Mode": "ack",
+ "X-Hermes-Screen-Url": screen_url or "",
+ "Content-Type": "audio/wav",
+ }
+ return web.FileResponse(ack_path, headers=headers)
+
+ try:
+ from tools.transcription_tools import transcribe_audio
+ stt_result = await asyncio.to_thread(transcribe_audio, str(audio_path))
+ except Exception as exc:
+ logger.exception("[api_server] ESP32 transcription failed")
+ failed_event = dict(audit_event)
+ failed_event.update({"status": "transcription_exception", "error": str(exc), "timestamp": time.time()})
+ _append_audit(failed_event)
+ return web.json_response(_openai_error(f"Transcription failed: {exc}"), status=500)
+
+ if not stt_result.get("success"):
+ failed_event = dict(audit_event)
+ failed_event.update({
+ "status": "transcription_failed",
+ "error": stt_result.get("error") or "Transcription failed",
+ "timestamp": time.time(),
+ })
+ _append_audit(failed_event)
+ return web.json_response(
+ _openai_error(stt_result.get("error") or "Transcription failed", code="transcription_failed"),
+ status=502,
+ )
+
+ transcript = str(stt_result.get("transcript") or "").strip()
+ if not transcript:
+ empty_event = dict(audit_event)
+ empty_event.update({"status": "empty_transcript", "transcript": "", "timestamp": time.time()})
+ _append_audit(empty_event)
+ return web.json_response(_openai_error("Transcription was empty", code="empty_transcript"), status=422)
+
+ transcript_event = dict(audit_event)
+ transcript_event.update({"status": "transcribed", "transcript": transcript, "timestamp": time.time()})
+ _append_audit(transcript_event)
+
+ instructions = request.headers.get("X-Hermes-Instructions") or None
+ try:
+ response_data, final_text = await self._run_esp32_voice_turn(
+ transcript=transcript,
+ device_id=device_id,
+ instructions=instructions,
+ )
+ except Exception as exc:
+ logger.exception("[api_server] ESP32 agent turn failed")
+ return web.json_response(_openai_error(f"Agent turn failed: {exc}"), status=500)
+
+ if not final_text.strip():
+ final_text = self._extract_response_text(response_data) or "I heard you, but I do not have a spoken answer."
+
+ try:
+ from tools.tts_tool import text_to_speech_tool
+ tts_raw = await asyncio.to_thread(text_to_speech_tool, final_text)
+ tts_result = json.loads(tts_raw)
+ except Exception as exc:
+ logger.exception("[api_server] ESP32 TTS failed")
+ return web.json_response(
+ _openai_error(f"TTS failed after response was generated: {exc}", code="tts_failed"),
+ status=500,
+ headers={
+ "X-Hermes-Transcript": self._safe_audio_header(transcript, 1000),
+ "X-Hermes-Text-Response": self._safe_audio_header(final_text, 2000),
+ },
+ )
+
+ if not tts_result.get("success"):
+ return web.json_response(
+ _openai_error(tts_result.get("error") or "TTS failed", code="tts_failed"),
+ status=502,
+ headers={
+ "X-Hermes-Transcript": self._safe_audio_header(transcript, 1000),
+ "X-Hermes-Text-Response": self._safe_audio_header(final_text, 2000),
+ },
+ )
+
+ tts_path = str(tts_result.get("file_path") or "")
+ if not tts_path or not Path(tts_path).exists():
+ return web.json_response(
+ _openai_error("TTS succeeded but no audio file was produced", code="tts_file_missing"),
+ status=500,
+ )
+
+ try:
+ response_audio_path = self._convert_audio_to_esp32_wav(
+ tts_path,
+ output_dir=audio_dir,
+ device_id=device_id,
+ )
+ except Exception as exc:
+ logger.exception("[api_server] ESP32 WAV normalization failed")
+ return web.json_response(
+ _openai_error(f"Could not convert TTS audio to ESP32 WAV: {exc}", code="wav_conversion_failed"),
+ status=500,
+ headers={
+ "X-Hermes-Transcript": self._safe_audio_header(transcript, 1000),
+ "X-Hermes-Text-Response": self._safe_audio_header(final_text, 2000),
+ },
+ )
+
+ headers = {
+ "X-Hermes-Transcript": self._safe_audio_header(transcript, 1000),
+ "X-Hermes-Text-Response": self._safe_audio_header(final_text, 2000),
+ "X-Hermes-Response-Id": str(response_data.get("id") or ""),
+ "X-Hermes-Conversation": f"esp32:{device_id}",
+ "X-Hermes-TTS-Source-Type": self._audio_response_mime(tts_path),
+ "Content-Type": "audio/wav",
+ }
+ return web.FileResponse(
+ response_audio_path,
+ headers=headers,
+ )
+
+ # ------------------------------------------------------------------
+ # Cron jobs API
+ # ------------------------------------------------------------------
+
+ _JOB_ID_RE = __import__("re").compile(r"[a-f0-9]{12}")
+ # Allowed fields for update — prevents clients injecting arbitrary keys
+ _UPDATE_ALLOWED_FIELDS = {"name", "schedule", "prompt", "deliver", "skills", "skill", "repeat", "enabled"}
+ _MAX_NAME_LENGTH = 200
+ _MAX_PROMPT_LENGTH = 5000
+
+ @staticmethod
+
+
async def _handle_models(self, request: "web.Request") -> "web.Response":
"""GET /v1/models — list hermes-agent and any configured model_routes aliases."""
auth_err = self._check_auth(request)
@@ -5018,6 +6548,9 @@ class APIServerAdapter(BasePlatformAdapter):
# bootstrap shims use this key as a feature-detection hook; registering
# native routes first lets those shims no-op instead of shadowing the
# upstream session-control handlers.
+ # Local ESP32 push-to-talk endpoints: POST complete audio, or stream PCM over WebSocket.
+ self._app.router.add_post("/api/esp32/voice", self._handle_esp32_voice)
+ self._app.router.add_get("/api/esp32/voice/ws", self._handle_esp32_voice_ws)
self._app["api_server_adapter"] = self
# Start background sweep to clean up orphaned (unconsumed) run streams
Reference in New Issue View Git Blame Copy Permalink