Bolt-Pet/bolt_pet/config.py

"""Environment configuration for the Bolt desktop pet.

Same lightweight ".env next to the script" pattern as desk_client/bolt_desk.py
in the main tmn-api repo, so this project can be copied anywhere (it does not
import anything from that repo) and configured the same way.
"""

from __future__ import annotations

import os
import platform
from pathlib import Path

HERE = Path(__file__).resolve().parent.parent  # project root (one above bolt_pet/)


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 _node_name() -> str:
    try:
        return platform.node() or "desktop"
    except Exception:
        return "desktop"


# ── server / identity ───────────────────────────────────────────────────────

SERVER_URL = os.environ.get("BOLT_SERVER_URL", "").rstrip("/")
API_KEY = os.environ.get("DESK_API_KEY", "")
SESSION_ID = os.environ.get("DESK_SESSION_ID", "pet-" + _node_name())

# ── wake word ────────────────────────────────────────────────────────────────
# openWakeWord model trained specifically for "thunderbolt" — same pattern as
# desk_client/bolt_desk.py's WAKE_MODEL_FILE (bolt.onnx / "hey bolt") in the
# main repo, resolved relative to the project root so it sits next to .env.

WAKE_MODEL_PATH = str(HERE / os.environ.get("WAKE_MODEL_FILE", "thunderbolt.onnx"))
WAKE_WORD_THRESHOLD = float(os.environ.get("WAKE_WORD_THRESHOLD", "0.5"))
# How often (independent of prediction, which runs every frame) the wake
# listener yields control back to its caller via on_tick — e.g. the
# heartbeat poll in controller.py during quiet stretches with no wake word.
WAKE_CHECK_INTERVAL_SECONDS = float(os.environ.get("WAKE_CHECK_INTERVAL_SECONDS", "1.2"))

# ── STT (Deepgram, same as bolt_desk.py) ────────────────────────────────────

DEEPGRAM_API_KEY = os.environ.get("DEEPGRAM_API_KEY", "")
DEEPGRAM_MODEL = os.environ.get("DEEPGRAM_MODEL", "nova-3")

# ── TTS (ElevenLabs, requested as raw PCM so playback needs no external
# player binary — cross-platform via sounddevice instead of shelling out to
# mpv/ffplay like bolt_desk.py does on Linux) ───────────────────────────────

ELEVENLABS_API_KEY = os.environ.get("ELEVENLABS_API_KEY", "")
ELEVENLABS_VOICE_ID = os.environ.get("ELEVENLABS_VOICE_ID", "")
ELEVENLABS_MODEL_ID = os.environ.get("ELEVENLABS_MODEL_ID", "eleven_flash_v2")
# eleven_flash_v2 is English-only, and the two cases that swap the voice
# (server-picked `speak_as`, or a reply with non-ASCII in it) are usually
# exactly the cases where the reply isn't English — see tts.model_for().
ELEVENLABS_MULTILINGUAL_MODEL_ID = os.environ.get(
    "ELEVENLABS_MULTILINGUAL_MODEL_ID", "eleven_flash_v2_5"
)
# ElevenLabs PCM output formats are named pcm_<sample_rate>.
TTS_SAMPLE_RATE = int(os.environ.get("TTS_SAMPLE_RATE", "24000"))

# Does a voice the server picks (its speak_as marker — "talk like a pirate",
# "say that in Japanese") stay on for later replies, or last one reply only?
# Sticky by default: the server tags a single reply and does *not* keep the
# voice id in its history, so a one-reply-only voice can't be re-used when
# you say "keep talking like that" — it would have to search for a voice
# again. Reset it from the tray ("Use default voice") or by restarting.
VOICE_STICKY = os.environ.get("VOICE_STICKY", "true").lower() in ("1", "true", "yes", "on")

# ── multi-voice dialogue (ElevenLabs Text to Dialogue) ──────────────────────
# Lets Bolt play a short scene in several voices with delivery tags the v3
# model acts on ("[cheerfully] Hello"), instead of one voice reading a line.
# Driven by the server through the `dialoguectl` relayed command — see
# dialogue.py. Costs a separate (slower, whole-clip) request per scene, so
# it's a set piece, not the normal reply path.
#
# DIALOGUE_VOICES names the cast: "narrator:9BWtsMINqrJLrRacOk9x,villain:IKne3meq5aSn9XLyUdCD".
# The name "self" always resolves to the voice the pet is currently using,
# including one the server picked with speak_as.

DIALOGUE = os.environ.get("DIALOGUE", "true").lower() in ("1", "true", "yes", "on")
DIALOGUE_MODEL_ID = os.environ.get("DIALOGUE_MODEL_ID", "eleven_v3")
DIALOGUE_VOICES = os.environ.get("DIALOGUE_VOICES", "")

# ── mic / VAD (same tuning knobs as bolt_desk.py) ───────────────────────────

MIC_DEVICE = os.environ.get("MIC_DEVICE", "") or None  # sounddevice name/index
SPEAKER_DEVICE = os.environ.get("SPEAKER_DEVICE", "") or None

RMS_THRESHOLD = int(os.environ.get("VAD_RMS_THRESHOLD", "300"))
SILENCE_END_SEC = float(os.environ.get("VAD_SILENCE_END_SEC", "1.2"))
MAX_UTTERANCE_S = float(os.environ.get("VAD_MAX_UTTERANCE_SECONDS", "15"))
MIN_UTTERANCE_S = float(os.environ.get("VAD_MIN_UTTERANCE_SECONDS", "0.4"))
# How long to wait for you to *start* talking before giving up on a turn.
GRACE_SECONDS = float(os.environ.get("VAD_GRACE_SECONDS", "4"))

# ── follow-up listening ─────────────────────────────────────────────────────
# When a reply ends on a question, the pet keeps listening for the answer
# instead of dropping back to idle and making you say the wake word again.
# The grace period is longer than a normal turn's because you were asked
# something and may need a beat to think. FOLLOW_UP_MAX_TURNS caps how many
# question-and-answer rounds can chain without you re-triggering it — a stop
# on runaway loops if the server ends every reply with a question and the mic
# keeps feeding it noise. 0 means no cap.

FOLLOW_UP_LISTEN = os.environ.get("FOLLOW_UP_LISTEN", "true").lower() in ("1", "true", "yes", "on")
FOLLOW_UP_MAX_TURNS = int(os.environ.get("FOLLOW_UP_MAX_TURNS", "10"))
FOLLOW_UP_GRACE_SECONDS = float(os.environ.get("FOLLOW_UP_GRACE_SECONDS", "7"))

# ── local intents ───────────────────────────────────────────────────────────
# A short, closed list of utterances the pet answers itself instead of paying a
# server round trip for: "stop", "come here", "go to sleep", "say that again",
# "use your normal voice". Matched whole and exact (see intents.py), never
# during a follow-up turn, so a real request is never swallowed. Turn it off to
# route absolutely everything through Bolt.
LOCAL_INTENTS = os.environ.get("LOCAL_INTENTS", "true").lower() in ("1", "true", "yes", "on")

COMMAND_TIMEOUT_SECONDS = int(os.environ.get("COMMAND_TIMEOUT_SECONDS", "30"))

# ── sudo password prompts ───────────────────────────────────────────────────
# The pet has no terminal, so a relayed `sudo` would block on a tty nobody is
# watching. With this on, bare `sudo` is rewritten to `sudo -A` and the
# password is collected in a desktop dialog (a real askpass binary if one is
# installed, otherwise a generated zenity/kdialog wrapper). Turn it off and
# sudo commands simply fail, which is the safer default if you'd rather Bolt
# never be able to ask for root at all.
SUDO_ASKPASS_PROMPT = os.environ.get("SUDO_ASKPASS_PROMPT", "true").lower() in ("1", "true", "yes", "on")
SUDO_ASKPASS_HELPER = os.environ.get("SUDO_ASKPASS_HELPER", "")  # blank = auto-detect
# Longer than COMMAND_TIMEOUT_SECONDS because a person has to notice the
# dialog, read it, and type — 30s is nowhere near enough for that.
SUDO_COMMAND_TIMEOUT_SECONDS = int(os.environ.get("SUDO_COMMAND_TIMEOUT_SECONDS", "180"))
HEARTBEAT_INTERVAL_SECONDS = float(os.environ.get("HEARTBEAT_INTERVAL_SECONDS", "60"))

# ── self-restart ────────────────────────────────────────────────────────────
# `petctl self_restart` lets Bolt restart the pet after editing its code, so
# he can see his own change running instead of waiting for someone to restart
# it by hand. The code is import-checked in a subprocess first, and the reason
# is carried across the restart so the new process can report back — see
# self_restart.py. SELF_RESTART_MAX/_WINDOW_SECONDS bound the crash-loop case.
SELF_RESTART = os.environ.get("SELF_RESTART", "true").lower() in ("1", "true", "yes", "on")

# ── barge-in (interrupt playback while the pet is talking) ──────────────────
# The mic stays live while the pet talks. BARGE_IN_MODE decides what counts
# as an interruption:
#   wake   — only the wake word cuts playback (default). Immune to coughs,
#            doors, and the TV, at the cost of ~a word of extra latency.
#   energy — any sustained noise above BARGE_IN_RMS_THRESHOLD does. Faster,
#            but interrupts on background noise. That threshold is well above
#            VAD_RMS_THRESHOLD because the mic also hears the pet's own voice
#            through the speakers — raise it further if playback keeps
#            interrupting itself.
# Set BARGE_IN=false to make playback uninterruptible either way.

BARGE_IN = os.environ.get("BARGE_IN", "true").lower() in ("1", "true", "yes", "on")
BARGE_IN_MODE = os.environ.get("BARGE_IN_MODE", "wake")
BARGE_IN_RMS_THRESHOLD = int(os.environ.get("BARGE_IN_RMS_THRESHOLD", str(RMS_THRESHOLD * 4)))
BARGE_IN_FRAMES = int(os.environ.get("BARGE_IN_FRAMES", "4"))  # consecutive loud frames (80ms each)
# Wake-mode sensitivity. Blank means "track the live WAKE_WORD_THRESHOLD from
# the tray tuner"; set a number to make interrupting deliberately harder than
# waking the pet from idle (useful if Bolt's own voice trips the model).
BARGE_IN_WAKE_THRESHOLD = float(os.environ.get("BARGE_IN_WAKE_THRESHOLD") or 0) or None

# ── streaming TTS ───────────────────────────────────────────────────────────
# ElevenLabs' /stream endpoint + chunked playback: the pet starts talking
# after the first PCM chunk instead of after the whole clip is synthesized.

TTS_STREAMING = os.environ.get("TTS_STREAMING", "true").lower() in ("1", "true", "yes", "on")

# ── screen context ──────────────────────────────────────────────────────────
# Appends the active window's title to what you say, so "what's this error?"
# has a referent. The desk API takes text only, so this is a text annotation
# (no screenshot upload).

SCREEN_CONTEXT = os.environ.get("SCREEN_CONTEXT", "true").lower() in ("1", "true", "yes", "on")

# ── monitors ────────────────────────────────────────────────────────────────
# A one-line note about the screen layout (how many, their sizes, which one
# the pet is standing on) rides along with each utterance, so Bolt can decide
# to `petctl jump` somewhere without asking you what you've got plugged in.
# Cheap — the list comes from the UI, nothing is probed per turn.

MONITOR_CONTEXT = os.environ.get("MONITOR_CONTEXT", "true").lower() in (
    "1", "true", "yes", "on"
)

# ── screen text (OCR) ───────────────────────────────────────────────────────
# Lets Bolt actually read a monitor, via `petctl read`. Pull-only: nothing is
# captured unless the server asks for it, and every read is logged. Needs the
# optional capture/OCR extras — see the comments in requirements.txt.
#
# This widens what can leave the machine more than any other switch here: the
# recognised text of a whole screen goes to the server. It is *not* a new
# capability (the shell relay could already run a screenshot tool and OCR it),
# but it is a much easier one to use by accident. Set SCREEN_TEXT=false to
# take it away entirely.

SCREEN_TEXT = os.environ.get("SCREEN_TEXT", "true").lower() in ("1", "true", "yes", "on")
SCREEN_TEXT_MAX_CHARS = int(os.environ.get("SCREEN_TEXT_MAX_CHARS", "4000"))

# ── quiet hours / do-not-disturb ────────────────────────────────────────────
# Comma-separated HH:MM-HH:MM ranges (wrapping midnight is fine). While
# napping the pet dims, stops wandering, and makes no proactive noise —
# wake word and click-to-talk still work.

QUIET_HOURS = os.environ.get("QUIET_HOURS", "")
DND_ON_FULLSCREEN = os.environ.get("DND_ON_FULLSCREEN", "true").lower() in ("1", "true", "yes", "on")

# ── desktop notification bridge ─────────────────────────────────────────────
# Mirrors desktop notifications to the server so Bolt can react to them.
# Off by default: every forwarded notification costs a converse() round trip.

NOTIFICATION_BRIDGE = os.environ.get("NOTIFICATION_BRIDGE", "false").lower() in ("1", "true", "yes", "on")
# Regex matched against "<app>: <summary> <body>"; empty means "everything".
NOTIFICATION_FILTER = os.environ.get("NOTIFICATION_FILTER", "")
NOTIFICATION_MIN_INTERVAL_SECONDS = float(os.environ.get("NOTIFICATION_MIN_INTERVAL_SECONDS", "60"))
# Notifications arrive on the watcher thread and are forwarded from the
# heartbeat, which doesn't run while the pet is napping — so they queue. Both
# limits exist to stop an overnight backlog turning into a burst of round trips
# and a monologue at 8am: the queue is bounded (oldest dropped first) and
# anything staler than the age limit is discarded at drain time, because
# "Firefox finished downloading" is not news nine hours later.
NOTIFICATION_QUEUE_LIMIT = int(os.environ.get("NOTIFICATION_QUEUE_LIMIT", "20"))
NOTIFICATION_MAX_AGE_SECONDS = float(os.environ.get("NOTIFICATION_MAX_AGE_SECONDS", "900"))

# ── file delivery ────────────────────────────────────────────────────────
# The server's deliver_files tool (ai/desk_api.py in the main tmn-api repo)
# queues workspace files on this session — e.g. "send me that report" — for
# the client to fetch via GET /desk/files. Downloading a file dequeues it
# server-side, so each one lands here exactly once.

RECEIVE_FILES = os.environ.get("RECEIVE_FILES", "true").lower() in ("1", "true", "yes", "on")
DELIVERED_FILES_DIR = Path(
    os.environ.get("DELIVERED_FILES_DIR") or str(Path.home() / "Downloads" / "Bolt")
).expanduser()

# ── conversation history ────────────────────────────────────────────────────

HISTORY_LIMIT = int(os.environ.get("HISTORY_LIMIT", "100"))

# ── push-to-talk ────────────────────────────────────────────────────────────
# Global hotkey (needs `pynput`; unavailable on most Wayland sessions, in
# which case it logs once and the wake word / tray still work). Empty to
# disable.

PUSH_TO_TALK_HOTKEY = os.environ.get("PUSH_TO_TALK_HOTKEY", "ctrl+alt+space")

# ── wake-word tuning ────────────────────────────────────────────────────────
# Scores this far below the threshold are recorded as "near misses" and shown
# in the tray's wake-word tuner, so the threshold can be set from evidence.

WAKE_NEAR_MISS_MARGIN = float(os.environ.get("WAKE_NEAR_MISS_MARGIN", "0.2"))
WAKE_NEAR_MISS_LIMIT = int(os.environ.get("WAKE_NEAR_MISS_LIMIT", "40"))

# ── auto-update ─────────────────────────────────────────────────────────────
# Watches the Gitea releases API for a tag newer than bolt_pet.__version__,
# then `git checkout`s it in place and restarts (see updater.py). The install
# has to be a git clone with a clean working tree — a dirty tree is skipped
# rather than stashed, so local edits are never thrown away. Any failure
# after checkout rolls back to the ref that was checked out before.

AUTO_UPDATE = os.environ.get("AUTO_UPDATE", "true").lower() in ("1", "true", "yes", "on")
UPDATE_REPO_API = os.environ.get(
    "UPDATE_REPO_API",
    "https://git.themajesticnetwork.com/api/v1/repos/TheMajesticNetwork/Bolt-Pet",
).rstrip("/")
UPDATE_CHECK_INTERVAL_SECONDS = float(os.environ.get("UPDATE_CHECK_INTERVAL_SECONDS", "3600"))
UPDATE_GIT_REMOTE = os.environ.get("UPDATE_GIT_REMOTE", "origin")
# Only needed if the repo is private — releases on a public repo read fine
# anonymously. A Gitea access token with read:repository.
UPDATE_TOKEN = os.environ.get("UPDATE_TOKEN", "")
# Reinstall requirements.txt when an update changes it. Off means a release
# that adds a dependency will roll straight back on the import smoke test.
UPDATE_INSTALL_DEPS = os.environ.get("UPDATE_INSTALL_DEPS", "true").lower() in ("1", "true", "yes", "on")

SAMPLE_RATE = 16000  # mic capture / STT rate
FRAME_LEN = 1280  # 80ms @ 16kHz — matches bolt_desk.py's chunking

# ── pet window ───────────────────────────────────────────────────────────────

PET_SIZE = int(os.environ.get("PET_SIZE", "160"))  # on-screen pixel size (square)
PET_START_X = os.environ.get("PET_START_X", "")  # blank = bottom-right of primary screen
PET_START_Y = os.environ.get("PET_START_Y", "")
PET_ALWAYS_ON_TOP = os.environ.get("PET_ALWAYS_ON_TOP", "true").lower() in ("1", "true", "yes", "on")
IDLE_ANIMATION_FPS = float(os.environ.get("IDLE_ANIMATION_FPS", "6"))

# ── wandering ────────────────────────────────────────────────────────────────
# The pet strolls to a random spot on its own while idle. Only ever moves when
# it's IDLE and not speaking/being dragged, so it never walks out from under a
# speech bubble mid-sentence.

PET_WANDER = os.environ.get("PET_WANDER", "true").lower() in ("1", "true", "yes", "on")
# Average seconds of standing still between strolls (each wait is randomized
# to 0.5x–1.5x this, so the pet doesn't move on an obvious metronome).
PET_WANDER_INTERVAL_SECONDS = float(os.environ.get("PET_WANDER_INTERVAL_SECONDS", "45"))
PET_WANDER_SPEED = float(os.environ.get("PET_WANDER_SPEED", "90"))  # pixels/second
# Cap on how far one stroll can be, so it doesn't teleport across a 4K screen.
PET_WANDER_MAX_DISTANCE = float(os.environ.get("PET_WANDER_MAX_DISTANCE", "600"))
PET_WANDER_MARGIN = int(os.environ.get("PET_WANDER_MARGIN", "20"))  # keep off screen edges

# ── mouse behaviour ─────────────────────────────────────────────────────────
# The sprite is drawn in a square translucent window, so its transparent
# corners would otherwise swallow clicks meant for whatever is underneath.
# PET_SHAPED_INPUT masks the window's input region to the sprite's own opaque
# pixels; PET_CLICK_THROUGH goes further and makes the whole pet ignore the
# mouse (tray-only control until you turn it back off).

PET_SHAPED_INPUT = os.environ.get("PET_SHAPED_INPUT", "true").lower() in ("1", "true", "yes", "on")
PET_CLICK_THROUGH = os.environ.get("PET_CLICK_THROUGH", "false").lower() in ("1", "true", "yes", "on")
# Snap flush to a screen edge when dropped/parked within this many pixels of it.
PET_EDGE_SNAP = os.environ.get("PET_EDGE_SNAP", "true").lower() in ("1", "true", "yes", "on")
PET_SNAP_MARGIN = int(os.environ.get("PET_SNAP_MARGIN", "48"))


def is_configured() -> bool:
    return bool(SERVER_URL and API_KEY)


def missing_config() -> list[str]:
    missing = []
    if not SERVER_URL:
        missing.append("BOLT_SERVER_URL")
    if not API_KEY:
        missing.append("DESK_API_KEY")
    return missing