"""A simulation-only pick/place controller demonstrating VeraPulse policy protocol v1. Python 3.10+, standard library only. No robot transport, learned weights or training. Run: python3 reference_policy.py --port 8791 """ import argparse import hmac import json import math import os import threading import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer SPEEDS = {'shoulder_pan': .65, 'shoulder_lift': .55, 'elbow_flex': .65, 'wrist_flex': .7, 'wrist_roll': .8, 'gripper': .9} MOVES = [('gripper', .1, False), ('shoulder_lift', -.15, True), ('shoulder_pan', .55, True), ('shoulder_lift', .278052849, True), ('gripper', .8, False), ('shoulder_lift', -.15, False)] METADATA = {'protocol': 'verapulse-policy-v1', 'name': 'Reference pick and place', 'version': '1.0', 'robotId': 'so101', 'actionType': 'joint_velocity', 'observationKeys': ['robot']} ORIGINS = {'https://verapulse.ai', 'https://www.verapulse.ai'} | { f'http://{host}:{port}' for host in ('localhost', '127.0.0.1') for port in range(4320, 4341)} class Controller: def __init__(self): self.index = 0 self.hold = 0 self.sequence = 0 self.touched = time.monotonic() def act(self, body): if type(body.get('sequence')) is not int or body['sequence'] != self.sequence: raise ValueError('Out-of-order action request.') if body.get('dt') != .05: raise ValueError('Expected a 0.05-second control interval.') robot = body['observation']['robot'] positions, held = robot['positions'], robot['gripper']['heldObjectId'] if any(type(positions.get(j)) not in (float, int) or not math.isfinite(positions[j]) for j in SPEEDS): raise ValueError('Invalid joint state.') action, done = None, False if self.index == len(MOVES): self.hold += .05 done = self.hold >= 1 else: joint, target, requires_grasp = MOVES[self.index] if requires_grasp and held != 'task-cube': raise ValueError('The cube was not held. Reset and try again.') error = target - positions[joint] pinched = joint == 'gripper' and error < 0 and held == 'task-cube' if abs(error) < .002 or pinched: self.hold += .05 if self.hold >= .5: self.index += 1; self.hold = 0 else: self.hold = 0 speed = SPEEDS[joint] * .5 action = {'type': 'joint_velocity', 'joint': joint, 'velocity': max(-speed, min(speed, error * 10))} self.sequence += 1 self.touched = time.monotonic() return {'runId': body['runId'], 'sequence': body['sequence'], 'action': action, 'done': done} def make_server(port=8791, key=''): runs, lock = {}, threading.Lock() class Handler(BaseHTTPRequestHandler): def log_message(self, *args): pass # Never log observations or access keys. def reply(self, status, body=None): data = json.dumps(body, allow_nan=False).encode() if body is not None else b'' self.send_response(status) origin = self.headers.get('Origin') if origin in ORIGINS: self.send_header('Access-Control-Allow-Origin', origin) self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type, Authorization') self.send_header('Access-Control-Allow-Private-Network', 'true') self.send_header('Vary', 'Origin') self.send_header('Cache-Control', 'no-store') self.send_header('Content-Type', 'application/json') self.send_header('Content-Length', str(len(data))) self.end_headers() self.wfile.write(data) def allowed(self, preflight=False): origin = self.headers.get('Origin') if origin and origin not in ORIGINS: self.reply(403, {'error': 'Origin not allowed.'}); return False if not preflight and key and not hmac.compare_digest(self.headers.get('Authorization', ''), f'Bearer {key}'): self.reply(401, {'error': 'Model access key required.'}); return False return True def do_OPTIONS(self): if self.allowed(preflight=True): self.reply(204) def do_GET(self): if self.allowed(): self.reply(200 if self.path == '/metadata' else 404, METADATA if self.path == '/metadata' else {'error': 'Not found.'}) def do_POST(self): if not self.allowed(): return try: size = int(self.headers.get('Content-Length', '0')) if not 0 < size <= 65536 or self.headers.get('Content-Type') != 'application/json': raise ValueError('Expected a JSON request under 64 KB.') self.connection.settimeout(8) body = json.loads(self.rfile.read(size)) run_id = body['runId'] if not isinstance(run_id, str) or not 1 <= len(run_id) <= 100 or body.get('protocol') != 'verapulse-policy-v1': raise ValueError('Invalid protocol or run ID.') with lock: for old in list(runs): if time.monotonic() - runs[old].touched > 300: del runs[old] if self.path == '/reset': if body['robot']['id'] != 'so101' or body['environment'].get('task', {}).get('id') != 'pick-place-v1': raise ValueError('This reference controller requires the SO-101 pick/place task.') if len(runs) >= 32 and run_id not in runs: raise ValueError('Too many active runs.') runs[run_id] = Controller() result = {'runId': run_id, 'ready': True} elif self.path == '/act': if run_id not in runs: raise ValueError('Reset this run first.') result = runs[run_id].act(body) else: self.reply(404, {'error': 'Not found.'}); return self.reply(200, result) except (ValueError, KeyError, TypeError, AttributeError, TimeoutError): self.reply(400, {'error': 'Invalid or out-of-order model request. Reset and try again.'}) return ThreadingHTTPServer(('127.0.0.1', port), Handler) if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--port', type=int, default=8791) args = parser.parse_args() server = make_server(args.port, os.environ.get('VERAPULSE_MODEL_KEY', '')) print(f'Simulation reference model ready at http://127.0.0.1:{server.server_port}', flush=True) try: server.serve_forever() except KeyboardInterrupt: server.server_close()