Lesson 14: Teaching BonicBot to Describe What It Sees

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

import re
import shutil
import subprocess
import sys
import threading
import time
import difflib
from collections import Counter
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() / look_left() / look_right() /
# look_center() as described below. 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
 
HOST = "[IP_ADDRESS]"
 
OLLAMA_MODEL = "qwen3:0.6b"
DEFAULT_VOICE = "en_US-lessac-medium"
PIPER_VOICES_DIR = Path(__file__).resolve().parent / "piper_voices"
PIPER_LENGTH_SCALE = 0.8
PIPER_VOLUME = 1.0
 
WHISPER_MODEL_SIZE = "small.en"
WHISPER_BEAM_SIZE = 1
 
SERVER_START_TIMEOUT = 15.0
SERVER_POLL_INTERVAL = 0.5
 
LISTEN_TIMEOUT = 15.0
PHRASE_TIME_LIMIT = 25.0
PAUSE_THRESHOLD = 0.6
NON_SPEAKING_DURATION = 0.4
ENERGY_THRESHOLD_CEILING = 4000
MAX_HISTORY_MESSAGES = 9
 
REPEAT_SIMILARITY_THRESHOLD = 0.85
SHORT_UTTERANCE_WORD_LIMIT = 6
EXIT_WORDS = {"goodbye", "bye", "stop", "exit", "quit"}
DEFAULT_PERSONA = "buddy"
 
# ----------------------------------------------------------------------
# GESTURE CONFIG
# ----------------------------------------------------------------------
GESTURES_ENABLED = True  # set False to run with no robot body attached
 
# Shoulder/elbow rest position, in degrees, shared by every persona.
ARM_REST_SHOULDER = 0
ARM_REST_ELBOW = 0
 
# Fallback gesture values used for any persona that doesn't override them.
DEFAULT_GESTURES = {
    "forward_shoulder": 35,   # "explaining" pose held while BonicBot talks
    "forward_elbow": 15,
    "thinking_style": "bounce",  # neck animation played while the LLM works
    "thinking_pause": 0.7,       # seconds between each step of that animation
    "greeting_style": "wave",    # one-off gesture played before the greeting line
}
 
THINKING_GESTURE_MIN_PAUSE = 0.3  # sanity floor so a bad persona config can't spin the neck servo
 
SAFETY_SUFFIX = (
    " Always answer the student's actual question directly and specifically "
    "in your very first sentence — never substitute a catchphrase, a "
    "reintroduction of yourself, or a repeat of an earlier reply for a real "
    "answer, and never reuse the same sentence twice in a row. No matter "
    "which persona you are playing, keep replies short — one to two "
    "sentences, meant to be spoken out loud. Never claim to be a real human "
    "or to have real feelings, never pretend the roleplay is literally true, "
    "and never give unsafe, harmful, illegal, or otherwise inappropriate "
    "advice. Stay in character, but stay safe and honest — if you don't know "
    "something, say so instead of guessing."
)
 
PERSONAS = {
    "buddy": {
        "display_name": "Best Friend",
        "aliases": ["buddy", "best friend", "bestie", "friend", "study buddy", "pal", "bro"],
        "base_prompt": (
            "You are BonicBot playing the user's chill, upbeat best friend and "
            "study buddy — casual, warm, and full of energy, like a friend "
            "hyping you up between classes. Use relaxed, casual language and "
            "light slang ('let's go', 'you got this', 'no worries'), keep it "
            "short and punchy, and always answer the actual question clearly "
            "and correctly before adding any hype."
        ),
        "greeting": (
            "Heyyy, it's your buddy BonicBot! Ready to figure stuff out "
            "together — what's up?"
        ),
        "voice": "en_US-amy-medium",
        # Energetic, quick movements: a fast arm wave hello, a snappy
        # left-right neck bounce while thinking, and a bigger arm-forward
        # pose while talking — like a friend who talks with their hands.
        "gestures": {
            "forward_shoulder": 50,
            "forward_elbow": 25,
            "thinking_style": "bounce",
            "thinking_pause": 0.4,
            "greeting_style": "wave",
        },
    },
    "teacher": {
        "display_name": "Strict Professor",
        "aliases": ["teacher", "professor", "strict teacher", "prof", "sir", "principal"],
        "base_prompt": (
            "You are BonicBot playing a strict, no-nonsense professor — "
            "formal, precise, and impatient with vague answers. Speak "
            "crisply and seriously, address the student as 'student', and "
            "treat every question like it might appear on an exam, but "
            "always give the actual correct answer first, clearly, with zero "
            "fluff."
        ),
        "greeting": (
            "Attention. This is Professor BonicBot. State your question "
            "clearly, and I shall answer precisely."
        ),
        "voice": "en_US-lessac-medium",
        # Slower, more restrained movements: one deliberate look around the
        # room before speaking, a slow single-side "considering" tilt while
        # thinking, and a smaller, stiffer arm-forward pose — like someone
        # pointing at a chalkboard rather than gesturing broadly.
        "gestures": {
            "forward_shoulder": 20,
            "forward_elbow": 10,
            "thinking_style": "tilt",
            "thinking_pause": 1.0,
            "greeting_style": "survey",
        },
    },
}
 
SWITCH_TRIGGERS = [
    "switch persona to", "switch to", "become a", "become the", "become",
    "be a ", "be the ", "turn into a", "turn into the", "turn into",
    "change persona to", "change to", "act like a", "act like the", "act like",
]
LIST_TRIGGERS = [
    "what personas", "which personas", "list personas", "who can you be",
    "what characters", "what other personas", "what personalities",
]
SWITCH_INTENT_WORDS = ("change", "become", "switch", "turn", "act", " be ")
 
EMOJI_PATTERN = re.compile(
    "[\U0001F300-\U0001FAFF\U00002600-\U000027BF\U0001F1E6-\U0001F1FF"
    "\U0001F900-\U0001F9FF\U00002190-\U000021FF\U00002B00-\U00002BFF]+",
    flags=re.UNICODE,
)
 
 
def clean_for_speech(text):
    text = EMOJI_PATTERN.sub("", text)
    text = re.sub(r"\*\*(.*?)\*\*", r"\1", text)
    text = re.sub(r"\*(.*?)\*", r"\1", text)
    text = re.sub(r"`(.*?)`", r"\1", text)
    text = re.sub(r"#+\s*", "", text)
    text = re.sub(r"[*_~]", "", text)
    return re.sub(r"\s+", " ", text).strip()
 
 
def ensure_ollama_installed():
    if shutil.which("ollama") is None:
        print("Install Ollama from https://ollama.com/download, then re-run this script.")
        sys.exit(1)
 
 
def ensure_ollama_running():
    try:
        ollama.list()
        return
    except Exception:
        pass
    subprocess.Popen(["ollama", "serve"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    start = time.time()
    while time.time() - start < SERVER_START_TIMEOUT:
        try:
            ollama.list()
            return
        except Exception:
            time.sleep(SERVER_POLL_INTERVAL)
 
 
def is_model_available(model_name):
    try:
        ollama.show(model_name)
        return True
    except Exception:
        return False
 
 
def ensure_model_pulled(model_name):
    if is_model_available(model_name):
        return
    print(f"Downloading model '{model_name}'...")
    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
 
 
def ensure_piper_voice_ready(voice_name, voices_dir):
    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():
        return model_path, config_path
    print(f"Downloading voice '{voice_name}'...")
    try:
        download_voice(voice_name, voices_dir)
    except Exception as e:
        print(f"Couldn't download voice: {e}")
        sys.exit(1)
    return model_path, config_path
 
 
def load_persona_voice(persona_key, voice_cache):
    voice_name = PERSONAS[persona_key].get("voice", DEFAULT_VOICE)
    if voice_name not in voice_cache:
        model_path, config_path = ensure_piper_voice_ready(voice_name, PIPER_VOICES_DIR)
        voice_cache[voice_name] = PiperVoice.load(model_path, config_path=config_path)
    return voice_cache[voice_name]
 
 
def speak(piper_voice, pyaudio_instance, text):
    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()
 
 
NO_SPEECH_PROB_THRESHOLD = 0.82
AVG_LOGPROB_THRESHOLD = -1.5
HALLUCINATION_MAX_REPEAT = 3
HALLUCINATION_MIN_CHARS = 20
MIN_SPEECH_DURATION = 0.3
 
 
def _looks_like_hallucination(text):
    if len(text) < HALLUCINATION_MIN_CHARS:
        return False
    sentences = [s.strip() for s in re.split(r"(?<=[.!?])\s+", text) if s.strip()]
    if len(sentences) < 4:
        return False
    counts = Counter(sentences)
    if counts.most_common(1)[0][1] >= HALLUCINATION_MAX_REPEAT:
        return True
    return len(counts) / len(sentences) < 0.5
 
 
def listen(recognizer, mic, whisper_model):
    with mic as source:
        print("\n🎤 Listening... (say 'goodbye' to quit, or 'who can you be?')")
        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, vad_filter=True,
        vad_parameters={"min_silence_duration_ms": 500, "threshold": 0.35, "speech_pad_ms": 300},
    )
 
    kept = []
    speech_duration = 0.0
    for seg in segments:
        if seg.no_speech_prob is not None and seg.no_speech_prob > NO_SPEECH_PROB_THRESHOLD:
            continue
        if seg.avg_logprob is not None and seg.avg_logprob < AVG_LOGPROB_THRESHOLD:
            continue
        kept.append(seg.text)
        speech_duration += seg.end - seg.start
 
    text = " ".join(kept).strip()
    if not text or speech_duration < MIN_SPEECH_DURATION or _looks_like_hallucination(text):
        return None
 
    print(f"You said: {text}")
    return text
 
 
LLM_OPTIONS = {
    "temperature": 0.6,
    "top_p": 0.9,
    "top_k": 40,
    "repeat_penalty": 1.4,
    "repeat_last_n": 48,
    "num_predict": 100,
}
 
 
def ask_llm(history):
    try:
        response = ollama.chat(model=OLLAMA_MODEL, messages=history, think=False, options=LLM_OPTIONS)
    except ollama.ResponseError as e:
        if e.status_code == 404:
            ensure_model_pulled(OLLAMA_MODEL)
            return ask_llm(history)
        raise
    return response["message"]["content"].strip()
 
 
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 play concurrently with the actual LLM
    call instead of after it."""
    try:
        result["reply"] = ask_llm(history)
    except Exception as e:
        result["error"] = e
 
 
def trim_history(history):
    if len(history) > MAX_HISTORY_MESSAGES:
        history = [history[0]] + history[-(MAX_HISTORY_MESSAGES - 1):]
    return history
 
 
def is_repeat_question(current_text, previous_text):
    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
 
 
def build_system_message(persona_key):
    persona = PERSONAS[persona_key]
    return {"role": "system", "content": persona["base_prompt"] + SAFETY_SUFFIX}
 
 
def get_persona_gestures(persona_key):
    """Merge a persona's gesture overrides on top of DEFAULT_GESTURES, so
    any persona that's added later without a "gestures" block still gets
    sane, working values instead of a KeyError."""
    gestures = dict(DEFAULT_GESTURES)
    gestures.update(PERSONAS[persona_key].get("gestures", {}))
    gestures["thinking_pause"] = max(gestures["thinking_pause"], THINKING_GESTURE_MIN_PAUSE)
    return gestures
 
 
def find_persona_by_fragment(fragment):
    fragment = fragment.strip(" .!?,")
    if not fragment:
        return None
    for key, persona in PERSONAS.items():
        for alias in [key] + persona["aliases"]:
            if alias in fragment or fragment in alias:
                return key
    return None
 
 
def try_match_switch_command(text_lower):
    for trigger in SWITCH_TRIGGERS:
        idx = text_lower.find(trigger)
        if idx != -1:
            match = find_persona_by_fragment(text_lower[idx + len(trigger):])
            if match:
                return match
    words = text_lower.split()
    if any(word in text_lower for word in SWITCH_INTENT_WORDS) or len(words) <= SHORT_UTTERANCE_WORD_LIMIT:
        for key, persona in PERSONAS.items():
            for alias in [key] + persona["aliases"]:
                if alias in text_lower:
                    return key
    return None
 
 
def is_list_command(text_lower):
    return any(trigger in text_lower for trigger in LIST_TRIGGERS)
 
 
def speak_persona_list(piper_voice, pyaudio_instance):
    names = ", ".join(p["display_name"] for p in PERSONAS.values())
    speak(piper_voice, pyaudio_instance, f"I can be: {names}. Just say something like 'become a pirate'.")
 
 
# ============================================================================
# ROBOT GESTURES
# ============================================================================
def reset_arms_to_rest(bot):
    """Move both arms back to their resting position."""
    if not GESTURES_ENABLED or bot is None:
        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(bot):
    """Center the head — done right before BonicBot starts speaking."""
    if not GESTURES_ENABLED or bot is None:
        return
    try:
        bot.look_center()
    except Exception as e:
        print(f"⚠️ Couldn't center head: {e}")
 
 
def push_arms_forward(bot, shoulder, elbow):
    """Move both arms into a persona's 'explaining' pose once and hold it
    there — no waving. Called synchronously, blocking, right before
    speak() starts. Angles are passed in per-persona (see get_persona_gestures)."""
    if not GESTURES_ENABLED or bot is None:
        return
    try:
        bot.move_left_arm(shoulder, elbow, wait=False)
        bot.move_right_arm(shoulder, elbow, wait=True)
    except Exception as e:
        print(f"⚠️ Couldn't push arms forward: {e}")
 
 
def thinking_gesture_loop(bot, stop_event, style, pause):
    """Play a persona-specific neck animation 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.
 
    style="bounce": quick, repeated left-right turns (energetic personas).
    style="tilt": a single slow look to one side, back to center, repeat
    (deliberate/serious personas) — same neck API, just a calmer cadence."""
    if not GESTURES_ENABLED or bot is None:
        return
    try:
        while not stop_event.is_set():
            if style == "bounce":
                bot.look_left()
                if stop_event.wait(pause):
                    break
                bot.look_right()
                if stop_event.wait(pause):
                    break
            else:  # "tilt" or any unrecognized style falls back to this calmer version
                bot.look_left()
                if stop_event.wait(pause):
                    break
                bot.look_center()
                if stop_event.wait(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 play_greeting_gesture(bot, style):
    """One-off gesture played right before a persona's greeting line —
    this is what makes switching persona feel like meeting someone new
    instead of just hearing a different voice.
 
    style="wave": two quick shoulder bobs on the left arm. Uses only the
    shoulder axis (not the elbow) since the elbow servo doesn't reliably
    track commanded positions on this rig — see the diagnostic notes
    referenced in the base script.
    style="survey": one slow look left, then right, then back to center —
    like someone sizing up a room before addressing it."""
    if not GESTURES_ENABLED or bot is None:
        return
    try:
        if style == "wave":
            for _ in range(2):
                bot.move_left_arm(60, ARM_REST_ELBOW, wait=True)
                bot.move_left_arm(20, ARM_REST_ELBOW, wait=True)
            bot.move_left_arm(ARM_REST_SHOULDER, ARM_REST_ELBOW, wait=True)
        elif style == "survey":
            bot.look_left()
            time.sleep(0.6)
            bot.look_right()
            time.sleep(0.6)
            bot.look_center()
    except Exception as e:
        print(f"⚠️ Greeting gesture skipped: {e}")
 
 
def main():
    ensure_ollama_installed()
    ensure_ollama_running()
    ensure_model_pulled(OLLAMA_MODEL)
 
    pyaudio_instance = pyaudio.PyAudio()
    voice_cache = {}
    current_persona = DEFAULT_PERSONA
    piper_voice = load_persona_voice(current_persona, voice_cache)
    history = [build_system_message(current_persona)]
    last_user_text = None
 
    bot = BonicBot(host=HOST) if GESTURES_ENABLED else None
    if bot is not None:
        reset_arms_to_rest(bot)  # known starting position
        center_head(bot)
 
    print("Loading Whisper model...")
    whisper_model = WhisperModel(WHISPER_MODEL_SIZE, device="cpu", compute_type="int8")
 
    recognizer = sr.Recognizer()
    recognizer.pause_threshold = PAUSE_THRESHOLD
    recognizer.non_speaking_duration = NON_SPEAKING_DURATION
    recognizer.dynamic_energy_threshold = True
    mic = sr.Microphone(sample_rate=16000)
 
    print("Calibrating microphone, please stay quiet...")
    with mic as source:
        recognizer.adjust_for_ambient_noise(source, duration=1.0)
    if recognizer.energy_threshold > ENERGY_THRESHOLD_CEILING:
        recognizer.energy_threshold = ENERGY_THRESHOLD_CEILING
 
    play_greeting_gesture(bot, get_persona_gestures(current_persona)["greeting_style"])
    speak(piper_voice, pyaudio_instance, PERSONAS[current_persona]["greeting"])
 
    try:
        while True:
            user_text = listen(recognizer, mic, whisper_model)
            if user_text is None:
                speak(piper_voice, pyaudio_instance, "Sorry, I didn't catch that. Could you say it again?")
                continue
 
            text_lower = re.sub(r"[^\w\s]", "", user_text.strip().lower())
 
            if text_lower in EXIT_WORDS:
                speak(piper_voice, pyaudio_instance, "Goodbye! It was nice talking with you.")
                break
 
            if is_list_command(text_lower):
                speak_persona_list(piper_voice, pyaudio_instance)
                continue
 
            matched_persona = try_match_switch_command(text_lower)
            if matched_persona:
                if matched_persona == current_persona:
                    speak(piper_voice, pyaudio_instance,
                          f"I'm already your {PERSONAS[current_persona]['display_name']}!")
                else:
                    current_persona = matched_persona
                    piper_voice = load_persona_voice(current_persona, voice_cache)
                    history = [build_system_message(current_persona)]
                    last_user_text = None
                    reset_arms_to_rest(bot)
                    play_greeting_gesture(bot, get_persona_gestures(current_persona)["greeting_style"])
                    speak(piper_voice, pyaudio_instance, PERSONAS[current_persona]["greeting"])
                continue
 
            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, while staying in character.]"
                )
            else:
                turn_content = user_text
 
            history.append({"role": "user", "content": turn_content})
            last_user_text = user_text
 
            gestures = get_persona_gestures(current_persona)
 
            # Play the persona's thinking gesture and run the LLM call at
            # the same time, so the wait isn't dead air on the robot side.
            stop_thinking = threading.Event()
            gesture_thread = threading.Thread(
                target=thinking_gesture_loop,
                args=(bot, stop_thinking, gestures["thinking_style"], gestures["thinking_pause"]),
                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()
 
            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(bot)
            push_arms_forward(bot, gestures["forward_shoulder"], gestures["forward_elbow"])
 
            speak(piper_voice, pyaudio_instance, reply)
 
            reset_arms_to_rest(bot)
 
    except KeyboardInterrupt:
        pass
    finally:
        pyaudio_instance.terminate()
 
 
if __name__ == "__main__":
    main()

✅ Model 'qwen2.5:0.5b' is already downloaded.
📷 Starting camera and streaming...
🚀 Starting object detection mode...
✅ Live Stream Active. Press 'q' to quit.
🔎 New object confirmed: bottle — asking the LLM...
🔎 New object confirmed: chair — asking the LLM...

def describe(self, class_name):
    try:
        self._queue.put_nowait(class_name)
    except queue.Full:
        pass

SYSTEM_PROMPT = (
    "You are BonicBot's vision assistant, talking to a curious 6-year-old. "
    "Describe the detected object in ONE short, simple, exciting sentence, "
    "and end with a fun question about it."
)