Clean out the Pi Zero leftovers, rewrite the README, and add the one installer
Wi-Fi-off acceptance passed: with Wi-Fi blocked, live and after a reboot, the Pi came up on the USB link alone, every node ran, and hermes mcp test robot and look() worked. (A temporary guard that would have unblocked Wi-Fi was the safety net and is removed; Wi-Fi is back on.) Delete what the ROS design superseded: the Picamera2 MCP server and its unit, the old installer and requirements, and the duplicate platform check. Move omarchy/ to host/omarchy/. Rewrite README.md for this build: parts, physical steps (including the Audio switch), setup, daily use, the troubleshooting learned the hard way, and the security model. pi/install.sh is now the single installer for a fresh Ubuntu 24.04 image, built from the steps done by hand: ROS, the libcamera fork at a pinned commit, camera_ros, the venv and models, the workspace, USB gadget, audio overlay, firewall (before PipeWire opens a port), PipeWire, mixer and services. Run twice on the live robot, the second run changed nothing and everything stayed healthy. It has not been run on a blank card. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01AFN8nRRCiDiMN7ChBpHJhs
This commit is contained in:
Claude Sonnet 5.5
+152
-83
@@ -1,96 +1,165 @@
|
||||
#!/usr/bin/env bash
|
||||
# Robot setup for Raspberry Pi OS Lite (64-bit) on a Pi Zero 2 W.
|
||||
# Run as your normal user (not root) from inside this "pi" folder:
|
||||
#!/bin/bash
|
||||
# Set up the robot's Raspberry Pi 4 from a fresh Ubuntu Server 24.04 (arm64) image.
|
||||
#
|
||||
# ./install.sh <omarchy-machine-ip> [--servos]
|
||||
# rsync -a ./ hermes@<pi>:~/hermes-bot/
|
||||
# ssh hermes@<pi> 'cd hermes-bot && sudo ./pi/install.sh'
|
||||
#
|
||||
# Safe to re-run. Reboot afterwards so the I2S sound card overlay takes effect.
|
||||
# Re-runnable: every step checks before it changes anything, and the heavy builds skip themselves if
|
||||
# they are already done. Needs the internet (apt, git, pip, model downloads). The first run takes a
|
||||
# while (the libcamera build alone is about 15 minutes). Reboot afterwards: it changes /boot/firmware/
|
||||
# config.txt for the USB gadget and the audio codec.
|
||||
#
|
||||
# The user (default "hermes", set ROBOT_USER) must already exist: create it in cloud-init user-data.
|
||||
# Order matters: the firewall goes in before PipeWire's network server, which listens on every
|
||||
# interface and is protected only by that firewall.
|
||||
set -euo pipefail
|
||||
|
||||
HOST_IP="${1:-}"
|
||||
SERVOS=0
|
||||
[[ "${2:-}" == "--servos" ]] && SERVOS=1
|
||||
ROBOT_USER="${ROBOT_USER:-hermes}"
|
||||
PI_DIR="$(cd "$(dirname "$0")" && pwd)"
|
||||
REPO="$(dirname "$PI_DIR")"
|
||||
# Raspberry Pi's fork of libcamera, at a commit known to capture on this kernel. Upstream libcamera,
|
||||
# Ubuntu's 0.2.0 and ROS's 0.7.2 all crash in the Pi image-processing plugin when capturing.
|
||||
LIBCAMERA_REF=6c1dd9d55573010f710c9e190a73e7e76f0d9432
|
||||
CAMERA_ROS_TAG=0.7.0
|
||||
MODEL_BASE=https://github.com/opencv/opencv_zoo/raw/main/models # a plain raw URL returns a Git LFS pointer
|
||||
|
||||
if [[ -z "$HOST_IP" || ! "$HOST_IP" =~ ^[0-9]{1,3}(\.[0-9]{1,3}){3}$ ]]; then
|
||||
echo "usage: $0 <omarchy-machine-ipv4> [--servos]" >&2
|
||||
exit 1
|
||||
fi
|
||||
if [[ $EUID -eq 0 ]]; then
|
||||
echo "Run this as your normal user; it calls sudo where needed." >&2
|
||||
exit 1
|
||||
export DEBIAN_FRONTEND=noninteractive NEEDRESTART_MODE=a
|
||||
REBOOT=0
|
||||
|
||||
[ "$(id -u)" = 0 ] || { echo "run with sudo" >&2; exit 1; }
|
||||
HOME_DIR="$(getent passwd "$ROBOT_USER" | cut -d: -f6)"
|
||||
[ -n "$HOME_DIR" ] || { echo "no user '$ROBOT_USER': create it in cloud-init user-data first" >&2; exit 1; }
|
||||
grep -qs "Ubuntu 24.04" /etc/os-release || echo "warning: this is written for Ubuntu 24.04" >&2
|
||||
|
||||
step() { printf '\n==> %s\n' "$*"; }
|
||||
as_user() { sudo -u "$ROBOT_USER" -H "$@"; }
|
||||
cfg_sum() { sha256sum /boot/firmware/config.txt | cut -d' ' -f1; }
|
||||
USER_UID="$(id -u "$ROBOT_USER")"
|
||||
|
||||
step "Base packages"
|
||||
apt-get update -q
|
||||
apt-get -y -q install git curl rsync software-properties-common openssl rfkill device-tree-compiler i2c-tools \
|
||||
alsa-utils python3-numpy python3-pil python3-spidev python3-lgpio python3-rpi-lgpio python3-pip \
|
||||
python3-venv pipewire pipewire-pulse wireplumber libspa-0.2-modules pulseaudio-utils \
|
||||
meson ninja-build pkg-config python3-yaml python3-ply python3-jinja2 libyaml-dev libgnutls28-dev \
|
||||
libssl-dev libudev-dev libevent-dev libdrm-dev libjpeg-dev libpng-dev libtiff-dev libboost-dev v4l-utils
|
||||
|
||||
step "Groups for hardware access"
|
||||
for g in video audio i2c spi gpio dialout plugdev; do
|
||||
getent group "$g" >/dev/null && usermod -aG "$g" "$ROBOT_USER" || echo " (no group $g on this image)"
|
||||
done
|
||||
|
||||
step "ROS 2 Jazzy"
|
||||
if ! dpkg -s ros-jazzy-ros-base >/dev/null 2>&1; then
|
||||
add-apt-repository -y universe
|
||||
. /etc/os-release
|
||||
V="$(curl -s https://api.github.com/repos/ros-infrastructure/ros-apt-source/releases/latest | grep -F tag_name | awk -F\" '{print $4}')"
|
||||
curl -fL -o /tmp/ros2-apt-source.deb \
|
||||
"https://github.com/ros-infrastructure/ros-apt-source/releases/download/${V}/ros2-apt-source_${V}.${UBUNTU_CODENAME:-$VERSION_CODENAME}_all.deb"
|
||||
dpkg -i /tmp/ros2-apt-source.deb
|
||||
apt-get update -q
|
||||
fi
|
||||
# ros-jazzy-camera-ros is installed for its dependencies; the camera driver actually used is built below
|
||||
apt-get -y -q install ros-jazzy-ros-base ros-dev-tools python3-colcon-common-extensions \
|
||||
ros-jazzy-robot-state-publisher ros-jazzy-xacro ros-jazzy-camera-ros ros-jazzy-vision-msgs
|
||||
|
||||
HERE="$(cd "$(dirname "$0")" && pwd)"
|
||||
ME="$(id -un)"
|
||||
CONFIG_TXT=/boot/firmware/config.txt
|
||||
[[ -f $CONFIG_TXT ]] || CONFIG_TXT=/boot/config.txt
|
||||
|
||||
echo "==> Installing packages"
|
||||
sudo apt-get update
|
||||
sudo apt-get install -y --no-install-recommends \
|
||||
python3-picamera2 python3-venv \
|
||||
pipewire pipewire-pulse pipewire-alsa wireplumber libspa-0.2-modules \
|
||||
pulseaudio-utils alsa-utils nftables openssl
|
||||
|
||||
echo "==> Enabling the I2S microphone/amplifier overlay in $CONFIG_TXT"
|
||||
add_line() { grep -qxF "$1" "$CONFIG_TXT" || echo "$1" | sudo tee -a "$CONFIG_TXT" >/dev/null; }
|
||||
add_line "dtparam=i2s=on"
|
||||
add_line "dtoverlay=googlevoicehat-soundcard"
|
||||
if [[ $SERVOS -eq 1 ]]; then
|
||||
add_line "dtparam=i2c_arm=on"
|
||||
fi
|
||||
|
||||
echo "==> Installing the MCP server to /opt/robot-mcp"
|
||||
sudo mkdir -p /opt/robot-mcp
|
||||
sudo chown "$ME:$ME" /opt/robot-mcp
|
||||
cp "$HERE/robot_mcp_server.py" /opt/robot-mcp/
|
||||
# --system-site-packages so the venv can see the apt-installed picamera2/libcamera
|
||||
[[ -d /opt/robot-mcp/venv ]] || python3 -m venv --system-site-packages /opt/robot-mcp/venv
|
||||
/opt/robot-mcp/venv/bin/pip install --upgrade pip >/dev/null
|
||||
/opt/robot-mcp/venv/bin/pip install -r "$HERE/requirements.txt"
|
||||
if [[ $SERVOS -eq 1 ]]; then
|
||||
/opt/robot-mcp/venv/bin/pip install adafruit-circuitpython-servokit
|
||||
fi
|
||||
|
||||
echo "==> Writing /etc/robot-mcp.env"
|
||||
if ! sudo test -f /etc/robot-mcp.env; then
|
||||
TOKEN="$(openssl rand -hex 32)"
|
||||
sudo tee /etc/robot-mcp.env >/dev/null <<EOF
|
||||
ROBOT_MCP_TOKEN=$TOKEN
|
||||
ROBOT_MCP_PORT=8765
|
||||
ROBOT_CAM_SIZE=1280x720
|
||||
ROBOT_CAM_HFLIP=0
|
||||
ROBOT_CAM_VFLIP=0
|
||||
ROBOT_SERVOS=$SERVOS
|
||||
EOF
|
||||
sudo chown root:"$ME" /etc/robot-mcp.env
|
||||
sudo chmod 640 /etc/robot-mcp.env
|
||||
step "Raspberry Pi libcamera fork -> /opt/libcamera-rpi"
|
||||
if [ -x /opt/libcamera-rpi/bin/cam ]; then
|
||||
echo " already built"
|
||||
else
|
||||
sudo sed -i "s/^ROBOT_SERVOS=.*/ROBOT_SERVOS=$SERVOS/" /etc/robot-mcp.env
|
||||
rm -rf /var/tmp/libcamera-rpi
|
||||
git clone -q --no-checkout https://github.com/raspberrypi/libcamera.git /var/tmp/libcamera-rpi
|
||||
git -C /var/tmp/libcamera-rpi fetch -q --depth 1 origin "$LIBCAMERA_REF"
|
||||
git -C /var/tmp/libcamera-rpi checkout -q FETCH_HEAD
|
||||
(cd /var/tmp/libcamera-rpi \
|
||||
&& meson setup build --buildtype=release -Dprefix=/opt/libcamera-rpi -Dpipelines=rpi/vc4 -Dipas=rpi/vc4 \
|
||||
-Dcam=enabled -Dqcam=disabled -Dgstreamer=disabled -Dpycamera=disabled -Dlc-compliance=disabled \
|
||||
-Dtest=false -Ddocumentation=disabled -Dv4l2=false \
|
||||
&& ninja -C build -j4 && ninja -C build install)
|
||||
fi
|
||||
|
||||
echo "==> Installing the systemd service"
|
||||
sed "s/__USER__/$ME/g" "$HERE/robot-mcp.service" | sudo tee /etc/systemd/system/robot-mcp.service >/dev/null
|
||||
getent group i2c >/dev/null || sudo groupadd --system i2c
|
||||
sudo systemctl daemon-reload
|
||||
sudo systemctl enable robot-mcp.service
|
||||
step "Python environment (~/venv: display, OpenCV 4.10 for YuNet, MCP SDK)"
|
||||
# OpenCV's DNN needs 4.8+ for the YuNet model and apt has 4.6; numpy stays below 2 because the ROS
|
||||
# Python packages are built against 1.x.
|
||||
[ -d "$HOME_DIR/venv" ] || as_user python3 -m venv --system-site-packages "$HOME_DIR/venv"
|
||||
as_user "$HOME_DIR/venv/bin/pip" install -q adafruit-circuitpython-rgb-display adafruit-blinka \
|
||||
"opencv-python-headless==4.10.0.84" "numpy<2" "mcp>=2.3,<3"
|
||||
|
||||
echo "==> Configuring PipeWire to serve audio to $HOST_IP"
|
||||
mkdir -p ~/.config/pipewire/pipewire-pulse.conf.d
|
||||
cp "$HERE/pipewire/50-network-server.conf" ~/.config/pipewire/pipewire-pulse.conf.d/
|
||||
# Run the audio stack at boot without anyone logging in.
|
||||
sudo loginctl enable-linger "$ME"
|
||||
systemctl --user enable pipewire.socket pipewire-pulse.socket wireplumber.service 2>/dev/null || true
|
||||
step "Face models (~/models)"
|
||||
as_user mkdir -p "$HOME_DIR/models"
|
||||
fetch_model() { # name, path in the zoo
|
||||
local dest="$HOME_DIR/models/$1"
|
||||
if [ "$(stat -c %s "$dest" 2>/dev/null || echo 0)" -lt 100000 ]; then
|
||||
as_user curl -fsSL -o "$dest" "$MODEL_BASE/$2/$1"
|
||||
fi
|
||||
[ "$(stat -c %s "$dest")" -gt 100000 ] || { echo " $1 looks like an LFS pointer, not the model" >&2; exit 1; }
|
||||
}
|
||||
fetch_model face_detection_yunet_2023mar.onnx face_detection_yunet
|
||||
fetch_model face_recognition_sface_2021dec.onnx face_recognition_sface
|
||||
|
||||
echo "==> Restricting ports 4713 and 8765 to $HOST_IP"
|
||||
sed "s/__HOST_IP__/$HOST_IP/g" "$HERE/nftables-robot.nft" | sudo tee /etc/nftables-robot.nft >/dev/null
|
||||
grep -qxF 'include "/etc/nftables-robot.nft"' /etc/nftables.conf \
|
||||
|| echo 'include "/etc/nftables-robot.nft"' | sudo tee -a /etc/nftables.conf >/dev/null
|
||||
sudo systemctl enable nftables.service
|
||||
sudo systemctl restart nftables.service
|
||||
step "camera_ros $CAMERA_ROS_TAG, built against the libcamera fork (~/camera_ws)"
|
||||
if [ -d "$HOME_DIR/camera_ws/install/camera_ros" ]; then
|
||||
echo " already built"
|
||||
else
|
||||
as_user mkdir -p "$HOME_DIR/camera_ws/src"
|
||||
[ -d "$HOME_DIR/camera_ws/src/camera_ros" ] || as_user git clone -q --depth 1 --branch "$CAMERA_ROS_TAG" \
|
||||
https://github.com/christianrauch/camera_ros.git "$HOME_DIR/camera_ws/src/camera_ros"
|
||||
as_user bash -c "source /opt/ros/jazzy/setup.bash && export PKG_CONFIG_PATH=/opt/libcamera-rpi/lib/aarch64-linux-gnu/pkgconfig \
|
||||
&& cd '$HOME_DIR/camera_ws' && colcon build --symlink-install --cmake-args -DCMAKE_BUILD_TYPE=Release"
|
||||
fi
|
||||
|
||||
echo
|
||||
echo "Done. Reboot now: sudo reboot"
|
||||
echo
|
||||
echo "Your MCP token (put this in ~/.hermes/.env on the Omarchy machine as ROBOT_MCP_TOKEN):"
|
||||
sudo grep '^ROBOT_MCP_TOKEN=' /etc/robot-mcp.env | cut -d= -f2
|
||||
step "Environment script and the robot's ROS workspace (~/ros_ws)"
|
||||
install -o "$ROBOT_USER" -g "$ROBOT_USER" -m 644 "$PI_DIR/robot-env.sh" "$HOME_DIR/robot-env.sh"
|
||||
as_user mkdir -p "$HOME_DIR/ros_ws/src"
|
||||
rsync -a --delete --exclude __pycache__ "$REPO/ros/" "$HOME_DIR/ros_ws/src/"
|
||||
chown -R "$ROBOT_USER:$ROBOT_USER" "$HOME_DIR/ros_ws/src"
|
||||
as_user bash -c "source /opt/ros/jazzy/setup.bash && source '$HOME_DIR/camera_ws/install/setup.bash' \
|
||||
&& cd '$HOME_DIR/ros_ws' && colcon build --symlink-install"
|
||||
|
||||
step "USB gadget (NCM Ethernet + ACM serial) and the audio codec overlay"
|
||||
before="$(cfg_sum)"
|
||||
bash "$PI_DIR/install-usb-gadget.sh"
|
||||
bash "$PI_DIR/install-audio-overlay.sh"
|
||||
[ "$before" = "$(cfg_sum)" ] || REBOOT=1
|
||||
|
||||
step "Firewall (before anything listens on the network)"
|
||||
bash "$PI_DIR/install-firewall.sh"
|
||||
|
||||
step "PipeWire for $ROBOT_USER (audio server, reachable only over the USB link)"
|
||||
loginctl enable-linger "$ROBOT_USER"
|
||||
as_user mkdir -p "$HOME_DIR/.config/pipewire/pipewire-pulse.conf.d"
|
||||
install -o "$ROBOT_USER" -g "$ROBOT_USER" -m 644 "$PI_DIR/pipewire/50-network-server.conf" \
|
||||
"$HOME_DIR/.config/pipewire/pipewire-pulse.conf.d/50-network-server.conf"
|
||||
for _ in $(seq 20); do [ -S "/run/user/$USER_UID/bus" ] && break; sleep 1; done
|
||||
as_user env XDG_RUNTIME_DIR="/run/user/$USER_UID" systemctl --user enable --now \
|
||||
pipewire.socket pipewire-pulse.socket wireplumber.service \
|
||||
|| echo " (the user session is not up yet; PipeWire starts at the next boot)"
|
||||
|
||||
step "Audio codec mixer and microphone gain"
|
||||
if aplay -l 2>/dev/null | grep -q wm8960; then
|
||||
alsactl -f "$PI_DIR/wm8960_asound.state" restore wm8960soundcard && alsactl store wm8960soundcard
|
||||
# WirePlumber resets the codec's capture gain to +30 dB when it takes the device; 63% on PipeWire's
|
||||
# cubic scale is -12 dB from that, about +17 dB.
|
||||
as_user env XDG_RUNTIME_DIR="/run/user/$USER_UID" pactl set-source-volume @DEFAULT_SOURCE@ 63% \
|
||||
|| echo " (could not set the source volume: run pactl set-source-volume @DEFAULT_SOURCE@ 63% later)"
|
||||
else
|
||||
echo " no WM8960 card yet: reboot, then run this script again"
|
||||
REBOOT=1
|
||||
fi
|
||||
|
||||
step "Services: the robot (face, camera, vision, audio, MCP bridge), the CPU governor, trimmed boot"
|
||||
bash "$PI_DIR/install-robot-service.sh"
|
||||
bash "$PI_DIR/trim-boot.sh"
|
||||
|
||||
cat <<EOF
|
||||
|
||||
Done.$([ "$REBOOT" = 1 ] && echo " REBOOT NOW: the USB gadget and the audio overlay only take effect then.")
|
||||
|
||||
Next:
|
||||
- Slide the HAT's "Audio" switch (right edge) to ON, or every recording is silent.
|
||||
- Plug the USB-C data cable into your computer, then on it: sudo host/omarchy/robot-usb-link.sh
|
||||
- Hermes needs the MCP token. It is not printed here; read it with:
|
||||
ssh $ROBOT_USER@10.55.0.2 sudo cat /etc/robot-mcp.env
|
||||
then see host/omarchy/hermes-config-snippet.yaml.
|
||||
- Check with: hermes mcp test robot
|
||||
EOF
|
||||
|
||||
@@ -1,3 +0,0 @@
|
||||
# picamera2 and libcamera come from apt (python3-picamera2), not pip.
|
||||
# Written against the MCP Python SDK 2.x (MCPServer); 1.x used the older FastMCP name.
|
||||
mcp>=2.3,<3
|
||||
@@ -1,17 +0,0 @@
|
||||
[Unit]
|
||||
Description=Robot MCP server (camera and head for Hermes Agent)
|
||||
Wants=network-online.target
|
||||
After=network-online.target
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
User=__USER__
|
||||
Group=__USER__
|
||||
SupplementaryGroups=video i2c
|
||||
EnvironmentFile=/etc/robot-mcp.env
|
||||
ExecStart=/opt/robot-mcp/venv/bin/python /opt/robot-mcp/robot_mcp_server.py
|
||||
Restart=on-failure
|
||||
RestartSec=3
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
@@ -1,229 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Robot MCP server: runs on the Raspberry Pi and gives Hermes Agent the robot's eyes
|
||||
(and, optionally, a neck).
|
||||
|
||||
Tools exposed
|
||||
look() -> a JPEG still from the camera
|
||||
turn_head(pan, tilt) -> aim the pan/tilt servos (only if ROBOT_SERVOS=1)
|
||||
head_position() -> current pan/tilt (only if ROBOT_SERVOS=1)
|
||||
|
||||
Transport: MCP streamable HTTP at http://<pi>:<port>/mcp, guarded by a bearer token.
|
||||
Audio is NOT handled here; it travels over PipeWire (see README).
|
||||
|
||||
Configuration is by environment variable (normally set in /etc/robot-mcp.env):
|
||||
ROBOT_MCP_TOKEN required shared secret; Hermes sends "Authorization: Bearer <token>"
|
||||
ROBOT_MCP_PORT 8765
|
||||
ROBOT_MCP_BIND 0.0.0.0
|
||||
ROBOT_CAM_SIZE 1280x720 still resolution handed to the model
|
||||
ROBOT_CAM_QUALITY 85 JPEG quality
|
||||
ROBOT_CAM_HFLIP 0 set 1 if the camera is mounted mirrored
|
||||
ROBOT_CAM_VFLIP 0 set 1 (with HFLIP=1) if mounted upside down
|
||||
ROBOT_SERVOS 0 set 1 to enable turn_head via a PCA9685 board
|
||||
ROBOT_PAN_CHANNEL 0
|
||||
ROBOT_TILT_CHANNEL 1
|
||||
ROBOT_PAN_LIMIT 80 degrees either side of centre
|
||||
ROBOT_TILT_LIMIT 45
|
||||
ROBOT_PAN_INVERT 0 set 1 if "positive = left" comes out backwards
|
||||
ROBOT_TILT_INVERT 0
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hmac
|
||||
import io
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
|
||||
import uvicorn
|
||||
from mcp.server.mcpserver import Image, MCPServer
|
||||
from mcp.server.transport_security import TransportSecuritySettings
|
||||
|
||||
log = logging.getLogger("robot-mcp")
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
|
||||
|
||||
|
||||
def env_flag(name: str, default: str = "0") -> bool:
|
||||
return os.environ.get(name, default).strip().lower() in ("1", "true", "yes", "on")
|
||||
|
||||
|
||||
def env_int(name: str, default: int) -> int:
|
||||
return int(os.environ.get(name, str(default)))
|
||||
|
||||
|
||||
TOKEN = os.environ.get("ROBOT_MCP_TOKEN", "").strip()
|
||||
PORT = env_int("ROBOT_MCP_PORT", 8765)
|
||||
BIND = os.environ.get("ROBOT_MCP_BIND", "0.0.0.0")
|
||||
CAM_W, CAM_H = (int(v) for v in os.environ.get("ROBOT_CAM_SIZE", "1280x720").lower().split("x"))
|
||||
CAM_QUALITY = env_int("ROBOT_CAM_QUALITY", 85)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- camera
|
||||
|
||||
class Camera:
|
||||
"""Keeps Picamera2 running so a still is ready in well under a second."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
from libcamera import Transform # type: ignore[import-not-found]
|
||||
from picamera2 import Picamera2 # type: ignore[import-not-found]
|
||||
|
||||
self._lock = threading.Lock()
|
||||
self.cam = Picamera2()
|
||||
config = self.cam.create_still_configuration(
|
||||
main={"size": (CAM_W, CAM_H)},
|
||||
transform=Transform(hflip=env_flag("ROBOT_CAM_HFLIP"), vflip=env_flag("ROBOT_CAM_VFLIP")),
|
||||
buffer_count=2,
|
||||
)
|
||||
self.cam.configure(config)
|
||||
self.cam.options["quality"] = CAM_QUALITY
|
||||
self.has_autofocus = "AfMode" in self.cam.camera_controls
|
||||
self.cam.start()
|
||||
time.sleep(1.0) # let auto-exposure settle once at startup
|
||||
log.info("camera ready at %dx%d (autofocus: %s)", CAM_W, CAM_H, self.has_autofocus)
|
||||
|
||||
def snapshot(self) -> bytes:
|
||||
with self._lock:
|
||||
if self.has_autofocus:
|
||||
# Camera Module 3: run one focus sweep so whatever is in front is sharp.
|
||||
self.cam.autofocus_cycle()
|
||||
buf = io.BytesIO()
|
||||
self.cam.capture_file(buf, format="jpeg")
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- servos
|
||||
|
||||
class Head:
|
||||
"""Pan/tilt on a PCA9685 servo board. Angles are degrees from centre."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
from adafruit_servokit import ServoKit # type: ignore[import-not-found]
|
||||
|
||||
self._lock = threading.Lock()
|
||||
self.kit = ServoKit(channels=16)
|
||||
self.pan_ch = env_int("ROBOT_PAN_CHANNEL", 0)
|
||||
self.tilt_ch = env_int("ROBOT_TILT_CHANNEL", 1)
|
||||
self.pan_limit = env_int("ROBOT_PAN_LIMIT", 80)
|
||||
self.tilt_limit = env_int("ROBOT_TILT_LIMIT", 45)
|
||||
self.pan_sign = -1 if env_flag("ROBOT_PAN_INVERT") else 1
|
||||
self.tilt_sign = -1 if env_flag("ROBOT_TILT_INVERT") else 1
|
||||
self.pan = 0.0
|
||||
self.tilt = 0.0
|
||||
self.move(0.0, 0.0)
|
||||
log.info("servos ready (pan ch %d, tilt ch %d)", self.pan_ch, self.tilt_ch)
|
||||
|
||||
def move(self, pan: float, tilt: float) -> tuple[float, float]:
|
||||
pan = max(-self.pan_limit, min(self.pan_limit, pan))
|
||||
tilt = max(-self.tilt_limit, min(self.tilt_limit, tilt))
|
||||
with self._lock:
|
||||
self.kit.servo[self.pan_ch].angle = 90 + self.pan_sign * pan
|
||||
self.kit.servo[self.tilt_ch].angle = 90 + self.tilt_sign * tilt
|
||||
self.pan, self.tilt = pan, tilt
|
||||
time.sleep(0.4) # let the servos arrive and stop jittering before any photo
|
||||
return pan, tilt
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- MCP server
|
||||
|
||||
mcp = MCPServer(
|
||||
"robot",
|
||||
instructions=(
|
||||
"This server is a small physical robot on the user's desk. "
|
||||
"Call look() whenever you need to see what is in front of it; the image is current. "
|
||||
"If turn_head is available, aim first and then look."
|
||||
),
|
||||
)
|
||||
|
||||
camera: Camera | None = None
|
||||
head: Head | None = None
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def look() -> Image:
|
||||
"""Take a photograph with the robot's camera and return it. Use this to see the room,
|
||||
the desk, the person, or anything the user asks you to look at."""
|
||||
if camera is None:
|
||||
raise RuntimeError("camera is not available on the robot")
|
||||
t0 = time.monotonic()
|
||||
data = camera.snapshot()
|
||||
log.info("look(): %d bytes in %.2fs", len(data), time.monotonic() - t0)
|
||||
return Image(data=data, format="jpeg")
|
||||
|
||||
|
||||
def register_head_tools() -> None:
|
||||
@mcp.tool()
|
||||
def turn_head(pan: float = 0.0, tilt: float = 0.0) -> str:
|
||||
"""Aim the robot's head (and so its camera). Angles are in degrees from straight ahead.
|
||||
pan: positive turns to the robot's left, negative to its right.
|
||||
tilt: positive looks up, negative looks down.
|
||||
Values beyond the mechanical limits are clamped. Call look() afterwards to see."""
|
||||
assert head is not None
|
||||
p, t = head.move(pan, tilt)
|
||||
return f"head now at pan {p:+.0f} degrees, tilt {t:+.0f} degrees"
|
||||
|
||||
@mcp.tool()
|
||||
def head_position() -> str:
|
||||
"""Report where the robot's head is currently pointing."""
|
||||
assert head is not None
|
||||
return f"pan {head.pan:+.0f} degrees, tilt {head.tilt:+.0f} degrees"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- auth wrapper
|
||||
|
||||
class BearerAuth:
|
||||
"""Minimal ASGI middleware: reject any HTTP request without the right bearer token.
|
||||
Lifespan events pass straight through so the MCP session manager starts properly."""
|
||||
|
||||
def __init__(self, app, token: str) -> None:
|
||||
self.app = app
|
||||
self.expected = f"Bearer {token}".encode()
|
||||
|
||||
async def __call__(self, scope, receive, send):
|
||||
if scope["type"] == "http":
|
||||
supplied = dict(scope.get("headers") or []).get(b"authorization", b"")
|
||||
if not hmac.compare_digest(supplied, self.expected):
|
||||
await send({"type": "http.response.start", "status": 401,
|
||||
"headers": [(b"content-type", b"text/plain"),
|
||||
(b"www-authenticate", b"Bearer")]})
|
||||
await send({"type": "http.response.body", "body": b"unauthorized\n"})
|
||||
return
|
||||
await self.app(scope, receive, send)
|
||||
|
||||
|
||||
def build_app():
|
||||
app = mcp.streamable_http_app(
|
||||
host=BIND,
|
||||
# Bearer auth already stops drive-by requests; DNS-rebinding checks would only
|
||||
# get in the way of reaching the Pi by IP or by robot.local.
|
||||
transport_security=TransportSecuritySettings(enable_dns_rebinding_protection=False),
|
||||
)
|
||||
return BearerAuth(app, TOKEN)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
global camera, head
|
||||
|
||||
if len(TOKEN) < 16:
|
||||
sys.exit("ROBOT_MCP_TOKEN is missing or too short (generate one with: openssl rand -hex 32)")
|
||||
|
||||
try:
|
||||
camera = Camera()
|
||||
except Exception as exc: # keep serving servos even if the camera ribbon is loose
|
||||
log.error("camera failed to start: %s", exc)
|
||||
|
||||
if env_flag("ROBOT_SERVOS"):
|
||||
try:
|
||||
head = Head()
|
||||
register_head_tools()
|
||||
except Exception as exc:
|
||||
log.error("servos failed to start, turn_head disabled: %s", exc)
|
||||
|
||||
log.info("serving MCP on http://%s:%d/mcp", BIND, PORT)
|
||||
uvicorn.run(build_app(), host=BIND, port=PORT, log_level="warning")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user