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/hermes_plugin/__init__.py
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

305 lines
12 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
Reyna family ESP32 Voice Gateway - durable Hermes plugin
Survives `hermes update` which wipes in-tree patches to gateway/platforms/api_server.py.
What it does:
- On Hermes import (plugin load), checks if api_server.py has ESP32 routes.
- If missing (after update), auto-restores from bundled patch + logs actionable warning.
- Exposes tools for manual verify/restore from inside Hermes chat (via slash or tool).
The actual voice logic stays in api_server.py (patched file) because it's 1600+ lines
deeply integrated with APIServerAdapter internals (ResponseStore, _run_agent, TTS, transcription,
screen MCP). Extracting as runtime monkeypatch would be brittle. The durable strategy is:
1. Patch file lives in this repo + bundled in plugin as `patch/*.patch`
2. Plugin's register() does self-heal check (fast ~10ms)
3. Gateway auto-restarts handled by operator (can't restart self from inside gateway - see skill pitfalls)
Future: true runtime injection plugin that patches APIServerAdapter.connect to add routes
without touching file — feasible but needs tracking _handle_esp32_* methods. Kept separate as v2.
"""
import logging
import subprocess
import sys
from pathlib import Path
from typing import Optional
logger = logging.getLogger(__name__)
PLUGIN_DIR = Path(__file__).parent
PATCH_DIR = PLUGIN_DIR / "patch"
# repo root for dev checkout (symlinked plugin)
REPO_PATCH = PLUGIN_DIR.parent / "patch" / "api_server_esp32_live_draft_with_llm.patch"
def _find_hermes_agent_dir() -> Optional[Path]:
# Try standard locations
candidates = [
Path.home() / ".hermes" / "hermes-agent",
Path(__file__).resolve().parents[3], # if running from within hermes-agent/plugins/esp32-...
]
# Also check HERMES_AGENT_DIR env
import os
env_dir = os.getenv("HERMES_AGENT_DIR")
if env_dir:
candidates.insert(0, Path(env_dir))
# pip install location
try:
import hermes_cli
candidates.append(Path(hermes_cli.__file__).parents[1])
except Exception:
pass
for c in candidates:
if (c / "gateway" / "platforms" / "api_server.py").exists():
return c
return None
def _api_server_has_esp32() -> bool:
agent_dir = _find_hermes_agent_dir()
if not agent_dir:
return False
api_file = agent_dir / "gateway" / "platforms" / "api_server.py"
try:
txt = api_file.read_text()[:5000] # top matter + first routes
# quick check
if "ESP32_AUDIO_MAX_BYTES" not in api_file.read_text()[:20000]:
return False
# full route registration present?
content = api_file.read_text()
return "/api/esp32/voice" in content and "_handle_esp32_voice_ws" in content
except Exception:
return False
def _find_patch_file() -> Optional[Path]:
# Search order: bundled patch in plugin, repo patch, hermes-home fallback
for p in [
PATCH_DIR / "api_server_esp32_live_draft_with_llm.patch",
REPO_PATCH,
Path.home() / ".hermes" / "esp32_live_draft_with_llm_patch.patch",
Path.home() / ".hermes" / "esp32_voice_patch.patch",
PATCH_DIR / "esp32_live_draft_with_llm_patch.patch",
]:
if p.exists():
return p
# Also search inside plugin dir itself (dev checkout)
for pattern in ["*.patch"]:
for f in PATCH_DIR.glob(pattern):
if f.stat().st_size > 1000:
return f
return None
def _try_apply_patch(patch_path: Path, agent_dir: Path) -> bool:
"""Apply patch via git apply. Returns True if applied or already applied."""
api_file = agent_dir / "gateway" / "platforms" / "api_server.py"
if not api_file.exists():
logger.warning("[esp32-voice-gateway] api_server.py not found at %s", api_file)
return False
# Check if git
git_dir = agent_dir / ".git"
if not git_dir.exists():
logger.warning("[esp32-voice-gateway] not a git repo: %s", agent_dir)
return False
# Try git apply --check first
try:
# if already patched, check will fail with "already applied" type – we detect via has_esp32 already
res = subprocess.run(
["git", "apply", "--check", str(patch_path)],
cwd=str(agent_dir),
capture_output=True,
text=True,
timeout=15,
)
if res.returncode != 0:
# Could be already applied or conflict. Log but try to see if content already has marker
if "ESP32_AUDIO_MAX_BYTES" not in api_file.read_text()[:20000]:
logger.warning("[esp32-voice-gateway] patch check failed: %s %s", res.stdout[:500], res.stderr[:800])
# Try --3way if possible
res2 = subprocess.run(
["git", "apply", "--3way", str(patch_path)],
cwd=str(agent_dir),
capture_output=True,
text=True,
timeout=30,
)
if res2.returncode != 0:
logger.error("[esp32-voice-gateway] git apply --3way failed: %s %s", res2.stdout[-500:], res2.stderr[-1000:])
return False
else:
# already has marker, consider applied
logger.info("[esp32-voice-gateway] patch already present (has ESP32 marker)")
return True
else:
res_apply = subprocess.run(
["git", "apply", str(patch_path)],
cwd=str(agent_dir),
capture_output=True,
text=True,
timeout=30,
)
if res_apply.returncode != 0:
logger.error("[esp32-voice-gateway] git apply failed: %s %s", res_apply.stdout[-500:], res_apply.stderr[-1000:])
return False
# Verify
if _api_server_has_esp32():
logger.info("[esp32-voice-gateway] Successfully restored ESP32 routes from %s -> %s", patch_path, api_file)
return True
else:
logger.warning("[esp32-voice-gateway] Patch applied but route marker still missing – need manual check")
return False
except Exception as e:
logger.exception("[esp32-voice-gateway] Exception applying patch %s: %s", patch_path, e)
return False
def restore_if_needed() -> dict:
"""Check and auto-restore if needed. Safe to call multiple times."""
result = {"had_routes": False, "restored": False, "patch": None, "agent_dir": None, "error": None}
try:
agent_dir = _find_hermes_agent_dir()
result["agent_dir"] = str(agent_dir) if agent_dir else None
if not agent_dir:
result["error"] = "hermes-agent dir not found"
return result
if _api_server_has_esp32():
result["had_routes"] = True
return result
# missing -> try restore
patch_file = _find_patch_file()
result["patch"] = str(patch_file) if patch_file else None
if not patch_file:
result["error"] = "no patch file found (checked patch/*.patch + ~/.hermes/esp32*.patch)"
logger.error("[esp32-voice-gateway] Routes missing after update and no patch file found!")
return result
logger.warning("[esp32-voice-gateway] ESP32 routes missing (likely after `hermes update`) – auto-restoring from %s", patch_file)
ok = _try_apply_patch(patch_file, agent_dir)
result["restored"] = ok
if ok:
# also copy patch to ~/.hermes for next fallback
try:
dest = Path.home() / ".hermes" / "esp32_live_draft_with_llm_patch.patch"
if patch_file.resolve() != dest.resolve():
import shutil
shutil.copy2(patch_file, dest)
except Exception:
pass
return result
except Exception as e:
result["error"] = str(e)
logger.exception("[esp32-voice-gateway] restore_if_needed failed")
return result
# --- Plugin registration ---
def register(ctx):
"""Hermes plugin entry point."""
# On import, do fast self-heal check (10ms if routes present)
info = restore_if_needed()
# Register a tool so user can ask inside Hermes: "check ESP32 voice gateway"
import json
def _tool_check_esp32_gateway(args, **kwargs) -> str:
res = restore_if_needed()
# also run live curl checks if gateway running
checks = {}
try:
import socket
# try to find api_server port from config
port = 8642
try:
from hermes_cli.config import load_config
cfg = load_config()
g_port = ((cfg.get("gateway") or {}).get("api_server") or {}).get("port")
p_port = ((cfg.get("platforms") or {}).get("api_server") or {}).get("port")
raw_port = g_port or p_port or port
port = int(raw_port)
except Exception:
pass
# quick TCP check
s = socket.socket()
s.settimeout(1.0)
try:
s.connect(("127.0.0.1", port))
checks["tcp"] = f"127.0.0.1:{port} open"
except Exception as e:
checks["tcp"] = f"closed: {e}"
finally:
s.close()
except Exception as e:
checks["tcp_error"] = str(e)
return json.dumps({
"had_routes_before": info.get("had_routes"),
"restored": info.get("restored"),
"patch_used": info.get("patch"),
"agent_dir": info.get("agent_dir"),
"error": info.get("error"),
"has_routes_now": _api_server_has_esp32(),
"checks": checks,
"verify_curl": f'curl -H "Authorization: Bearer $API_SERVER_KEY" http://127.0.0.1:8642/api/esp32/voice/ws -v 2>&1 | grep "< HTTP" -> want 400 not 404'
}, indent=2)
def _tool_restore_esp32_gateway(args, **kwargs) -> str:
res = restore_if_needed()
# force reapply even if has routes if requested
force = bool(args.get("force"))
if force:
patch = _find_patch_file()
agent_dir = _find_hermes_agent_dir()
if patch and agent_dir:
# reset file first then reapply? just apply
ok = _try_apply_patch(patch, agent_dir)
res["force_restored"] = ok
return json.dumps(res, indent=2)
ctx.register_tool(
name="esp32_voice_gateway_status",
toolset="esp32_voice",
schema={
"name": "esp32_voice_gateway_status",
"description": "Check if ESP32 voice gateway routes (POST /api/esp32/voice + WS /api/esp32/voice/ws) are present after hermes update. Auto-restores from bundled patch if missing.",
"parameters": {"type": "object", "properties": {}, "required": []},
},
handler=_tool_check_esp32_gateway,
description="Check ESP32 voice gateway route status and auto-restore after update",
)
ctx.register_tool(
name="esp32_voice_gateway_restore",
toolset="esp32_voice",
schema={
"name": "esp32_voice_gateway_restore",
"description": "Force restore ESP32 voice gateway routes from bundled patch after hermes update wipes api_server.py",
"parameters": {
"type": "object",
"properties": {
"force": {"type": "boolean", "description": "Force re-apply even if routes seem present"},
},
"required": [],
},
},
handler=_tool_restore_esp32_gateway,
description="Force restore ESP32 routes from patch",
)
def _on_startup(session_id=None, **kw):
# Gateway hook – log status on startup for visibility
try:
if _api_server_has_esp32():
logger.info("[esp32-voice-gateway] Routes OK: /api/esp32/voice + /api/esp32/voice/ws present")
else:
logger.warning("[esp32-voice-gateway] Routes MISSING at startup – will try restore on next tool call; if you just ran `hermes update`, run: ~/Projects/hermes-esp32-voice-gateway/scripts/restore_after_update.py && hermes gateway restart")
except Exception:
pass
try:
ctx.register_hook("on_session_start", _on_startup)
except Exception:
pass
logger.info("[esp32-voice-gateway] Plugin loaded (had_routes=%s restored=%s has_now=%s)", info.get("had_routes"), info.get("restored"), _api_server_has_esp32())
Reference in New Issue View Git Blame Copy Permalink