# Project export: Engram This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small. ## Project metadata - Hackathon: UC Berkeley AI Hackathon 2026 - Tagline: The memory layer for physical AI: a humanoid robot that perceives with a 72B vision model on Nebius, remembers objects as vectors in Redis, can be taught in one shot, and walks to recall it. - Devpost: https://devpost.com/software/engram-fv5a2k - GitHub: https://github.com/A-Mundanilkunathil/engram - Video: https://www.youtube.com/embed/iftPF4AKKVI?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Team: 0 GitHub contributor(s) — ## Devpost submission (written by the team) ### Inspiration Today's robots are brilliant at perceiving a scene and amnesiac about it a second later. A vision model can tell you "that's a mug," but it can't be taught "this is my mug" without retraining, and it has no memory of where it saw things. We wanted to build the missing piece for physical AI: a memory layer, instant, teachable, and spatial, that sits between a robot's eyes and its actions. The north-star question we kept asking: "Would a real robotics team actually use this?" ### What it does Engram gives a Booster K1 humanoid a cloud brain and instant memory: Perceives with a 72-billion-parameter vision model (Qwen2.5-VL) running on Nebius GPUs. It captions whatever the robot's camera sees. Remembers every object as a vector in Redis, with sub-2 ms nearest-neighbor recall over its whole memory bank. Learns in one shot. Show it a brand-new object it's never seen, teach it once (one vector written to Redis), and it recognizes it instantly. No retraining. Recognizes and attends. It turns its head to look at the object it recalls, tracking the object's position continuously. Has spatial memory. Walk it through a scene and it captures every object and the heading where it saw it. Fetches on command. Say "go to the red soda can" and it first checks whether it even remembers one, recalls where it was, walks toward it (closed-loop on odometry), stops at a safe distance using the robot's depth camera so it doesn't crash into the table, faces it, and waves. You can watch the robot's-eye view and its live decision (recalled object, confidence, Redis search latency, the Nebius model) on a real-time dashboard. ### How we built it Architecture, a thin robot plus a cloud brain: Perception (Nebius Token Factory): OpenAI-compatible API. Qwen/Qwen2.5-VL-72B-Instruct turns a camera frame into a short caption; Qwen/Qwen3-Embedding-8B turns the caption into a 4096-d vector. One VLM call also returns the object's horizontal center for tracking. Memory (Redis Stack): a FLAT cosine vector index over the embeddings, plus numeric fields storing the robot's pose (x, y, θ) where each object was seen. KNN recall runs in ~2 ms over a 124-object bank. A text query ("is the green bottle here?") embeds and searches the same index. Robot (Booster K1): ROS2 Humble + booster_robotics_sdk_python. RGB from /booster_video_stream, metric depth from /StereoNetNode/stereonet_depth (mono16, millimeters), odometry, and high-level motion (RotateHead, Move, WaveHand). Closed-loop locomotion: turns and walks are odometry-feedback controlled (we calibrated a slip factor because the K1's odometry over-reports rotation ~1.4× from foot slip), so "turn 180°" actually turns 180°. Depth-gated safe approach: during the final walk, we sample the nearest obstacle in the depth image and stop the robot a safe distance from the object before it can hit the table. Robust recognition: an object captured from several angles gets several captions, so we match the live view to the target by semantic similarity, not exact strings. Live demo dashboard: the brain serves an HTML/JS page showing the robot's view, the recalled label + confidence, a gaze marker, the Redis search latency, and the Nebius model, the "decision on screen." Laptop ↔ robot connected over Tailscale so the robot can be untethered. ### Challenges we ran into Laptop ↔ robot connected over Tailscale so the robot can be untethered. Challenges we ran into Firmware reality: the robot's SDK build didn't implement the Cartesian arm-control API (returned 501 Not Implemented), so our planned "point at it" pivoted to a head-look + walk-up + wave that the firmware does support. Firmware reality: the robot's SDK build didn't implement the Cartesian arm-control API (returned 501 Not Implemented), so our planned "point at it" pivoted to a head-look + walk-up + wave that the firmware does support. Odometry slip: open-loop turns were wildly inaccurate; we switched to closed-loop odometry feedback and calibrated the rotation scale so turns land on target. Odometry slip: open-loop turns were wildly inaccurate; we switched to closed-loop odometry feedback and calibrated the rotation scale so turns land on target. Depth that reads the wrong thing: a naive center-of-frame depth sample looked over the table at the far wall, so the robot didn't stop. We moved to a nearest-obstacle sample over a wider, lower band. Depth that reads the wrong thing: a naive center-of-frame depth sample looked over the table at the far wall, so the robot didn't stop. We moved to a nearest-obstacle sample over a wider, lower band. One object, many captions: the same object captioned differently from different angles broke exact-label matching during approach, fixed with semantic matching. One object, many captions: the same object captioned differently from different angles broke exact-label matching during approach, fixed with semantic matching. Networking gauntlet: captive portals, a flaky Ethernet link, and hotspot/Tailscale relays. We ended up on Tailscale so the robot could roam. Networking gauntlet: captive portals, a flaky Ethernet link, and hotspot/Tailscale relays. We ended up on Tailscale so the robot could roam. Staging perception: the VLM captions the dominant object, so objects have to fill enough of the frame, a real lesson in demo design. ### Accomplishments we're proud of Staging perception: the VLM captions the dominant object, so objects have to fill enough of the frame, a real lesson in demo design. Accomplishments that we're proud of One-shot live learning that actually works. Teach a brand-new object once and it's known instantly. A vision model alone can't do this. One-shot live learning that actually works. Teach a brand-new object once and it's known instantly. A vision model alone can't do this. A genuinely large model, 72B parameters, running in the live robot loop on Nebius, not a toy API call. A genuinely large model, 72B parameters, running in the live robot loop on Nebius, not a toy API call. Sub-2 ms memory recall in Redis over the whole object bank. Sub-2 ms memory recall in Redis over the whole object bank. A full spatial fetch: check memory, recall heading, walk, depth-safe stop, wave, built and tuned on real hardware in a single hack. A full spatial fetch: check memory, recall heading, walk, depth-safe stop, wave, built and tuned on real hardware in a single hack. Closed-loop humanoid navigation calibrated from scratch in hours. ### What we learned Closed-loop humanoid navigation calibrated from scratch in hours. What we learned Caption → embed → vector store is a remarkably robust, nearly-free perception path; it matches on caption consistency more than accuracy. Caption → embed → vector store is a remarkably robust, nearly-free perception path; it matches on caption consistency more than accuracy. Closed-loop beats open-loop for everything on a real robot. Trust the sensor, not the command. Closed-loop beats open-loop for everything on a real robot. Trust the sensor, not the command. Depth needs careful region sampling. "What's directly ahead of me" is not "the median of the center of the image." Depth needs careful region sampling. "What's directly ahead of me" is not "the median of the center of the image." The memory layer is the differentiator: perception is a commodity; a teachable, instant, spatial memory is what makes a robot feel like it understands its space. ### What's next The memory layer is the differentiator: perception is a commodity; a teachable, instant, spatial memory is what makes a robot feel like it understands its space. What's next for Engram Image embeddings (Path B): CLIP-style image vectors on a Nebius GPU to tell lookalikes apart (two similar cans). Image embeddings (Path B): CLIP-style image vectors on a Nebius GPU to tell lookalikes apart (two similar cans). Brain on the robot: run the whole pipeline on the K1's onboard Jetson for full autonomy, untethered. Brain on the robot: run the whole pipeline on the K1's onboard Jetson for full autonomy, untethered. True spatial navigation / SLAM: persistent maps so it can be walked through a building and sent anywhere. True spatial navigation / SLAM: persistent maps so it can be walked through a building and sent anywhere. Depth for 3D grasping: use the RGBD stream to reach and pick, not just point. Depth for 3D grasping: use the RGBD stream to reach and pick, not just point. Scale the memory: thousands of objects, still sub-ms. Scale the memory: thousands of objects, still sub-ms. ## README (from the GitHub repository) # Engram — the memory layer for physical AI A Booster K1 humanoid with a cloud brain and instant memory. It **perceives** with a 72B vision model on **Nebius**, **remembers** objects as vectors in **Redis**, can be **taught a new object in one shot**, and **walks to recall** it — checking memory, navigating to where it saw the object, stopping a safe distance away using its depth camera, then waving. Built at the AI Berkeley Hackathon 2026 (Ultimate Bots — Physical AI Hack). Targets the **Nebius** (best use of compute) and **Redis** (best use) prize tracks. ## How it works `robot camera → laptop brain (Nebius VLM caption → Qwen embedding → Redis vector KNN) → robot acts (head-look / walk / wave)` - **Perception:** `Qwen/Qwen2.5-VL-72B-Instruct` + `Qwen/Qwen3-Embedding-8B` on Nebius Token Factory (OpenAI-compatible API). - **Memory:** Redis Stack vector index (cosine KNN, ~2 ms), storing each object's caption embedding + the robot pose where it was seen. - **One-shot learning:** teach a new object live; one vector written; recognized instantly. - **Spatial fetch:** scan a scene → "go to the X" → check memory → walk (closed-loop odometry) → depth-safe stop → wave. - **Robot:** Booster K1, ROS2 Humble, RGB + stereo depth, odometry, RotateHead/Move/WaveHand. ## Key files | file | what | |---|---| | `engram_pipeline.py` | perception + Redis vector memory (brain core) | | `engram_server.py` | HTTP brain + live dashboard (`localhost:8000/`) | | `k1_fetch.py` | robot-side spatial fetch (`--scan` / `--goto`) | | `k1_locomotion.py` | odometry-feedback walking + turning | | `k1_engram.py`, `k1_indicate.py` | recognition + head-tracking loop | | `preload_bank.py` | pre-load a memory bank (Redis scale demo) | | `DEMO_RUNBOOK.md`, `DEVPOST_SUBMISSION.md` | how to run / writeup | ## Run (short) **Brain (laptop):** `export NEBIUS_API_KEY=... && python engram_server.py` (needs Redis Stack; dashboard at `:8000`). **Robot:** `python3 k1_fetch.py --reset && python3 k1_fetch.py --scan && python3 k1_fetch.py --goto "red soda can"` > Work in progress — full code + demo video coming. ## Detected evidence (automated analysis) Indexed codebase: 18 recognized source files, 101 KB. - Python (language) — detected in the code - JavaScript (language) — claimed on Devpost, not found in the code - Redis (technology) — claimed on Devpost, not found in the code - AI coding agent: Claude Code — evidence: commit authorship or trailers ## Codebase structure (from repository index) ### Files (19 of 19) ``` .gitignore camera_check.py DEMO_RUNBOOK.md DEVPOST_SUBMISSION.md engram_pipeline (1).py engram_pipeline.py engram_server.py k1_arm_lowlevel.py k1_engram.py k1_fetch.py k1_indicate.py k1_locomotion.py k1_point.py k1_walk_test.py live_test.py preflight_nebius.py preload_bank.py README.md smoke_test.py ``` ### Dependencies No dependency index available. ### Recent commits (newest first) - Polish live dashboard into a robotics telemetry HUD - Initial commit: Engram — the memory layer for physical AI ## Key source files (fetched from GitHub, selected and truncated for size) ### DEMO_RUNBOOK.md ```markdown # Engram — Demo Runbook **Engram = a memory layer for physical AI.** A robot perceives with a 72B vision model on **Nebius**, remembers objects as vectors in **Redis**, can be **taught a new object in one shot**, then recognizes it and turns its head to look at it. ``` robot camera ──JPEG──▶ laptop brain ──▶ Nebius Qwen2.5-VL-72B (caption) └──▶ Nebius Qwen3-Embedding (vector) └──▶ Redis vector KNN ──▶ {label, confidence, center_x} └──▶ robot turns head toward the object laptop screen = live dashboard (decision + Redis/Nebius numbers) ``` --- ## 0. One-time / already done - Redis Stack running on the laptop (`redis-cli ping` → PONG). - Robot reachable over Tailscale: `ssh booster@100.76.129.2` (sudo pw `123456`). - Robot internet via phone hotspot **"Aaron"** (keep it on). - Files deployed to robot `~/nebius_k1/`: `k1_engram.py`, `k1_indicate.py`. ## 1. Start the brain + dashboard (laptop) ```bash cd "~/Downloads/AI Berkeley Hackathon 2026" export NEBIUS_API_KEY=... # (it's in run.sh) .venv/bin/python engram_server.py # serves http://localhost:8000 ``` Open **http://localhost:8000/** on the demo screen → the live dashboard. ## 2. Pre-load the memory bank (Redis scale flex) ```bash .venv/bin/python preload_bank.py # 124 distinct objects -> Redis ``` Dashboard "memories in Redis" jumps to 124; search stays ~2 ms. ## 3. Start the robot loop (robot — your SSH terminal, e-stop in hand) Stand the robot first, then: ```bash ssh booster@100.76.129.2 cd ~/nebius_k1 && source /opt/ros/humble/setup.bash python3 -u k1_engram.py --loop --point --interval 0.2 2>&1 | tee /tmp/engram_loop.log ``` Press **Enter** at the safety prompt. Use `-u` so output streams live. (Camera-only, no motion: drop `--point`.) --- ## The demo arc (~90s) **Staging that works (important):** hold ONE object **up close** to the camera so it fills the frame — the recognition is rock-solid then (1.0). On a table across the busy room, the *table* wins. Distinct objects only (color + shape differ). 1. **Hook — recall at scale.** Hold a common object (mug, banana, controller). The dashboard shows the label + "recalled from 124 memories in ~2 ms." Nebius + Redis working together, instantly. 2. **One-shot learning (the hero).** Hold an **unusual** object it won't know (badge, a weird gadget — *not* a generic item in the bank). It says **"no match."** Teach it once: ```bash # on the robot, separate terminal: python3 k1_engram.py --learn "my_thing" ``` Show it again → **recognized**, dashboard flashes **"✓ learned one-shot."** *"A vision model can't learn a new class from one example. Ours just did — live."* 3. **Prove it's real (anti-hardcode).** Glide the object left↔right → the head **follows its center** (continuous tracking, `center_x`). Swap objects → recognition switches. Live perception, not [truncated — 1439 more characters] ``` ### DEVPOST_SUBMISSION.md ```markdown # Engram — Devpost Submission > Hand this to the Claude Chrome extension to fill the Devpost form. Each `##` heading > matches a Devpost field. Replace anything in **[brackets]** before submitting. --- ## Project name Engram ## Tagline / Elevator pitch (≈ 200 chars) The memory layer for physical AI: a humanoid robot that perceives with a 72B vision model on Nebius, remembers objects as vectors in Redis, can be taught a new object in one shot, and walks to recall it. ## Thumbnail / one-liner A robot you can teach. Show it something once — it remembers, finds it in the room, and walks over to it. --- ## Inspiration Today's robots are brilliant at *perceiving* a scene and amnesiac about it a second later. A vision model can tell you "that's a mug," but it can't be *taught* "this is **my** mug" without retraining, and it has no memory of *where* it saw things. We wanted to build the missing piece for physical AI: a **memory layer** — instant, teachable, and spatial — that sits between a robot's eyes and its actions. The north-star question we kept asking: *"Would a real robotics team actually use this?"* ## What it does Engram gives a Booster K1 humanoid a **cloud brain and instant memory**: - **Perceives** with a 72-billion-parameter vision model (Qwen2.5-VL) running on **Nebius** GPUs — it captions whatever the robot's camera sees. - **Remembers** every object as a vector in **Redis**, with sub-2 ms nearest-neighbor recall over its whole memory bank. - **Learns in one shot** — show it a brand-new object it's never seen, teach it once (one vector written to Redis), and it recognizes it instantly. No retraining. - **Recognizes and attends** — it turns its head to look at the object it recalls, tracking the object's position continuously. - **Has spatial memory** — walk it through a scene and it captures every object *and the heading where it saw it*. - **Fetches on command** — say "go to the red soda can" and it first **checks whether it even remembers one**, recalls **where** it was, **walks toward it** (closed-loop on odometry), **stops at a safe distance using the robot's depth camera** so it doesn't crash into the table, faces it, and **waves**. You can watch the robot's-eye view and its live decision (recalled object, confidence, Redis search latency, the Nebius model) on a real-time dashboard. ## How we built it **Architecture — a thin robot + a cloud brain:** ``` K1 camera ──JPEG──▶ laptop "brain" ──▶ Nebius Qwen2.5-VL-72B (caption) ├──▶ Nebius Qwen3-Embedding-8B (vector) ├──▶ Redis vector KNN (recall + spatial memory) └──▶ {label, center_x, confidence, heading} ──▶ K1 acts ``` - **Perception (Nebius Token Factory):** OpenAI-compatible API. `Qwen/Qwen2.5-VL-72B-Instruct` turns a camera frame into a short caption; `Qwen/Qwen3-Embedding-8B` turns the caption into a 4096-d vector. One VLM call also returns the object's horizontal cent [truncated — 4801 more characters] ``` ### camera_check.py ```python """Non-interactive camera probe: opens cam index 0, grabs one frame, reports. No GUI.""" import cv2, sys cap = cv2.VideoCapture(0) if not cap.isOpened(): print("CAMERA: could not open index 0"); sys.exit(2) ok, frame = cap.read() cap.release() if not ok or frame is None: print("CAMERA: opened but frame read FAILED (likely permission)"); sys.exit(3) print(f"CAMERA OK: grabbed frame {frame.shape[1]}x{frame.shape[0]}") ``` ### live_test.py ```python """ Live end-to-end test on REAL webcam frames (no GUI, no keypresses). Grabs a frame -> caption -> embed -> store; grabs another -> identify. Proves the production path with real camera imagery. Flushes Redis after. """ import time import engram_pipeline as e cam = e.Camera() # warm up: first few frames are often dark while exposure settles for _ in range(8): cam.get_frame(); time.sleep(0.05) print("== LEARN from live frame ==") f1 = cam.get_frame() cap1 = e.caption_image(f1) print(f" caption: {cap1!r}") v1 = e.embed_text(cap1) e.ensure_index() e.store_memory("live-scene", cap1, v1) time.sleep(0.3) print("== IDENTIFY from a fresh live frame ==") f2 = cam.get_frame() cap2 = e.caption_image(f2) print(f" caption: {cap2!r}") v2 = e.embed_text(cap2) match = e.identify(v2) print(f" match -> {match}") cam.release() ok = match and match["label"] == "live-scene" print("\nRESULT:", "PASS — live camera->Nebius->Redis works ✅" if ok else "CHECK ⚠️") e.r.flushall() print("cleanup: Redis flushed, ready for your real run") ``` ### smoke_test.py ```python """ Headless smoke test for the Engram pipeline — NO webcam, NO GUI. Exercises the full path: synthetic image -> Nebius caption -> Nebius embed -> Redis store -> Redis KNN identify. Proves the Nebius+Redis wiring works before touching the camera. Cleans up after itself (flushes the test data). export NEBIUS_API_KEY=... .venv/bin/python smoke_test.py """ import numpy as np import engram_pipeline as e def red_circle(): img = np.full((240, 240, 3), 255, dtype=np.uint8) # white bg import cv2 cv2.circle(img, (120, 120), 70, (0, 0, 255), -1) # red (BGR) return img def blue_square(): img = np.full((240, 240, 3), 255, dtype=np.uint8) import cv2 cv2.rectangle(img, (60, 60), (180, 180), (255, 0, 0), -1) # blue (BGR) return img def learn(img, label): cap = e.caption_image(img) vec = e.embed_text(cap) e.store_memory(label, cap, vec) print(f" learned '{label}' (vec dim={vec.shape[0]}) caption: {cap!r}") return cap def identify(img): cap = e.caption_image(img) vec = e.embed_text(cap) return e.identify(vec), cap def main(): print("== ensure index ==") e.ensure_index() print("== LEARN two distinct objects ==") learn(red_circle(), "red circle") learn(blue_square(), "blue square") print("== IDENTIFY the red circle again ==") match, cap = identify(red_circle()) print(f" saw: {cap!r}") print(f" match -> {match}") ok = match and match["label"] == "red circle" print("\nRESULT:", "PASS ✅ full pipeline works" if ok else "CHECK ⚠️ unexpected match") print("== cleanup (flush test data) ==") e.r.flushall() print(" done — Redis is clean for your real run") if __name__ == "__main__": main() ``` ### k1_walk_test.py ```python """ k1_walk_test.py — GATED minimal locomotion test for the Booster K1. Runs ON THE ROBOT. This is the make-or-break test for the spatial/walk demo: can the robot walk safely? It does a tiny in-place turn, turns back, then one slow ~0.2 m forward step — pausing for your Enter before EACH motion. ================ SAFETY ================ Robot on the FLOOR (not a stand), ~2 m clear all around, e-stop IN HAND. Small/slow on purpose. Hit e-stop if it lurches, drifts, or looks unstable. ========================================= python3 k1_walk_test.py # interactive, gated """ import time import argparse from booster_robotics_sdk_python import ChannelFactory, B1LocoClient, RobotMode def move_for(client, vx, vy, vyaw, secs, hz=10): """Hold a velocity for `secs` then stop (Move is a velocity setpoint sent continuously).""" n = int(secs * hz) for _ in range(n): client.Move(vx, vy, vyaw) time.sleep(1.0 / hz) client.Move(0.0, 0.0, 0.0) # stop def main(): ap = argparse.ArgumentParser() ap.add_argument("interface", nargs="?", default="usb_eth0") args = ap.parse_args() ChannelFactory.Instance().Init(0, args.interface) c = B1LocoClient() c.Init() time.sleep(0.8) input("\n*** SAFETY: robot on the FLOOR, ~2 m clear, e-stop in hand. Enter to stand / Ctrl-C > ") print("kPrepare ..."); c.ChangeMode(RobotMode.kPrepare); time.sleep(2.0) print("kWalking ..."); c.ChangeMode(RobotMode.kWalking); time.sleep(1.5) print("standing in walk mode.") input("\n>>> TEST 1 — small IN-PLACE TURN LEFT (~20°). Enter to go (Ctrl-C aborts) > ") move_for(c, 0.0, 0.0, 0.3, 1.2) print(" turn-left done.") time.sleep(1.0) input("\n>>> TEST 2 — turn BACK RIGHT (~20°). Enter > ") move_for(c, 0.0, 0.0, -0.3, 1.2) print(" turn-right done.") time.sleep(1.0) input("\n>>> TEST 3 — one slow FORWARD step (~0.2 m). Enter > ") move_for(c, 0.12, 0.0, 0.0, 1.5) print(" forward done.") time.sleep(1.0) print("\n[done] robot standing. If that all looked stable, walking is viable.") if __name__ == "__main__": main() ``` ### k1_indicate.py ```python """ k1_indicate.py — high-level zone indication for the Booster K1 (runs ON THE ROBOT). This firmware's Cartesian arm API is unsupported (MoveHandEndEffector -> 501) and low-level kCustom is risky, but high-level RotateHead works in kWalking (rc=0, verified). So the robot INDICATES the recalled object by turning its HEAD toward the zone — body stays put, camera stays on the scene. Verified on this robot: interface usb_eth0; RotateHead rc=0 in kWalking. SAFETY: arm() does kPrepare -> kWalking (robot stands/balances). Stand/supported, e-stop ready. """ import time from booster_robotics_sdk_python import ( ChannelFactory, B1LocoClient, RobotMode, B1HandAction, ) # head yaw (radians): +yaw = LEFT, -yaw = RIGHT. Hardware range ~[-0.785, 0.785]. ZONE_YAW = {"left": 0.6, "center": 0.0, "right": -0.6} # legacy discrete fallback MAX_YAW = 0.7 # full-left / full-right deflection LOOK_DOWN_PITCH = 0.0 # level — objects are held up at camera height in the demo def yaw_from_center(cx): """Map object center_x in [0,1] (0=left edge, 1=right edge) to a continuous head yaw. cx=0 -> +MAX_YAW (look left), cx=0.5 -> 0, cx=1 -> -MAX_YAW (look right).""" cx = max(0.0, min(1.0, float(cx))) return round((0.5 - cx) * 2.0 * MAX_YAW, 3) class K1Indicator: def __init__(self, interface="usb_eth0"): print(f"[k1] init SDK on {interface} ...") ChannelFactory.Instance().Init(0, interface) self.client = B1LocoClient() self.client.Init() time.sleep(0.8) print("[k1] connected") def arm(self): """Stand + balance (kWalking) so RotateHead is accepted. ACTUATES.""" print("[k1] ChangeMode(kPrepare) ...") self.client.ChangeMode(RobotMode.kPrepare); time.sleep(2.0) print("[k1] ChangeMode(kWalking) ...") self.client.ChangeMode(RobotMode.kWalking); time.sleep(1.2) print("[k1] armed (standing; head control ready)") def indicate(self, zone=None, center_x=None, hold_s=0.0): """Turn the head toward the object. Prefers continuous center_x; falls back to zone.""" if center_x is not None: yaw = yaw_from_center(center_x) print(f"[k1] track center_x={center_x:.2f} -> yaw={yaw}") else: yaw = ZONE_YAW.get(zone, 0.0) print(f"[k1] indicate({zone}) -> yaw={yaw}") self.client.RotateHead(LOOK_DOWN_PITCH, yaw) if hold_s: time.sleep(hold_s) def rest(self): """Head back to center/level.""" self.client.RotateHead(0.0, 0.0) time.sleep(0.1) def wave(self): """Optional 'found it' flourish — wave then close so it doesn't loop.""" self.client.WaveHand(B1HandAction.kHandOpen) time.sleep(2.5) self.client.WaveHand(B1HandAction.kHandClose) if __name__ == "__main__": import argparse ap = argparse.ArgumentParser() ap.add_argument("interface", nargs="?", default="usb_eth0") ap.add_argument("--zone", choices=list(ZONE_YAW), default="right") ap.add_argument("--no-arm", action="store_true") args = ap.parse_args() k = K1Indicator(args.interface) if not args.no_arm: input("SAFETY: will stand (kPrepare->kWalking) + turn head. e-stop ready? Enter > ") k.arm() k.indicate(args.zone) k.rest() print("[k1] done") ``` ### preflight_nebius.py ```python """ Preflight for Engram Path A — verifies the 4 *** config values against what Nebius Token Factory *actually* serves right now (per CLAUDE_CODE_BRIEF task 2). It does NOT touch the pipeline. It just: 1. lists live models from /v1/models 2. shows candidate vision (VLM) and embedding models 3. makes ONE real embedding call and measures the true output dimension 4. makes ONE real vision call on a synthetic image to confirm image input works 5. prints a ready-to-paste CONFIG block for engram_pipeline.py Run: export NEBIUS_API_KEY=... .venv/bin/python preflight_nebius.py """ import os import base64 import numpy as np import cv2 from openai import OpenAI BASE_URL = "https://api.tokenfactory.nebius.com/v1/" # Preference order — first available wins. Edit if you want a specific one. VLM_PREFS = [ "Qwen/Qwen2.5-VL-72B-Instruct", "Qwen/Qwen2.5-VL-7B-Instruct", "Qwen/Qwen2-VL-72B-Instruct", "Qwen/Qwen2-VL-7B-Instruct", ] EMBED_PREFS = [ "Qwen/Qwen3-Embedding-8B", "intfloat/e5-mistral-7b-instruct", "BAAI/bge-en-icl", "BAAI/bge-multilingual-gemma2", ] def pick(prefs, available): for p in prefs: if p in available: return p return None def main(): key = os.environ.get("NEBIUS_API_KEY") if not key: raise SystemExit("NEBIUS_API_KEY not set. export NEBIUS_API_KEY=... first.") client = OpenAI(base_url=BASE_URL, api_key=key) print(f"[base_url] {BASE_URL}") ids = sorted(m.id for m in client.models.list().data) print(f"[models] {len(ids)} available\n") vlm_like = [m for m in ids if any(t in m.lower() for t in ("-vl", "vision", "vl-"))] emb_like = [m for m in ids if any(t in m.lower() for t in ("embed", "bge", "e5-"))] print(" vision-ish:", vlm_like or "(none matched — see full list below)") print(" embed-ish :", emb_like or "(none matched — see full list below)") vlm = pick(VLM_PREFS, ids) or (vlm_like[0] if vlm_like else None) emb = pick(EMBED_PREFS, ids) or (emb_like[0] if emb_like else None) print(f"\n[chosen] VLM = {vlm}") print(f"[chosen] EMBED = {emb}") if not vlm or not emb: print("\nFull model list:") for m in ids: print(" ", m) raise SystemExit("Could not auto-pick a VLM and/or embedding model — pick manually from the list above.") # --- measure true embedding dimension --- e = client.embeddings.create(model=emb, input="a red coffee mug") dim = len(e.data[0].embedding) print(f"\n[embed] {emb} -> dimension = {dim} (THIS is EMBED_DIM)") # --- confirm the VLM accepts image input --- img = np.zeros((64, 64, 3), dtype=np.uint8) cv2.rectangle(img, (10, 10), (54, 54), (0, 0, 255), -1) # red square (BGR) ok, buf = cv2.imencode(".jpg", img) b64 = base64.b64encode(buf).decode() resp = client.chat.completions.create( model=vlm, messages=[{"role": "user", "content": [ {"type": "text", "text": "One short phrase: what color is this square?"}, {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}, ]}], max_tokens=20, ) print(f"[vlm] {vlm} -> replied: {resp.choices[0].message.content.strip()!r}") print("\n" + "=" * 60) print("PASTE INTO engram_pipeline.py CONFIG:") print("=" * 60) print(f'NEBIUS_BASE_URL = "{BASE_URL}"') print(f'VLM_MODEL = "{vlm}"') print(f'EMBED_MODEL = "{emb}"') print(f'EMBED_DIM = {dim}') if __name__ == "__main__": main() ``` ### k1_locomotion.py ```python """ k1_locomotion.py — odometry-feedback locomotion primitives for the Booster K1. Runs ON THE ROBOT. Closed-loop: turns/walks until the MEASURED odometry pose hits the target, so angles/distances are accurate regardless of the velocity mapping. This is the navigation foundation for the spatial "go to the green bottle" demo: turn_to(heading) / turn_by(deg) / walk_forward(meters) ================ SAFETY ================ Robot on the FLOOR, clear space, e-stop in hand. Start small. ========================================= python3 k1_locomotion.py --turn 180 # turn exactly 180 deg (closed-loop) python3 k1_locomotion.py --turn 360 python3 k1_locomotion.py --forward 0.5 # walk 0.5 m forward """ import time import math import threading import argparse from booster_robotics_sdk_python import ( ChannelFactory, B1LocoClient, B1OdometerStateSubscriber, RobotMode, ) # Odometry over-reports rotation (foot slip): command this many odom-radians per # desired physical radian. Calibrate: if --turn 180 ends short, raise it; if long, lower. TURN_SCALE = 1.4 def ang_diff(a, b): """smallest signed difference a-b, wrapped to [-pi, pi]""" return (a - b + math.pi) % (2 * math.pi) - math.pi class K1Locomotion: def __init__(self, interface="usb_eth0"): ChannelFactory.Instance().Init(0, interface) self.x = self.y = self.theta = 0.0 self._have = False self.odom = B1OdometerStateSubscriber(self._on_odom) self.odom.InitChannel() self.client = B1LocoClient() self.client.Init() t0 = time.time() while not self._have and time.time() - t0 < 8: # DDS discovery can take a few s time.sleep(0.1) if not self._have: raise RuntimeError("no odometry received in 8s — wrong interface?") print(f"[loco] connected. pose x={self.x:.2f} y={self.y:.2f} th={self.theta:.2f}") def _on_odom(self, m): self.x, self.y, self.theta = m.x, m.y, m.theta self._have = True def stand(self): print("[loco] kPrepare -> kWalking") self.client.ChangeMode(RobotMode.kPrepare); time.sleep(2.0) self.client.ChangeMode(RobotMode.kWalking); time.sleep(1.5) def stop(self): self.client.Move(0.0, 0.0, 0.0) def turn_by(self, delta_rad, yaw_speed=0.5, tol=0.05): """Turn by delta_rad of PHYSICAL rotation (closed-loop, slip-corrected). +=left.""" target = abs(delta_rad) * TURN_SCALE # odom radians to command direction = 1.0 if delta_rad >= 0 else -1.0 turned = 0.0 prev = self.theta print(f"[loco] turn_by {math.degrees(delta_rad):.0f} deg (scale {TURN_SCALE})") t0 = time.time() while turned < target - tol and time.time() - t0 < 20: turned += abs(ang_diff(self.theta, prev)) prev = self.theta remaining = target - turned spd = yaw_speed if remaining > 0.35 else max(0.18, yaw_speed * 0.45) # ease in to stop self.client.Move(0.0, 0.0, direction * spd) time.sleep(0.04) self.stop() time.sleep(0.4) print(f"[loco] turned ~{math.degrees(turned):.0f} deg (final th={self.theta:.2f})") def turn_to(self, target_theta, **kw): self.turn_by(ang_diff(target_theta, self.theta), **kw) def walk_forward(self, dist, vx=0.15, tol=0.04): sx, sy = self.x, self.y print(f"[loco] walk_forward {dist:.2f} m") t0 = time.time() while math.hypot(self.x - sx, self.y - sy) < dist - tol and time.time() - t0 < 30: self.client.Move(vx, 0.0, 0.0) time.sleep(0.04) self.stop() time.sleep(0.3) moved = math.hypot(self.x - sx, self.y - sy) print(f"[loco] moved ~{moved:.2f} m") def main(): ap = argparse.ArgumentParser() ap.add_argument("interface", nargs="?", default="usb_eth0") ap.add_argument("--turn", type=float, help="turn N degrees (closed-loop)") ap.add_argument("--forward", type=float, help="walk N meters") ap.add_argument("--turn-scale", type=float, help="override slip-correction factor") ap.add_argument("--no-arm", action="store_true", help="already standing in kWalking") args = ap.parse_args() if args.turn_scale: global TURN_SCALE TURN_SCALE = args.turn_scale loco = K1Locomotion(args.interface) input("\n*** SAFETY: robot on FLOOR, clear space, e-stop in hand. Enter / Ctrl-C > ") if not args.no_arm: loco.stand() if args.turn is not None: input(f">>> turn {args.turn} deg — Enter > ") loco.turn_by(math.radians(args.turn)) if args.forward is not None: input(f">>> walk {args.forward} m — Enter > ") loco.walk_forward(args.forward) print("[loco] done — standing") if __name__ == "__main__": main() ``` ### k1_engram.py ```python """ k1_engram.py — Engram agent on the Booster K1 (runs ON THE ROBOT). Robot camera (ROS2) -> laptop brain (Nebius+Redis over Tailscale) -> {label,zone,confidence} -> point. Brain default = http://100.83.124.0:8000 (laptop's Tailscale IP, engram_server.py). Camera = /booster_video_stream (compressed) with /image_left_raw (nv12) fallback. Motion reuses k1_point.K1Pointer and ONLY happens with --point (else perception-only, safe). Modes (run with ROS sourced: `source /opt/ros/humble/setup.bash`): python3 k1_engram.py --learn "phone" # teach current view (NO motion) python3 k1_engram.py --identify # identify current view (NO motion) python3 k1_engram.py --identify --point # identify + point the arm python3 k1_engram.py --loop --point # autonomous identify+point loop """ import sys import time import json import argparse import urllib.parse import urllib.request import numpy as np import cv2 import rclpy from rclpy.node import Node from sensor_msgs.msg import CompressedImage, Image BRAIN = "http://100.83.124.0:8000" class Cam(Node): def __init__(self): super().__init__("engram_cam") self.frame = None self.create_subscription(CompressedImage, "/booster_video_stream", self._comp, 10) self.create_subscription(Image, "/image_left_raw", self._raw, 10) def _comp(self, m): self.frame = cv2.imdecode(np.frombuffer(m.data, np.uint8), cv2.IMREAD_COLOR) def _raw(self, m): if self.frame is None and m.encoding == "nv12": yuv = np.frombuffer(m.data, np.uint8).reshape(int(m.height * 3 / 2), m.width) self.frame = cv2.cvtColor(yuv, cv2.COLOR_YUV2BGR_NV12) def grab(self, timeout=6.0): self.frame = None t0 = time.time() while self.frame is None and time.time() - t0 < timeout: rclpy.spin_once(self, timeout_sec=0.3) return self.frame def post_image(path, frame): ok, buf = cv2.imencode(".jpg", frame) req = urllib.request.Request(BRAIN + path, data=buf.tobytes(), headers={"Content-Type": "image/jpeg"}, method="POST") with urllib.request.urlopen(req, timeout=30) as r: return json.loads(r.read().decode()) def main(): global BRAIN ap = argparse.ArgumentParser() ap.add_argument("--brain", default=BRAIN, help="laptop engram_server URL") ap.add_argument("--learn", metavar="LABEL", help="teach the current view") ap.add_argument("--identify", action="store_true") ap.add_argument("--loop", action="store_true") ap.add_argument("--point", action="store_true", help="ACTUALLY move the arm") ap.add_argument("--interface", default="usb_eth0") ap.add_argument("--interval", type=float, default=0.5) ap.add_argument("--min-conf", type=float, default=0.7) ap.add_argument("--yes", action="store_true", help="skip the safety prompt") args = ap.parse_args() BRAIN = args.brain rclpy.init() cam = Cam() print("[k1] warming camera ...") if cam.grab() is None: print("[k1] NO camera frame — is the camera bridge publishing?") return print("[k1] camera OK") pointer = None if args.point: from k1_indicate import K1Indicator pointer = K1Indicator(args.interface) if not args.yes: input("[k1] SAFETY: about to stand (kPrepare->kWalking) + head-look. " "Robot supported, space clear, e-stop ready? Enter to go / Ctrl-C to abort > ") pointer.arm() if args.learn: f = cam.grab() r = post_image("/learn_image?label=" + urllib.parse.quote(args.learn), f) print("[learn]", r) return if args.identify and not args.loop: f = cam.grab() r = post_image("/identify_image", f) print("[id]", {k: r.get(k) for k in ("label", "center_x", "confidence")}, "| saw:", r.get("caption")) if pointer and r.get("label") != "none" and r.get("confidence", 0) >= args.min_conf: pointer.indicate(center_x=r.get("center_x", 0.5)) return if args.loop: last = None print(f"[k1] loop every {args.interval}s (point={'on' if pointer else 'OFF'})") while True: f = cam.grab() if f is None: time.sleep(args.interval); continue try: r = post_image("/identify_image", f) except Exception as ex: print("[k1] brain error:", ex); time.sleep(args.interval); continue lbl, conf, cx = r.get("label"), r.get("confidence", 0), r.get("center_x", 0.5) print(f"[id] {lbl} center_x={cx:.2f} conf={conf} saw={r.get('caption')}") if pointer: if lbl != "none" and conf >= args.min_conf: pointer.indicate(center_x=cx) # continuous head tracking else: pointer.rest() time.sleep(args.interval) return print("nothing to do — use --learn LABEL | --identify | --loop (+ --point to move)") if __name__ == "__main__": main() ``` [7 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]