Claude's Corner: Servo7 - The Robot That Clocks In at the Dock Door

# How to Build a Servo7 Clone with Claude Code: A 7-Step Guide

A step-by-step guide for building an AI-powered warehouse container unloading robot system. This covers the full stack from perception to fleet management.

---

## Step 1: Set Up the Core Database Schema

Design your data model for robot fleets, deployments, telemetry, and learning data.

```sql
-- Facilities where robots are deployed
CREATE TABLE facilities (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  name TEXT NOT NULL,
  address TEXT,
  dock_count INTEGER,
  created_at TIMESTAMPTZ DEFAULT now()
);

-- Robot units
CREATE TABLE robots (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  serial_number TEXT UNIQUE NOT NULL,
  facility_id UUID REFERENCES facilities(id),
  firmware_version TEXT,
  model_version TEXT,
  status TEXT DEFAULT 'idle', -- idle, active, maintenance, error
  last_heartbeat TIMESTAMPTZ,
  created_at TIMESTAMPTZ DEFAULT now()
);

-- Container unloading sessions
CREATE TABLE sessions (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  robot_id UUID REFERENCES robots(id),
  facility_id UUID REFERENCES facilities(id),
  container_id TEXT, -- external reference
  started_at TIMESTAMPTZ DEFAULT now(),
  completed_at TIMESTAMPTZ,
  boxes_unloaded INTEGER DEFAULT 0,
  status TEXT DEFAULT 'in_progress', -- in_progress, completed, failed, manual_takeover
  error_code TEXT,
  demo_count INTEGER DEFAULT 0 -- number of human demonstrations used
);

-- Per-pick telemetry frames
CREATE TABLE pick_events (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  session_id UUID REFERENCES sessions(id),
  robot_id UUID REFERENCES robots(id),
  timestamp TIMESTAMPTZ DEFAULT now(),
  box_weight_kg DECIMAL(5,2),
  pick_duration_ms INTEGER,
  success BOOLEAN,
  failure_reason TEXT,
  perception_confidence DECIMAL(3,2), -- 0.0 to 1.0
  model_version TEXT
);

-- Human demonstration recordings
CREATE TABLE demonstrations (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  facility_id UUID REFERENCES facilities(id),
  recorded_at TIMESTAMPTZ DEFAULT now(),
  duration_seconds INTEGER,
  box_count INTEGER,
  storage_path TEXT NOT NULL, -- path to raw trajectory data
  processed BOOLEAN DEFAULT false,
  used_in_training BOOLEAN DEFAULT false
);
```

---

## Step 2: Build the Perception Pipeline

The core computer vision system for understanding the container scene in real time.

```python
# perception/scene_understanding.py
import numpy as np
import torch
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class BoxDetection:
    bbox_3d: np.ndarray  # [x, y, z, w, h, d] in robot frame
    confidence: float
    estimated_weight_kg: float
    label_visible: bool
    grasp_candidates: List[np.ndarray]  # candidate grasp poses

class SceneUnderstanding:
    def __init__(self, model_path: str, depth_camera_config: dict):
        self.detector = self._load_detection_model(model_path)
        self.camera_config = depth_camera_config
        self.point_cloud_buffer = []

    def process_frame(
        self,
        rgb_image: np.ndarray,
        depth_image: np.ndarray,
        robot_pose: np.ndarray
    ) -> List[BoxDetection]:
        # 1. Fuse RGB + depth into organized point cloud
        point_cloud = self._fuse_rgbd(rgb_image, depth_image)

        # 2. Transform to robot base frame
        point_cloud_robot = self._transform_to_robot_frame(point_cloud, robot_pose)

        # 3. Run 3D object detection
        detections_raw = self.detector(point_cloud_robot, rgb_image)

        # 4. Estimate grasp candidates for each detection
        detections = []
        for det in detections_raw:
            grasps = self._compute_grasp_candidates(det, point_cloud_robot)
            detections.append(BoxDetection(
                bbox_3d=det['bbox_3d'],
                confidence=det['confidence'],
                estimated_weight_kg=self._estimate_weight(det['volume_m3']),
                label_visible=det['label_visible'],
                grasp_candidates=grasps
            ))

        return sorted(detections, key=lambda d: self._pick_priority(d))

    def _pick_priority(self, detection: BoxDetection) -> float:
        # Prefer top-layer, accessible, high-confidence boxes
        # Lower score = pick first
        z_score = -detection.bbox_3d[2]  # higher boxes first
        confidence_penalty = 1.0 - detection.confidence
        return z_score + confidence_penalty * 0.5

    def _estimate_weight(self, volume_m3: float) -> float:
        # Average cardboard box density ~150kg/m3 for mixed goods
        DENSITY_KG_M3 = 150
        return min(volume_m3 * DENSITY_KG_M3, 23.0)  # cap at 23kg limit

    def _compute_grasp_candidates(self, detection, point_cloud) -> List[np.ndarray]:
        # Generate top-surface and side-surface grasp poses
        # Returns list of 4x4 homogeneous transforms (grasp frame in robot frame)
        raise NotImplementedError("Implement grasp candidate generation")

    def _fuse_rgbd(self, rgb, depth):
        raise NotImplementedError("Implement RGBD fusion")

    def _transform_to_robot_frame(self, cloud, robot_pose):
        raise NotImplementedError("Implement coordinate transform")

    def _load_detection_model(self, path):
        raise NotImplementedError("Load your 3D detection model")
```

---

## Step 3: Implement the Imitation Learning System

Collect human demonstrations and train a manipulation policy via behavioral cloning.

```python
# learning/imitation.py
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
from pathlib import Path
import numpy as np

class DemonstrationDataset(Dataset):
    """Dataset of human-recorded manipulation trajectories."""

    def __init__(self, demo_paths: List[Path], augment: bool = True):
        self.demos = []
        for path in demo_paths:
            trajectory = np.load(path, allow_pickle=True)
            self.demos.extend(self._segment_trajectory(trajectory))
        self.augment = augment

    def _segment_trajectory(self, traj) -> List[dict]:
        # Segment full trajectory into (observation, action) pairs
        pairs = []
        for i in range(len(traj['observations']) - 1):
            pairs.append({
                'obs': traj['observations'][i],   # RGB + depth + robot state
                'action': traj['actions'][i],     # target joint positions / EE pose
            })
        return pairs

    def __len__(self):
        return len(self.demos)

    def __getitem__(self, idx):
        sample = self.demos[idx]
        if self.augment:
            sample = self._augment(sample)
        return sample

    def _augment(self, sample):
        # Apply color jitter, depth noise, and small spatial perturbations
        # to improve generalization from limited demonstrations
        return sample  # implement augmentation here


class ManipulationPolicy(nn.Module):
    """Behavioral cloning policy: obs -> action."""

    def __init__(self, obs_dim: int, action_dim: int, hidden_dim: int = 512):
        super().__init__()
        self.visual_encoder = self._build_visual_encoder()
        self.state_encoder = nn.Linear(obs_dim - 512, 128)
        self.policy_head = nn.Sequential(
            nn.Linear(512 + 128, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, action_dim)
        )

    def _build_visual_encoder(self):
        # Use a pre-trained ResNet or ViT backbone for visual features
        import torchvision.models as models
        backbone = models.resnet50(pretrained=True)
        return nn.Sequential(*list(backbone.children())[:-1])

    def forward(self, obs):
        visual_features = self.visual_encoder(obs['image']).flatten(1)
        state_features = self.state_encoder(obs['state'])
        combined = torch.cat([visual_features, state_features], dim=1)
        return self.policy_head(combined)


def train_policy(demo_paths: List[Path], epochs: int = 100, lr: float = 1e-4):
    dataset = DemonstrationDataset(demo_paths, augment=True)
    loader = DataLoader(dataset, batch_size=32, shuffle=True, num_workers=4)

    policy = ManipulationPolicy(obs_dim=640, action_dim=7)
    optimizer = torch.optim.AdamW(policy.parameters(), lr=lr, weight_decay=1e-4)

    for epoch in range(epochs):
        total_loss = 0
        for batch in loader:
            pred_actions = policy(batch['obs'])
            loss = nn.MSELoss()(pred_actions, batch['action'])
            optimizer.zero_grad()
            loss.backward()
            torch.nn.utils.clip_grad_norm_(policy.parameters(), 1.0)
            optimizer.step()
            total_loss += loss.item()
        print(f"Epoch {epoch}: loss={total_loss/len(loader):.4f}")

    return policy
```

---

## Step 4: Build the Motion Planning and Control Layer

Real-time collision-aware trajectory planning for the robot arm.

```python
# control/motion_planner.py
from dataclasses import dataclass
import numpy as np
from typing import Optional, Tuple

@dataclass
class MotionPlan:
    waypoints: np.ndarray  # [N, joint_dim] trajectory
    duration_s: float
    collision_free: bool
    estimated_success_prob: float

class MotionPlanner:
    def __init__(self, robot_model_urdf: str, collision_checker):
        self.robot = self._load_robot(robot_model_urdf)
        self.collision_checker = collision_checker
        self.workspace_limits = None  # updated as container empties

    def plan_pick(
        self,
        current_joints: np.ndarray,
        grasp_pose: np.ndarray,  # 4x4 homogeneous transform
        workspace_mask: np.ndarray,  # 3D occupancy grid of container
    ) -> Optional[MotionPlan]:
        # 1. Update reachability given current container state
        self.workspace_limits = self._compute_reachable_workspace(workspace_mask)

        # 2. Validate grasp pose is reachable
        if not self._is_pose_reachable(grasp_pose):
            return None

        # 3. Plan approach trajectory with RRT-Connect or similar
        approach_pose = self._compute_approach_pose(grasp_pose, standoff_m=0.15)
        approach_traj = self._rrt_connect(current_joints, approach_pose, workspace_mask)
        if approach_traj is None:
            return None

        # 4. Plan grasp insertion (straight-line in Cartesian space)
        grasp_traj = self._cartesian_interpolate(approach_pose, grasp_pose, steps=20)

        # 5. Plan retract trajectory
        retract_traj = self._cartesian_interpolate(grasp_pose, approach_pose, steps=20)

        full_traj = np.concatenate([approach_traj, grasp_traj, retract_traj])

        return MotionPlan(
            waypoints=full_traj,
            duration_s=len(full_traj) * 0.05,  # 50ms per step
            collision_free=True,
            estimated_success_prob=self._estimate_success(grasp_pose, workspace_mask)
        )

    def _rrt_connect(self, start, goal_pose, obstacles):
        raise NotImplementedError("Implement RRT-Connect or use MoveIt/OMPL")

    def _cartesian_interpolate(self, start_pose, end_pose, steps):
        raise NotImplementedError("Implement Cartesian interpolation with IK")

    def _compute_reachable_workspace(self, workspace_mask):
        raise NotImplementedError("Compute workspace from occupancy grid")

    def _is_pose_reachable(self, pose):
        raise NotImplementedError("Check IK solvability and joint limits")

    def _compute_approach_pose(self, grasp_pose, standoff_m):
        approach = grasp_pose.copy()
        approach[:3, 3] -= grasp_pose[:3, 2] * standoff_m  # back off along z
        return approach

    def _estimate_success(self, grasp_pose, workspace_mask) -> float:
        raise NotImplementedError("Estimate grasp success probability")

    def _load_robot(self, urdf_path):
        raise NotImplementedError("Load URDF with robotics framework")
```

---

## Step 5: Fleet Management Backend API

REST + WebSocket backend for multi-robot coordination, OTA updates, and telemetry.

```python
# backend/main.py  (FastAPI)
from fastapi import FastAPI, WebSocket, HTTPException, Depends
from fastapi.middleware.cors import CORSMiddleware
import asyncio
import json
from uuid import UUID
from datetime import datetime

app = FastAPI(title="Servo7-Clone Fleet API")
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])

# In-memory robot state (replace with Redis in production)
robot_connections: dict[str, WebSocket] = {}
robot_states: dict[str, dict] = {}

@app.websocket("/ws/robot/{robot_serial}")
async def robot_websocket(websocket: WebSocket, robot_serial: str):
    """Persistent connection from each deployed robot unit."""
    await websocket.accept()
    robot_connections[robot_serial] = websocket
    try:
        while True:
            data = await websocket.receive_json()
            await handle_robot_message(robot_serial, data)
    except Exception:
        del robot_connections[robot_serial]

async def handle_robot_message(serial: str, msg: dict):
    msg_type = msg.get("type")
    if msg_type == "heartbeat":
        robot_states[serial] = {**robot_states.get(serial, {}), "last_seen": datetime.utcnow().isoformat(), **msg.get("state", {})}
    elif msg_type == "pick_complete":
        await log_pick_event(serial, msg)
    elif msg_type == "session_complete":
        await finalize_session(serial, msg)
    elif msg_type == "error":
        await handle_robot_error(serial, msg)

@app.post("/api/sessions/start")
async def start_session(robot_id: UUID, facility_id: UUID, container_ref: str):
    """Dispatch a new unloading session to a robot."""
    robot = await get_robot(robot_id)
    if robot["status"] != "idle":
        raise HTTPException(400, "Robot not available")

    session_id = await create_session_record(robot_id, facility_id, container_ref)

    # Push start command to robot via WebSocket
    if robot["serial"] in robot_connections:
        await robot_connections[robot["serial"]].send_json({
            "type": "start_session",
            "session_id": str(session_id),
            "model_version": robot["model_version"]
        })

    return {"session_id": session_id}

@app.post("/api/robots/{robot_id}/update-model")
async def push_model_update(robot_id: UUID, model_version: str, model_url: str):
    """OTA model update to a deployed robot."""
    robot = await get_robot(robot_id)
    if robot["status"] == "active":
        raise HTTPException(409, "Cannot update model during active session")

    if robot["serial"] in robot_connections:
        await robot_connections[robot["serial"]].send_json({
            "type": "update_model",
            "version": model_version,
            "download_url": model_url,
            "checksum": await get_model_checksum(model_url)
        })

    return {"status": "update_dispatched"}

@app.get("/api/facilities/{facility_id}/metrics")
async def facility_metrics(facility_id: UUID, days: int = 7):
    """Returns throughput, success rates, and downtime per facility."""
    return await compute_facility_metrics(facility_id, days)

# Stub implementations - replace with real DB calls
async def get_robot(robot_id): raise NotImplementedError
async def create_session_record(robot_id, facility_id, container_ref): raise NotImplementedError
async def log_pick_event(serial, msg): raise NotImplementedError
async def finalize_session(serial, msg): raise NotImplementedError
async def handle_robot_error(serial, msg): raise NotImplementedError
async def get_model_checksum(url): raise NotImplementedError
async def compute_facility_metrics(facility_id, days): raise NotImplementedError
```

---

## Step 6: Operator Dashboard Frontend

React dashboard for facility managers to monitor robots, review sessions, and trigger demonstrations.

```typescript
// frontend/src/components/RobotDashboard.tsx
import React, { useEffect, useState } from 'react';

interface RobotStatus {
  serialNumber: string;
  status: 'idle' | 'active' | 'maintenance' | 'error';
  currentSession?: {
    sessionId: string;
    boxesUnloaded: number;
    startedAt: string;
    estimatedCompletion: string;
  };
  modelVersion: string;
  lastSeen: string;
  successRate7d: number;
}

interface FacilityMetrics {
  containersThisWeek: number;
  avgUnloadTimeMinutes: number;
  pickSuccessRate: number;
  manualInterventions: number;
  robotsActive: number;
}

export function RobotDashboard({ facilityId }: { facilityId: string }) {
  const [robots, setRobots] = useState<RobotStatus[]>([]);
  const [metrics, setMetrics] = useState<FacilityMetrics | null>(null);
  const [ws, setWs] = useState<WebSocket | null>(null);

  useEffect(() => {
    // Live telemetry via WebSocket
    const socket = new WebSocket(`wss://api.yourcompany.com/ws/facility/${facilityId}`);
    socket.onmessage = (event) => {
      const update = JSON.parse(event.data);
      if (update.type === 'robot_state') {
        setRobots(prev => prev.map(r =>
          r.serialNumber === update.serial ? { ...r, ...update.state } : r
        ));
      }
    };
    setWs(socket);

    // Initial data fetch
    fetch(`/api/facilities/${facilityId}/robots`)
      .then(r => r.json()).then(setRobots);
    fetch(`/api/facilities/${facilityId}/metrics?days=7`)
      .then(r => r.json()).then(setMetrics);

    return () => socket.close();
  }, [facilityId]);

  async function startSession(robotSerial: string, containerRef: string) {
    await fetch('/api/sessions/start', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ robotSerial, facilityId, containerRef })
    });
  }

  return (
    <div className="dashboard">
      {metrics && <MetricsBar metrics={metrics} />}
      <div className="robot-grid">
        {robots.map(robot => (
          <RobotCard
            key={robot.serialNumber}
            robot={robot}
            onStartSession={(ref) => startSession(robot.serialNumber, ref)}
          />
        ))}
      </div>
    </div>
  );
}

function MetricsBar({ metrics }: { metrics: FacilityMetrics }) {
  return (
    <div className="metrics-bar">
      <Metric label="Containers (7d)" value={metrics.containersThisWeek} />
      <Metric label="Avg Unload Time" value={`${metrics.avgUnloadTimeMinutes}m`} />
      <Metric label="Pick Success" value={`${(metrics.pickSuccessRate * 100).toFixed(1)}%`} />
      <Metric label="Manual Interventions" value={metrics.manualInterventions} />
    </div>
  );
}

function RobotCard({ robot, onStartSession }: { robot: RobotStatus; onStartSession: (ref: string) => void }) {
  return (
    <div className={`robot-card status-${robot.status}`}>
      <h3>{robot.serialNumber}</h3>
      <span className="status-badge">{robot.status}</span>
      {robot.currentSession && (
        <div className="session-info">
          <span>Boxes: {robot.currentSession.boxesUnloaded}</span>
          <span>ETA: {robot.currentSession.estimatedCompletion}</span>
        </div>
      )}
      <div className="robot-meta">
        <span>Model: {robot.modelVersion}</span>
        <span>7d success: {(robot.successRate7d * 100).toFixed(0)}%</span>
      </div>
      {robot.status === 'idle' && (
        <button onClick={() => {
          const ref = prompt('Container reference:');
          if (ref) onStartSession(ref);
        }}>
          Start Session
        </button>
      )}
    </div>
  );
}

function Metric({ label, value }: { label: string; value: string | number }) {
  return <div className="metric"><span className="label">{label}</span><span className="value">{value}</span></div>;
}
```

---

## Step 7: Edge Deployment, OTA Updates, and Safety Certification

Production deployment strategy for robots in warehouse environments.

### Edge Runtime Architecture

```yaml
# docker-compose.yml for on-robot edge stack
version: '3.8'
services:
  perception:
    image: your-registry/perception:latest
    runtime: nvidia  # GPU inference on Jetson or similar
    volumes:
      - /dev/video0:/dev/video0  # depth camera
    environment:
      - MODEL_PATH=/models/scene_understanding_v2.1.pt
      - INFERENCE_FPS=10

  policy:
    image: your-registry/policy:latest
    runtime: nvidia
    depends_on: [perception]
    environment:
      - MODEL_PATH=/models/manipulation_policy_v3.0.pt
      - MAX_WEIGHT_KG=23

  control:
    image: your-registry/control:latest
    volumes:
      - /dev/ttyUSB0:/dev/ttyUSB0  # robot arm serial
    network_mode: host  # low-latency robot bus access

  telemetry:
    image: your-registry/telemetry:latest
    environment:
      - FLEET_API_URL=wss://api.yourcompany.com
      - ROBOT_SERIAL=${ROBOT_SERIAL}
      - RETRY_INTERVAL_S=30  # handles intermittent warehouse WiFi

  ota-agent:
    image: your-registry/ota-agent:latest
    volumes:
      - /var/lib/robot/models:/models
    environment:
      - FLEET_API_URL=https://api.yourcompany.com
      - UPDATE_CHECK_INTERVAL_S=3600
```

### OTA Update Protocol

```python
# ota/agent.py
import hashlib, requests, shutil, subprocess, time
from pathlib import Path

class OTAAgent:
    def __init__(self, fleet_api_url: str, robot_serial: str, models_dir: Path):
        self.api_url = fleet_api_url
        self.serial = robot_serial
        self.models_dir = models_dir

    def check_and_apply(self):
        resp = requests.get(f"{self.api_url}/api/robots/{self.serial}/pending-update", timeout=30)
        if resp.status_code != 200 or not resp.json().get("update_available"):
            return

        update = resp.json()
        model_path = self.models_dir / f"{update['model_name']}_{update['version']}.pt"

        # Download with integrity check
        self._download_with_verify(update["download_url"], model_path, update["checksum"])

        # Atomic swap: keep old model as fallback
        current = self.models_dir / f"{update['model_name']}_current.pt"
        backup = self.models_dir / f"{update['model_name']}_backup.pt"
        if current.exists():
            shutil.copy2(current, backup)
        shutil.copy2(model_path, current)

        # Restart relevant service (no reboot needed)
        subprocess.run(["docker", "restart", f"robot_{update['model_name']}"], check=True)
        self._report_success(update["version"])

    def _download_with_verify(self, url: str, dest: Path, expected_sha256: str):
        resp = requests.get(url, stream=True, timeout=300)
        sha256 = hashlib.sha256()
        with open(dest, 'wb') as f:
            for chunk in resp.iter_content(chunk_size=8192):
                f.write(chunk)
                sha256.update(chunk)
        actual = sha256.hexdigest()
        if actual != expected_sha256:
            dest.unlink()
            raise ValueError(f"Checksum mismatch: got {actual}, expected {expected_sha256}")

    def _report_success(self, version: str):
        requests.post(f"{self.api_url}/api/robots/{self.serial}/update-complete",
                      json={"version": version, "timestamp": time.time()})
```

### Safety Certification Checklist

Before deploying robots in human-present warehouse environments, target these standards:

- **ISO 10218-1/2**: Safety requirements for industrial robots and robot systems
- **ISO/TS 15066**: Collaborative robot safety (if operating near workers)
- **CE marking** (EU): Required for placing machinery on European market
- **OSHA 29 CFR 1910.217** (US): Machine guarding requirements
- **IEC 62061**: Safety of machinery functional safety requirements

Implementation requirements:
1. Emergency stop accessible within 1 second from any worker in the vicinity
2. Force/torque sensing on robot arm to detect unexpected contact
3. Speed reduction to collaborative mode (<250mm/s) when humans enter the safety zone
4. Dual-channel safety relay for stop circuits (not software-only)
5. Annual third-party safety audit for each facility deployment
6. Incident reporting and near-miss logging in telemetry system