120 lines
3.9 KiB
Python
120 lines
3.9 KiB
Python
"""Bolt Assistant: talk while the mic is open, hear the answer. (Setting
|
|
REQUIRE_WAKE_WORD=true puts "Hey Bolt" back in front of each request.)
|
|
|
|
python -m bolt_assistant listen on the default mic
|
|
python -m bolt_assistant --verbose also show what was heard but ignored
|
|
python -m bolt_assistant --type type requests instead of speaking (testing)
|
|
python -m bolt_assistant --say "hello" speak a line through the speaker and exit
|
|
python -m bolt_assistant --devices list microphones and speakers
|
|
python -m bolt_assistant --check check settings, server, mic and voice
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import threading
|
|
|
|
from . import __version__, audio, config, server
|
|
|
|
|
|
def _devices() -> int:
|
|
import sounddevice as sd
|
|
|
|
print(sd.query_devices())
|
|
print("\nDefault input/output:", sd.default.device,
|
|
"\nSet MIC_DEVICE / SPEAKER_DEVICE in .env to an index or a name to choose another.")
|
|
return 0
|
|
|
|
|
|
def _check() -> int:
|
|
from .stt import connect
|
|
|
|
ok = True
|
|
|
|
def row(name, good, detail):
|
|
nonlocal ok
|
|
ok = ok and good
|
|
print(f"[{' ok ' if good else 'FAIL'}] {name:12} {detail}")
|
|
|
|
missing = config.missing_config()
|
|
row("settings", not missing, "all set" if not missing else "missing " + ", ".join(missing))
|
|
try:
|
|
server.check_health()
|
|
row("server", True, config.SERVER_URL)
|
|
except Exception as exc:
|
|
row("server", False, f"{config.SERVER_URL}: {exc}")
|
|
try:
|
|
import sounddevice as sd
|
|
|
|
mic = sd.query_devices(config.MIC_DEVICE, "input")["name"]
|
|
out = sd.query_devices(config.SPEAKER_DEVICE, "output")["name"]
|
|
row("microphone", True, mic)
|
|
row("speaker", True, out)
|
|
except Exception as exc:
|
|
row("audio", False, f"{exc} (pip install sounddevice; Linux: libportaudio2)")
|
|
try:
|
|
connect().close()
|
|
row("speech-to-text", True, "/desk/stt reachable")
|
|
except Exception as exc:
|
|
row("speech-to-text", False, f"{exc} (pip install websocket-client)")
|
|
row("commands", True, f"ASSISTANT_COMMANDS={config.COMMANDS}")
|
|
return 0 if ok else 1
|
|
|
|
|
|
def main(argv=None) -> int:
|
|
argv = list(sys.argv[1:] if argv is None else argv)
|
|
for stream in (sys.stdout, sys.stderr):
|
|
try: # replies carry dashes and curly quotes; a cp1252 console must not crash on them
|
|
stream.reconfigure(errors="replace")
|
|
except Exception:
|
|
pass
|
|
if "-h" in argv or "--help" in argv:
|
|
print(__doc__.strip())
|
|
return 0
|
|
if "--devices" in argv:
|
|
return _devices()
|
|
if "--check" in argv:
|
|
return _check()
|
|
missing = config.missing_config()
|
|
if missing:
|
|
print(f"Missing {', '.join(missing)}. Copy .env.example to .env and fill them in.")
|
|
return 2
|
|
|
|
from .assistant import Assistant
|
|
|
|
assistant = Assistant(verbose="--verbose" in argv)
|
|
print(f"Bolt Assistant {__version__} | {config.SERVER_URL} | session {config.SESSION_ID}")
|
|
|
|
if "--say" in argv:
|
|
assistant.speak(" ".join(argv[argv.index("--say") + 1:]) or "Hello, I'm Bolt.")
|
|
return 0
|
|
|
|
if "--type" in argv: # the whole pipeline minus the microphone
|
|
hint = " (including the wake phrase)" if assistant.wake_required else ""
|
|
print(f"Type what you'd say{hint}. Ctrl+C to quit.")
|
|
try:
|
|
while True:
|
|
assistant.heard(input("say> "))
|
|
except (EOFError, KeyboardInterrupt):
|
|
return 0
|
|
|
|
stop = threading.Event()
|
|
try:
|
|
mic = audio.open_mic(assistant.on_mic_frame)
|
|
except Exception as exc:
|
|
print(f"Couldn't open the microphone: {exc}\nTry --devices, and MIC_DEVICE in .env.")
|
|
return 1
|
|
try:
|
|
assistant.run(stop)
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
stop.set()
|
|
mic.stop()
|
|
mic.close()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|