ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

simplify(compat): tools/transcription_tools — drop 47 re-exports/aliases, repoint 4 callers + 9 test files

Browse Source
This commit is contained in:
Teknium
2026-09-03 13:17:25 -07:00
parent d179f28307
commit 62732c7d8b
19 changed files with 127 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/web_server_config.py
+11 -11
View File
@@ -246,7 +246,7 @@ CONFIG_SCHEMA = _config_schema_with_virtual_fields()
def _is_command_provider_block(value: Any) -> bool:
"""True when *value* declares a command-type voice provider.
Mirrors the runtime discriminators (``tools.tts_tool._is_command_provider_config`` /
Mirrors the runtime discriminators (``tools.tts_command_provider._is_command_provider_config`` /
``tools.transcription_tools._is_command_stt_provider_config``) and the desktop's
``isCommandProvider``: ``type`` is OPTIONAL and case/space-insensitive (absent or
normalizing to ``"command"``); ``command`` MUST be a non-empty string.
@@ -278,7 +278,7 @@ def _custom_provider_options(kind: str, builtin_names: List[str], cfg: Dict[str,
if kind == "tts":
from tools.tts_tool import BUILTIN_TTS_PROVIDERS as _runtime_builtins
else:
from tools.transcription_tools import BUILTIN_STT_PROVIDERS as _runtime_builtins
from tools.transcription_common import BUILTIN_STT_PROVIDERS as _runtime_builtins
def _add(name: Any) -> None:
stripped = name.strip() if isinstance(name, str) else ""
@@ -319,7 +319,6 @@ def _custom_provider_options(kind: str, builtin_names: List[str], cfg: Dict[str,
def _memory_provider_schema_options(cfg: Dict[str, Any]) -> List[str]:
"""Discovered memory providers plus the currently-configured one, so a value that is no
longer discoverable (e.g. plugin removed from disk) never vanishes from the dropdown."""
from hermes_cli.web_server import _memory_provider_options
options = _memory_provider_options()
memory = cfg.get("memory")
current = _normalize_memory_provider_name(memory.get("provider") if isinstance(memory, dict) else None)
@@ -343,7 +342,8 @@ def _schema_with_dynamic_provider_options() -> Dict[str, Dict[str, Any]]:
that reads the schema. ``CONFIG_SCHEMA`` is never mutated; changed entries are
shallow-copied onto a copied mapping.
"""
from hermes_cli.web_server import _plugin_terminal_backend_rows, load_config
from hermes_cli.web_server_profiles import _plugin_terminal_backend_rows
from hermes_cli.config import load_config
try:
cfg = load_config()
except Exception: # pragma: no cover - schema must survive config errors
@@ -393,7 +393,7 @@ def _normalize_main_model_assignment(provider: str, model: str) -> tuple[str, st
2. Model-format normalization for the resolved provider via
``normalize_model_for_provider`` (custom/user providers keep the model verbatim).
"""
from hermes_cli.web_server import load_config
from hermes_cli.config import load_config
from hermes_cli.config import get_compatible_custom_providers
from hermes_cli.models import _AGGREGATOR_PROVIDERS, _KNOWN_PROVIDER_NAMES, normalize_provider
from hermes_cli.model_normalize import normalize_model_for_provider
@@ -595,7 +595,7 @@ def _register_custom_endpoint(base_url: str, api_key: str, model: str) -> None:
``hermes model`` custom flow) so the picker gets a proper ready row instead of a "needs
setup" dead-end. Dedups by base_url; never blocks the already-persisted assignment."""
try:
from hermes_cli.main import _auto_provider_name, _save_custom_provider
from hermes_cli.main_provider_setup import _auto_provider_name, _save_custom_provider
_save_custom_provider(base_url, api_key, model, name=_auto_provider_name(base_url))
except Exception:
@@ -627,7 +627,7 @@ def _stale_aux_pins(cfg: dict, new_provider: str) -> list:
def _cron_model_impact(cfg: dict, provider: str, model: str) -> Any:
from hermes_cli.web_server import load_config
from hermes_cli.config import load_config
try:
effective_config = load_config()
effective_provider, effective_model = resolve_cron_model_drift_defaults(effective_config)
@@ -642,7 +642,7 @@ def _cron_model_impact(cfg: dict, provider: str, model: str) -> Any:
def _apply_main_assignment_sync(cfg: dict, provider: str, model: str, base_url: str, api_key: str) -> dict:
from hermes_cli.web_server import save_config
from hermes_cli.config import save_config
if not provider or not model:
raise HTTPException(status_code=400, detail="provider and model required for main")
provider, model = _normalize_main_model_assignment(provider, model)
@@ -673,7 +673,7 @@ def _apply_main_assignment_sync(cfg: dict, provider: str, model: str, base_url:
def _apply_aux_assignment_sync(cfg: dict, provider: str, model: str, task: str, base_url: str, api_key: str) -> dict:
from hermes_cli.web_server import save_config
from hermes_cli.config import save_config
aux = cfg.get("auxiliary")
if not isinstance(aux, dict):
aux = {}
@@ -736,7 +736,7 @@ def _apply_model_assignment_sync(
Runs inside ``_profile_scope`` (worker thread) so every load_config/save_config lands in
the requested profile. Raises HTTPException for validation errors.
"""
from hermes_cli.web_server import load_config
from hermes_cli.config import load_config
cfg = load_config()
if scope == "main":
return _apply_main_assignment_sync(cfg, provider, model, base_url, api_key)
@@ -790,7 +790,7 @@ def _denormalize_config_from_web(config: Dict[str, Any]) -> Dict[str, Any]:
removed). A partial update (Settings autosave diff) that OMITS the key means "unchanged"
and must leave the on-disk override alone — not be treated as an explicit 0.
"""
from hermes_cli.web_server import load_config
from hermes_cli.config import load_config
config = dict(config)
config.pop("_model_meta", None)
plugins/platforms/discord/ffmpeg_utils.py
+2 -2
View File
@@ -1,4 +1,4 @@
"""ffmpeg discovery for Discord voice: ``tools.transcription_tools`` owns the shared lookup
"""ffmpeg discovery for Discord voice: ``tools.transcription_audio`` owns the shared lookup
(PATH + Homebrew/local prefixes); this layers an explicit ``FFMPEG_PATH`` override and a
Windows winget fallback (installs that never touch PATH) on top."""
@@ -12,7 +12,7 @@ from pathlib import Path
def _shared_find_ffmpeg():
"""Delegate to the repo-wide ffmpeg discovery helper when importable."""
try:
from tools.transcription_tools import _find_ffmpeg_binary
from tools.transcription_audio import _find_ffmpeg_binary
except ImportError: # standalone plugin import (tests / sandboxes)
return shutil.which("ffmpeg")
return _find_ffmpeg_binary()
tests/hermes_cli/test_stt_picker.py
+1 -1
View File
@@ -89,7 +89,7 @@ class TestActiveDetection:
class TestModelPicker:
def test_catalog_matches_runtime_model_sets(self):
from tools.transcription_tools import GROQ_MODELS, OPENAI_MODELS
from tools.transcription_common import GROQ_MODELS, OPENAI_MODELS
assert set(STT_MODEL_CATALOG["openai"]) == OPENAI_MODELS
assert set(STT_MODEL_CATALOG["groq"]) == GROQ_MODELS
tests/tools/test_pre_transcription_hook.py
+3 -2
View File
@@ -383,9 +383,10 @@ class TestSttPromptConfig:
assert result["success"] is True # truncation never errors
_, kwargs = backend.call_args
from tools import transcription_command
max_chars = (
transcription_tools._WHISPER_PROMPT_TOKEN_CAP
* transcription_tools._PROMPT_CHARS_PER_TOKEN
transcription_command._WHISPER_PROMPT_TOKEN_CAP
* transcription_command._PROMPT_CHARS_PER_TOKEN
)
assert len(kwargs["prompt"]) == max_chars
# Tail survives — whisper conditions on the final context window.
tests/tools/test_strict_provider_selection.py
+4 -4
View File
@@ -212,7 +212,7 @@ class TestSttStrictSelection:
with patch.object(tt, "_load_stt_config", return_value={"openai": {"api_key": "sk-direct"}}), \
patch("tools.tool_backend_helpers.read_selection", return_value="nous"), \
patch.object(tt, "resolve_managed_tool_gateway", return_value=MANAGED):
patch("tools.managed_tool_gateway.resolve_managed_tool_gateway", return_value=MANAGED):
api_key, base_url = tt._resolve_openai_audio_client_config()
assert api_key == "managed-token"
assert base_url.startswith("https://gateway.nousresearch.com")
@@ -222,8 +222,8 @@ class TestSttStrictSelection:
with patch.object(tt, "_load_stt_config", return_value={}), \
patch("tools.tool_backend_helpers.read_selection", return_value="openai"), \
patch.object(tt, "resolve_openai_audio_api_key", return_value=""), \
patch.object(tt, "resolve_managed_tool_gateway") as gw:
patch("tools.tool_backend_helpers.resolve_openai_audio_api_key", return_value=""), \
patch("tools.managed_tool_gateway.resolve_managed_tool_gateway") as gw:
with pytest.raises(ValueError) as exc:
tt._resolve_openai_audio_client_config()
gw.assert_not_called()
@@ -235,7 +235,7 @@ class TestSttStrictSelection:
with patch.object(tt, "_load_stt_config", return_value={}), \
patch("tools.tool_backend_helpers.read_selection", return_value=None), \
patch.object(tt, "resolve_openai_audio_api_key", return_value="sk-env"):
patch("tools.tool_backend_helpers.resolve_openai_audio_api_key", return_value="sk-env"):
api_key, base_url = tt._resolve_openai_audio_client_config()
assert api_key == "sk-env"
tests/tools/test_stt_cloud_trim.py
+14 -15
View File
@@ -36,18 +36,17 @@ if "faster_whisper" not in sys.modules:
faster_whisper_stub.__spec__ = ModuleSpec("faster_whisper", loader=None)
sys.modules["faster_whisper"] = faster_whisper_stub
from tools.transcription_tools import (
CLOUD_STT_PROVIDERS,
BUILTIN_STT_PROVIDERS,
from tools.transcription_common import BUILTIN_STT_PROVIDERS, CLOUD_STT_PROVIDERS
from tools.transcription_audio import (
_cloud_trim_settings,
_CLOUD_TRIM_KEEP_MS_DEFAULT,
_CLOUD_TRIM_MIN_INPUT_SECONDS,
_CLOUD_TRIM_THRESHOLD_DB_DEFAULT,
_trim_silence_for_cloud_stt,
)
import tools.transcription_tools as tt_module
# The E2E fixtures below must be past the short-clip input gate.
_GATE = tt_module._CLOUD_TRIM_MIN_INPUT_SECONDS
_GATE = _CLOUD_TRIM_MIN_INPUT_SECONDS
_HAS_FFMPEG = bool(shutil.which("ffmpeg")) and bool(shutil.which("ffprobe"))
@@ -231,13 +230,13 @@ class TestTrimFallbacks:
def test_missing_ffmpeg_returns_none(self, tmp_path):
wav = _write_wav(tmp_path / "a.wav", [("tone", 1)])
with patch("tools.transcription_tools._find_ffmpeg_binary", return_value=None):
with patch("tools.transcription_audio._find_ffmpeg_binary", return_value=None):
assert _trim_silence_for_cloud_stt(wav, {}) is None
def test_missing_ffprobe_returns_none(self, tmp_path):
wav = _write_wav(tmp_path / "a.wav", [("tone", 1)])
with patch("tools.transcription_tools._find_ffmpeg_binary", return_value="/bin/ffmpeg"), \
patch("tools.transcription_tools._find_ffprobe_binary", return_value=None):
with patch("tools.transcription_audio._find_ffmpeg_binary", return_value="/bin/ffmpeg"), \
patch("tools.transcription_audio._find_ffprobe_binary", return_value=None):
assert _trim_silence_for_cloud_stt(wav, {}) is None
def test_ffmpeg_failure_returns_none_and_cleans_up(self, tmp_path):
@@ -247,16 +246,16 @@ class TestTrimFallbacks:
def probe(path):
return 60.0 # past the short-clip gate so the encode is attempted
with patch("tools.transcription_tools._find_ffmpeg_binary", return_value="/bin/ffmpeg"), \
patch("tools.transcription_tools._probe_audio_duration", side_effect=probe), \
patch("tools.transcription_tools.subprocess.run",
with patch("tools.transcription_audio._find_ffmpeg_binary", return_value="/bin/ffmpeg"), \
patch("tools.transcription_audio._probe_audio_duration", side_effect=probe), \
patch("tools.transcription_audio.subprocess.run",
side_effect=sp.CalledProcessError(1, "ffmpeg")):
assert _trim_silence_for_cloud_stt(wav, {}) is None
def test_unprobeable_source_returns_none(self, tmp_path):
wav = _write_wav(tmp_path / "a.wav", [("tone", 1)])
with patch("tools.transcription_tools._find_ffmpeg_binary", return_value="/bin/ffmpeg"), \
patch("tools.transcription_tools._probe_audio_duration", return_value=None):
with patch("tools.transcription_audio._find_ffmpeg_binary", return_value="/bin/ffmpeg"), \
patch("tools.transcription_audio._probe_audio_duration", return_value=None):
assert _trim_silence_for_cloud_stt(wav, {}) is None
@@ -274,7 +273,7 @@ class TestTrimE2E:
tmp_path / "pauses.wav",
[("tone", 2), ("silence", 6), ("tone", 2), ("silence", 4)],
)
from tools.transcription_tools import _probe_audio_duration
from tools.transcription_audio import _probe_audio_duration
trimmed = _trim_silence_for_cloud_stt(wav, {})
assert trimmed is not None
try:
@@ -306,7 +305,7 @@ class TestTrimE2E:
wav = _write_wav(
tmp_path / "short.wav", [("tone", 2), ("silence", 4), ("tone", 2)]
)
with patch.object(tt_module, "_run_ffmpeg_stt_encode") as mock_encode:
with patch("tools.transcription_audio._run_ffmpeg_stt_encode") as mock_encode:
assert _trim_silence_for_cloud_stt(wav, {}) is None
mock_encode.assert_not_called()
tests/tools/test_stt_silence_hallucinations.py
+1 -1
View File
@@ -12,7 +12,7 @@ three-layer fix at every local whisper call site:
from types import SimpleNamespace
from tools.transcription_tools import (
from tools.transcription_local import (
_LOGPROB_THRESHOLD_DEFAULT,
_NO_SPEECH_PROB_THRESHOLD_DEFAULT,
_is_hallucinated_segment,
tests/tools/test_transcription.py
+2 -1
View File
@@ -71,7 +71,8 @@ class TestValidateAudioFile:
def test_too_large(self, tmp_path):
f = tmp_path / "big.ogg"
f.write_bytes(b"x")
from tools.transcription_tools import _validate_audio_file, MAX_FILE_SIZE
from tools.transcription_tools import _validate_audio_file
from tools.transcription_common import MAX_FILE_SIZE
real_stat = f.stat()
with patch.object(type(f), "stat", return_value=os.stat_result((
real_stat.st_mode, real_stat.st_ino, real_stat.st_dev,
tests/tools/test_transcription_command_providers.py
+4 -2
View File
@@ -25,8 +25,8 @@ from pathlib import Path
from unittest.mock import patch
from tools.transcription_tools import (
BUILTIN_STT_PROVIDERS,
from tools.transcription_common import BUILTIN_STT_PROVIDERS
from tools.transcription_command import (
COMMAND_STT_OUTPUT_FORMATS,
DEFAULT_COMMAND_STT_LANGUAGE,
DEFAULT_COMMAND_STT_OUTPUT_FORMAT,
@@ -37,6 +37,8 @@ from tools.transcription_tools import (
_render_command_stt_template,
_resolve_command_stt_provider_config,
_transcribe_command_stt,
)
from tools.transcription_tools import (
transcribe_audio,
)
tests/tools/test_transcription_plugin_dispatch.py
+2 -1
View File
@@ -212,7 +212,8 @@ class TestTranscribeAudioE2E:
transcription_registry.register_provider(provider)
audio_path = tmp_path / "oversized.mp3"
with audio_path.open("wb") as audio_file:
audio_file.seek(transcription_tools.MAX_FILE_SIZE)
from tools.transcription_common import MAX_FILE_SIZE
audio_file.seek(MAX_FILE_SIZE)
audio_file.write(b"\0")
with patch("tools.transcription_tools._load_stt_config", return_value={"provider": "openrouter"}), \
tests/tools/test_transcription_tools.py
+26 -25
View File
@@ -65,7 +65,7 @@ def sample_silk(tmp_path):
@pytest.fixture
def oversized_wav(tmp_path):
"""Create a sparse WAV-shaped file just above the remote upload cap."""
from tools.transcription_tools import MAX_FILE_SIZE
from tools.transcription_common import MAX_FILE_SIZE
wav_path = tmp_path / "oversized.wav"
with wav_path.open("wb") as audio_file:
@@ -248,7 +248,7 @@ class TestTranscribeLocalCommand:
monkeypatch.setattr("tools.tts_command_provider.subprocess.Popen", fake_popen)
from tools.transcription_tools import _run_command_stt
from tools.transcription_command import _run_command_stt
result = _run_command_stt("echo hi", timeout=1)
@@ -292,10 +292,10 @@ class TestTranscribeLocalCommand:
returncode = 0
return R()
monkeypatch.setattr("tools.transcription_tools.tempfile.TemporaryDirectory", fake_tempdir)
monkeypatch.setattr("tools.transcription_tools.subprocess.run", fake_run)
monkeypatch.setattr("tools.transcription_local.tempfile.TemporaryDirectory", fake_tempdir)
monkeypatch.setattr("tools.transcription_audio.subprocess.run", fake_run)
monkeypatch.setattr(
"tools.transcription_tools._prepare_local_audio",
"tools.transcription_local._prepare_local_audio",
lambda *a, **k: (str(sample_wav), None),
)
@@ -334,9 +334,9 @@ class TestTranscribeLocalCommand:
(out_dir / "test.txt").write_text("hello from local command\n", encoding="utf-8")
return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="")
monkeypatch.setattr("tools.transcription_tools.tempfile.TemporaryDirectory", fake_tempdir)
monkeypatch.setattr("tools.transcription_tools._find_ffmpeg_binary", lambda: "/opt/homebrew/bin/ffmpeg")
monkeypatch.setattr("tools.transcription_tools.subprocess.run", fake_run)
monkeypatch.setattr("tools.transcription_local.tempfile.TemporaryDirectory", fake_tempdir)
monkeypatch.setattr("tools.transcription_audio._find_ffmpeg_binary", lambda: "/opt/homebrew/bin/ffmpeg")
monkeypatch.setattr("tools.transcription_audio.subprocess.run", fake_run)
from tools.transcription_tools import _transcribe_local_command
@@ -509,7 +509,8 @@ class TestValidateAudioFileEdgeCases:
def test_all_supported_formats_accepted(self, tmp_path):
from tools.transcription_tools import _validate_audio_file, SUPPORTED_FORMATS
from tools.transcription_tools import _validate_audio_file
from tools.transcription_common import SUPPORTED_FORMATS
for fmt in SUPPORTED_FORMATS:
f = tmp_path / f"test{fmt}"
f.write_bytes(b"data")
@@ -925,7 +926,7 @@ class TestTranscribeAudioElevenLabsDispatch:
class TestExtractTranscriptText:
def test_strips_qwen3_asr_language_envelope(self):
from tools.transcription_tools import _extract_transcript_text
from tools.transcription_cloud import _extract_transcript_text
result = _extract_transcript_text(
"language zh\n<audio_language>zh</audio_language>\n<asr_text>你好，世界",
@@ -934,7 +935,7 @@ class TestExtractTranscriptText:
assert result == "你好，世界"
def test_keeps_non_envelope_marker_literal(self):
from tools.transcription_tools import _extract_transcript_text
from tools.transcription_cloud import _extract_transcript_text
result = _extract_transcript_text(
"The user literally said <asr_text> while reading markup.",
@@ -951,10 +952,10 @@ class TestShellSafety:
import shlex
monkeypatch.delenv("HERMES_LOCAL_STT_COMMAND", raising=False)
monkeypatch.setattr(
"tools.transcription_tools._find_whisper_binary",
"tools.transcription_local._find_whisper_binary",
lambda: "/usr/bin/whisper",
)
from tools.transcription_tools import _get_local_command_template
from tools.transcription_local import _get_local_command_template
template = _get_local_command_template()
assert template is not None
cmd = template.format(
@@ -970,10 +971,10 @@ class TestShellSafety:
def test_env_var_template_metacharacters_are_literal_argv(
self, monkeypatch, sample_wav, tmp_path
):
from hermes_cli._subprocess_compat import windows_hide_flags
from tools.transcription_tools import (
LOCAL_STT_COMMAND_ENV,
_transcribe_local_command,
windows_hide_flags,
)
output_dir = tmp_path / "transcript-output"
@@ -1005,9 +1006,9 @@ class TestShellSafety:
return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
monkeypatch.setattr(
"tools.transcription_tools.tempfile.TemporaryDirectory", fake_tempdir
"tools.transcription_local.tempfile.TemporaryDirectory", fake_tempdir
)
monkeypatch.setattr("tools.transcription_tools.subprocess.run", fake_run)
monkeypatch.setattr("tools.transcription_audio.subprocess.run", fake_run)
result = _transcribe_local_command(sample_wav, "base")
@@ -1102,7 +1103,7 @@ class TestLocalBaseUrlNoApiKey:
def test_is_local_or_private_url(self):
from tools.transcription_tools import _is_local_or_private_url
from tools.transcription_cloud import _is_local_or_private_url
assert _is_local_or_private_url("http://localhost:8504/v1")
assert _is_local_or_private_url("http://127.0.0.1:9000")
assert _is_local_or_private_url("http://10.0.0.5/v1")
@@ -1131,7 +1132,7 @@ class TestCafConversion:
return MagicMock(returncode=0)
monkeypatch.setattr(
"tools.transcription_tools._find_ffmpeg_binary",
"tools.transcription_audio._find_ffmpeg_binary",
lambda: "/usr/bin/ffmpeg",
)
monkeypatch.setattr(subprocess, "run", fake_run)
@@ -1199,7 +1200,7 @@ class TestRunCommandSttIdleTimeout:
def test_stderr_progress_extends_beyond_timeout(self, tmp_path):
"""A slow-but-alive command that keeps emitting output survives an
idle timeout shorter than its total runtime."""
from tools.transcription_tools import _run_command_stt
from tools.transcription_command import _run_command_stt
script = tmp_path / "progress_then_exit.py"
script.write_text(
@@ -1225,7 +1226,7 @@ class TestRunCommandSttIdleTimeout:
def test_silent_stall_still_times_out(self, tmp_path):
"""A silently stalled command is killed once the idle window elapses,
and pre-stall output is preserved on the TimeoutExpired."""
from tools.transcription_tools import _run_command_stt
from tools.transcription_command import _run_command_stt
script = tmp_path / "progress_then_hang.py"
script.write_text(
@@ -1265,11 +1266,11 @@ class TestExplicitOpenaiSelectionError:
monkeypatch.delenv("VOICE_TOOLS_OPENAI_KEY", raising=False)
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
monkeypatch.setattr(
"tools.transcription_tools.resolve_openai_audio_api_key",
"tools.tool_backend_helpers.resolve_openai_audio_api_key",
lambda: None,
)
monkeypatch.setattr(
"tools.transcription_tools.resolve_managed_tool_gateway",
"tools.managed_tool_gateway.resolve_managed_tool_gateway",
lambda vendor: None,
)
@@ -1278,7 +1279,7 @@ class TestExplicitOpenaiSelectionError:
):
self._no_openai_credentials(monkeypatch)
monkeypatch.setattr(
"tools.transcription_tools.managed_nous_tools_enabled", lambda: True
"tools.tool_backend_helpers.managed_nous_tools_enabled", lambda: True
)
monkeypatch.setattr(
"tools.transcription_tools._load_stt_config", lambda: {}
@@ -1303,7 +1304,7 @@ class TestExplicitOpenaiSelectionError:
hint."""
self._no_openai_credentials(monkeypatch)
monkeypatch.setattr(
"tools.transcription_tools.managed_nous_tools_enabled", lambda: True
"tools.tool_backend_helpers.managed_nous_tools_enabled", lambda: True
)
monkeypatch.setattr(
"tools.transcription_tools._load_stt_config", lambda: {}
@@ -1311,7 +1312,7 @@ class TestExplicitOpenaiSelectionError:
with patch("tools.transcription_tools._HAS_OPENAI", True), \
patch("tools.transcription_tools._HAS_FASTER_WHISPER", False), \
patch(
"tools.transcription_tools.nous_tool_gateway_unavailable_message",
"tools.tool_backend_helpers.nous_tool_gateway_unavailable_message",
lambda what: f"managed route down for {what}; run `hermes tools`",
):
from tools.transcription_tools import _dispatch_stt_provider
tools/transcription_audio.py
+2 -8
View File
@@ -2,9 +2,8 @@
Binary discovery, the shared ffmpeg m4a encode (transcode + silence trim),
source/format validation, WeChat .silk decoding, CAF conversion and the
best-effort cloud pre-upload silence trim. Every name is re-imported by
``tools/transcription_tools.py`` (patch surface), which is imported lazily here
so origin patches still intercept.
best-effort cloud pre-upload silence trim. Facade-owned state (``_HAS_PILK``,
``_safe_find_spec``) is read lazily from ``tools.transcription_tools``.
"""
from __future__ import annotations
@@ -66,7 +65,6 @@ def _transcode_audio_for_stt(file_path: str, work_dir: str) -> tuple[Optional[st
"""Transcode to a compact 16 kHz mono AAC/m4a for STT upload; ``(converted_path, None)`` or ``(None, error)``.
Newer OpenAI models reject containers ``whisper-1`` accepted (notably Ogg/Opus voice notes) and
gateway downloads may carry a misleading extension."""
from tools.transcription_tools import _find_ffmpeg_binary, _run_ffmpeg_stt_encode
ffmpeg = _find_ffmpeg_binary()
if not ffmpeg:
return None, "audio needs transcoding for the STT API, but ffmpeg was not found"
@@ -144,7 +142,6 @@ def _prepare_audio_for_transcription(file_path: str) -> tuple[Optional[str], Opt
def _prepare_local_audio(file_path: str, work_dir: str) -> tuple[Optional[str], Optional[str]]:
"""Normalize audio for local CLI STT when needed."""
from tools.transcription_tools import _find_ffmpeg_binary
audio_path = Path(file_path)
if audio_path.suffix.lower() in LOCAL_NATIVE_AUDIO_FORMATS:
return file_path, None
@@ -166,7 +163,6 @@ def _prepare_local_audio(file_path: str, work_dir: str) -> tuple[Optional[str],
def _convert_caf_to_wav(file_path: str) -> Optional[str]:
"""Convert CAF to WAV using ffmpeg or afconvert (macOS)."""
from tools.transcription_tools import _find_ffmpeg_binary
audio_path = Path(file_path)
wav_path = os.path.join(audio_path.parent, f"{audio_path.stem}.wav")
ffmpeg = _find_ffmpeg_binary()
@@ -209,7 +205,6 @@ def _probe_audio_duration(file_path: str) -> Optional[float]:
"""Return the audio duration in seconds via ffprobe, or None. Canonical sync probe;
``gateway/run.py._probe_audio_duration`` and the Telegram adapter carry local variants — keep
the command shape in sync."""
from tools.transcription_tools import _find_ffprobe_binary
ffprobe = _find_ffprobe_binary()
if not ffprobe:
return None
@@ -234,7 +229,6 @@ def _cloud_trim_settings(stt_config: Dict[str, Any]) -> tuple[bool, int, int]:
def _trim_silence_for_cloud_stt(file_path: str, stt_config: Dict[str, Any]) -> Optional[str]:
"""Return a silence-trimmed copy of *file_path* for cloud upload, or None (= upload the original).
On success the caller owns deleting the returned file's parent directory."""
from tools.transcription_tools import _find_ffmpeg_binary, _probe_audio_duration, _run_ffmpeg_stt_encode
enabled, threshold_db, keep_ms = _cloud_trim_settings(stt_config)
if not enabled:
return None
tools/transcription_cloud.py
+11 -10
View File
@@ -2,9 +2,9 @@
OpenAI-SDK-shaped backends (groq, openai, deepinfra), Mistral Voxtral, REST multipart
backends (xAI, ElevenLabs), and OpenAI audio credential resolution (config > keyless
local server > env > managed Nous gateway). Every name is re-imported by
``tools/transcription_tools.py`` (patch surface), imported lazily here so origin
patches still intercept.
local server > env > managed Nous gateway). Facade-owned state and helpers
(``_HAS_OPENAI``, ``_resolve_provider_key``, ``_resolve_stt_language``, ``_load_stt_config``,
``get_env_value``) are read lazily from ``tools.transcription_tools``.
"""
from __future__ import annotations
@@ -111,7 +111,7 @@ def _transcribe_openai(
"""Transcribe via the OpenAI ``audio.transcriptions.create`` SDK shape, shared by every
OpenAI-compatible endpoint (DeepInfra etc.): explicit ``api_key``/``base_url`` skip the
OpenAI-only auth chain; ``provider_label`` names the response's provider."""
from tools.transcription_tools import _HAS_OPENAI, _resolve_openai_audio_client_config, _resolve_stt_language
from tools.transcription_tools import _HAS_OPENAI, _resolve_stt_language
if api_key is None:
try:
api_key, fallback_base = _resolve_openai_audio_client_config()
@@ -329,7 +329,7 @@ def _transcribe_deepinfra(
file_path: str, model_name: str, *, language: Optional[str] = None, prompt: Optional[str] = None
) -> Dict[str, Any]:
"""Resolve DeepInfra credentials/model (shared ``hermes_cli.models`` helpers), then delegate to :func:`_transcribe_openai`."""
from tools.transcription_tools import _load_stt_config, _resolve_provider_key, _transcribe_openai
from tools.transcription_tools import _load_stt_config, _resolve_provider_key
api_key = _resolve_provider_key("DEEPINFRA_API_KEY", "deepinfra")
if not api_key:
return _error_result("DEEPINFRA_API_KEY not set")
@@ -364,7 +364,7 @@ def _is_local_or_private_url(url: str) -> bool:
def _direct_openai_credentials(cfg_api_key: str, cfg_base_url: str) -> Optional[tuple[str, str]]:
"""Direct-credential ladder: config key > keyless local base_url (placeholder key so the SDK
constructs a client) > env key; None if none apply."""
from tools.transcription_tools import resolve_openai_audio_api_key
from tools.tool_backend_helpers import resolve_openai_audio_api_key
if cfg_api_key:
return cfg_api_key, (cfg_base_url or OPENAI_BASE_URL)
# A local OpenAI-compatible server needs no key — send a placeholder so the SDK doesn't refuse to
@@ -380,10 +380,11 @@ def _resolve_openai_audio_client_config() -> tuple[str, str]:
``"nous"`` -> managed gateway ONLY (a direct OPENAI_API_KEY must NOT override it); any other
stored provider -> direct credentials ONLY (no silent managed fallback); never-configured ->
legacy ladder: direct credentials, then the managed gateway. Failures raise ValueError."""
from tools.transcription_tools import (
_load_stt_config, managed_nous_tools_enabled, nous_tool_gateway_unavailable_message,
resolve_managed_tool_gateway)
from tools.tool_backend_helpers import NOUS_MANAGED_PROVIDER, read_selection, selection_error
from tools.transcription_tools import _load_stt_config
from tools.managed_tool_gateway import resolve_managed_tool_gateway
from tools.tool_backend_helpers import (
NOUS_MANAGED_PROVIDER, managed_nous_tools_enabled, nous_tool_gateway_unavailable_message,
read_selection, selection_error)
openai_cfg = _load_stt_config().get("openai") or {}
selected = read_selection("stt")
tools/transcription_command.py
+2 -3
View File
@@ -2,9 +2,8 @@
``stt.providers.<name>: type: command`` registry, plugin-registered
``TranscriptionProvider`` dispatch, and the ``pre_transcription`` hook that
threads prompt/language/model overrides into every backend. Every name is
re-imported by ``tools/transcription_tools.py`` (patch surface), which is
imported lazily here so origin patches still intercept.
threads prompt/language/model overrides into every backend. ``_resolve_stt_language``
is read lazily from ``tools.transcription_tools``.
"""
from __future__ import annotations
tools/transcription_common.py
+2 -6
View File
@@ -1,8 +1,4 @@
"""Constants, result envelopes and tiny config readers shared by every STT module.
Every name is re-imported by ``tools/transcription_tools.py``, so
``tools.transcription_tools.<name>`` keeps resolving (and monkeypatching).
"""
"""Constants, result envelopes and tiny config readers shared by every STT module."""
from __future__ import annotations
@@ -11,7 +7,7 @@ import os
import subprocess # noqa: F401 (type annotation only)
from typing import Any, Dict
from tools.tts_command_provider import _get_provider_section as _get_stt_section # noqa: F401 (re-exported)
from tools.tts_command_provider import _get_provider_section as _get_stt_section
# Log-record parity with the origin module.
logger = logging.getLogger("tools.transcription_tools")
tools/transcription_local.py
+3 -5
View File
@@ -3,8 +3,7 @@
faster-whisper loading (CUDA->CPU fallback, Apple Silicon pinning), the
anti-hallucination transcribe kwargs and segment gate, and the local whisper CLI
(``local_command``) provider. The cached-model singleton and idle-unload watcher
stay in ``transcription_tools`` (module state), which re-imports every name here
(patch surface) and is imported lazily so origin patches still intercept.
stay in ``transcription_tools`` (module state) and are read from it lazily.
"""
from __future__ import annotations
@@ -19,7 +18,7 @@ import importlib.util as _ilu
from pathlib import Path
from typing import Any, Dict, Optional
from tools.transcription_audio import _run_quiet
from tools.transcription_audio import _find_whisper_binary, _prepare_local_audio, _run_quiet
from tools.transcription_common import (
DEFAULT_LOCAL_MODEL, DEFAULT_LOCAL_STT_LANGUAGE, GROQ_MODELS, LOCAL_STT_COMMAND_ENV,
OPENAI_MODELS, _config_number, _error_result, _log_prompt_unsupported, _ok_result,
@@ -30,7 +29,6 @@ logger = logging.getLogger("tools.transcription_tools")
def _get_local_command_template() -> Optional[str]:
from tools.transcription_tools import _find_whisper_binary
configured = os.getenv(LOCAL_STT_COMMAND_ENV, "").strip()
if configured:
return configured
@@ -225,7 +223,7 @@ def _transcribe_local_command(
file_path: str, model_name: str, *, language: Optional[str] = None, prompt: Optional[str] = None
) -> Dict[str, Any]:
"""Run the configured local STT command template and read back a .txt transcript."""
from tools.transcription_tools import _prepare_local_audio, _resolve_stt_language
from tools.transcription_tools import _resolve_stt_language
if prompt:
_log_prompt_unsupported("STT provider 'local_command'")
command_template = _get_local_command_template()
tools/transcription_tools.py
+17 -33
View File
@@ -6,55 +6,39 @@ Built-in providers: local (faster-whisper, default/free), local_command, groq, o
user-declared command providers and plugin providers. ``transcribe_audio(path)`` returns
``{"success", "transcript", "error"?, "provider"?}``. This module owns provider resolution,
the dispatcher and the cached local model + idle-unload state; backends live in
``transcription_{common,audio,local,cloud,command}`` and are re-imported here so
``tools.transcription_tools.<name>`` stays the patch surface.
``transcription_{common,audio,local,cloud,command}``.
"""
import logging
import os
import shutil
import subprocess # noqa: F401 (tests patch tools.transcription_tools.subprocess.run)
import tempfile # noqa: F401 (tests patch tools.transcription_tools.tempfile.TemporaryDirectory)
import threading
import time
import importlib.util as _ilu
from pathlib import Path
from typing import Optional, Dict, Any
from hermes_cli._subprocess_compat import windows_hide_flags # noqa: F401 (imported by tests)
from utils import is_truthy_value
from tools.managed_tool_gateway import resolve_managed_tool_gateway # noqa: F401 (patched by tests)
from tools.tool_backend_helpers import ( # noqa: F401 (patched by tests; read lazily by transcription_cloud)
managed_nous_tools_enabled, nous_tool_gateway_unavailable_message, resolve_openai_audio_api_key)
from tools.transcription_common import ( # noqa: F401 (re-exported; tests patch tools.transcription_tools.<name>)
from tools.transcription_common import (
BUILTIN_STT_PROVIDERS, CLOUD_STT_PROVIDERS, DEFAULT_ELEVENLABS_STT_MODEL,
DEFAULT_GROQ_STT_MODEL, DEFAULT_LOCAL_MODEL, DEFAULT_MISTRAL_STT_MODEL, DEFAULT_PROVIDER,
DEFAULT_STT_MODEL, ELEVENLABS_STT_BASE_URL, GROQ_MODELS, LOCAL_STT_COMMAND_ENV,
LOCAL_STT_LANGUAGE_ENV, MAX_FILE_SIZE, OPENAI_MODELS, SUPPORTED_FORMATS, XAI_STT_BASE_URL,
_error_result, _get_stt_section, _ok_result)
from tools.transcription_audio import ( # noqa: F401 (re-exported; tests patch tools.transcription_tools.<name>)
_CLOUD_TRIM_KEEP_MS_DEFAULT, _CLOUD_TRIM_MIN_INPUT_SECONDS, _CLOUD_TRIM_THRESHOLD_DB_DEFAULT,
_cloud_trim_settings, _convert_caf_to_wav, _find_ffmpeg_binary, _find_ffprobe_binary,
_find_whisper_binary, _prepare_audio_for_transcription, _prepare_local_audio,
_probe_audio_duration, _run_ffmpeg_stt_encode, _trim_silence_for_cloud_stt,
DEFAULT_STT_MODEL, LOCAL_STT_COMMAND_ENV, LOCAL_STT_LANGUAGE_ENV, _error_result,
_get_stt_section, _ok_result)
from tools.transcription_audio import (
_convert_caf_to_wav, _prepare_audio_for_transcription, _trim_silence_for_cloud_stt,
_validate_audio_file, _validate_audio_file_size, _validate_audio_source_file)
from tools.transcription_local import ( # noqa: F401 (re-exported; tests patch tools.transcription_tools.<name>)
_LOGPROB_THRESHOLD_DEFAULT, _NO_SPEECH_PROB_THRESHOLD_DEFAULT, _get_idle_unload_seconds,
_get_local_command_template, _has_local_command, _is_hallucinated_segment,
_join_confident_segments, _load_local_whisper_model, _looks_like_cuda_lib_error,
_normalize_local_model, _transcribe_local_command, _try_lazy_install_stt,
build_local_transcribe_kwargs)
from tools.transcription_cloud import ( # noqa: F401 (re-exported; tests patch tools.transcription_tools.<name>)
_extract_transcript_text, _has_xai_stt_credentials, _is_local_or_private_url,
_resolve_openai_audio_client_config, _transcribe_deepinfra, _transcribe_elevenlabs,
_transcribe_groq, _transcribe_mistral, _transcribe_openai, _transcribe_xai)
from tools.transcription_command import ( # noqa: F401 (re-exported; tests patch tools.transcription_tools.<name>)
COMMAND_STT_OUTPUT_FORMATS, DEFAULT_COMMAND_STT_LANGUAGE, DEFAULT_COMMAND_STT_OUTPUT_FORMAT,
DEFAULT_COMMAND_STT_TIMEOUT_SECONDS, _PROMPT_CHARS_PER_TOKEN, _WHISPER_PROMPT_TOKEN_CAP,
from tools.transcription_local import (
_get_idle_unload_seconds, _has_local_command, _join_confident_segments,
_load_local_whisper_model, _looks_like_cuda_lib_error, _normalize_local_model,
_transcribe_local_command, _try_lazy_install_stt, build_local_transcribe_kwargs)
# The ``_transcribe_<provider>`` handlers are looked up in this module's globals by _dispatch_stt_provider.
from tools.transcription_cloud import ( # noqa: F401 (handlers dispatched via globals())
_has_xai_stt_credentials, _resolve_openai_audio_client_config, _transcribe_deepinfra,
_transcribe_elevenlabs, _transcribe_groq, _transcribe_mistral, _transcribe_openai,
_transcribe_xai)
from tools.transcription_command import (
_apply_pre_transcription_hook, _dispatch_to_plugin_provider, _enforce_prompt_length_limit,
_get_command_stt_output_format, _get_command_stt_timeout, _get_named_stt_provider_config,
_render_command_stt_template, _resolve_command_stt_provider_config, _run_command_stt,
_transcribe_command_stt, _unregistered_stt_provider_error)
_resolve_command_stt_provider_config, _transcribe_command_stt, _unregistered_stt_provider_error)
logger = logging.getLogger(__name__)
tools/voice_client_config.py
+18 -16
View File
@@ -69,8 +69,8 @@ def _deepinfra_model(section: Dict[str, Any], kind: str) -> Optional[str]:
# ── STT ──
# provider -> (env var, default-model attr on transcription_tools, base_url).
# ``base_url`` is a transcription_tools attr name or a literal URL.
# provider -> (env var, default-model attr on transcription_common, base_url).
# ``base_url`` is a transcription_common attr name or a literal URL.
_STT_KEYED: Dict[str, tuple[str, str, str]] = {
"groq": ("GROQ_API_KEY", "DEFAULT_GROQ_STT_MODEL", "GROQ_BASE_URL"),
"mistral": ("MISTRAL_API_KEY", "DEFAULT_MISTRAL_STT_MODEL", "https://api.mistral.ai/v1"),
@@ -78,6 +78,7 @@ _STT_KEYED: Dict[str, tuple[str, str, str]] = {
def _resolve_stt_client_config() -> Dict[str, Any]:
from tools import transcription_common as tc
from tools import transcription_tools as tt
stt_config = tt._load_stt_config()
@@ -88,7 +89,7 @@ def _resolve_stt_client_config() -> Dict[str, Any]:
# declared command providers, and anything plugin-registered.
if tt._is_local_stt_provider(provider, stt_config):
return _relay("local provider")
if provider not in tt.BUILTIN_STT_PROVIDERS:
if provider not in tc.BUILTIN_STT_PROVIDERS:
return _relay("command/plugin provider")
language = tt._resolve_stt_language(
@@ -106,8 +107,8 @@ def _resolve_stt_client_config() -> Dict[str, Any]:
api_key = tt._resolve_provider_key(env_var, provider)
if not api_key:
return _relay("no credentials")
return direct(STT_WIRE_OPENAI, getattr(tt, base, base), api_key,
section.get("model") or getattr(tt, default_model))
return direct(STT_WIRE_OPENAI, getattr(tc, base, base), api_key,
section.get("model") or getattr(tc, default_model))
if provider == "openai":
# Covers the Nous-managed selection too: the resolver returns the user's
# own gateway token + managed base URL — exactly what the client should use.
@@ -115,20 +116,20 @@ def _resolve_stt_client_config() -> Dict[str, Any]:
api_key, base_url = tt._resolve_openai_audio_client_config()
except ValueError as exc:
return _relay(f"openai resolution failed: {exc}")
return direct(STT_WIRE_OPENAI, base_url, api_key, section.get("model") or tt.DEFAULT_STT_MODEL)
return direct(STT_WIRE_OPENAI, base_url, api_key, section.get("model") or tc.DEFAULT_STT_MODEL)
if provider == "xai":
# API key only: an xAI OAuth bearer refreshes server-side mid-session and
# would strand the client on the first 401.
api_key = str(tt.get_env_value("XAI_API_KEY") or "").strip()
if not api_key:
return _relay("xai oauth (server-managed) or no credentials")
return direct(STT_WIRE_XAI, env_base_url("XAI_STT_BASE_URL", tt.XAI_STT_BASE_URL), api_key, None)
return direct(STT_WIRE_XAI, env_base_url("XAI_STT_BASE_URL", tc.XAI_STT_BASE_URL), api_key, None)
if provider == "elevenlabs":
api_key = tt._resolve_provider_key("ELEVENLABS_API_KEY", "elevenlabs")
if not api_key:
return _relay("no credentials")
return direct(STT_WIRE_ELEVENLABS, env_base_url("ELEVENLABS_STT_BASE_URL", tt.ELEVENLABS_STT_BASE_URL),
api_key, section.get("model") or tt.DEFAULT_ELEVENLABS_STT_MODEL)
return direct(STT_WIRE_ELEVENLABS, env_base_url("ELEVENLABS_STT_BASE_URL", tc.ELEVENLABS_STT_BASE_URL),
api_key, section.get("model") or tc.DEFAULT_ELEVENLABS_STT_MODEL)
if provider == "deepinfra":
api_key = tt._resolve_provider_key("DEEPINFRA_API_KEY", "deepinfra")
if not api_key:
@@ -144,6 +145,7 @@ def _resolve_stt_client_config() -> Dict[str, Any]:
# ── TTS ──
def _resolve_tts_client_config() -> Dict[str, Any]:
from tools import tts_tool as tts
from tools import tts_tool_openai, tts_tool_providers
tts_config = tts._load_tts_config()
provider = tts._get_provider(tts_config)
@@ -153,24 +155,24 @@ def _resolve_tts_client_config() -> Dict[str, Any]:
if provider == "openai":
# Covers the direct-key, custom-base_url, and Nous-managed selections.
try:
api_key, base_url, is_managed = tts._resolve_openai_audio_client_config()
api_key, base_url, is_managed = tts_tool_openai._resolve_openai_audio_client_config()
except ValueError as exc:
return _relay(f"openai resolution failed: {exc}")
oai = _section(tts_config, "openai")
model = oai.get("model") or tts.DEFAULT_OPENAI_MODEL
model = oai.get("model") or tts_tool_openai.DEFAULT_OPENAI_MODEL
config_base = oai.get("base_url")
base_url = config_base or base_url
# The managed gateway only proxies MANAGED_OPENAI_TTS_MODELS — same
# coercion text_to_speech applies server-side.
if is_managed and not config_base and model not in tts.MANAGED_OPENAI_TTS_MODELS:
model = tts.DEFAULT_OPENAI_MODEL
if is_managed and not config_base and model not in tts_tool_openai.MANAGED_OPENAI_TTS_MODELS:
model = tts_tool_openai.DEFAULT_OPENAI_MODEL
speed_default = tts_config.get("speed", 1.0) if isinstance(tts_config, dict) else 1.0
try:
speed = float(oai.get("speed", speed_default))
except (TypeError, ValueError):
speed = 1.0
return _direct(TTS_WIRE_OPENAI, "openai", base_url, api_key, model,
voice=oai.get("voice") or tts.DEFAULT_OPENAI_VOICE, speed=speed)
voice=oai.get("voice") or tts_tool_openai.DEFAULT_OPENAI_VOICE, speed=speed)
if provider == "elevenlabs":
api_key = tts._resolve_provider_key("ELEVENLABS_API_KEY", "elevenlabs")
if not api_key:
@@ -178,8 +180,8 @@ def _resolve_tts_client_config() -> Dict[str, Any]:
el = _section(tts_config, "elevenlabs")
return _direct(TTS_WIRE_ELEVENLABS, "elevenlabs",
str(el.get("base_url") or "https://api.elevenlabs.io/v1").rstrip("/"),
api_key, el.get("model_id") or tts.DEFAULT_ELEVENLABS_MODEL_ID,
voice=el.get("voice_id") or tts.DEFAULT_ELEVENLABS_VOICE_ID, speed=None)
api_key, el.get("model_id") or tts_tool_providers.DEFAULT_ELEVENLABS_MODEL_ID,
voice=el.get("voice_id") or tts_tool_providers.DEFAULT_ELEVENLABS_VOICE_ID, speed=None)
if provider == "deepinfra":
api_key = tts._resolve_provider_key("DEEPINFRA_API_KEY", "deepinfra")
if not api_key:
tools/voice_mode.py
+2 -1
View File
@@ -848,7 +848,8 @@ def create_audio_recorder() -> AudioRecorder | TermuxAudioRecorder:
def transcribe_recording(wav_path: str, model: Optional[str] = None) -> Dict[str, Any]:
"""Transcribe a WAV via ``transcribe_audio()``, filtering Whisper hallucinations;
returns ``{success, transcript[, error]}``."""
from tools.transcription_tools import MAX_FILE_SIZE, transcribe_audio
from tools.transcription_common import MAX_FILE_SIZE
from tools.transcription_tools import transcribe_audio
result = transcribe_audio(wav_path, model=model, source="voice_mode")
# Only chunk when the provider itself reports "File too large" — local