### Setup Wired Network Connection Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/source/installation.md Run the setup command to configure the network bridge for wired communication over USB-C. This is provided by the avp_stream package and offers lower latency. ```bash setup-avp-wired ``` -------------------------------- ### Quick Start: VisionProStreamer with Hand Tracking and Video Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/source/index.md Initialize the VisionProStreamer with the Vision Pro's IP address. Optionally, start streaming video from a specified device and format. The loop continuously prints tracking data. ```python from avp_stream import VisionProStreamer avp_ip = "10.31.181.201" # example IP s = VisionProStreamer(ip=avp_ip) # Optional: Start video streaming s.start_streaming(device="/dev/video0", format="v4l2", size="640x480", fps=30, stereo=False) while True: r = s.latest print(r['head'], r['right_wrist'], r['right_fingers']) ``` -------------------------------- ### Install Wrangler CLI Source: https://github.com/improbable-ai/visionproteleop/blob/main/infrastructure/signaling-server/README.md Install the Wrangler CLI globally to manage Cloudflare Workers. ```bash npm install -g wrangler ``` -------------------------------- ### Start Video Streaming with VisionProStreamer Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/build/html/index.html Optionally start streaming video from a specified device. Configure resolution, frame rate, and stereo mode. ```python # Optional: Start video streaming s.start_streaming(device="/dev/video0", format="v4l2", size="640x480", fps=30, stereo=False) ``` -------------------------------- ### Configure Video and Start WebRTC Source: https://github.com/improbable-ai/visionproteleop/blob/main/README.md This snippet shows how to configure video input and start WebRTC streaming using the avp_stream library. It's used for real-world or simulation teleoperation. ```python from avp_stream import VisionProStreamer # Instead of IP address, use the room code shown on Vision Pro s = VisionProStreamer(ip="ABC-1234") # Everything else works exactly the same s.configure_video(device="/dev/video0", format="v4l2", size="1280x720", fps=30) s.start_webrtc() while True: r = s.get_latest() # ... ``` -------------------------------- ### VisionProStreamer Quick Start Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/build/html/index.html Initialize VisionProStreamer with the robot's IP address. This sets up the connection for receiving tracking data. ```python from avp_stream import VisionProStreamer avp_ip = "10.31.181.201" # example IP s = VisionProStreamer(ip=avp_ip) ``` -------------------------------- ### Start Video Streaming with VisionProStreamer Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/build/html/quickstart.html Configure and start video streaming from a specified device to your Vision Pro. This snippet also shows how to access tracking data concurrently. ```python from avp_stream import VisionProStreamer avp_ip = "10.31.181.201" s = VisionProStreamer(ip=avp_ip) # Start video streaming from a camera s.start_streaming( device="/dev/video0", format="v4l2", size="640x480", fps=30, stereo_video=False ) # Now access tracking data as usual while True: r = s.latest # Your robot control code here time.sleep(1/30.) # Maintain desired control rate ``` -------------------------------- ### Start Streaming - Linux USB Camera Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/build/html/video_streaming.html Common configuration for streaming from a USB camera on Linux. ```python s.start_streaming( device="/dev/video0", format="v4l2", size="640x480", fps=30 ) ``` -------------------------------- ### Registering and Streaming Simulation Frames Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/build/html/video_streaming.html Use this snippet to get rendered frames from a simulator and start streaming them. Ensure your simulator API is correctly integrated for `sim.render_camera()`. ```python def get_simulation_frame(): """Get rendered frame from simulator.""" # Get camera image from simulator camera_image = sim.render_camera() # Your simulator API # Convert to RGB numpy array if needed frame = np.array(camera_image) return frame s.register_frame_callback(get_simulation_frame) s.start_streaming( device=None, format=None, size="1280x720", fps=30 ) ``` -------------------------------- ### Deploy Signaling Server Source: https://github.com/improbable-ai/visionproteleop/blob/main/infrastructure/signaling-server/README.md Navigate to the signaling server directory, install dependencies, and deploy the worker. ```bash cd infrastructure/signaling-server npm install wrangler deploy ``` -------------------------------- ### Start Streaming - Windows USB Camera Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/build/html/video_streaming.html Common configuration for streaming from a USB camera on Windows. ```python s.start_streaming( device="0", format="dshow", size="640x480", fps=30 ) ``` -------------------------------- ### Install and Use avp_stream Python Library Source: https://github.com/improbable-ai/visionproteleop/blob/main/website/index.html Install the avp_stream Python package using pip and use the VisionProStreamer class to connect to your Vision Pro device and receive tracking data. Ensure your Vision Pro is on the same network and its IP address is correctly specified. ```python # Install the package pip install --upgrade avp_stream # Basic usage from avp_stream import VisionProStreamer avp_ip = "10.31.181.201" # Your Vision Pro IP s = VisionProStreamer(ip=avp_ip) while True: r = s.latest print(r['head'], r['right_wrist'], r['right_fingers']) ``` -------------------------------- ### Start Streaming from Python Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/build/html/video_streaming.html Initiates video streaming from a specified device with given format, size, and FPS. Ensure the VisionProStreamer is imported. ```python from avp_stream import VisionProStreamer avp_ip = "10.31.181.201" s = VisionProStreamer(ip=avp_ip) s.start_streaming( device="/dev/video0", format="v4l2", size="640x480", fps=30, stereo=False ) ``` -------------------------------- ### Example Output of Test Script Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/source/video_streaming.md Illustrates the output format of the test_video_devices.py script, showing detected devices and their tested configurations. ```text Found devices: /dev/video0 - USB Camera /dev/video2 - Integrated Webcam Testing /dev/video0... ✓ v4l2 @ 640x480 @ 30fps - Working! ✓ v4l2 @ 1280x720 @ 30fps - Working! ✗ v4l2 @ 1920x1080 @ 60fps - Failed ``` -------------------------------- ### Start Streaming - macOS Built-in Camera Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/build/html/video_streaming.html Common configuration for streaming from the built-in camera on macOS. ```python s.start_streaming( device="0", format="avfoundation", size="1280x720", fps=30 ) ``` -------------------------------- ### Install avp_stream Python Package Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/source/index.md Install the avp_stream Python package using pip. This command upgrades the package to the latest version. ```bash pip install --upgrade avp_stream ``` -------------------------------- ### Start Video and Audio Streaming Source: https://github.com/improbable-ai/visionproteleop/blob/main/README.md Initiates WebRTC streaming for video from a robot camera and retrieves tracking data. Use this to stream your robot's camera feed to Vision Pro for teleoperation. ```python from avp_stream import VisionProStreamer avp_ip = "10.31.181.201" # Vision Pro IP (shown in the app) s = VisionProStreamer(ip=avp_ip) # Configure video streaming from robot camera s.configure_video(device="/dev/video0", format="v4l2", size="1280x720", fps=30) s.start_webrtc() while True: r = s.get_latest() # Use tracking data to control your robot head_pose = r['head'] right_wrist = r['right_wrist'] right_fingers = r['right_fingers'] ``` -------------------------------- ### BridgeSetupError Exception Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/build/html/license.html Custom exception raised for errors encountered during VisionProBridge setup. ```APIDOC ## Exception: BridgeSetupError ### Description Custom exception raised when errors occur during the setup or operation of the VisionProBridge. ``` -------------------------------- ### List and Download Public Recordings Source: https://github.com/improbable-ai/visionproteleop/blob/main/README.md Use the avp_stream.datasets module to list available public recordings and download a specific recording to a local directory. Ensure you have the necessary libraries installed. ```python from avp_stream.datasets import list_public_recordings, download_recording # List all public recordings recordings = list_public_recordings() for rec in recordings: print(f"{rec.title} - {rec.duration:.1f}s, {rec.frame_count} frames") print(f" Data: video={rec.has_video}, hands={rec.has_left_hand or rec.has_right_hand}") # Download a recording download_path = download_recording(recordings[0], dest_dir="./downloads") ``` -------------------------------- ### BridgeSetupError Exception Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/build/html/py-modindex.html Custom exception raised during the bridge setup process when errors occur. ```APIDOC ## Exception: BridgeSetupError ### Description An exception class used to indicate errors encountered during the setup or operation of the VisionPro bridge. ``` -------------------------------- ### VisionProBridge Class Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/build/html/video_streaming.html The VisionProBridge class handles the setup and execution of commands for the VisionPro system bridge. ```APIDOC ## Class VisionProBridge Handles the setup and execution of commands for the VisionPro system bridge. ### Methods - **check_prerequisites()**: Verifies that all necessary prerequisites are met. - **run_command(command)**: Executes a given command on the bridge. - **detect_primary_interface_macos()**: Detects the primary network interface on macOS. ``` -------------------------------- ### Get All Tracked Images (ArUco and Custom) Source: https://github.com/improbable-ai/visionproteleop/blob/main/README.md Retrieve a unified list of all tracked images, including ArUco markers and custom registered images. Provides name, type, tracking status, and pose. ```python images = s.get_tracked_images() for image_id, info in images.items(): # image_id format: "aruco_0_5" or "custom_0" print(f"{info['name']}: type={info['image_type']}, tracked={info['is_tracked']}") position = info["pose"][:3, 3] ``` -------------------------------- ### VisionProBridge Methods Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/build/html/genindex.html Methods for configuring and managing the VisionPro bridge, including network interface detection and setup for different operating systems. ```APIDOC ## VisionProBridge.detect_primary_interface_linux() ### Description Detects the primary network interface on a Linux system. ### Method (Not specified, likely a Python method call) ### Endpoint (Not applicable, Python method) ### Parameters None ### Response (Not specified) ## VisionProBridge.is_wireless_interface() ### Description Checks if a given network interface is wireless. ### Method (Not specified, likely a Python method call) ### Endpoint (Not applicable, Python method) ### Parameters (Not specified) ### Response (Not specified) ## VisionProBridge.detect_configuration() ### Description Detects the current system configuration for the VisionPro bridge. ### Method (Not specified, likely a Python method call) ### Endpoint (Not applicable, Python method) ### Parameters None ### Response (Not specified) ## VisionProBridge.setup_bridge_macos() ### Description Sets up the VisionPro bridge on a macOS system. ### Method (Not specified, likely a Python method call) ### Endpoint (Not applicable, Python method) ### Parameters None ### Response (Not specified) ## VisionProBridge.setup_bridge_linux() ### Description Sets up the VisionPro bridge on a Linux system. ### Method (Not specified, likely a Python method call) ### Endpoint (Not applicable, Python method) ### Parameters None ### Response (Not specified) ## VisionProBridge.create_cleanup_script_macos() ### Description Creates a cleanup script for the VisionPro bridge on macOS. ### Method (Not specified, likely a Python method call) ### Endpoint (Not applicable, Python method) ### Parameters None ### Response (Not specified) ## VisionProBridge.create_cleanup_script_linux() ### Description Creates a cleanup script for the VisionPro bridge on Linux. ### Method (Not specified, likely a Python method call) ### Endpoint (Not applicable, Python method) ### Parameters None ### Response (Not specified) ## VisionProBridge.setup() ### Description Sets up the VisionPro bridge. ### Method (Not specified, likely a Python method call) ### Endpoint (Not applicable, Python method) ### Parameters None ### Response (Not specified) ``` -------------------------------- ### Run Camera Configuration Test Script Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/build/html/quickstart.html Execute the provided Python script to test and identify suitable camera parameters. This helps in setting the correct device, format, size, and FPS for your camera. ```python python test_video_devices.py --live ``` -------------------------------- ### VisionProBridge Class Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/build/html/index.html The VisionProBridge class handles the setup and execution of commands for the bridge connection. ```APIDOC ## Class VisionProBridge ### Description Handles the setup and execution of commands for the bridge connection. ### Methods - **check_prerequisites()**: Checks if all necessary prerequisites are met. - **run_command(command)**: Executes a given command. - **detect_primary_interface_macos()**: Detects the primary network interface on macOS. ``` -------------------------------- ### Stream from Camera Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/source/examples.md Demonstrates basic video streaming from a physical camera to Vision Pro. Requires VisionProStreamer and time modules. ```python from avp_stream import VisionProStreamer import time # Vision Pro IP address avp_ip = "10.31.181.201" # Initialize streamer streamer = VisionProStreamer(ip=avp_ip) print("Connected to Vision Pro") # Start video streaming from camera streamer.start_streaming( device="/dev/video0", format="v4l2", size="640x480", fps=30, stereo=False ) print("Video streaming started") try: while True: # Get latest tracking data data = streamer.latest # Print head position head_pos = data['head'][0, :3, 3] print(f"Head: x={head_pos[0]:.2f}, y={head_pos[1]:.2f}, z={head_pos[2]:.2f}") time.sleep(0.1) except KeyboardInterrupt: print("Stopping...") streamer.stop() ``` -------------------------------- ### Configure Video with Overlay Processing Source: https://github.com/improbable-ai/visionproteleop/blob/main/README.md Sets up video streaming with a custom overlay function applied to each frame. This is useful for adding annotations or processing frames before they are sent. ```python def add_overlay(frame): return cv2.putText(frame, "Robot View", (50, 50), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2) s = VisionProStreamer(ip=avp_ip) s.register_frame_callback(add_overlay) s.configure_video(device="/dev/video0", format="v4l2", size="640x480", fps=30) s.start_webrtc() ``` -------------------------------- ### Main Functions Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/build/html/genindex.html Entry points for the bridge and latency testing modules. ```APIDOC ## main() (avp_stream.bridge_avp) ### Description Main function for the bridge module. ### Method (Not specified, likely a Python function call) ### Endpoint (Not applicable, Python function) ### Parameters None ### Response (Not specified) ## main() (avp_stream.latency_test) ### Description Main function for the latency testing module. ### Method (Not specified, likely a Python function call) ### Endpoint (Not applicable, Python function) ### Parameters None ### Response (Not specified) ``` -------------------------------- ### Plot Benchmark Results Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/benchmark.md Generate a comparison image from collected benchmark JSON files. The output will be saved as comparison.png. ```bash python avp_stream/plot_benchmarks.py benchmarks/*.json --output comparison.png ``` -------------------------------- ### Run Latency Benchmark Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/benchmark.md Execute the main latency test script. Replace with the actual IP address of the Vision Pro device. ```bash python avp_stream/latency_test.py --ip --sweep ``` -------------------------------- ### Get Detected ArUco Markers Source: https://github.com/improbable-ai/visionproteleop/blob/main/README.md Retrieve a dictionary of detected ArUco markers, including their pose, fixed status, and tracking status. ```python markers = s.get_markers() for marker_id, info in markers.items(): pose = info["pose"] # (4, 4) transform matrix position = pose[:3, 3] # XYZ position is_fixed = info["is_fixed"] # Whether pose is frozen is_tracked = info["is_tracked"] # Whether actively tracked by ARKit aruco_dict = info["dict"] # ArUco dictionary type (e.g., 0 = DICT_4X4_50) ``` -------------------------------- ### Configure Audio Streaming with Microphone Source: https://github.com/improbable-ai/visionproteleop/blob/main/README.md Sets up audio streaming using the default microphone input. This allows for two-way audio communication during teleoperation. ```python s = VisionProStreamer(ip=avp_ip) s.configure_video(device="/dev/video0", format="v4l2", size="1280x720", fps=30) s.configure_audio(device=":0", stereo=True) # Default mic s.start_webrtc() ``` -------------------------------- ### VisionProStreamer Class Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/build/html/index.html The VisionProStreamer class provides methods for managing the Vision Pro teleoperation stream, including starting, stopping, configuring, and retrieving data. ```APIDOC ## Class VisionProStreamer ### Description Provides methods for managing the Vision Pro teleoperation stream, including starting, stopping, configuring, and retrieving data. ### Methods - **__init__()**: Initializes the VisionProStreamer. - **stream()**: Starts the streaming process. - **get_latest()**: Retrieves the latest available data. - **get_recording()**: Retrieves recorded data. - **set_origin()**: Sets the origin point for the stream. - **get_sync_timestamp()**: Gets the synchronization timestamp. - **get_local_ip()**: Gets the local IP address. - **register_frame_callback(callback)**: Registers a callback function for frame updates. - **register_audio_callback(callback)**: Registers a callback function for audio updates. - **update_frame()**: Updates the frame data. - **configure_video(config)**: Configures video streaming settings. - **configure_audio(config)**: Configures audio streaming settings. - **configure_mujoco(config)**: Configures MuJoCo simulation integration. - **configure_sim(config)**: Configures general simulation integration. - **configure_isaac(config)**: Configures Isaac Sim integration. - **update_sim(data)**: Updates simulation data. - **serve()**: Starts serving the stream. - **start_webrtc()**: Initiates WebRTC connection. - **wait_for_sim_channel()**: Waits for the simulation channel to be available. - **wait_for_connection()**: Waits for a client connection. - **is_connected()**: Checks if a client is connected. - **reset_benchmark_epoch()**: Resets the benchmark epoch. - **enable_sim_benchmark()**: Enables simulation benchmarking. - **update_stream_resolution(resolution)**: Updates the stream resolution. - **wait_for_benchmark_event()**: Waits for a benchmark event. - **start_streaming()**: Starts the streaming process. ``` -------------------------------- ### Accessing Hand Data via Dictionary Style Source: https://github.com/improbable-ai/visionproteleop/blob/main/README.md Legacy dictionary-style access for hand and head data is still supported. ```python data["head"] ``` ```python data["right_wrist"] ``` ```python data["right_fingers"] ``` ```python data["right_arm"] ``` ```python data["right_pinch_distance"] ``` -------------------------------- ### Multiple Frame Processing Steps Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/build/html/video_streaming.html Chains multiple processing functions, applying preprocessing before adding information overlays. Each function should accept and return a frame. ```python def preprocessing(frame): """Adjust brightness/contrast.""" import cv2 alpha = 1.2 # Contrast beta = 10 # Brightness return cv2.convertScaleAbs(frame, alpha=alpha, beta=beta) def add_info(frame): """Add information overlay.""" import cv2 import time timestamp = time.strftime("%H:%M:%S") cv2.putText(frame, timestamp, (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2) return frame # Apply preprocessing first processed = preprocessing(frame) # Then add overlay final = add_info(processed) ``` -------------------------------- ### Utility Functions Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/build/html/genindex.html Utility functions for injecting benchmark payloads and running main application logic. ```APIDOC ## Utility Functions ### Description Provides utility functions for testing and application entry points. ### Functions - `inject_benchmark_payload(payload)`: Injects a benchmark payload. - `main()`: The main entry point for the bridge or latency test module. ``` -------------------------------- ### Positioning Simulation in AR Source: https://github.com/improbable-ai/visionproteleop/blob/main/README.md Demonstrates how to use the `relative_to` parameter to position a simulation in AR space. Supports both 4-dimensional (translation + Z-axis yaw) and 7-dimensional (full quaternion) orientation specifications. ```python # 4-dim: [x, y, z, yaw°] — translation + rotation around z-axis (degrees) # 7-dim: [x, y, z, qw, qx, qy, qz] — full quaternion orientation ``` -------------------------------- ### Login to Cloudflare Source: https://github.com/improbable-ai/visionproteleop/blob/main/infrastructure/signaling-server/README.md Authenticate with your Cloudflare account using the Wrangler CLI. ```bash wrangler login ``` -------------------------------- ### Configure Synthetic Audio Streaming Source: https://github.com/improbable-ai/visionproteleop/blob/main/README.md Enables synthetic audio generation via a callback function. This can be used for generating feedback sounds or custom audio streams. ```python def beep_on_pinch(audio_frame): # Generate audio based on hand tracking state return audio_frame s = VisionProStreamer(ip=avp_ip) s.register_audio_callback(beep_on_pinch) s.configure_video(size="1280x720", fps=60) s.configure_audio(sample_rate=48000, stereo=True) s.start_webrtc() ``` -------------------------------- ### Create Ground Point and Height Line Entities Source: https://github.com/improbable-ai/visionproteleop/blob/main/Tracking Streamer/Supporting files/Archive/HandToGround.md Initializes two `Entity` objects: `heightLineEntity` and `groundPointEntity`. The `groundPointEntity` is a sphere with an occlusion component to represent a point on the ground. ```swift let heightLineEntity = Entity() let groundPointEntity: Entity = { let radius: Float = 0.03 let value = ModelEntity(mesh: .generateSphere(radius: radius), materials: [SimpleMaterial(color: .yellow, isMetallic: false)]) let occlusion = ModelEntity(mesh: .generateCylinder(height: radius, radius: radius), materials: [OcclusionMaterial()]) occlusion.position.y -= radius / 2 value.addChild(occlusion) return value }() ``` -------------------------------- ### Simulation Integration with VisionProStreamer Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/source/examples.md Integrates VisionProStreamer with a simulation environment to update avatar poses based on tracked data. Requires 'avp_stream' and a custom 'simulation_env' library. ```python from avp_stream import VisionProStreamer import simulation_env # Your simulation environment avp_ip = "10.31.181.201" streamer = VisionProStreamer(ip=avp_ip) sim = simulation_env.Simulation() def get_sim_camera(): """Get rendered frame from simulation.""" return sim.render_camera() # Register simulation camera streamer.register_frame_callback(get_sim_camera) streamer.start_streaming( device=None, format=None, size="1280x720", fps=30 ) while True: data = streamer.latest # Update simulation with tracked poses sim.set_avatar_head(data['head'][0]) sim.set_avatar_hand(data['right_wrist'][0]) # Step simulation sim.step() ``` -------------------------------- ### Configure MuJoCo with World Frame Offset Source: https://github.com/improbable-ai/visionproteleop/blob/main/README.md Configures MuJoCo simulation with a specified world frame position and rotation. Use this to place the simulation origin relative to the physical ground. ```python s.configure_mujoco("robot.xml", model, data, relative_to=[0, 0, 0.8, 90]) ``` -------------------------------- ### Process Frames Before Streaming Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/source/examples.md Shows how to process video frames, such as adding overlays and timestamps, before streaming them to Vision Pro. Uses OpenCV for image manipulation. ```python from avp_stream import VisionProStreamer import cv2 import numpy as np import time avp_ip = "10.31.181.201" def process_frame(frame): """Add overlay and processing to video frame.""" # frame is RGB numpy array (H, W, 3) # Add timestamp timestamp = time.strftime("%H:%M:%S") cv2.putText(frame, timestamp, (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 2) # Add a robot status indicator status = "ACTIVE" cv2.putText(frame, f"Status: {status}", (10, 70), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2) # Draw a crosshair in the center h, w = frame.shape[:2] cv2.line(frame, (w//2 - 20, h//2), (w//2 + 20, h//2), (255, 0, 0), 2) cv2.line(frame, (w//2, h//2 - 20), (w//2, h//2 + 20), (255, 0, 0), 2) return frame # Initialize and register callback streamer = VisionProStreamer(ip=avp_ip) streamer.register_frame_callback(process_frame) # Start streaming with processing streamer.start_streaming( device="/dev/video0", format="v4l2", size="640x480", fps=30 ) try: while True: data = streamer.latest # Your robot control logic here time.sleep(0.1) except KeyboardInterrupt: streamer.stop() ``` -------------------------------- ### Setting Points for Debugging Source: https://github.com/improbable-ai/visionproteleop/blob/main/Tracking Streamer/Supporting files/Archive/DebugView.md Handles the placement and visualization of two points ('1' and '2') in the AR scene upon tapping. It manages the creation and updating of these point entities, including their visual representation. ```swift fileprivate extension 👆DebugView { func setPoints() { guard let pointer = rootEntity?.findEntity(named: "POINTER") else { return } if rootEntity?.findEntity(named: "1") == nil { let entity = Entity() entity.name = "1" entity.position = pointer.position entity.components.set(ModelComponent(mesh: .generateSphere(radius: 0.025), materials: [SimpleMaterial(color: .red, isMetallic: false)])) rootEntity?.addChild(entity) } else { if let entity2 = rootEntity?.findEntity(named: "2") { rootEntity?.removeChild(entity2) } let entity = Entity() entity.name = "2" entity.position = pointer.position entity.components.set(ModelComponent(mesh: .generateSphere(radius: 0.025), materials: [SimpleMaterial(color: .green, isMetallic: false)])) rootEntity?.addChild(entity) } } ``` -------------------------------- ### Enable Video Streaming with VisionProTeleop Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/source/quickstart.md Configure VisionProTeleop to stream video from a device to your Vision Pro. Ensure the camera device and settings are correctly specified. ```python from avp_stream import VisionProStreamer import time avp_ip = "10.31.181.201" s = VisionProStreamer(ip=avp_ip) # Start video streaming from a camera s.start_streaming( device="/dev/video0", format="v4l2", size="640x480", fps=30, stereo_video=False ) # Now access tracking data as usual while True: r = s.latest # Your robot control code here time.sleep(1/30.) # Maintain desired control rate ``` -------------------------------- ### Data Recording with VisionProStreamer Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/source/examples.md Records tracking data from VisionProStreamer to a JSON file with timestamps. Requires 'avp_stream', 'numpy', 'json', and 'datetime'. ```python from avp_stream import VisionProStreamer import numpy as np import json from datetime import datetime avp_ip = "10.31.181.201" streamer = VisionProStreamer(ip=avp_ip) # Start recording recording = [] start_time = datetime.now() try: while True: data = streamer.latest # Save tracking data with timestamp timestamp = (datetime.now() - start_time).total_seconds() recording.append({ 'timestamp': timestamp, 'head': data['head'].tolist(), 'right_wrist': data['right_wrist'].tolist(), 'left_wrist': data['left_wrist'].tolist(), 'right_fingers': data['right_fingers'].tolist(), 'left_fingers': data['left_fingers'].tolist(), 'right_pinch': data['right_pinch_distance'], 'left_pinch': data['left_pinch_distance'], }) except KeyboardInterrupt: # Save recording output_file = f"recording_{start_time.strftime('%Y%m%d_%H%M%S')}.json" with open(output_file, 'w') as f: json.dump(recording, f) print(f"Saved recording to {output_file}") streamer.stop() ``` -------------------------------- ### Generate Synthetic Frames Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/source/examples.md Generates synthetic frames with an animated test pattern for streaming to Vision Pro, useful when a physical camera is unavailable. Uses NumPy for frame generation. ```python from avp_stream import VisionProStreamer import numpy as np import time avp_ip = "10.31.181.201" # State for animation frame_count = 0 def generate_synthetic_frame(): """Generate an animated test pattern.""" global frame_count height, width = 720, 1280 frame = np.zeros((height, width, 3), dtype=np.uint8) # Create animated gradient t = frame_count / 60.0 # Time in seconds for i in range(height): for j in range(width): # Animated color pattern r = int(127 + 127 * np.sin(t + i * 0.01)) g = int(127 + 127 * np.sin(t + j * 0.01)) b = int(127 + 127 * np.sin(t + (i + j) * 0.005)) frame[i, j] = [r, g, b] frame_count += 1 return frame # Initialize and register callback streamer = VisionProStreamer(ip=avp_ip) streamer.register_frame_callback(generate_synthetic_frame) # Start streaming synthetic frames streamer.start_streaming( device=None, format=None, size="1280x720", fps=60 ) try: while True: data = streamer.latest time.sleep(0.1) except KeyboardInterrupt: streamer.stop() ``` -------------------------------- ### Transforming Points from Wrist to Ground Frame Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/source/data_format.md Illustrates how to combine transformation matrices to transform a point (e.g., a finger joint) from the wrist's local coordinate system to the ground frame. Matrix multiplication is used for this operation. ```python # Transform a point from wrist frame to ground frame finger_in_wrist = data['right_fingers'][5] # Index finger metacarpal wrist_in_ground = data['right_wrist'][0] # Combine transformations finger_in_ground = wrist_in_ground @ finger_in_wrist ``` -------------------------------- ### Register Frame Callback for Robot Control Source: https://github.com/improbable-ai/visionproteleop/blob/main/examples/README.md Use the callback method to decouple video frame processing from your main control loop. This allows the video to stream at a different rate (e.g., 60Hz) than your control logic (e.g., 100Hz), ensuring consistent control performance. ```python streamer.register_frame_callback(visualizer_callback()) streamer.start_streaming(device=None, fps=60) while True: # Control loop runs at 100Hz robot_state = get_robot_state() action = compute_action(robot_state) robot.execute(action) time.sleep(1/100.) # Control rate: 100Hz # Video automatically streams at 60Hz in background ``` -------------------------------- ### Stream Isaac Lab Simulation Source: https://github.com/improbable-ai/visionproteleop/blob/main/README.md Configures and streams an Isaac Lab simulation to Vision Pro. Allows for positioning the simulation in AR and selecting specific environments to stream. ```python from avp_stream import VisionProStreamer # After creating your Isaac Lab environment... streamer = VisionProStreamer(ip=avp_ip) streamer.configure_isaac( scene=env.scene, relative_to=[0, 0, 0.8, 90], include_ground=False, env_indices=[0], # Stream only first environment ) streamer.start_webrtc() while simulation_app.is_running(): env.step(action) streamer.update_sim() # Stream updated poses to Vision Pro ``` -------------------------------- ### Multi-Modal Control with VisionProStreamer Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/source/examples.md Combines head and hand tracking for robot control, using head orientation for base movement and hands for manipulation and mode switching. Requires 'avp_stream' and 'robot_interface'. ```python from avp_stream import VisionProStreamer import robot_interface import numpy as np avp_ip = "10.31.181.201" streamer = VisionProStreamer(ip=avp_ip) robot = robot_interface.Robot() while True: data = streamer.latest # Use head orientation to control robot base head_matrix = data['head'][0] head_forward = head_matrix[:3, 2] # Z-axis # Project to ground plane forward_2d = head_forward[:2] / np.linalg.norm(head_forward[:2]) robot.set_base_direction(forward_2d) # Use right hand for manipulation robot.set_arm_pose(data['right_wrist'][0]) robot.set_gripper(data['right_pinch_distance']) # Use left hand for mode switching if data['left_pinch_distance'] < 0.02: robot.switch_mode() # Toggle between modes ``` -------------------------------- ### Configure Stereo Video Streaming Source: https://github.com/improbable-ai/visionproteleop/blob/main/README.md Configures video streaming for stereo (3D) output. Ensure your camera and capture settings support the specified resolution for stereo capture. ```python s = VisionProStreamer(ip=avp_ip) s.configure_video(device="/dev/video0", format="v4l2", size="1920x1080", fps=30, stereo=True) s.start_webrtc() ``` -------------------------------- ### VisionProStreamer Class Methods Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/build/html/search.html This section details the various methods available for the VisionProStreamer class, which is central to managing video and audio streams, configurations, and synchronization. ```APIDOC ## VisionProStreamer Methods ### `__init__()` Initializes the VisionProStreamer. ### `stream()` Starts the streaming process. ### `get_latest()` Retrieves the latest available frame or data. ### `get_recording()` Gets the recorded data. ### `set_origin()` Sets the origin for the stream. ### `get_sync_timestamp()` Gets the synchronization timestamp. ### `get_local_ip()` Retrieves the local IP address. ### `register_frame_callback()` Registers a callback function for frames. ### `register_audio_callback()` Registers a callback function for audio. ### `update_frame()` Updates the frame data. ### `configure_video()` Configures video settings. ### `configure_audio()` Configures audio settings. ### `configure_mujoco()` Configures MuJoCo simulation settings. ### `configure_sim()` Configures general simulation settings. ### `configure_isaac()` Configures Isaac Sim settings. ### `update_sim()` Updates the simulation state. ### `serve()` Starts serving the stream. ### `start_webrtc()` Initiates WebRTC connection. ### `wait_for_sim_channel()` Waits for the simulation channel to be ready. ### `wait_for_connection()` Waits for a client connection. ### `is_connected()` Checks if a client is connected. ### `reset_benchmark_epoch()` Resets the benchmark epoch. ### `enable_sim_benchmark()` Enables simulation benchmarking. ### `update_stream_resolution()` Updates the stream resolution. ### `wait_for_benchmark_event()` Waits for a benchmark event. ### `start_streaming()` Starts the actual streaming process. ``` -------------------------------- ### Connect and Track Hands with VisionProStreamer Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/build/html/quickstart.html Connect to your Vision Pro device and stream hand tracking data. Replace '10.31.181.201' with your Vision Pro's IP address. ```python from avp_stream import VisionProStreamer # Replace with your Vision Pro's IP address avp_ip = "10.31.181.201" s = VisionProStreamer(ip=avp_ip) # Access the latest tracking data while True: r = s.latest print(f"Head position: {r['head']}") print(f"Right wrist: {r['right_wrist']}") print(f"Right fingers: {r['right_fingers']}") ``` -------------------------------- ### Frame Processing with Overlay Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/build/html/video_streaming.html Registers a callback function to add custom overlays to frames before streaming. The callback receives the frame as a NumPy array. ```python def add_overlay(frame): """Add information overlay to frame.""" import cv2 # frame is a numpy array (H, W, 3) in RGB format cv2.putText(frame, "Robot View", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 2) # Draw bounding boxes, add filters, etc. return frame s.register_frame_callback(add_overlay) s.start_streaming( device="/dev/video0", format="v4l2", size="640x480", fps=30 ) ``` -------------------------------- ### Robot Teleoperation with VisionProStreamer Source: https://github.com/improbable-ai/visionproteleop/blob/main/docs/source/examples.md Streams a robot's camera to Vision Pro and uses hand tracking data to control robot actions like gripper and end-effector pose. Assumes a robot_interface module is available. ```python from avp_stream import VisionProStreamer import robot_interface # Your robot's API avp_ip = "10.31.181.201" streamer = VisionProStreamer(ip=avp_ip) robot = robot_interface.Robot() # Stream robot's camera back to Vision Pro streamer.start_streaming( device="/dev/video0", format="v4l2", size="640x480", fps=30 ) while True: data = streamer.latest # Use right hand to control gripper gripper_opening = data['right_pinch_distance'] robot.set_gripper(gripper_opening) # Use right wrist to control end-effector pose target_pose = data['right_wrist'][0] robot.move_to_pose(target_pose) ```