Capstone 4 — BonicBot Weather Mood Agent

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

import difflib
import shutil
import subprocess
import sys
import time
from pathlib import Path
from typing import Optional
 
import ollama
import pyaudio
import requests
from pydantic import BaseModel, ValidationError
from piper import PiperVoice, SynthesisConfig
from piper.download_voices import download_voice
from bonicbot_bridge import BonicBot
from bonicbot_bridge.exceptions import BonicBotError
 
 
# ============================================================================
# CONFIG
# ============================================================================
HOST = 'localhost'
 
OLLAMA_MODEL = "qwen3.5:0.8b"      # bigger model, natively supports tool-calling + thinking
PIPER_VOICE_NAME = "en_US-lessac-medium"
PIPER_VOICES_DIR = Path(__file__).resolve().parent / "piper_voices"
 
SERVER_START_TIMEOUT = 15.0
SERVER_POLL_INTERVAL = 0.5
 
WEATHER_TIMEOUT = 6.0
 
# Classification thresholds -- deterministic, code decides this, not the SLM
HOT_THRESHOLD_C = 30
COLD_THRESHOLD_C = 15
RAIN_CODES = {51, 53, 55, 56, 57, 61, 63, 65, 66, 67, 80, 81, 82, 95, 96, 99}
 
SYSTEM_PROMPT = (
    "You are BonicBot playing a Weather Mood game with a student. The "
    "student will name a real city. Your job is to call the get_weather "
    "tool using the city name the student just typed. If their spelling "
    "has an obvious typo, correct it to the real city's proper spelling "
    "(for example 'trivaandrum' -> 'Trivandrum') -- but never substitute "
    "a DIFFERENT real city than the one they meant. Never invent or "
    "guess weather information yourself. If their message doesn't "
    "clearly name a city, ask them to name one instead of calling the "
    "tool."
)
 
 
# ============================================================================
# STATE — fallback structured output if the model doesn't call the tool
# ============================================================================
class CityGuess(BaseModel):
    city: str
 
 
# ============================================================================
# EXTERNAL LIBRARY — Open-Meteo, not part of the BonicBot Bridge SDK
# ============================================================================
def get_weather(city):
    """Looks up the current weather for a city using Open-Meteo (free,
    no API key). Returns {'temp_c', 'weather_code', 'precipitation'} or
    None if the city can't be found or the API fails."""
    try:
        geo_resp = requests.get(
            "https://geocoding-api.open-meteo.com/v1/search",
            params={"name": city, "count": 1},
            timeout=WEATHER_TIMEOUT,
        )
        if geo_resp.status_code != 200:
            return None
        results = geo_resp.json().get("results")
        if not results:
            return None
        lat, lon = results[0]["latitude"], results[0]["longitude"]
 
        wx_resp = requests.get(
            "https://api.open-meteo.com/v1/forecast",
            params={
                "latitude": lat,
                "longitude": lon,
                "current": "temperature_2m,precipitation,weather_code",
            },
            timeout=WEATHER_TIMEOUT,
        )
        if wx_resp.status_code != 200:
            return None
        current = wx_resp.json().get("current", {})
        if "temperature_2m" not in current or "weather_code" not in current:
            return None
 
        return {
            "temp_c": current["temperature_2m"],
            "weather_code": current["weather_code"],
            "precipitation": current.get("precipitation", 0.0),
        }
    except requests.RequestException:
        return None
 
 
def classify_weather(wx):
    """Deterministic classification -- no SLM involved, just thresholds
    on numeric/structured tool output."""
    if wx["weather_code"] in RAIN_CODES or wx["precipitation"] > 0.1:
        return "rainy"
    if wx["temp_c"] >= HOT_THRESHOLD_C:
        return "hot"
    if wx["temp_c"] <= COLD_THRESHOLD_C:
        return "cold"
    return "clear"
 
 
# The tool description Ollama uses to decide WHEN to call get_weather.
WEATHER_TOOL = {
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": (
            "Get the current temperature and weather condition for a "
            "city the student named. Always use this instead of guessing "
            "what the weather might be."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "The city the student typed, with obvious typos corrected to the real spelling. Must still be the same city they meant -- do not substitute a different one.",
                }
            },
            "required": ["city"],
        },
    },
}
 
 
# ============================================================================
# OLLAMA + PIPER SETUP — same helpers as Lessons 16-19
# ============================================================================
def ensure_ollama_installed():
    if shutil.which("ollama") is None:
        print("⚠️ Install Ollama once from https://ollama.com/download, then re-run.")
        sys.exit(1)
 
 
def ensure_ollama_running():
    try:
        ollama.list()
        return
    except Exception:
        pass
    print("Starting Ollama server...")
    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)
 
 
def ensure_model_pulled(model_name):
    try:
        ollama.show(model_name)
        return
    except Exception:
        pass
    print(f"Downloading model '{model_name}'...")
    for _ in ollama.pull(model_name, stream=True):
        pass
    print(f"✅ Model '{model_name}' ready.")
 
 
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}'...")
    download_voice(voice_name, voices_dir)
    return model_path, config_path
 
 
def speak_step(piper_voice, pyaudio_instance, text):
    print(f"  🗣️  BonicBot says: \"{text}\"")
    syn_config = SynthesisConfig(length_scale=1.0, volume=1.0)
    stream = None
    try:
        for chunk in piper_voice.synthesize(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()
 
 
# ============================================================================
# DECIDE — does the agent call the weather tool, and on what city?
# ============================================================================
def is_plausible_correction(candidate, raw_input, threshold=0.5):
    """True if `candidate` looks like a spelling fix of `raw_input` rather
    than a completely different city. Uses fuzzy similarity instead of an
    exact substring check, since a corrected spelling (e.g. 'Trivandrum')
    won't literally appear inside the student's typo ('trivaandrum')."""
    candidate = candidate.strip().lower()
    raw = raw_input.strip().lower()
    if not candidate:
        return False
    return difflib.SequenceMatcher(None, candidate, raw).ratio() >= threshold
 
 
def extract_city_fallback(history):
    """If the model didn't call the tool on its own, ask it directly to
    pull out just the city name using structured output."""
    response = ollama.chat(
        model=OLLAMA_MODEL,
        think=False,
        format=CityGuess.model_json_schema(),
        messages=history + [
            {"role": "user", "content": "Extract just the city name the student mentioned, correcting any obvious typo."}
        ],
    )
    try:
        return CityGuess.model_validate_json(response["message"]["content"]).city
    except ValidationError:
        return None
 
 
def decide_and_fetch_weather(history, raw_input):
    """The agent's decision each round: figure out which city the student
    named, and call get_weather for it. Mutates history in place so later
    steps can see the tool result.
 
    Small models occasionally ignore the real input and echo back an
    example city instead (a known failure mode). As a safety net, if the
    model's extracted city isn't a plausible spelling-corrected match for
    what the student typed, we trust the student's raw text over the
    model's guess."""
    response = ollama.chat(model=OLLAMA_MODEL, think=False, messages=history, tools=[WEATHER_TOOL])
    message = response["message"]
    tool_calls = message.get("tool_calls")
 
    if tool_calls:
        history.append({"role": "assistant", "content": message.get("content", ""), "tool_calls": tool_calls})
        city = ""
        for call in tool_calls:
            city = call["function"]["arguments"].get("city", "")
    else:
        city = extract_city_fallback(history) or ""
 
    if not is_plausible_correction(city, raw_input):
        if city:
            print(f"  ⚠️ Model suggested '{city}', which doesn't match what you typed -- using your input instead.")
        city = raw_input
 
    print(f"  🔎 Looking up weather for: {city}")
    wx = get_weather(city)
    result_text = str(wx) if wx else f"Could not find weather for '{city}'."
    history.append({"role": "tool", "content": result_text})
    return city, wx
 
 
# ============================================================================
# ACT — one gesture per weather bucket, each announcing itself first
# ============================================================================
WEATHER_LINES = {
    "hot":   "Phew, it's hot out there!",
    "cold":  "Brr, that sounds freezing!",
    "rainy": "Better grab an umbrella!",
    "clear": "What a beautiful, clear day!",
}
 
 
def gesture_hot(bot, piper_voice, pyaudio_instance):
    speak_step(piper_voice, pyaudio_instance, WEATHER_LINES["hot"])
    bot.move_left_arm(shoulder=150, elbow=40, wait=True)
    bot.move_right_arm(shoulder=150, elbow=40, wait=True)
    for _ in range(2):
        bot.look_left(); time.sleep(0.2)
        bot.look_right(); time.sleep(0.2)
    bot.look_center()
    bot.servo.reset_all_servos()
 
 
def gesture_cold(bot, piper_voice, pyaudio_instance):
    speak_step(piper_voice, pyaudio_instance, WEATHER_LINES["cold"])
    bot.move_left_arm(shoulder=90, elbow=50, wait=True)
    time.sleep(0.3)
    bot.move_right_arm(shoulder=90, elbow=50, wait=True)
    for _ in range(3):
        bot.turn_left(speed=60, duration=0.5)
        bot.turn_right(speed=60, duration=0.5)
    bot.stop()
    bot.servo.reset_all_servos()
 
 
def gesture_rainy(bot, piper_voice, pyaudio_instance):
    speak_step(piper_voice, pyaudio_instance, WEATHER_LINES["rainy"])
    bot.move_right_arm(shoulder=180, elbow=0, wait=True)
    bot.look_left()
    time.sleep(2)
    bot.look_right()
    bot.servo.reset_all_servos()
 
 
def gesture_clear(bot, piper_voice, pyaudio_instance):
    speak_step(piper_voice, pyaudio_instance, WEATHER_LINES["clear"])
    bot.move_left_arm(shoulder=170, elbow=10, wait=True)
    bot.move_right_arm(shoulder=170, elbow=10, wait=True)
    bot.turn_left(speed=60, duration=2.0)
    bot.turn_right(speed=60, duration=2.0)
    bot.stop()
    bot.servo.reset_all_servos()
 
 
GESTURES = {
    "hot": gesture_hot,
    "cold": gesture_cold,
    "rainy": gesture_rainy,
    "clear": gesture_clear,
}
 
 
# ============================================================================
# MAIN — GOAL + STATE + the LOOP that ties every ingredient together
# ============================================================================
def main():
    ensure_ollama_installed()
    ensure_ollama_running()
    ensure_model_pulled(OLLAMA_MODEL)
 
    model_path, config_path = ensure_piper_voice_ready(PIPER_VOICE_NAME, PIPER_VOICES_DIR)
    piper_voice = PiperVoice.load(model_path, config_path=config_path)
    pyaudio_instance = pyaudio.PyAudio()
 
    history = [{"role": "system", "content": SYSTEM_PROMPT}]   # STATE starts here
 
    try:
        with BonicBot(host=HOST, port=9090, timeout=10) as bot:
            print("✅ Connected. Name a real city and BonicBot will show you how the weather feels there!")
            print("   (Type 'quit' at any point to stop.)\n")
 
            while True:                                          # LOOP
                user_input = input("Name a city> ").strip()
                if user_input.lower() in {"quit", "exit"}:
                    break
                if not user_input:
                    continue
 
                history.append({"role": "user", "content": user_input})  # STATE updates
 
                city, wx = decide_and_fetch_weather(history, user_input)  # DECIDE
 
                if not wx:
                    msg = "Sorry, I couldn't find the weather for that. Try another city."
                    speak_step(piper_voice, pyaudio_instance, msg)
                    history.append({"role": "assistant", "content": msg})
                    continue
 
                bucket = classify_weather(wx)                      # classify (deterministic)
                print(f"  📄 {city}: {wx['temp_c']}°C, code {wx['weather_code']} -> {bucket}")
                history.append({"role": "assistant", "content": f"The weather in {city} is {bucket}."})
 
                GESTURES[bucket](bot, piper_voice, pyaudio_instance)  # ACT
 
    except BonicBotError as e:
        print(f"⚠️ Robot error: {e}")
    finally:
        pyaudio_instance.terminate()
 
 
if __name__ == "__main__":
    main()

✅ Connected. Name a real city and BonicBot will show you how the weather feels there!
   (Type 'quit' at any point to stop.)

Name a city> Trivandrum
  🔎 Looking up weather for: Trivandrum
  📄 Trivandrum: 31.4°C, code 0 -> hot
  🗣️  BonicBot says: "Phew, it's hot out there!"

Name a city> London
  🔎 Looking up weather for: London
  📄 London: 12.1°C, code 61 -> rainy
  🗣️  BonicBot says: "Better grab an umbrella!"

Name a city> quit