Capstone 2 — BonicBot Library Checkout Assistant

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

import time, csv, pickle, threading, queue, sys, subprocess, importlib, os, wave, io
from pathlib import Path
import cv2, numpy as np
import sounddevice as sd
from piper import PiperVoice, SynthesisConfig
 
# ---- config ----
PIPER_VOICE = "en_US-lessac-medium"
PIPER_VOICE_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "piper_voices")
FACE_DB_PATH, LOG_PATH = 'face_database.pkl', 'checkout_log.csv'
FACE_MATCH_THRESHOLD, DETECTOR_BACKEND = 0.30, "opencv"
RECOGNIZE_EVERY_N = 5
SESSION_TIMEOUT = 30.0
GREEN_HUE_MIN, GREEN_HUE_MAX = 35, 85
BLUE_HUE_MIN, BLUE_HUE_MAX = 100, 130
FACE_MATCH_WINDOW, FACE_MATCH_RETRY_INTERVAL = 10.0, 0.15
FACE_EARLY_EXIT_SCORE = 0.45  # stop polling early once we're confidently past threshold
 
latest_frame, frame_lock = None, threading.Lock()
 
def set_latest_frame(frame):
    global latest_frame
    with frame_lock: latest_frame = frame.copy()
 
def get_latest_frame():
    with frame_lock: return None if latest_frame is None else latest_frame.copy()
 
def ensure_piper_voice(voice):
    os.makedirs(PIPER_VOICE_DIR, exist_ok=True)
    model_path = os.path.join(PIPER_VOICE_DIR, f"{voice}.onnx")
    if not (os.path.exists(model_path) and os.path.exists(model_path + ".json")):
        subprocess.run([sys.executable, "-m", "piper.download_voices", "--data-dir", PIPER_VOICE_DIR, voice], check=True)
    return model_path
 
# ---- speech ----
class VoiceSpeaker:
    def __init__(self, voice, volume=2.5):
        self._voice, self._cfg, self._q = voice, SynthesisConfig(length_scale=0.85, noise_scale=0.9, volume=volume), queue.Queue()
        threading.Thread(target=self._run, daemon=True).start()
    def _run(self):
        while (text := self._q.get()) is not None:
            try: self._speak(text)
            except Exception as ex: print(f"⚠️ TTS error: {ex}")
    def _speak(self, text):
        with io.BytesIO() as buf:
            with wave.open(buf, "wb") as wf: self._voice.synthesize_wav(text, wf, syn_config=self._cfg)
            buf.seek(0)
            with wave.open(buf, "rb") as wf:
                ch, rate = wf.getnchannels(), wf.getframerate()
                frames = wf.readframes(wf.getnframes())
        audio = np.frombuffer(frames, dtype=np.int16)
        if ch > 1: audio = audio.reshape(-1, ch)
        sd.play(audio, samplerate=rate); sd.wait()
    def speak(self, text): self._q.put(text)
    def stop(self): self._q.put(None)
 
# ---- body language (left-arm gesture) ----
ARM_REST, ARM_GESTURE, GESTURE_HOLD = (0, 0), (40, 25), 0.6
gesture_busy = threading.Event()
 
def _gesture_worker(bot):
    try:
        bot.move_left_arm(*ARM_GESTURE, wait=True); time.sleep(GESTURE_HOLD); bot.move_left_arm(*ARM_REST, wait=True)
    except Exception as e: print(f"⚠️ Gesture error: {e}")
    finally: gesture_busy.clear()
 
def gesture(bot):
    if bot is None or gesture_busy.is_set(): return
    gesture_busy.set(); threading.Thread(target=_gesture_worker, args=(bot,), daemon=True).start()
 
def announce(speaker, text, bot=None):
    print(f"🔊 {text}"); speaker.speak(text); gesture(bot)

# ---- badge detection ----
def badge_color(frame, lo, hi):
    h, w = frame.shape[:2]; cx, cy, half = w//2, h//2, 40
    patch = frame[max(0,cy-half):cy+half, max(0,cx-half):cx+half]
    if patch.size == 0: return False
    hsv = cv2.cvtColor(patch, cv2.COLOR_BGR2HSV)
    sat, val, hue = hsv[:,:,1].mean(), hsv[:,:,2].mean(), hsv[:,:,0].mean()
    return sat >= 120 and val >= 100 and lo <= hue <= hi
 
# ---- face setup ----
def ensure(mod, pkg=None):
    try: return importlib.import_module(mod)
    except ImportError:
        subprocess.check_call([sys.executable, "-m", "pip", "install", pkg or mod]); return importlib.import_module(mod)
 
def pick_detector_backend(DeepFace):
    if hasattr(cv2, "CascadeClassifier"): return "opencv"
    for backend, pkg, mod in [("mediapipe","mediapipe","mediapipe"), ("mtcnn","mtcnn","mtcnn"), ("retinaface","retina-face","retinaface")]:
        try:
            ensure(mod, pkg)
            DeepFace.represent(np.zeros((100,100,3),np.uint8), model_name="ArcFace", detector_backend=backend, enforce_detection=False, align=True)
            print(f"✅ Using '{backend}' detector backend."); return backend
        except Exception as e: print(f"   ↳ '{backend}' failed: {e}")
    sys.exit("❌ No working face detector found.")
 
def load_face_database(path):
    if not Path(path).exists(): return {}
    with open(path, "rb") as f: raw = pickle.load(f)
    if isinstance(raw, dict) and "identities" in raw and "embeddings" in raw:
        db = {}
        for n, e in zip(raw["identities"], raw["embeddings"]):
            v = np.array(e, dtype=np.float32); norm = np.linalg.norm(v)
            if norm: db.setdefault(n, []).append(v / norm)
        return db
    if isinstance(raw, dict):
        # Defensive: raw dict format is NOT guaranteed to be pre-normalized.
        # match_face() uses a plain dot product assuming unit vectors, so
        # un-normalized entries here will silently produce wrong similarity
        # scores. Normalize defensively and warn if anything was off.
        db, fixed = {}, 0
        for name, vecs in raw.items():
            norm_vecs = []
            for e in vecs:
                v = np.array(e, dtype=np.float32); norm = np.linalg.norm(v)
                if norm:
                    if abs(norm - 1.0) > 1e-3: fixed += 1
                    norm_vecs.append(v / norm)
            db[name] = norm_vecs
        if fixed:
            print(f"⚠️ face_database.pkl: {fixed} embedding(s) were not unit-normalized; normalized on load. "
                  f"Consider re-saving the pkl with normalization baked in to avoid ambiguity.")
        return db
    sys.exit(f"❌ Bad face_database.pkl: {type(raw)}")
 
def match_face(frame, face_db):
    """Returns (best_name, best_score). best_name is None only if no face
    was detected at all. If a face is detected but doesn't clear the
    threshold, best_name will still be the closest match with its score,
    and threshold decisions are left to the caller."""
    from deepface import DeepFace
    try: r = DeepFace.represent(frame, model_name="ArcFace", detector_backend=DETECTOR_BACKEND, enforce_detection=True)
    except ValueError: return None, None
    emb = np.array(r[0]["embedding"], dtype=np.float32); emb /= np.linalg.norm(emb) + 1e-9
    best_name, best_score = None, -1.0
    for name, vecs in face_db.items():
        for v in vecs:
            s = float(np.dot(emb, v))
            if s > best_score: best_name, best_score = name, s
    return best_name, best_score

# ---- session lock (UNKNOWN/no-face never calls start_session, so active stays False -> blue badge retry works) ----
busy = threading.Event()
session_lock = threading.Lock()
session = {"name": None, "active": False, "last_activity": 0.0, "items": set(), "last_item": None}
 
def start_session(name):
    with session_lock:
        session["name"], session["active"], session["last_activity"], session["items"], session["last_item"] = name, True, time.time(), set(), None
 
def end_session():
    with session_lock:
        session["name"], session["active"], session["items"], session["last_item"] = None, False, set(), None
 
def touch_session():
    with session_lock: session["last_activity"] = time.time()
 
def session_snapshot():
    with session_lock: return session["active"], session["name"], session["last_item"]
 
def session_timed_out():
    with session_lock: return session["active"] and (time.time() - session["last_activity"] > SESSION_TIMEOUT)
 
def mark_checked_this_session(item_id):
    with session_lock: session["items"].add(item_id)
 
def checked_this_session(item_id):
    with session_lock: return item_id in session["items"]
 
def face_worker(get_frame_fn, face_db, speaker, bot):
    """Polls for up to FACE_MATCH_WINDOW seconds, tracking the BEST score
    seen across all frames rather than stopping at the first frame that
    produces a detection. This matters because per-frame ArcFace similarity
    naturally jitters (blink, micro-angle, motion blur, compression noise)
    even under identical lighting on the same face -- a single low-scoring
    frame should not end the attempt early."""
    deadline = time.time() + FACE_MATCH_WINDOW
    best_name, best_score, frames_with_face, attempts = None, -1.0, 0, 0
 
    while time.time() < deadline:
        frame = get_frame_fn()
        if frame is not None:
            attempts += 1
            try:
                cand_name, cand_score = match_face(frame, face_db)
            except Exception as e:
                print(f"⚠️ match_face error on attempt {attempts}: {e}")
                cand_name, cand_score = None, None
            if cand_name is not None:
                frames_with_face += 1
                print(f"   ↳ [{frames_with_face}] candidate={cand_name} score={cand_score:.3f}")
                if cand_score > best_score:
                    best_name, best_score = cand_name, cand_score
                if best_score >= FACE_EARLY_EXIT_SCORE:
                    break
        time.sleep(FACE_MATCH_RETRY_INTERVAL)
 
    print(f"🔍 face_worker done: attempts={attempts} frames_with_face={frames_with_face} "
          f"best={best_name} best_score={(f'{best_score:.3f}' if best_score > -1.0 else 'n/a')}")
 
    if best_score >= FACE_MATCH_THRESHOLD:
        name = best_name
        start_session(name)
        items = person_items.get(name)
        announce(speaker, f"Welcome back {name}, you have {len(items)} item(s) out. Scan to return." if items else f"Hi {name}, go ahead and scan your items.", bot)
    elif frames_with_face > 0:
        announce(speaker, "I don't recognize you, please try again or register.", bot)
    else:
        announce(speaker, "I couldn't see a face clearly, please try again.", bot)
    busy.clear()

# ---- checkout logic ----
checked_out, person_items = {}, {}
 
def rebuild_state():
    checked_out, person_items = {}, {}
    if not Path(LOG_PATH).exists(): return checked_out, person_items
    with open(LOG_PATH, newline="") as f:
        for row in csv.DictReader(f):
            item, name = row["item"], row["name"]
            if row["action"] == "checkout":
                checked_out[item] = {"name": name, "time": row["timestamp"]}; person_items.setdefault(name, set()).add(item)
            elif row["action"] == "return":
                checked_out.pop(item, None); person_items.get(name, set()).discard(item)
    return checked_out, person_items
 
def log_event(action, item_id, name):
    exists = Path(LOG_PATH).exists()
    with open(LOG_PATH, "a", newline="") as f:
        w = csv.writer(f)
        if not exists: w.writerow(["timestamp","action","item","name"])
        w.writerow([time.strftime("%Y-%m-%d %H:%M:%S"), action, item_id, name])
 
def handle_scan(item_id, name, speaker, bot):
    if checked_this_session(item_id): return
    mark_checked_this_session(item_id)
    with session_lock: session["last_item"] = item_id
    if item_id in checked_out:
        checked_out.pop(item_id); person_items.get(name, set()).discard(item_id)
        log_event("return", item_id, name); announce(speaker, f"{item_id} returned, thanks {name}.", bot)
    else:
        checked_out[item_id] = {"name": name, "time": time.time()}
        person_items.setdefault(name, set()).add(item_id)
        log_event("checkout", item_id, name); announce(speaker, f"{item_id} checked out for {name}.", bot)

# ---- main loop ----
def main():
    global checked_out, person_items, DETECTOR_BACKEND
    checked_out, person_items = rebuild_state()
    face_db = load_face_database(FACE_DB_PATH)
    from deepface import DeepFace
    DETECTOR_BACKEND = pick_detector_backend(DeepFace)
 
    print("🔧 Warming up face recognition model...")
    try:
        DeepFace.represent(np.zeros((100, 100, 3), np.uint8), model_name="ArcFace", detector_backend=DETECTOR_BACKEND, enforce_detection=False, align=True)
        print("✅ Face model warmed up.")
    except Exception as e: print(f"⚠️ Warmup failed: {e}")
 
    from bonicbot_bridge import BonicBot, BonicBotError
    speaker = VoiceSpeaker(PiperVoice.load(ensure_piper_voice(PIPER_VOICE)))
    qr_detector = cv2.QRCodeDetector()
    seen_tag, frame_count = None, 0
 
    try:
        with BonicBot(host='localhost', port=9090, timeout=10) as bot:
            bot.system.start_camera(); bot.start_camera(); bot.camera.wait_for_image(timeout=5.0)
            print("✅ Checkout assistant online. Press 'q' to stop.")
 
            while True:
                frame = bot.get_image()
                if frame is None: continue
                frame_count += 1
                set_latest_frame(frame)
 
                active, locked_name, last_item = session_snapshot()
 
                if not active:
                    if badge_color(frame, BLUE_HUE_MIN, BLUE_HUE_MAX) and not busy.is_set():
                        busy.set()
                        threading.Thread(target=face_worker, args=(get_latest_frame, face_db, speaker, bot), daemon=True).start()
                else:
                    if badge_color(frame, GREEN_HUE_MIN, GREEN_HUE_MAX):
                        log_event("release", "-", locked_name)
                        announce(speaker, f"Thanks {locked_name}, ready for the next person.", bot)
                        end_session()
                    else:
                        found, texts, points, _ = qr_detector.detectAndDecodeMulti(frame)
                        tag = next((t.strip().upper() for t in (texts or []) if t and t.strip()), None) if found else None
                        if tag and tag != seen_tag:
                            handle_scan(tag, locked_name, speaker, bot); touch_session()
                        seen_tag = tag
 
                    if session_timed_out():
                        announce(speaker, f"Session timed out for {locked_name}.", bot); end_session()
 
                active, locked_name, last_item = session_snapshot()
                text = (f"{locked_name} | {last_item}" if last_item else locked_name) if active else \
                       "Verifying..." if busy.is_set() else \
                       "Section started" if badge_color(frame, BLUE_HUE_MIN, BLUE_HUE_MAX) else "Waiting for next section"
 
                cv2.putText(frame, text, (10,30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0,255,0), 2)
                h, w = frame.shape[:2]; cx, cy, half = w // 2, h // 2, 40
                cv2.rectangle(frame, (cx - half, cy - half), (cx + half, cy + half), (255, 255, 0), 2)
                cv2.imshow("Lab Checkout", frame)
                if cv2.waitKey(1) & 0xFF == ord('q'): break
 
            bot.stop_camera(); bot.system.stop_camera()
    except BonicBotError as e: print(f"⚠️ Robot error: {e}")
    finally: cv2.destroyAllWindows(); speaker.stop()
 
if __name__ == "__main__":
    main()

🔧 Warming up face recognition model...
✅ Face model warmed up.
✅ Checkout assistant online. Press 'q' to stop.
   ↳ [1] candidate=Alice score=0.812
🔍 face_worker done: attempts=2 frames_with_face=2 best=Alice best_score=0.812
🔊 Hi Alice, go ahead and scan your items.
🔊 BOOK001 checked out for Alice.
🔊 BOOK002 checked out for Alice.
🔊 Thanks Alice, ready for the next person.

timestamp,action,item,name
2026-07-24 10:15:32,checkout,BOOK001,Alice
2026-07-24 10:16:05,checkout,BOOK002,Alice
2026-07-24 10:20:41,release,-,Alice

KNOWN_BOOKS = {
    "BOOK001",
    "BOOK002",
    "BOOK003",
    # ...
}