Lesson 12: Talking to BonicBot

python3 -m venv bonicbot-env
source bonicbot-env/bin/activate   # On Windows: bonicbot-env\Scripts\activate
pip install bonicbot-bridge opencv-python numpy ollama speechrecognition piper-tts faster-whisper pyaudio

import difflib
import random
import re
import shutil
import subprocess
import sys
import threading
import time
from pathlib import Path
 
import numpy as np
import ollama
import pyaudio
import speech_recognition as sr
from faster_whisper import WhisperModel
from piper import PiperVoice, SynthesisConfig
from piper.download_voices import download_voice
 
# NOTE: this assumes your BonicBot control class lives here and exposes
# move_left_arm() / move_right_arm() / set_neck() / look_left() / etc. as
# described. Swap this import (and the `bot = BonicBot()` line in main())
# for whatever actually creates your `bot` object if it's different.
from bonicbot_bridge import BonicBot
 
 
# ============================================================================
# CONFIG
# ============================================================================
OLLAMA_MODEL = "qwen3:0.6b"  # small (~400MB), fast on CPU, good for a first LLM
HOST = "[IP_ADDRESS]"
 
PIPER_VOICE = "en_US-lessac-medium"  # ~60MB, natural-sounding, good default
# Other voices worth trying (just change the string above, everything else
# auto-downloads the new one). Listen before you pick:
# https://rhasspy.github.io/piper-samples/
#   "en_US-amy-medium"   - US English, female
#   "en_GB-alba-medium"  - British English, female
#   "en_US-lessac-low"   - smaller & faster to synthesize, more robotic
PIPER_VOICES_DIR = Path(__file__).resolve().parent / "piper_voices"
PIPER_LENGTH_SCALE = 0.8  # > 1 = slower speech, < 1 = faster (1.0 = voice's own pace)
PIPER_VOLUME = 1.0  # > 1 = louder, < 1 = quieter
PIPER_USE_CUDA = False  # set True if you have an NVIDIA GPU + onnxruntime-gpu
 
WHISPER_MODEL_SIZE = "base.en"  # tiny.en (fastest) / base.en (balanced) / small.en (most accurate, slower)
WHISPER_DEVICE = "cpu"
WHISPER_COMPUTE_TYPE = "int8"
WHISPER_SAMPLE_RATE = 16000  # matches what Whisper expects
WHISPER_BEAM_SIZE = 1  # 1 = greedy decoding, noticeably faster than beam search;
# bump back up to 3-5 only if you start seeing more misheard words than you can live with
 
SERVER_START_TIMEOUT = 15.0  # seconds to wait for the Ollama server to come up
SERVER_POLL_INTERVAL = 0.5
 
LISTEN_TIMEOUT = 60.0  # seconds to wait for the person to start speaking
PHRASE_TIME_LIMIT = 30.0  # max seconds for a single thing they say
MAX_HISTORY_MESSAGES = 13  # system message + ~6 back-and-forth exchanges
 
PAUSE_THRESHOLD = 1.2  # seconds of silence before a phrase is considered done —
# this is the biggest lever for perceived lag: every turn waits this long
# after you stop talking before transcription even starts. Lower = snappier
# but risks cutting someone off if they pause mid-sentence to think; 0.8-1.5
# is a reasonable range to try. The old 3.0 meant 3 full seconds of dead air
# on every single turn.
NON_SPEAKING_DURATION = 0.5  # silence padding kept on either side of speech
DYNAMIC_ENERGY_THRESHOLD = True  # auto-adapt to room noise instead of a fixed cutoff
AMBIENT_NOISE_CALIBRATION_DURATION = 2.0
 
REPEAT_SIMILARITY_THRESHOLD = 0.85
 
EXIT_WORDS = {"goodbye", "bye", "stop", "exit", "quit"}
 
# ----------------------------------------------------------------------
# GESTURE CONFIG
# ----------------------------------------------------------------------
GESTURES_ENABLED = True  # set False to run with no robot body attached
 
# Left/right arm positions, in degrees, within the joint limits you gave:
#   shoulder pitch -45–180, elbow 0–50
ARM_REST_SHOULDER = 0    # resting position for both arms
ARM_REST_ELBOW = 0
 
# "Explaining" pose: both arms move here, once, and HOLD for the whole
# reply while BonicBot talks — no elbow waving. (The elbow servo doesn't
# track commanded positions on this rig right now — see diagnostic notes —
# so it's been dropped rather than animate something that doesn't move.)
ARM_FORWARD_SHOULDER = 40
ARM_FORWARD_ELBOW = 20
 
THINKING_GESTURE_PAUSE = 0.7  # seconds between each left/right neck turn
 
# Said out loud (briefly) while the LLM works in the background, so the
# pause doesn't feel like dead air. Kept short since it's just filler.
THINKING_PHRASES = [
    "Hmm, let me think about that.",
    "Good question, thinking...",
    "Let me see.",
    "One moment, thinking it over.",
]
 
SYSTEM_PROMPT = (
    "You are BonicBot, a friendly voice assistant running on a small, "
    "lightweight local AI model. You are good at simple things: basic facts, "
    "definitions, simple math, and short explanations of AI/robotics/NLP "
    "concepts a beginner would ask. You are NOT good at long stories, complex "
    "multi-step reasoning, coding, or detailed technical questions — if asked "
    "something like that, briefly say it's a bit much for you and suggest the "
    "student ask something simpler instead. Keep answers short — one to three "
    "sentences, spoken out loud. Use simple words. If you don't know something, "
    "say so honestly instead of guessing. Never pretend to be a human, and "
    "never give unsafe, harmful, or inappropriate advice."
)
 
EMOJI_PATTERN = re.compile(
    "["
    "\U0001F300-\U0001FAFF"
    "\U00002600-\U000027BF"
    "\U0001F1E6-\U0001F1FF"
    "\U0001F900-\U0001F9FF"
    "\U00002190-\U000021FF"
    "\U00002B00-\U00002BFF"
    "]+",
    flags=re.UNICODE,
)
 
 
def clean_for_speech(text):
    """Strip markdown formatting and emoji before handing text to Piper —
    the model doesn't know it's writing for a voice, so it still reaches
    for **bold**, `code`, and emoji out of habit."""
    text = EMOJI_PATTERN.sub("", text)
    text = re.sub(r"\*\*(.*?)\*\*", r"\1", text)  # **bold**
    text = re.sub(r"\*(.*?)\*", r"\1", text)  # *italic*
    text = re.sub(r"`(.*?)`", r"\1", text)  # `code`
    text = re.sub(r"#+\s*", "", text)  # headers
    text = re.sub(r"[*_~]", "", text)  # stray leftover symbols
    text = re.sub(r"\s+", " ", text).strip()
    return text
 
 
# ============================================================================
# OLLAMA SETUP — all handled from Python, no manual `ollama` commands
# ============================================================================
def ensure_ollama_installed():
    """Confirm the Ollama app is installed. This is the one thing pip can't
    do — Ollama is a background service, not a Python package."""
    if shutil.which("ollama") is None:
        print("⚠️ The Ollama app isn't installed on this computer yet.")
        print("This is a one-time setup step, done once per computer:")
        print("  1. Go to https://ollama.com/download")
        print("  2. Download and run the installer for your OS")
        print("  3. Re-run this script afterwards")
        sys.exit(1)
 
 
def ensure_ollama_running():
    """Make sure the Ollama server is up. If it's not, start it as a
    background process from Python — the student never types `ollama serve`."""
    try:
        ollama.list()
        return
    except Exception:
        pass
 
    print("Ollama server isn't running yet — starting it now...")
    subprocess.Popen(
        ["ollama", "serve"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
 
    start = time.time()
    while time.time() - start < SERVER_START_TIMEOUT:
        try:
            ollama.list()
            print("✅ Ollama server is up.")
            return
        except Exception:
            time.sleep(SERVER_POLL_INTERVAL)
 
    print("⚠️ Couldn't confirm the server started in time — continuing anyway.")
 
 
def is_model_available(model_name):
    """Return True if the model has already been downloaded."""
    try:
        ollama.show(model_name)
        return True
    except Exception:
        return False
 
 
def ensure_model_pulled(model_name):
    """Download the model via the Python client if it isn't already present.
    Same job as typing `ollama pull <model>` in a terminal, just from code."""
    if is_model_available(model_name):
        print(f"✅ Model '{model_name}' is already downloaded.")
        return
 
    print(
        f"Model '{model_name}' not found locally — downloading now "
        f"(one-time, a few hundred MB)..."
    )
    last_status = None
    for progress in ollama.pull(model_name, stream=True):
        status = progress.get("status", "")
        if status and status != last_status:
            print(f"  {status}")
            last_status = status
    print(f"✅ Model '{model_name}' ready.")
 
 
# ============================================================================
# PIPER VOICE SETUP — auto-downloads the voice, no manual `piper` commands
# ============================================================================
def ensure_piper_voice_ready(voice_name, voices_dir):
    """Download the Piper voice (an .onnx model + matching .onnx.json config
    file) if it isn't on disk yet, and return their paths. Mirrors
    ensure_model_pulled() above — same "the student never types a download
    command" philosophy, just for the TTS voice instead of the LLM."""
    voices_dir.mkdir(parents=True, exist_ok=True)
    model_path = voices_dir / f"{voice_name}.onnx"
    config_path = voices_dir / f"{voice_name}.onnx.json"
 
    if model_path.exists() and config_path.exists():
        print(f"✅ Voice '{voice_name}' is already downloaded.")
        return model_path, config_path
 
    print(
        f"Voice '{voice_name}' not found locally — downloading now "
        f"(one-time, roughly 60MB)..."
    )
    try:
        download_voice(voice_name, voices_dir)
    except Exception as e:
        print(f"⚠️ Couldn't download the Piper voice: {e}")
        print("Check your internet connection and the voice name, then try again.")
        sys.exit(1)
 
    print(f"✅ Voice '{voice_name}' ready.")
    return model_path, config_path
 
 
# ============================================================================
# TEXT-TO-SPEECH (Piper)
# ============================================================================
def speak(piper_voice, pyaudio_instance, text):
    """Synthesize text with Piper and stream it out through PyAudio."""
    print(f"BonicBot: {text}")
    speech_text = clean_for_speech(text)
    syn_config = SynthesisConfig(length_scale=PIPER_LENGTH_SCALE, volume=PIPER_VOLUME)
 
    stream = None
    try:
        for chunk in piper_voice.synthesize(speech_text, syn_config=syn_config):
            if stream is None:
                stream = pyaudio_instance.open(
                    format=pyaudio_instance.get_format_from_width(chunk.sample_width),
                    channels=chunk.sample_channels,
                    rate=chunk.sample_rate,
                    output=True,
                )
            stream.write(chunk.audio_int16_bytes)
    finally:
        if stream is not None:
            stream.stop_stream()
            stream.close()
 
 
# ============================================================================
# SPEECH-TO-TEXT (faster-whisper, fully offline)
# ============================================================================
def warm_up_whisper(whisper_model):
    """Run one throwaway transcription on a second of silence right after
    loading the model. The very first real transcribe() call in a session
    is noticeably slower than the rest (kernels/caches aren't warm yet) —
    this eats that one-time cost during startup instead of during the
    student's first question."""
    dummy_audio = np.zeros(WHISPER_SAMPLE_RATE, dtype=np.float32)
    segments, _ = whisper_model.transcribe(dummy_audio, language="en", beam_size=WHISPER_BEAM_SIZE)
    list(segments)  # force evaluation, segments is a lazy generator
 
 
def listen(recognizer, mic, whisper_model):
    """Record one utterance from the microphone and transcribe it locally
    with Whisper. Returns the text, or None if nothing usable was heard."""
    with mic as source:
        print("\n🎤 Listening... (say 'goodbye' to quit)")
        try:
            audio = recognizer.listen(
                source, timeout=LISTEN_TIMEOUT, phrase_time_limit=PHRASE_TIME_LIMIT
            )
        except sr.WaitTimeoutError:
            return None
 
    audio_np = np.frombuffer(audio.get_raw_data(), dtype=np.int16).astype(np.float32) / 32768.0
    segments, _ = whisper_model.transcribe(
        audio_np,
        language="en",
        beam_size=WHISPER_BEAM_SIZE,
        condition_on_previous_text=False,  # each utterance here is a fresh,
        # standalone recording, not a continuation of the last one — this
        # skips Whisper carrying context forward between turns, which is
        # unnecessary work here and occasionally causes it to loop/hallucinate
    )
    text = " ".join(seg.text for seg in segments).strip()
 
    if not text:
        return None
    print(f"You said: {text}")
    return text
 
 
# ============================================================================
# LOCAL LLM (via Ollama)
# ============================================================================
def ask_llm(history):
    """Send the running conversation to the local model and return its reply.
    Auto-pulls the model if Ollama reports it's missing (belt-and-suspenders,
    in case ensure_model_pulled() was somehow skipped)."""
    try:
        response = ollama.chat(model=OLLAMA_MODEL, messages=history, think=False)
    except ollama.ResponseError as e:
        if e.status_code == 404:
            ensure_model_pulled(OLLAMA_MODEL)
            response = ollama.chat(model=OLLAMA_MODEL, messages=history, think=False)
        else:
            raise
    return response["message"]["content"].strip()
 
 
def trim_history(history):
    """Keep the conversation short by retaining the system prompt plus
    the most recent exchanges."""
    if len(history) > MAX_HISTORY_MESSAGES:
        history = [history[0]] + history[-(MAX_HISTORY_MESSAGES - 1):]
    return history
 
 
def is_repeat_question(current_text, previous_text):
    """Detect whether the current question is essentially a re-ask of the
    previous one, so the LLM can be nudged to try a different answer."""
    if not previous_text:
        return False
    a, b = current_text.strip().lower(), previous_text.strip().lower()
    return difflib.SequenceMatcher(None, a, b).ratio() >= REPEAT_SIMILARITY_THRESHOLD
 
 
# ============================================================================
# ROBOT GESTURES
# ============================================================================
def thinking_gesture_loop(bot, stop_event):
    """Turn the neck left and right in a slow loop until stop_event is set —
    this is what plays while the LLM is generating its reply in the
    background. Runs on its own thread; always leaves the neck centered
    when it exits, even if something above it goes wrong."""
    if not GESTURES_ENABLED:
        return
    try:
        while not stop_event.is_set():
            bot.look_left()
            if stop_event.wait(THINKING_GESTURE_PAUSE):
                break
            bot.look_right()
            if stop_event.wait(THINKING_GESTURE_PAUSE):
                break
    except Exception as e:
        print(f"⚠️ Thinking gesture skipped: {e}")
    finally:
        try:
            bot.look_center()
        except Exception as e:
            print(f"⚠️ Couldn't re-center neck: {e}")
 
 
def reset_arms_to_rest(bot):
    """Move both arms back to their resting position."""
    if not GESTURES_ENABLED:
        return
    try:
        bot.move_left_arm(ARM_REST_SHOULDER, ARM_REST_ELBOW, wait=False)
        bot.move_right_arm(ARM_REST_SHOULDER, ARM_REST_ELBOW, wait=True)
    except Exception as e:
        print(f"⚠️ Couldn't reset arms to rest: {e}")
 
 
def center_head_for_response(bot):
    """Center the head — done right before BonicBot starts speaking."""
    if not GESTURES_ENABLED:
        return
    try:
        bot.look_center()
    except Exception as e:
        print(f"⚠️ Couldn't center head: {e}")
 
 
def push_arms_forward(bot):
    """Move both arms into the 'explaining' pose (shoulders forward, elbows
    bent) once, and just hold it there — no waving. Called synchronously,
    blocking, right before speak() starts."""
    if not GESTURES_ENABLED:
        return
    try:
        bot.move_left_arm(ARM_FORWARD_SHOULDER, ARM_FORWARD_ELBOW, wait=False)
        bot.move_right_arm(ARM_FORWARD_SHOULDER, ARM_FORWARD_ELBOW, wait=True)
    except Exception as e:
        print(f"⚠️ Couldn't push arms forward: {e}")
 
 
def ask_llm_in_background(history, result):
    """Run ask_llm() on a worker thread and stash the outcome in `result`
    (a plain dict) so the main thread can pick it up once ready — this is
    what lets the thinking gesture and filler phrase play concurrently with
    the actual LLM call instead of after it."""
    try:
        result["reply"] = ask_llm(history)
    except Exception as e:
        result["error"] = e
 
 
# ============================================================================
# MAIN LOOP
# ============================================================================
def main():
    ensure_ollama_installed()
    ensure_ollama_running()
    ensure_model_pulled(OLLAMA_MODEL)
 
    model_path, config_path = ensure_piper_voice_ready(PIPER_VOICE, PIPER_VOICES_DIR)
    print("Loading Piper voice into memory...")
    piper_voice = PiperVoice.load(model_path, config_path=config_path, use_cuda=PIPER_USE_CUDA)
    pyaudio_instance = pyaudio.PyAudio()
 
    print("Loading Whisper model into memory...")
    whisper_model = WhisperModel(
        WHISPER_MODEL_SIZE, device=WHISPER_DEVICE, compute_type=WHISPER_COMPUTE_TYPE
    )
    print("Warming up Whisper (one-time cost, keeps your first question fast)...")
    warm_up_whisper(whisper_model)
 
    bot = BonicBot(host=HOST) if GESTURES_ENABLED else None
    if bot is not None:
        reset_arms_to_rest(bot)  # known starting position
        try:
            bot.look_center()
        except Exception as e:
            print(f"⚠️ Couldn't center neck on startup: {e}")
 
    recognizer = sr.Recognizer()
    recognizer.pause_threshold = PAUSE_THRESHOLD
    recognizer.non_speaking_duration = NON_SPEAKING_DURATION
    recognizer.dynamic_energy_threshold = DYNAMIC_ENERGY_THRESHOLD
    mic = sr.Microphone(sample_rate=WHISPER_SAMPLE_RATE)
 
    history = [{"role": "system", "content": SYSTEM_PROMPT}]
    last_user_text = None
 
    print("Calibrating microphone for background noise, please stay quiet...")
    with mic as source:
        recognizer.adjust_for_ambient_noise(source, duration=AMBIENT_NOISE_CALIBRATION_DURATION)
 
    speak(
        piper_voice,
        pyaudio_instance,
        "Hello! I'm BonicBot, a small local voice assistant. I'm best at "
        "simple stuff, like quick facts, easy definitions, or short "
        "questions about AI and robotics. Try me with something short! "
    )
 
    try:
        while True:
            user_text = listen(recognizer, mic, whisper_model)
 
            if user_text is None:
                continue
 
            if user_text.strip().lower() in EXIT_WORDS:
                speak(piper_voice, pyaudio_instance, "Goodbye! It was nice talking with you.")
                break
 
            if is_repeat_question(user_text, last_user_text):
                turn_content = (
                    f"{user_text}\n\n"
                    "[The user just asked this again — your previous answer was "
                    "likely wrong, unclear, or not what they needed. Do not repeat "
                    "it. If you're not confident of the correct answer, say so "
                    "honestly instead of guessing again.]"
                )
            else:
                turn_content = user_text
 
            history.append({"role": "user", "content": turn_content})
            last_user_text = user_text
 
            print("Thinking...")
 
            # Kick off the neck "thinking" gesture and the actual LLM call
            # on their own threads so they run at the same time...
            stop_thinking = threading.Event()
            gesture_thread = threading.Thread(
                target=thinking_gesture_loop, args=(bot, stop_thinking), daemon=True
            )
            gesture_thread.start()
 
            llm_result = {}
            llm_thread = threading.Thread(
                target=ask_llm_in_background, args=(history, llm_result), daemon=True
            )
            llm_thread.start()
 
            # ...and say a short filler out loud while both of those run in
            # the background, so the wait doesn't feel like dead air.
            speak(piper_voice, pyaudio_instance, random.choice(THINKING_PHRASES))
 
            llm_thread.join()
            stop_thinking.set()
            gesture_thread.join()
 
            if "error" in llm_result:
                raise llm_result["error"]
            reply = llm_result["reply"]
 
            history.append({"role": "assistant", "content": reply})
            history = trim_history(history)
 
            center_head_for_response(bot)
            push_arms_forward(bot)
 
            speak(piper_voice, pyaudio_instance, reply)
 
            reset_arms_to_rest(bot)
 
    except KeyboardInterrupt:
        print("\nShutting down...")
    finally:
        pyaudio_instance.terminate()
 
 
if __name__ == "__main__":
    main()

✅ Model 'qwen3:0.6b' is already downloaded.
✅ Voice 'en_US-lessac-medium' is already downloaded.
Loading Piper voice into memory...
Loading Whisper model into memory...
Warming up Whisper (one-time cost, keeps your first question fast)...
Calibrating microphone for background noise, please stay quiet...
BonicBot: Hello! I'm BonicBot, a small local voice assistant. I'm best at simple stuff, like quick facts, easy definitions, or short questions about AI and robotics. Try me with something short! 

🎤 Listening... (say 'goodbye' to quit)
You said: what is natural language processing
Thinking...
BonicBot: Hmm, let me think about that.
BonicBot: Natural language processing, or NLP, is how computers understand and work with human language, like text or speech.

🎤 Listening... (say 'goodbye' to quit)
You said: goodbye
BonicBot: Goodbye! It was nice talking with you.

for chunk in piper_voice.synthesize(speech_text, syn_config=syn_config):
    ...
    stream.write(chunk.audio_int16_bytes)

PIPER_VOICE = "en_GB-alba-medium"

WHISPER_MODEL_SIZE = "tiny.en"