Lesson 11: Syncing BonicBot’s Head Movements

Learning Objective

Give BonicBot a responsive head-tracking brain — one that watches you through the camera, detects when you turn your head using pose estimation, thinks up quirky observations using a local Ollama LLM, speaks them offline via Piper TTS, and physically mirrors your movement, all running completely locally.

Introduction

So far, BonicBot has learned to see the world — detecting objects, faces, hand gestures, and ArUco markers. In this lesson, we combine vision, reasoning, physical motion, and speech: BonicBot will watch which way you turn your head, think up a quirky observation, speak it aloud, and physicalize the action by mirroring your look, entirely offline.

BonicBot’s sync pipeline has four stages:


camera -> YOLO-pose (ear visibility check) -> head direction confirmation -> Ollama local LLM -> Piper TTS & neck movement

Pose Estimation: A local YOLO-pose model (yolo11n-pose.pt) runs on a background thread (PoseWorker) to find 17 body keypoints, specifically tracking the visibility of your left and right ears.
Head Direction: The relative visibility of your ears is monitored. If only one ear is visible, it signals you have turned your head. A debouncing state machine confirms the direction (“left” or “right”) once held steady.
Local LLM: When a direction is confirmed, it is sent to a local language model running via Ollama (e.g., qwen3:0.6b). The model generates a short, dramatic robotic observation.
Speech & Motion: The generated text is spoken using Piper, a fast local TTS engine, while BonicBot physically turns its neck (look_left() or look_right()) and then returns to center, matching your action in real-time.

The only step a student has to do manually, once per computer, is install the Ollama background app itself. Everything else — downloading the model, the Piper voice, and running the real-time loop — is handled automatically by the script.

Setup: Installing Packages

Before running the code, make sure your computer has the required Python packages and local AI services installed. Open a terminal and run:


pip install bonicbot-bridge opencv-python numpy ollama pyaudio piper-tts ultralytics

What each package / tool does

Package / Tool	Purpose
`bonicbot-bridge`	The official BonicBot SDK. Provides the `BonicBot` class used to connect to the robot, control its camera feed via `bot.system.start_camera()` / `bot.start_camera()`, and send head movement commands (`bot.look_left()`, `bot.look_right()`, `bot.look_center()`). Also provides `BonicBotError` for safe error handling.
`opencv-python`	Imported as `cv2`. Used to display the live video feed (`cv2.imshow`), render the pose skeleton overlay (`cv2.line`, `cv2.circle`), draw status text, and overlay LLM subtitles on a black banner at the bottom of the window.
`numpy`	Imported as `np`. Used for keypoint coordinate array operations, confidence scoring, and warming up model tensors.
`ollama`	Python client for Ollama. Used by `gen_line()` to query the local `qwen3:0.6b` LLM model offline to generate quirky robotic observations when a head turn is detected.
`pyaudio`	Audio I/O library. Used by `say()` to stream synthesized audio bytes directly to system speakers in real-time as Piper generates them.
`piper-tts`	Provides local neural text-to-speech synthesis (`PiperVoice`, `SynthesisConfig`, `download_voice`). Converts LLM text output into spoken audio.
`ultralytics`	Provides the YOLO framework (`YOLO`). Used by `PoseWorker` to load `yolo11n-pose.pt` and estimate 17 body keypoints in real time.

If pip install fails, try pip3 install ... instead, or use a virtual environment:


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

How to run the program

Install Ollama (one-time setup):
- Windows / Mac: Download and run the installer from ollama.com/download .
- Linux: Run curl -fsSL https://ollama.com/install.sh | sh in your terminal.
Pull the LLM model: Ensure the Ollama background service is running, then run:
```
ollama pull qwen3:0.6b
```
Find your BonicBot’s IP address — via the robot’s display, companion app, or router.
Save the code into a file, e.g., lesson11_head_sync.py.
Update HOST: If connecting to a physical BonicBot, replace HOST = "localhost" with your robot’s IP address (e.g., HOST = "172.20.10.2").
Download the Piper voice model (or let the script download it automatically on first run):
```
python3 -m piper.download_voices --data-dir ./piper_voices en_US-lessac-medium
```
Run the script:
```
python lesson11_head_sync.py
```
A window titled “BonicBot Head Sync” will open showing your video feed, skeleton tracking, detected head direction, and live subtitles.
Turn your head left or right and hold it for a moment: BonicBot will detect the turn, query the local LLM, speak the observation, and mirror your head movement physically.
Press q with the video window focused to exit cleanly.

Don’t have a physical BonicBot? Try it in simulation (optional)

This lesson’s code uses HOST = "localhost" by default, so it can run directly against the ROS 2 simulation environment:

Launch the BonicBot simulation:
```
ros2 launch my_bot robot_system.launch.py use_sim_time:=true world:=obsworld.sdf use_real_camera:=True
```
- use_real_camera:=True — the camera feed comes from your laptop’s webcam, allowing pose estimation and ear visibility checks to track your actual head turns while simulating BonicBot’s head movements.
- use_real_camera:=False — the camera feed comes from Gazebo instead.
Leave HOST = "localhost" unchanged.
Everything else — YOLO pose tracking, Ollama local LLM generation, Piper TTS audio, and BonicBot neck movement — works identically.

Code

Click to view the complete program


import random
import re
import threading
import time
from pathlib import Path
 
import cv2
import numpy as np
import ollama
import pyaudio
from bonicbot_bridge import BonicBot, BonicBotError
from piper import PiperVoice, SynthesisConfig
from piper.download_voices import download_voice
from ultralytics import YOLO
 
HOST, MODEL = "localhost", "qwen3:0.6b"
POSE_MODEL_PATH, POSE_IMG_SIZE = "yolo11n-pose.pt", 256
PERSON_CONF_THRESHOLD, VIS_THRESHOLD = 0.5, 0.5
MIN_CONFIRM_SAMPLES, CONFIRM_HOLD_SECONDS, COOLDOWN = 2, 0.35, 3.0
RAW_LOG_INTERVAL, MAX_LINE_WORDS = 30, 12
 
NOSE, LEFT_EYE, RIGHT_EYE, LEFT_EAR, RIGHT_EAR = 0, 1, 2, 3, 4
COCO_SKELETON = [
    (0, 1), (0, 2), (1, 3), (2, 4), (0, 5), (0, 6), (5, 6),
    (5, 7), (7, 9), (6, 8), (8, 10), (5, 11), (6, 12), (11, 12),
    (11, 13), (13, 15), (12, 14), (14, 16),
]
 
PIPER_VOICE = "en_US-lessac-medium"
PIPER_VOICES_DIR = Path(__file__).resolve().parent / "piper_voices"
PIPER_LENGTH_SCALE, PIPER_VOLUME, PIPER_USE_CUDA = 0.8, 1.0, False
 
EMOJI_PATTERN = re.compile(
    r"[\U0001F300-\U0001FAFF\U00002600-\U000027BF\U0001F1E6-\U0001F1FF"
    r"\U0001F900-\U0001F9FF\U00002190-\U000021FF\U00002B00-\U00002BFF]+"
)
FALLBACK_LINES = {
    "left": ["Looking left!", "Whoa, something's on my left!", "Left it is!"],
    "right": ["Looking right!", "Something's caught my eye on the right!", "Right this way!"],
}
 
 
def clean_for_speech(text):
    text = EMOJI_PATTERN.sub("", text)
    text = re.sub(r"[*_~`#]", "", text)
    return re.sub(r"\s+", " ", text).strip()
 
 
def ensure_piper_voice_ready(voice_name, voices_dir):
    voices_dir.mkdir(parents=True, exist_ok=True)
    m_path, c_path = voices_dir / f"{voice_name}.onnx", voices_dir / f"{voice_name}.onnx.json"
    if not (m_path.exists() and c_path.exists()):
        print(f"[TTS] downloading voice '{voice_name}'...")
        download_voice(voice_name, voices_dir)
    print(f"[TTS] voice '{voice_name}' ready")
    return m_path, c_path
 
 
def say(piper_voice, pyaudio_instance, text):
    print(f"[TTS] speaking: '{text}'")
    syn_config = SynthesisConfig(length_scale=PIPER_LENGTH_SCALE, volume=PIPER_VOLUME)
    stream = None
    try:
        for chunk in piper_voice.synthesize(clean_for_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()
    print("[TTS] done speaking")
 
 
def _fallback_line(direction):
    return random.choice(FALLBACK_LINES.get(direction, [f"Looking {direction}!"]))
 
 
def _sanitize_line(raw):
    line = raw.strip().splitlines()[0].strip() if raw.strip() else ""
    line = re.sub(r'^(sure|okay|ok|here you go|here\'s one|here is one)[:,]?\s*', "", line, flags=re.IGNORECASE)
    return line.strip('"\'').strip()
 
 
def _is_valid_line(line, direction):
    if not line:
        return False
    words = line.split()
    # Reject if too short (< 3 words) or too long (> MAX_LINE_WORDS)
    if len(words) < 3 or len(words) > MAX_LINE_WORDS:
        return False
    if direction.lower() not in line.lower():
        return False
    if any(bad in line.lower() for bad in ("i cannot", "i can't", "as an ai", "i'm sorry")):
        return False
    return True
 
 
TRIGGERS = [
    "you heard a strange squeak",
    "you saw something shiny move",
    "you caught a sudden shadow",
    "you thought someone dropped a snack",
    "you heard a mystery whisper",
]
 
def gen_line(direction):
    trigger = random.choice(TRIGGERS)
 
    system_prompt = (
        "You are a curious, quirky robot exploring a room. You speak in concise, "
        "dramatic observations. Output ONLY one line. No markdown, quotes, or emoji."
    )
    
    # Use {direction} in the examples so the model sees the correct target word!
    user_prompt = (
        f"You turned your head to the {direction} because {trigger}.\n\n"
        f"Examples of how you sound when looking {direction}:\n"
        f"- Whoa, what shiny thing is on my {direction}?\n"
        f"- Did you hear that noise on my {direction}?\n"
        f"- Hey, something definitely moved on my {direction}!\n\n"
        f"Write ONE short line (4 to 8 words) for looking to the {direction}:"
    )
 
    print(f"[LLM] requesting line for direction='{direction}' (trigger: '{trigger}')")
    try:
        r = ollama.chat(
            model=MODEL,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt},
            ],
            think=False,
            options={"temperature": 0.6, "top_p": 0.9, "num_predict": 30},
        )
        line = _sanitize_line(r["message"]["content"])
 
        if not _is_valid_line(line, direction):
            print(f"[LLM] output failed validation ('{line}') — using fallback")
            return _fallback_line(direction)
 
        print(f"[LLM] got: '{line}'")
        return line
    except Exception as e:
        print(f"[LLM] error: {e} — falling back to default line")
        return _fallback_line(direction)
 
current_speech_text = ""  # Place near top of script with other state variables
 
def act(bot, direction, busy, piper_voice, pyaudio_instance):
    global current_speech_text
    print(f"[ACT] triggered — direction={direction}")
    
    # Generate line and update OpenCV display text
    line = gen_line(direction)
    current_speech_text = line
    
    threading.Thread(
        target=say,
        args=(piper_voice, pyaudio_instance, line),
        daemon=True,
    ).start()
    
    print(f"[ACT] issuing neck move: look_{direction}()")
    (bot.look_left if direction == "left" else bot.look_right)()
    time.sleep(1.5)
    print("[ACT] returning neck to center")
    bot.look_center()
    busy.clear()
    print("[ACT] done, busy cleared")
 
 
def head_direction(kconf):
    l_vis, r_vis = float(kconf[LEFT_EAR]), float(kconf[RIGHT_EAR])
    if r_vis < VIS_THRESHOLD <= l_vis:
        return "left"
    if l_vis < VIS_THRESHOLD <= r_vis:
        return "right"
    return "center"
 
 
def get_primary_person(results):
    r = results[0]
    if r.keypoints is None or r.boxes is None or len(r.boxes) == 0:
        return None
    confs = r.boxes.conf.cpu().numpy()
    best_idx = int(np.argmax(confs))
    if confs[best_idx] < PERSON_CONF_THRESHOLD:
        return None
    xy = r.keypoints.xy[best_idx].cpu().numpy()
    kconf = r.keypoints.conf[best_idx].cpu().numpy() if r.keypoints.conf is not None else np.ones(17)
    return xy, kconf, float(confs[best_idx])
 
 
def draw_person(frame, xy, kconf):
    for a, b in COCO_SKELETON:
        if kconf[a] >= VIS_THRESHOLD and kconf[b] >= VIS_THRESHOLD:
            cv2.line(frame, tuple(xy[a].astype(int)), tuple(xy[b].astype(int)), (0, 255, 0), 2)
    for i, conf in enumerate(kconf):
        if conf >= VIS_THRESHOLD:
            cv2.circle(frame, tuple(xy[i].astype(int)), 4, (0, 255, 0), -1)
 
 
class PoseWorker:
    def __init__(self, model, img_size):
        self.model, self.img_size = model, img_size
        self._lock = threading.Lock()
        self._latest_frame, self._direction, self._person = None, "center", None
        self._seq, self._inference_count, self._none_person_count = 0, 0, 0
        self._running = True
        self._thread = threading.Thread(target=self._loop, daemon=True)
        self._thread.start()
 
    def submit_frame(self, frame):
        with self._lock:
            self._latest_frame = frame
 
    def get_state(self):
        with self._lock:
            return self._direction, self._seq, self._person
 
    def _loop(self):
        while self._running:
            with self._lock:
                frame, self._latest_frame = self._latest_frame, None
            if frame is None:
                time.sleep(0.005)
                continue
 
            results = self.model.predict(frame, imgsz=self.img_size, verbose=False)
            found = get_primary_person(results)
            self._inference_count += 1
 
            if found is None:
                self._none_person_count += 1
                direction, person = "center", None
            else:
                xy, kconf, _ = found
                direction, person = head_direction(kconf), (xy, kconf)
 
            if RAW_LOG_INTERVAL > 0 and self._inference_count % RAW_LOG_INTERVAL == 0:
                print(f"[DEBUG] inference#{self._inference_count} person_found={found is not None} direction={direction}")
 
            with self._lock:
                self._direction, self._person, self._seq = direction, person, self._seq + 1
 
    def stop(self):
        self._running = False
        self._thread.join(timeout=2.0)
 
 
print("[INIT] loading and warming up pose model...")
pose_model = YOLO(POSE_MODEL_PATH)
pose_model.predict(np.zeros((480, 640, 3), dtype=np.uint8), imgsz=POSE_IMG_SIZE, verbose=False)
 
print("[INIT] setting up Piper voice...")
piper_m_path, piper_c_path = ensure_piper_voice_ready(PIPER_VOICE, PIPER_VOICES_DIR)
piper_voice = PiperVoice.load(piper_m_path, config_path=piper_c_path, use_cuda=PIPER_USE_CUDA)
pyaudio_instance = pyaudio.PyAudio()
 
pose_worker = PoseWorker(pose_model, POSE_IMG_SIZE)
 
with BonicBot(host=HOST) as bot:
    print("[INIT] starting camera...")
    bot.system.start_camera()
    bot.start_camera()
    bot.camera.wait_for_image(timeout=5.0)
    print("[INIT] camera ready. Look left/right. Press 'q' to quit.")
 
    busy = threading.Event()
    confirm_count, last_dir, confirm_start_time = 0, None, None
    last_trigger, last_time, last_seen_seq = None, 0.0, -1
    frame_count, none_frame_count = 0, 0
 
    try:
        while True:
            frame_count += 1
            frame = bot.get_image()
            if frame is None:
                none_frame_count += 1
                continue
 
            pose_worker.submit_frame(frame)
            direction, seq, person = pose_worker.get_state()
 
            if seq != last_seen_seq:
                last_seen_seq = seq
                now = time.time()
                if not busy.is_set() and direction in ("left", "right"):
                    if direction == last_dir:
                        confirm_count += 1
                    else:
                        last_dir, confirm_count, confirm_start_time = direction, 1, now
 
                    held_seconds = now - confirm_start_time
                    on_cooldown = (direction == last_trigger and now - last_time < COOLDOWN)
 
                    if confirm_count >= MIN_CONFIRM_SAMPLES and held_seconds >= CONFIRM_HOLD_SECONDS and not on_cooldown:
                        print(f"[STATE] CONFIRMED {direction} (held {held_seconds:.2f}s, {confirm_count} samples) — firing action")
                        busy.set()
                        threading.Thread(target=act, args=(bot, direction, busy, piper_voice, pyaudio_instance), daemon=True).start()
                        last_trigger, last_time = direction, now
                        confirm_count, last_dir, confirm_start_time = 0, None, None
                else:
                    confirm_count, last_dir, confirm_start_time = 0, None, None
 
            display = frame
            if person is not None:
                draw_person(display, *person)
 
            # Top label for Head direction
            cv2.putText(display, f"Head: {direction}", (20, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2)
 
            # Render LLM response text at the bottom
            if current_speech_text:
                h, w = display.shape[:2]
                # Black banner overlay for contrast
                cv2.rectangle(display, (0, h - 45), (w, h), (0, 0, 0), -1)
                # Yellow subtitle text centered near the bottom
                cv2.putText(
                    display, 
                    current_speech_text, 
                    (15, h - 15), 
                    cv2.FONT_HERSHEY_SIMPLEX, 
                    0.6, 
                    (0, 255, 255), 
                    2, 
                    cv2.LINE_AA
                )
 
            cv2.imshow("BonicBot Head Sync", display)
            if cv2.waitKey(1) & 0xFF == ord("q"):
                print("[LOOP] quit key pressed")
                break
    except BonicBotError as e:
        print(f"[ERROR] Robot error: {e}")
    finally:
        print(f"[SUMMARY] frames={frame_count} none_frames={none_frame_count} inferences={pose_worker._inference_count} none_person={pose_worker._none_person_count}")
        print("[SHUTDOWN] cleaning up...")
        pose_worker.stop()
        cv2.destroyAllWindows()
        bot.stop_camera()
        bot.system.stop_camera()
        pyaudio_instance.terminate()
        print("[SHUTDOWN] done")

Before running the program, make sure Ollama is installed on your computer (one-time step):

Windows / Mac: download and run the installer from ollama.com/download .
Linux: run curl -fsSL https://ollama.com/install.sh | sh once in a terminal.

Also download the Piper voice before your first run:


python3 -m piper.download_voices --data-dir ./piper_voices en_US-lessac-medium

And install the required Python packages:


pip install ollama opencv-python numpy piper-tts pyaudio ultralytics --break-system-packages

The first run will also automatically download the yolo11n-pose.pt weights file.

Code Walkthrough

Line-by-line explanation

Imports & Globals (lines 38–78) — Imports cv2, numpy, ollama, pyaudio, PiperVoice, YOLO, BonicBot, and regex helpers. Sets configuration defaults including HOST, MODEL = "qwen3:0.6b", yolo11n-pose.pt, COCO skeleton keypoints (NOSE, LEFT_EAR, RIGHT_EAR), and fallback responses.
clean_for_speech(text) (lines 80–84) — Strips emojis, markdown syntax (*, _, #), and redundant whitespace from LLM text so Piper TTS receives clean readable text.
ensure_piper_voice_ready(voice_name, voices_dir) (lines 86–94) — Checks if the requested .onnx voice model and config exist locally; downloads them automatically if missing.
say(piper_voice, pyaudio_instance, text) (lines 96–114) — Synthesizes speech using piper_voice.synthesize() and streams raw audio chunks directly to an active pyaudio output stream.
_sanitize_line, _is_valid_line, and gen_line(direction) (lines 119–187) — Queries the local Ollama LLM (ollama.chat) with system and user prompts to generate a 4–8 word observation for the detected look direction. Validates output formatting and returns a fallback line if validation fails.
act(bot, direction, busy, piper_voice, pyaudio_instance) (lines 190–211) — Action dispatcher triggered upon head turn confirmation. Spawns gen_line(), updates on-screen subtitles (current_speech_text), initiates speech on a daemon thread, and commands the robot neck (bot.look_left() or bot.look_right()) before centering (bot.look_center()).
head_direction(kconf) (lines 213–219) — Evaluates keypoint confidence scores for left and right ears (kconf[LEFT_EAR] vs kconf[RIGHT_EAR]). If right ear confidence drops below VIS_THRESHOLD while left ear remains visible, infers a head turn to the left (and vice versa).
get_primary_person(results) & draw_person(frame, xy, kconf) (lines 222–242) — Selects the person with the highest detection score from YOLO output and draws COCO skeleton connections and keypoints on the display frame.
PoseWorker class (lines 244–290) — Asynchronous worker thread that runs YOLO-pose inference on camera frames in a background loop (_loop()), ensuring pose estimation never slows down the main display/camera thread.
Initialization & Main Loop (lines 292–386) — Warms up yolo11n-pose.pt, prepares Piper voice and PyAudio, starts PoseWorker, opens BonicBot connection and camera stream, and monitors look direction debouncing (MIN_CONFIRM_SAMPLES and CONFIRM_HOLD_SECONDS). When confirmed, launches act() on a background thread and renders HUD elements with live subtitles.

Expected Output

Click to see expected output

Visual Output:

📌 Note: This visual demonstration is using the ROS 2 simulation with use_real_camera:=True.

Terminal Output:


[INIT] loading and warming up pose model...
[INIT] setting up Piper voice...
[INIT] starting camera...
[INIT] camera ready. Look left/right. Press 'q' to quit.

A window titled “BonicBot Head Sync” opens showing the camera feed with:

A green skeleton overlay showing the tracked person’s pose
The detected head direction overlay (Head: left/right/center) in the top-left corner
Spoken subtitles displayed on a black banner at the bottom of the window

When you turn your head left or right and hold it, BonicBot confirms the direction, fires the physical head movement (look_left or look_right), and speaks a quirky robotic observation, for example:


[STATE] CONFIRMED left (held 0.45s, 3 samples) — firing action
[LLM] requesting line for direction='left' (trigger: 'you caught a sudden shadow')
[LLM] got: 'Did you hear that noise on my left?'
[TTS] speaking: 'Did you hear that noise on my left?'

Press q in the OpenCV window at any time to quit the program.

🔧 Under the Hood

How does BonicBot mirror head movements and talk without the cloud?

Unlike systems that rely on cloud APIs, BonicBot does everything — seeing, deciding, moving, and speaking — entirely on your own computer.

Seeing: The YOLO-pose model estimates 17 keypoints per frame. The PoseWorker runs these predictions asynchronously in a background thread to prevent the camera loop from freezing. head_direction() determines look direction by checking the visibility of your left and right ears (LEFT_EAR and RIGHT_EAR). If only one ear is visible, it infers the head is turned in that direction.

Tracking & Debouncing: In the main loop, we track consecutive frames of the same head direction. To prevent accidental triggers from nose/ear detection noise, the look direction must be held for a minimum number of samples (MIN_CONFIRM_SAMPLES) and duration (CONFIRM_HOLD_SECONDS) before BonicBot acts. A cooldown timer (COOLDOWN) prevents rapid duplicate triggers.

Thinking: Once a head movement is confirmed, the main thread spawns a background thread to call act(). This invokes gen_line(), which uses Ollama locally to feed a system prompt and a user prompt (based on a random trigger and look direction) to the qwen3:0.6b model. The output is validated to ensure it contains the target direction.

Speaking & Moving: BonicBot runs the TTS engine (Piper) on a background thread via pyaudio to announce the LLM’s quirky observation. Simultaneously, the neck turns physically (look_left() or look_right()) using BonicBot API calls, pauses for 1.5 seconds, and then centers back (look_center()).

This see-think-speak-act pipeline showcases a real-time multimodal loop running entirely on edge hardware.

Student Challenge

Modify the program so BonicBot has more complex reactions or behaves differently based on head tracking.

For example:

Add a double-take action: If the user looks in the same direction twice in a short period, make BonicBot do a double-take (rapidly look, return to center, and look again).
Incorporate a head nod/shake: Track the nose relative to the ears to detect vertical head nodding (yes) or horizontal shaking (no), and have BonicBot say a corresponding response.
Custom robotic persona: Rewrite the system_prompt in gen_line() to give BonicBot a completely different personality, like an anxious butler or a sassy security bot.

Hint

For a double-take, you can add a timestamp tracker last_trigger_time and check if the new trigger direction matches the previous one within a 5-second window:


if direction == last_trigger and now - last_trigger_time < 5.0:
    # Trigger double-take neck sequence
    bot.look_left()
    time.sleep(0.5)
    bot.look_center()
    time.sleep(0.2)
    bot.look_left()

Reflection Question

Why is running PoseWorker on a separate thread from the camera and physical action loop important? What would happen to the video stream and responsiveness if all predictions and actions ran sequentially on the main loop thread?