feat: add Hermes WebUI provisioning and English Dragon HD Omni voices

web/voice_catalog.py

@@ -0,0 +1,133 @@
+"""Azure English voice catalog, independent of the speech agent.
+
+Omni names follow Microsoft's persona naming rule (remove final ``Neural``
+and append ``:DragonHDOmniLatestNeural``):
+https://learn.microsoft.com/en-us/azure/ai-services/speech-service/high-definition-voices
+Only the final Neural suffix is removed: Multilingual is retained because the
+rule does not document removing it. Advertised Omni identifiers take precedence
+for a matching persona. Derived variants follow the documented convention, not
+an individual synthesis/region availability probe. MultiTalker composites are
+excluded because the single-voice picker cannot supply their dialogue SSML.
+"""
+import asyncio
+import copy
+import os
+import re
+import time
+
+import httpx
+
+DEFAULT_VOICE = "en-US-Ava:DragonHDOmniLatestNeural"
+OMNI_SUFFIX = ":DragonHDOmniLatestNeural"
+CACHE_TTL = 3600
+FAILURE_RETRY_TTL = 60
+_REGION = re.compile(r"[a-z][a-z0-9]{0,62}\Z")
+
+
+def normalize_default_voice(voice: str | None = None) -> str:
+    """Read optional AZURE_TTS_VOICE and migrate a legacy DragonHD default.
+
+    Explicit regular Neural choices and already-Omni choices are preserved.
+    This helper has no dependency on, or side effects in, the agent.
+    """
+    value = (os.getenv("AZURE_TTS_VOICE", "") if voice is None else voice).strip()
+    if not value:
+        return DEFAULT_VOICE
+    if value.endswith(":DragonHDLatestNeural"):
+        return value.removesuffix(":DragonHDLatestNeural") + OMNI_SUFFIX
+    return value
+
+
+def _english(locale) -> bool:
+    return isinstance(locale, str) and (locale.lower() == "en" or locale.lower().startswith("en-"))
+
+
+def _entry(voice_id, name, locale, gender):
+    tag = "Omni" if "dragonhdomni" in voice_id.lower() else (
+        "HD" if "dragonhd" in voice_id.lower() else (
+            "Neural" if voice_id.endswith("Neural") else "Standard"))
+    return {"id": voice_id, "name": name, "locale": locale, "gender": gender, "tag": tag}
+
+
+def english_voices(payload: list) -> list[dict]:
+    """Keep advertised English-capable voices and add ordinary Neural Omni variants."""
+    if not isinstance(payload, list):
+        raise ValueError("Invalid catalog")
+    voices = {}
+    for row in payload:
+        if not isinstance(row, dict):
+            continue
+        voice_id, locale = row.get("ShortName"), row.get("Locale")
+        secondary = row.get("SecondaryLocaleList", [])
+        if not isinstance(secondary, list):
+            secondary = []
+        if not isinstance(voice_id, str) or not voice_id or "multitalker" in voice_id.lower():
+            continue
+        if not isinstance(locale, str) or not (_english(locale) or any(_english(s) for s in secondary)):
+            continue
+        name = row.get("DisplayName")
+        gender = row.get("Gender")
+        voices[voice_id.lower()] = _entry(voice_id, name if isinstance(name, str) else voice_id,
+                                          locale, gender if isinstance(gender, str) else "Unknown")
+    if not voices:
+        raise ValueError("No English voices in catalog")
+    # Advertised entries win (including identifier spelling and display metadata).
+    advertised_personas = {v["id"].split(":", 1)[0].lower() for v in voices.values() if v["tag"] == "Omni"}
+    for voice in list(voices.values()):
+        voice_id = voice["id"]
+        if ":" not in voice_id and voice_id.endswith("Neural"):
+            persona = voice_id.removesuffix("Neural")
+            if persona.lower() not in advertised_personas:
+                omni_id = persona + OMNI_SUFFIX
+                voices.setdefault(omni_id.lower(), _entry(omni_id, voice["name"], voice["locale"], voice["gender"]))
+    voices.setdefault(DEFAULT_VOICE.lower(), _entry(DEFAULT_VOICE, "Ava", "en-US", "Female"))
+    return sorted(voices.values(), key=lambda v: (v["id"].lower() != DEFAULT_VOICE.lower(),
+                  v["tag"] != "Omni", v["locale"].lower(), v["name"].lower(), v["id"].lower()))
+
+
+class VoiceCatalog:
+    """Process-local 1h cache; coalesces refreshes and retains last good data.
+
+    Failed refreshes are retried after 60s, not by every concurrent caller.
+    Failures always report catalog_available=False even when returning stale data.
+    No upstream exception, response body, header, or key is logged/returned.
+    """
+    def __init__(self, region: str, speech_key: str, default_voice: str | None = None):
+        self.region = region
+        self.speech_key = speech_key
+        self.default_voice = normalize_default_voice(default_voice)
+        self._lock = asyncio.Lock()
+        self._expires = 0.0
+        self._good = None
+        self._result = None
+
+    async def get(self) -> dict:
+        if self._result is not None and time.monotonic() < self._expires:
+            return copy.deepcopy(self._result)
+        async with self._lock:
+            if self._result is not None and time.monotonic() < self._expires:
+                return copy.deepcopy(self._result)
+            try:
+                # A single DNS label, never a URL, port, userinfo or path. Redirects
+                # are disabled so the subscription header cannot leave Azure.
+                if not self.speech_key or not isinstance(self.region, str) or not _REGION.fullmatch(self.region):
+                    raise ValueError("Invalid speech configuration")
+                async with httpx.AsyncClient(timeout=10, follow_redirects=False) as client:
+                    response = await client.get(
+                        f"https://{self.region}.tts.speech.microsoft.com/cognitiveservices/voices/list",
+                        headers={"Ocp-Apim-Subscription-Key": self.speech_key})
+                    response.raise_for_status()
+                    voices = english_voices(response.json())
+                self._good = voices
+                self._result = {"default_voice": self.default_voice, "voices": voices, "catalog_available": True}
+                self._expires = time.monotonic() + CACHE_TTL
+            except (httpx.HTTPError, ValueError, TypeError):
+                voices = self._good or [_entry(DEFAULT_VOICE, "Ava", "en-US", "Female")]
+                self._result = {"default_voice": self.default_voice, "voices": voices, "catalog_available": False,
+                                "message": "Azure voice catalog unavailable. Using last known voices." if self._good else
+                                           "Azure voice catalog unavailable. Using the default voice."}
+                self._expires = time.monotonic() + FAILURE_RETRY_TTL
+            return copy.deepcopy(self._result)
+
+
+catalog = VoiceCatalog(os.getenv("AZURE_SPEECH_REGION", "eastus"), os.getenv("AZURE_SPEECH_KEY", ""))