Claude's Corner: General Astronautics, The Robot Replacing $130K/hr Astronaut Labor

# Build Guide: Space Robotics Lab Automation Platform (General Astronautics Clone)

A step-by-step guide to building an autonomous robotics-as-a-service platform for in-orbit lab automation. Follow each step with Claude Code.

---

## Step 1: Database Schema

Design and deploy the core data model in PostgreSQL (Supabase).

```sql
-- Customers and their organizations
CREATE TABLE organizations (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  name TEXT NOT NULL,
  contact_email TEXT NOT NULL,
  tier TEXT DEFAULT 'standard', -- standard | enterprise
  created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Robot units deployed to specific station modules
CREATE TABLE robot_units (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  serial_number TEXT UNIQUE NOT NULL,
  station_name TEXT NOT NULL,        -- e.g. "Axiom-1", "ISS-US-Lab"
  module_slot TEXT NOT NULL,         -- physical location identifier
  firmware_version TEXT,
  status TEXT DEFAULT 'offline',     -- offline | standby | busy | error
  last_heartbeat_at TIMESTAMPTZ,
  capabilities JSONB DEFAULT '[]',   -- ["pipette","plate_handler","centrifuge"]
  created_at TIMESTAMPTZ DEFAULT NOW(),
  updated_at TIMESTAMPTZ DEFAULT NOW()
);

-- Experiment protocols defined by customers
CREATE TABLE protocols (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  org_id UUID REFERENCES organizations(id),
  name TEXT NOT NULL,
  description TEXT,
  steps JSONB NOT NULL,   -- ordered array of action steps
  estimated_duration_mins INT,
  sample_type TEXT,       -- "protein_crystal" | "semiconductor" | "biological"
  created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Scheduled and executing experiment runs
CREATE TABLE experiment_runs (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  org_id UUID REFERENCES organizations(id),
  protocol_id UUID REFERENCES protocols(id),
  robot_unit_id UUID REFERENCES robot_units(id),
  status TEXT DEFAULT 'queued',   -- queued | uplinking | executing | completed | failed
  scheduled_at TIMESTAMPTZ,
  started_at TIMESTAMPTZ,
  completed_at TIMESTAMPTZ,
  step_cursor INT DEFAULT 0,
  telemetry_log JSONB DEFAULT '[]',
  result_summary JSONB,
  error_message TEXT,
  created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Telemetry frames streamed from robot
CREATE TABLE telemetry_frames (
  id BIGSERIAL PRIMARY KEY,
  robot_unit_id UUID REFERENCES robot_units(id),
  run_id UUID REFERENCES experiment_runs(id),
  captured_at TIMESTAMPTZ NOT NULL,
  frame_type TEXT,       -- "image" | "force_torque" | "joint_state" | "sensor"
  payload JSONB NOT NULL,
  created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Usage metering for billing
CREATE TABLE usage_records (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  org_id UUID REFERENCES organizations(id),
  run_id UUID REFERENCES experiment_runs(id),
  robot_minutes NUMERIC NOT NULL,
  billed_at TIMESTAMPTZ,
  amount_usd_cents INT
);
```

**Prompt**: "Create this schema in Supabase, add RLS policies so organizations only see their own runs, and generate TypeScript types."

---

## Step 2: Robot Onboard Firmware (Python / ROS 2)

Build the on-board autonomy stack that runs on the robot's compute unit.

**Architecture:**
- `controller/`, ROS 2 nodes for motor control, safety watchdog
- `perception/`, depth camera + force-torque sensor fusion
- `planner/`, step executor that reads protocol JSON and dispatches actions
- `comms/`, ground uplink/downlink over MQTT with store-and-forward queue

**Key algorithm, Microgravity Pipette Controller:**

```python
class MicrogravityPipetteNode(Node):
    """
    Compensates for surface tension dominance and liquid sphere behavior in 0g.
    Uses force-torque feedback to detect liquid contact instead of gravity-based detection.
    """
    def __init__(self):
        super().__init__('pipette_controller')
        self.ft_sub = self.create_subscription(WrenchStamped, '/ft_sensor', self.ft_callback, 10)
        self.cmd_pub = self.create_publisher(JointTrajectory, '/arm/trajectory', 10)
        self.contact_threshold_N = 0.05   # microgravity requires much lower threshold than terrestrial

    def aspirate(self, volume_ul: float, tip_position: Pose):
        """
        Aspirate volume using force feedback rather than gravity-dependent depth sensing.
        Approach slowly until surface tension breaks (detected via FT spike), then pull.
        """
        # Phase 1: approach with vision-guided positioning
        self.move_to_pose(tip_position, speed=0.02)  # slow approach
        # Phase 2: detect meniscus via force-torque inflection
        self.wait_for_ft_contact(threshold=self.contact_threshold_N)
        # Phase 3: aspirate at calibrated rate for zero-g fluid dynamics
        self.actuate_syringe(volume_ul, rate_ul_per_sec=5.0)
```

**Prompt**: "Scaffold a ROS 2 Python package called `generalastro_autonomy` with nodes for pipette control, plate handler, and a mission executor that reads JSON protocol steps. Include unit tests using pytest and a simulation environment using Gazebo with a zero-gravity world plugin."

---

## Step 3: Ground Control Backend API

Build the server that bridges Earth-side customers with orbital robots.

**Stack**: FastAPI + PostgreSQL + Redis (job queue) + MQTT broker (HiveMQ or EMQX for space comm simulation)

```python
# Key endpoints
POST   /api/v1/protocols          # Create experiment protocol
POST   /api/v1/runs               # Schedule a run on a specific robot
GET    /api/v1/runs/{run_id}      # Poll run status and telemetry
GET    /api/v1/runs/{run_id}/frames  # Stream telemetry frames
PATCH  /api/v1/runs/{run_id}/abort   # Emergency abort
POST   /api/v1/robots/{id}/heartbeat # Robot health ping (called by onboard)
GET    /api/v1/robots/{id}/queue     # Robot fetches its pending run queue

# MQTT topics (robot-side)
robot/{unit_id}/telemetry        # Robot publishes telemetry
robot/{unit_id}/commands         # Ground publishes commands
robot/{unit_id}/ack              # Robot acknowledges command receipt
```

**Latency-tolerant design**: commands are persisted before being published to MQTT. If the comm window is closed, the command sits in Redis until the robot's next uplink. The robot fetches its queue on reconnect rather than waiting for a push.

**Prompt**: "Build a FastAPI backend with the endpoints above. Use asyncpg for PostgreSQL, Redis for the command queue, and an MQTT client (aiomqtt) for robot communication. Add a background worker that promotes queued runs to 'uplinking' when the robot comes online. Include OpenAPI docs and Docker Compose with Postgres, Redis, and EMQX."

---

## Step 4: Perception & Manipulation ML

Train the vision models that guide robot manipulation.

**Data pipeline:**
1. Collect synthetic training data: simulate microgravity pipette scenes in Isaac Sim or Blender
2. Augment with ground-based real images from a parabolic flight camera rig (3-4 seconds of real 0g per flight arc)
3. Fine-tune a YOLO-based object detector for lab consumables (pipette tips, well plates, vials)
4. Train a pose estimation model (FoundationPose or custom ViT) for 6-DOF object pose in 0g

**Key challenge, domain randomization for microgravity:**

```python
def randomize_microgravity_scene(scene):
    """
    In zero-g, liquid droplets and small objects float.
    Randomize their positions to train robust perception.
    """
    for obj in scene.floating_objects:
        obj.position += np.random.normal(0, 0.02, 3)  # 2cm positional noise
        obj.rotation = Rotation.random().as_quat()      # any orientation valid
    # Simulate liquid sphere at random position above container
    scene.liquid_sphere.position = scene.container.position + np.random.uniform(-0.01, 0.01, 3)
    scene.liquid_sphere.radius = np.random.uniform(0.003, 0.008)  # 3-8mm droplet
    return scene
```

**Prompt**: "Set up a PyTorch training pipeline using NVIDIA Isaac Sim for synthetic data generation. Fine-tune FoundationPose on lab consumable objects with microgravity domain randomization. Export to ONNX for deployment on the Jetson Orin NX onboard compute. Include a validation script that measures pose estimation accuracy on a held-out set."

---

## Step 5: Customer Experiment Portal (Frontend)

Build the web interface where researchers design, submit, and monitor experiments.

**Stack**: Next.js 15 + TypeScript + shadcn/ui + Supabase Realtime

**Key screens:**
1. **Protocol Builder**, drag-and-drop step editor (pipette X µL from well A1 to B1, centrifuge at N RPM, image sample)
2. **Run Scheduler**, pick robot, time window, priority; shows estimated queue position
3. **Live Monitor**, realtime telemetry feed, camera frames streamed via WebSocket, step progress indicator
4. **Results**, downloadable telemetry logs, annotated images, auto-generated experiment report

```typescript
// Realtime run monitor using Supabase subscriptions
function useRunStatus(runId: string) {
  const [run, setRun] = useState<ExperimentRun | null>(null);
  
  useEffect(() => {
    const channel = supabase
      .channel(`run-${runId}`)
      .on('postgres_changes', {
        event: 'UPDATE',
        schema: 'public',
        table: 'experiment_runs',
        filter: `id=eq.${runId}`
      }, payload => setRun(payload.new as ExperimentRun))
      .subscribe();
    return () => { supabase.removeChannel(channel); };
  }, [runId]);
  
  return run;
}
```

**Prompt**: "Build a Next.js 15 app with three main routes: /protocols (builder), /runs (scheduler + list), and /runs/[id] (live monitor). Use shadcn/ui components, Supabase for auth and realtime, and React Flow for the protocol step drag-and-drop editor. The live monitor should stream telemetry frames as JPEGs via Supabase Storage URLs updated in real time."

---

## Step 6: Billing & Usage Metering

Implement Stripe-based metered billing for robot-time consumption.

```typescript
// Stripe metered billing: report usage after each completed run
async function recordRunUsage(runId: string) {
  const run = await db.experimentRuns.findById(runId);
  const robotMinutes = differenceInMinutes(run.completedAt, run.startedAt);
  
  // Create usage record
  await db.usageRecords.create({
    orgId: run.orgId,
    runId: run.id,
    robotMinutes,
    amountUsdCents: robotMinutes * 500  // $5/min standard rate
  });
  
  // Report to Stripe for metered subscription
  await stripe.subscriptionItems.createUsageRecord(
    run.org.stripeSubscriptionItemId,
    { quantity: robotMinutes, action: 'increment' }
  );
}
```

**Pricing tiers:**
- **Researcher** ($500/mo base + $5/robot-min): Access to standard protocols, 48hr queue
- **Enterprise** (custom): Dedicated robot time blocks, priority queue, custom protocol development, SLA

**Prompt**: "Add Stripe metered billing to the FastAPI backend. Create Stripe products for Researcher and Enterprise tiers, implement a webhook handler for subscription events, and add a usage dashboard to the Next.js frontend showing month-to-date robot minutes consumed vs. plan limit."

---

## Step 7: Deployment & Hardware Integration

Deploy the ground stack and integrate with actual (or simulated) robot hardware.

**Ground infrastructure (AWS):**
```yaml
# docker-compose.prod.yml skeleton
services:
  api:
    image: generalastro/ground-control:latest
    environment:
      - DATABASE_URL=${SUPABASE_DB_URL}
      - REDIS_URL=redis://redis:6379
      - MQTT_BROKER=emqx:1883
    deploy:
      replicas: 3

  mqtt_bridge:
    image: generalastro/mqtt-bridge:latest
    # Bridges EMQX (ground) to actual space comm relay (Iridium/TDRS)
    # In dev: simulates latency with configurable delay middleware

  worker:
    image: generalastro/mission-worker:latest
    # Processes run queue, handles uplink scheduling
```

**Hardware integration checklist:**
- [ ] Flash Jetson Orin NX with custom BSP including ROS 2 and autonomy nodes
- [ ] Radiation tolerance: use COTS components with total ionizing dose (TID) characterization; add watchdog timer for single-event upsets (SEUs)
- [ ] Thermal management: design for 0°C, 50°C operating range inside pressurized module
- [ ] Power budget: target <200W average draw (standard station utility allocation)
- [ ] Integration testing: vibration table at 14.1 Grms random (launch qualification), thermal vacuum chamber at -40°C / +85°C

**Comm latency simulation for local dev:**
```python
# Middleware that adds configurable latency to MQTT messages
# Run locally with: COMM_LATENCY_MS=600 python simulate_space_link.py
import asyncio, aiomqtt, os

LATENCY = int(os.getenv('COMM_LATENCY_MS', 600))

async def relay_with_latency(src_topic, dst_topic):
    async with aiomqtt.Client('localhost') as client:
        await client.subscribe(src_topic)
        async for msg in client.messages:
            await asyncio.sleep(LATENCY / 1000)
            await client.publish(dst_topic, msg.payload)
```

**Prompt**: "Write Terraform for deploying the ground control stack to AWS (ECS Fargate for API and worker, ElastiCache for Redis, MSK for MQTT, RDS for Postgres). Add a GitHub Actions CI/CD pipeline that runs tests, builds Docker images, and deploys on merge to main. Include a local dev setup script that starts all services with `docker compose up` and runs the latency simulator at 600ms."