TheMajesticMagicianandClaude Code
751d1c8b69
The client never called /desk/report_status, so Bolt's reminders and any files he delivered to this session sat in the server's outbox forever - the same dead-air failure the pet client showed when its heartbeat stopped. Now a heartbeat thread (HEARTBEAT_SECONDS, default 60; 0 disables) reports in, speaks announcements through the assistant's queue when the mic is idle, and downloads the outbox to BOLT_DOWNLOAD_DIR (default ~/Downloads/Bolt) via .part-then-rename, never overwriting. A startup health probe warns loudly when the server refuses the DESK_API_KEY. Also: add the .env.example the README's setup step references (it was missing, so `cp .env.example .env` failed for everyone), and gitignore .claude/ so Claude Code's worktree directory stays out of the repo. Tests: tests/test_heartbeat.py (12) plus announcement tests in tests/test_assistant.py (6); full suite 135 passed. Co-Authored-By: Claude Code <noreply@anthropic.com>
133 lines
6.6 KiB
Python
133 lines
6.6 KiB
Python
"""Settings for Bolt Assistant, from a `.env` next to the project (or the
|
|
real environment, which wins). Everything else reads config from here."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import platform
|
|
from pathlib import Path
|
|
|
|
HERE = Path(__file__).resolve().parent.parent # project root
|
|
|
|
|
|
def _load_env() -> None:
|
|
env_path = HERE / ".env"
|
|
if not env_path.exists():
|
|
return
|
|
for line in env_path.read_text(encoding="utf-8").splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, _, value = line.partition("=")
|
|
os.environ.setdefault(key.strip(), value.strip())
|
|
|
|
|
|
_load_env()
|
|
|
|
|
|
def _flag(name: str, default: str) -> bool:
|
|
return os.environ.get(name, default).strip().lower() in ("1", "true", "yes", "on")
|
|
|
|
|
|
def _device(raw: str):
|
|
"""An input/output device: blank = system default, digits = an index,
|
|
anything else = a name substring (sounddevice's rules)."""
|
|
raw = (raw or "").strip()
|
|
if not raw:
|
|
return None
|
|
return int(raw) if raw.lstrip("-").isdigit() else raw
|
|
|
|
|
|
# ── server ──────────────────────────────────────────────────────────────────
|
|
|
|
SERVER_URL = os.environ.get("BOLT_SERVER_URL", "").rstrip("/")
|
|
API_KEY = os.environ.get("DESK_API_KEY", "")
|
|
# Its own conversation, separate from the pet's and the console's. The "pet-"
|
|
# prefix gets the desk API's voice-client behaviour (short, speakable replies).
|
|
SESSION_ID = os.environ.get("DESK_SESSION_ID") or "pet-assistant-" + (platform.node() or "desktop")
|
|
# The voice the server's /desk/tts speaks in (an ElevenLabs voice id).
|
|
VOICE_ID = os.environ.get("ELEVENLABS_VOICE_ID", "")
|
|
TTS_SAMPLE_RATE = int(os.environ.get("TTS_SAMPLE_RATE", "16000"))
|
|
STREAMING_REPLIES = _flag("STREAMING_REPLIES", "true")
|
|
|
|
# ── heartbeat: reminders and delivered files ─────────────────────────────────
|
|
# Every HEARTBEAT_SECONDS the assistant reports in to /desk/report_status —
|
|
# the channel Bolt's reminders and any files he delivers to this client
|
|
# arrive on. Announcements are spoken when the mic is idle (see
|
|
# assistant.announce), and files download to DOWNLOAD_DIR without
|
|
# overwriting what is already there. Set HEARTBEAT_SECONDS to 0 to turn the
|
|
# heartbeat off entirely.
|
|
HEARTBEAT_SECONDS = max(0, int(os.environ.get("HEARTBEAT_SECONDS", "60")))
|
|
DOWNLOAD_DIR = Path(os.environ.get("BOLT_DOWNLOAD_DIR", "").strip()
|
|
or str(Path.home() / "Downloads" / "Bolt"))
|
|
|
|
# ── audio devices ───────────────────────────────────────────────────────────
|
|
|
|
MIC_DEVICE = _device(os.environ.get("MIC_DEVICE", ""))
|
|
SPEAKER_DEVICE = _device(os.environ.get("SPEAKER_DEVICE", ""))
|
|
|
|
# ── wake phrases ────────────────────────────────────────────────────────────
|
|
# With REQUIRE_WAKE_WORD off (the default) the mic itself is the activation:
|
|
# every utterance the local detector passes on is treated as a request, and
|
|
# whatever is said near an open mic gets answered. Set it to true to go back
|
|
# to needing a phrase before each request.
|
|
REQUIRE_WAKE_WORD = _flag("REQUIRE_WAKE_WORD", "false")
|
|
|
|
# Heard through the server's speech-to-text, so these are matched against
|
|
# *text*. A phrase counts only near the start of what was said ("hey bolt,
|
|
# what's the time"), not buried in a sentence ("tighten that bolt"). Speech
|
|
# recognisers often write "Bolt" as "bolts"/"volt"/"bold"; those spellings
|
|
# only count after "hey"/"ok", where they can't be ordinary words. Only used
|
|
# when REQUIRE_WAKE_WORD is on (a phrase said anyway is still stripped).
|
|
|
|
WAKE_WORDS = [w.strip().lower() for w in os.environ.get(
|
|
"WAKE_WORDS",
|
|
"hey bolt,bolt,thunderbolt,hey thunderbolt,ok bolt,okay bolt,yo bolt,hi bolt,"
|
|
"hey assistant,ok assistant,okay assistant,"
|
|
"hey bolts,hey volt,hey bold,hey boat,okay volt,hey thunder bolt,thunder bolt",
|
|
).split(",") if w.strip()]
|
|
# These count anywhere in what was said, not just at the start ("what's the
|
|
# weather, assistant?"), and the whole sentence is sent as the request.
|
|
ANYWHERE_WAKE_WORDS = [w.strip().lower() for w in os.environ.get(
|
|
"ANYWHERE_WAKE_WORDS", "assistant").split(",") if w.strip()]
|
|
|
|
# ── listening (local voice-activity detection, before anything is sent) ────
|
|
# Nothing leaves the machine until this says someone is talking: silence and
|
|
# background hum are never sent (or billed). Raise VAD_RMS_THRESHOLD if
|
|
# background noise keeps getting transcribed; lower it if you have to shout.
|
|
|
|
VAD_RMS_THRESHOLD = int(os.environ.get("VAD_RMS_THRESHOLD", "500"))
|
|
VAD_START_FRAMES = int(os.environ.get("VAD_START_FRAMES", "3")) # 80 ms each
|
|
VAD_SILENCE_END_SEC = float(os.environ.get("VAD_SILENCE_END_SEC", "0.8"))
|
|
MIN_UTTERANCE_SEC = float(os.environ.get("MIN_UTTERANCE_SEC", "0.35")) # shorter: never sent
|
|
MAX_UTTERANCE_SEC = float(os.environ.get("MAX_UTTERANCE_SEC", "12"))
|
|
# After a bare "Hey Bolt", how long to wait for the actual request.
|
|
COMMAND_WAIT_SEC = float(os.environ.get("COMMAND_WAIT_SEC", "8"))
|
|
# When a reply ends on a question, keep listening for the answer without the
|
|
# wake word, for this long, at most this many times in a row.
|
|
FOLLOW_UP_SEC = float(os.environ.get("FOLLOW_UP_SEC", "8"))
|
|
FOLLOW_UP_MAX_TURNS = int(os.environ.get("FOLLOW_UP_MAX_TURNS", "5"))
|
|
CHIMES = _flag("CHIMES", "true")
|
|
|
|
# ── what Bolt may do on this machine ────────────────────────────────────────
|
|
# The server can relay commands back. With no screen to approve them on:
|
|
# off nothing runs; Bolt is told it can't here
|
|
# read (default) listing and reading files only, via filectl
|
|
# all everything, including shell commands and file writes — like
|
|
# Bolt-Pet. Only if you trust every request that reaches your mic.
|
|
|
|
COMMANDS = os.environ.get("ASSISTANT_COMMANDS", "read").strip().lower()
|
|
COMMAND_TIMEOUT_SECONDS = int(os.environ.get("COMMAND_TIMEOUT_SECONDS", "60"))
|
|
|
|
SAMPLE_RATE = 16000
|
|
FRAME_SAMPLES = 1280 # 80 ms
|
|
|
|
|
|
def is_configured() -> bool:
|
|
return bool(SERVER_URL and API_KEY)
|
|
|
|
|
|
def missing_config() -> list:
|
|
return [name for name, value in (("BOLT_SERVER_URL", SERVER_URL), ("DESK_API_KEY", API_KEY),
|
|
("ELEVENLABS_VOICE_ID", VOICE_ID)) if not value]
|