Buckets:
| """Record a fresh simulated calibration and RGB-guided block-sorting run. | |
| Start the simulator separately in its fresh block scene, then run: | |
| uv run python video/record_run.py | |
| The server and this client both respect SO101_PORT (default 8877). | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import os | |
| from pathlib import Path | |
| import sys | |
| import time | |
| import urllib.request | |
| ROOT = Path(__file__).resolve().parents[1] | |
| WORK = ROOT / 'work/video' | |
| def main(): | |
| parser = argparse.ArgumentParser(description=__doc__) | |
| parser.add_argument('--port', type=int, default=int(os.environ.get('SO101_PORT', '8877'))) | |
| args = parser.parse_args() | |
| if not 1 <= args.port <= 65535: | |
| parser.error('--port must be between 1 and 65535') | |
| os.environ['SO101_PORT'] = str(args.port) | |
| sys.path.insert(0, str(ROOT / 'scripts')) | |
| import cv2 | |
| import numpy as np | |
| import calibrate_camera | |
| import sort_blocks | |
| from block_vision import detect_blocks | |
| WORK.mkdir(parents=True, exist_ok=True) | |
| command = calibrate_camera.command | |
| base_url = f'http://127.0.0.1:{args.port}' | |
| def state(): | |
| with urllib.request.urlopen(base_url + '/state', timeout=10) as response: | |
| return json.load(response) | |
| def checked_command(endpoint, **payload): | |
| result = command(endpoint, **payload) | |
| if 'error' in result: | |
| raise RuntimeError(result['error']) | |
| return result | |
| def hold(seconds): | |
| end = float(state()['time']) + seconds | |
| while float(state()['time']) < end: | |
| time.sleep(.1) | |
| def observed_command(endpoint, **payload): | |
| result = checked_command(endpoint, **payload) | |
| if endpoint == '/camera': | |
| # This pause occurs in the actual recording, before the next | |
| # command, leaving the observation visible for the edited video. | |
| hold(1.7) | |
| return result | |
| current = state() | |
| if current.get('scene', 'lego') != 'lego': | |
| raise RuntimeError('Restart the simulator with the fresh block scene before recording.') | |
| if any(current.get('warning_counts', [])): | |
| raise RuntimeError('Restart the simulator: it has an existing physics warning.') | |
| checked_command('/record/start', path=str(WORK / 'front-camera.mp4'), fps=30) | |
| print('Recording the front camera to work/video/front-camera.mp4', flush=True) | |
| original_argv = sys.argv[:] | |
| original_sort_command = sort_blocks.command | |
| recording_error = None | |
| try: | |
| checked_command('/record/mark', label='calibrating') | |
| hold(1.8) | |
| calibrate_camera.main() | |
| hold(1.2) | |
| checked_command('/record/mark', label='picking') | |
| sort_blocks.command = observed_command | |
| sys.argv = ['sort_blocks.py', '--block', 'all'] | |
| sort_blocks.main() | |
| controller = sort_blocks.Controller() | |
| rgb, _ = controller.observe('verify-all-blocks-in-box') | |
| _, box = controller.locate_box(rgb) | |
| hull = cv2.convexHull(box.contour.astype(np.float32)) | |
| detections = detect_blocks(rgb) | |
| verification = {} | |
| for color in ('blue', 'red', 'green'): | |
| detection = detections.get(color) | |
| inside = detection is not None and cv2.pointPolygonTest( | |
| hull, tuple(map(float, detection.centroid)), False) >= 0 | |
| verification[color] = { | |
| 'visible_inside_box': bool(inside), | |
| 'pixel': detection.centroid.tolist() if detection is not None else None, | |
| } | |
| if not all(item['visible_inside_box'] for item in verification.values()): | |
| raise RuntimeError(f'Final RGB verification failed: {verification}') | |
| if any(state()['warning_counts']): | |
| raise RuntimeError('Simulator reported a physics warning') | |
| (WORK / 'verification.json').write_text(json.dumps(verification, indent=2) + '\n') | |
| print('All three blocks verified inside the box in the final RGB image.', flush=True) | |
| checked_command('/record/mark', label='finished') | |
| hold(4.5) | |
| except BaseException as exc: | |
| recording_error = exc | |
| raise | |
| finally: | |
| sys.argv = original_argv | |
| sort_blocks.command = original_sort_command | |
| try: | |
| checked_command('/record/stop') | |
| print('Saved work/video/front-camera.mp4 and work/video/timeline.json', flush=True) | |
| except Exception as stop_error: | |
| if recording_error is None: | |
| raise | |
| print(f'Could not finalize the recording: {stop_error}', file=sys.stderr) | |
| if __name__ == '__main__': | |
| main() | |
Xet Storage Details
- Size:
- 4.62 kB
- Xet hash:
- 2ffe726f682c8c9895c8394f31e1c23ad567d6ae0bc9313cd2d02ab99779af7d
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.