Programming Your First AI Robot with Python

# Create a virtual environment for Reachy development python -m venv reachy_env source reachy_env/bin/activate # On Windows: reachy_env\Scripts\activate # Install the complete Reachy development stack pip install reachy-sdk pip install opencv-python pip install numpy pip install scipy pip install matplotlib # For data visualization pip install jupyter # For interactive development # Install additional AI/ML libraries pip install transformers # For Hugging Face model integration pip install torch # PyTorch for deep learning pip install speechrecognition # For voice commands

from reachy_sdk import ReachySDK import time import math # Connect to your Reachy Mini reachy = ReachySDK(host='reachy-mini.local') def basic_head_movements(): """Demonstrate basic head control capabilities.""" print("Starting basic head movements...") # Reset to neutral position reachy.head.look_at(x=0, y=0, z=50, duration=2.0) time.sleep(2) # Look around - simulate curiosity positions = [ (30, 0, 50), # Look right (-30, 0, 50), # Look left (0, 20, 50), # Look up (0, -20, 50), # Look down (0, 0, 50) # Return to center ] for x, y, z in positions: reachy.head.look_at(x=x, y=y, z=z, duration=1.5) time.sleep(1.5) print("Basic movements complete!") # Run the demonstration basic_head_movements()

def natural_head_tracking(): """Create natural head movements that simulate human-like behavior.""" def smooth_sine_movement(duration=10, frequency=0.5): """Create smooth, sine-wave based head movements.""" start_time = time.time() while time.time() - start_time < duration: current_time = time.time() - start_time # Calculate smooth positions using sine waves x = 20 * math.sin(frequency * current_time) y = 15 * math.sin(frequency * current_time * 0.7) z = 50 + 10 * math.sin(frequency * current_time * 0.3) reachy.head.look_at(x=x, y=y, z=z, duration=0.1) time.sleep(0.1) def simulate_attention(): """Simulate paying attention to different objects.""" attention_points = [ {"name": "Screen", "position": (0, 10, 40), "duration": 3}, {"name": "Keyboard", "position": (0, -15, 30), "duration": 2}, {"name": "User", "position": (0, 0, 50), "duration": 4}, {"name": "Side object", "position": (25, 5, 45), "duration": 2} ] for point in attention_points: print(f"Looking at: {point['name']}") x, y, z = point['position'] reachy.head.look_at(x=x, y=y, z=z, duration=1.0) time.sleep(point['duration']) # Add small random movements to simulate natural micro-movements for _ in range(3): offset_x = x + random.uniform(-2, 2) offset_y = y + random.uniform(-2, 2) reachy.head.look_at(x=offset_x, y=offset_y, z=z, duration=0.5) time.sleep(0.5) # Demonstrate natural movements import random natural_head_tracking()

import cv2 import numpy as np def setup_camera_processing(): """Initialize camera and set up basic computer vision pipeline.""" # Access the camera camera = reachy.camera def process_video_frame(frame): """Process a single video frame with basic CV operations.""" # Convert to grayscale for processing gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) # Apply Gaussian blur to reduce noise blurred = cv2.GaussianBlur(gray, (5, 5), 0) # Detect edges edges = cv2.Canny(blurred, 50, 150) # Find contours contours, _ = cv2.findContours(edges, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) # Draw contours on the original frame cv2.drawContours(frame, contours, -1, (0, 255, 0), 2) return frame, len(contours) # Process video stream for frame in camera.stream(): processed_frame, object_count = process_video_frame(frame) # Display frame (if running locally) cv2.imshow('Reachy Vision', processed_frame) # Print detection info print(f"Objects detected: {object_count}") # Break on 'q' key press if cv2.waitKey(1) & 0xFF == ord('q'): break cv2.destroyAllWindows() # Start vision processing setup_camera_processing()

def sound_localization_demo(): """Demonstrate sound source localization and tracking.""" # Access the microphone array mic_array = reachy.microphones def detect_sound_direction(): """Analyze audio input to determine sound source direction.""" # Record audio from all 4 microphones audio_data = mic_array.record(duration=1.0) # Calculate time delays between microphones # This is a simplified example - actual implementation would use # more sophisticated signal processing techniques delays = [] for i in range(1, 4): # Cross-correlation to find delay correlation = np.correlate(audio_data[0], audio_data[i], mode='full') delay = np.argmax(correlation) - len(audio_data[i]) + 1 delays.append(delay) # Convert delays to approximate direction # This is a basic approximation avg_delay = np.mean(delays) if avg_delay > 10: return "left" elif avg_delay < -10: return "right" else: return "center" def respond_to_sound(): """Make the robot respond to sound direction.""" print("Listening for sounds...") for _ in range(10): # Listen for 10 iterations direction = detect_sound_direction() print(f"Sound detected from: {direction}") # Turn head toward sound source if direction == "left": reachy.head.look_at(x=-30, y=0, z=50, duration=1.0) elif direction == "right": reachy.head.look_at(x=30, y=0, z=50, duration=1.0) else: reachy.head.look_at(x=0, y=0, z=50, duration=1.0) time.sleep(2) respond_to_sound() # Demonstrate sound localization sound_localization_demo()

import speech_recognition as sr from transformers import pipeline class VoiceAssistant: def __init__(self, reachy): self.reachy = reachy self.recognizer = sr.Recognizer() self.microphone = sr.Microphone() # Initialize AI models self.sentiment_analyzer = pipeline("sentiment-analysis") self.qa_model = pipeline("question-answering") # Calibrate microphone with self.microphone as source: self.recognizer.adjust_for_ambient_noise(source) def listen_for_command(self): """Listen for voice commands and return recognized text.""" try: with self.microphone as source: print("Listening...") # Animate antennas to show listening state self.reachy.antennas.listening() audio = self.recognizer.listen(source, timeout=5, phrase_time_limit=3) print("Processing speech...") command = self.recognizer.recognize_google(audio) print(f"Heard: {command}") return command.lower() except sr.WaitTimeoutError: print("No speech detected") return None except sr.UnknownValueError: print("Could not understand audio") return None except sr.RequestError as e: print(f"Error with speech recognition: {e}") return None def analyze_sentiment(self, text): """Analyze the sentiment of the input text.""" result = self.sentiment_analyzer(text) return result[0] def respond_with_emotion(self, sentiment): """Respond with appropriate emotional expression.""" if sentiment['label'] == 'POSITIVE': # Happy response self.reachy.antennas.happy() self.reachy.head.look_at(x=0, y=10, z=50, duration=1.0) self.reachy.voice.say("I'm happy to help!") elif sentiment['label'] == 'NEGATIVE': # Sad/concerned response self.reachy.antennas.sad() self.reachy.head.look_at(x=0, y=-10, z=50, duration=1.0) self.reachy.voice.say("I sense you're upset. How can I help?") else: # Neutral response self.reachy.antennas.neutral() self.reachy.head.look_at(x=0, y=0, z=50, duration=1.0) self.reachy.voice.say("I'm here to assist you.") def handle_command(self, command): """Process and respond to voice commands.""" # Analyze sentiment first sentiment = self.analyze_sentiment(command) print(f"Sentiment: {sentiment}") # Command processing if "hello" in command or "hi" in command: self.reachy.voice.say("Hello! Nice to meet you!") self.respond_with_emotion(sentiment) elif "how are you" in command: self.reachy.voice.say("I'm functioning perfectly! Thank you for asking.") self.reachy.antennas.happy() elif "dance" in command: self.perform_dance() elif "look at me" in command: self.reachy.head.look_at(x=0, y=0, z=50, duration=1.0) self.reachy.voice.say("I'm looking at you now!") elif "weather" in command: self.reachy.voice.say("I don't have access to weather data, but I hope it's a beautiful day!") self.reachy.antennas.neutral() else: # Unknown command - respond with curiosity self.reachy.antennas.curious() self.reachy.voice.say("I'm not sure I understood that. Could you repeat it?") def perform_dance(self): """Perform a simple dance routine.""" self.reachy.voice.say("Let me show you my moves!") dance_moves = [ (20, 15, 45), # Right up (-20, 15, 45), # Left up (20, -15, 55), # Right down (-20, -15, 55), # Left down (0, 0, 50) # Center ] for x, y, z in dance_moves * 2: # Repeat twice self.reachy.head.look_at(x=x, y=y, z=z, duration=0.5) self.reachy.antennas.happy() time.sleep(0.5) def run(self): """Main loop for voice assistant.""" self.reachy.voice.say("Voice assistant activated! Say hello to begin.") while True: command = self.listen_for_command() if command: if "goodbye" in command or "stop" in command: self.reachy.voice.say("Goodbye! It was nice talking with you.") self.reachy.antennas.goodbye() break else: self.handle_command(command) time.sleep(0.5) # Create and run voice assistant assistant = VoiceAssistant(reachy) assistant.run()

from enum import Enum import threading import queue class RobotState(Enum): IDLE = "idle" LISTENING = "listening" PROCESSING = "processing" RESPONDING = "responding" TRACKING = "tracking" class StateMachine: def __init__(self, reachy): self.reachy = reachy self.current_state = RobotState.IDLE self.state_queue = queue.Queue() self.running = True # State-specific behaviors self.state_behaviors = { RobotState.IDLE: self.idle_behavior, RobotState.LISTENING: self.listening_behavior, RobotState.PROCESSING: self.processing_behavior, RobotState.RESPONDING: self.responding_behavior, RobotState.TRACKING: self.tracking_behavior } def transition_to_state(self, new_state): """Safely transition to a new state.""" print(f"Transitioning from {self.current_state.value} to {new_state.value}") self.current_state = new_state self.state_queue.put(new_state) def idle_behavior(self): """Robot behavior when idle - subtle movements.""" # Gentle breathing-like movement for i in range(5): y_offset = 2 * math.sin(i * 0.5) self.reachy.head.look_at(x=0, y=y_offset, z=50, duration=2.0) time.sleep(2.0) def listening_behavior(self): """Robot behavior when actively listening.""" self.reachy.antennas.listening() # Slight forward lean to show attention self.reachy.head.look_at(x=0, y=5, z=45, duration=1.0) def processing_behavior(self): """Robot behavior when processing information.""" self.reachy.antennas.thinking() # Small side-to-side movements for _ in range(3): self.reachy.head.look_at(x=5, y=0, z=50, duration=0.3) self.reachy.head.look_at(x=-5, y=0, z=50, duration=0.3) def responding_behavior(self): """Robot behavior when responding to user.""" self.reachy.antennas.happy() self.reachy.head.look_at(x=0, y=0, z=50, duration=1.0) def tracking_behavior(self): """Robot behavior when tracking objects or faces.""" # This would integrate with computer vision pass def run(self): """Main state machine loop.""" while self.running: # Execute current state behavior if self.current_state in self.state_behaviors: self.state_behaviors[self.current_state]() # Check for state transitions try: new_state = self.state_queue.get_nowait() # State transition logic here except queue.Empty: pass time.sleep(0.1) # Usage example state_machine = StateMachine(reachy) threading.Thread(target=state_machine.run, daemon=True).start() # Trigger state changes based on events state_machine.transition_to_state(RobotState.LISTENING) time.sleep(5) state_machine.transition_to_state(RobotState.PROCESSING) time.sleep(3) state_machine.transition_to_state(RobotState.RESPONDING)

from transformers import pipeline, AutoTokenizer, AutoModel import torch class AIIntegration: def __init__(self, reachy): self.reachy = reachy # Initialize various AI models self.setup_ai_models() def setup_ai_models(self): """Initialize AI models from Hugging Face Hub.""" print("Loading AI models...") # Text generation for conversations self.text_generator = pipeline( "text-generation", model="microsoft/DialoGPT-small" ) # Image classification for object recognition self.image_classifier = pipeline( "image-classification", model="google/vit-base-patch16-224" ) # Sentiment analysis for emotional responses self.sentiment_analyzer = pipeline( "sentiment-analysis", model="cardiffnlp/twitter-roberta-base-sentiment-latest" ) print("AI models loaded successfully!") def analyze_camera_view(self): """Analyze what the robot sees and respond accordingly.""" # Capture frame from camera frame = self.reachy.camera.capture_frame() # Classify objects in the image results = self.image_classifier(frame) # Get top prediction top_prediction = results[0] object_name = top_prediction['label'] confidence = top_prediction['score'] print(f"I see: {object_name} (confidence: {confidence:.2f})") # Respond based on what was detected if confidence > 0.7: response = f"I can see a {object_name}! That's interesting." self.reachy.voice.say(response) # Look at the detected object with interest self.reachy.antennas.curious() self.reachy.head.look_at(x=0, y=5, z=45, duration=1.0) else: self.reachy.voice.say("I'm not sure what I'm looking at. Could you help me understand?") self.reachy.antennas.confused() def intelligent_conversation(self, user_input): """Generate intelligent responses using language models.""" # Analyze sentiment of user input sentiment = self.sentiment_analyzer(user_input)[0] # Generate contextual response response = self.text_generator( user_input, max_length=100, num_return_sequences=1, pad_token_id=50256 )[0]['generated_text'] # Clean up the response (remove user input from beginning) clean_response = response.replace(user_input, "").strip() # Respond with appropriate emotion based on sentiment self.respond_with_emotion(sentiment, clean_response) return clean_response def respond_with_emotion(self, sentiment, text): """Respond with emotional expression matching the sentiment.""" # Map sentiment to robot expressions emotion_mapping = { 'POSITIVE': 'happy', 'NEGATIVE': 'sad', 'NEUTRAL': 'neutral' } emotion = emotion_mapping.get(sentiment['label'], 'neutral') # Set antenna expression getattr(self.reachy.antennas, emotion)() # Adjust head position based on emotion if emotion == 'happy': self.reachy.head.look_at(x=0, y=10, z=50, duration=1.0) elif emotion == 'sad': self.reachy.head.look_at(x=0, y=-10, z=50, duration=1.0) else: self.reachy.head.look_at(x=0, y=0, z=50, duration=1.0) # Speak the response self.reachy.voice.say(text) # Usage example ai_integration = AIIntegration(reachy) # Analyze what the robot sees ai_integration.analyze_camera_view() # Have an intelligent conversation response = ai_integration.intelligent_conversation("How are you feeling today?") print(f"Robot response: {response}")

import logging import traceback from functools import wraps # Set up logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) def safe_robot_operation(func): """Decorator for safe robot operations with error handling.""" @wraps(func) def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except Exception as e: logger.error(f"Error in {func.__name__}: {e}") logger.error(traceback.format_exc()) # Safe robot state recovery try: # Return robot to neutral position args[0].head.look_at(x=0, y=0, z=50, duration=1.0) args[0].antennas.neutral() args[0].voice.say("I encountered an error, but I'm okay now.") except: logger.error("Failed to recover robot to safe state") return None return wrapper class RobustRobot: def __init__(self, host='reachy-mini.local'): self.reachy = None self.connect_with_retry(host) def connect_with_retry(self, host, max_retries=3): """Connect to robot with automatic retry.""" for attempt in range(max_retries): try: self.reachy = ReachySDK(host=host) logger.info(f"Connected to Reachy Mini at {host}") return except Exception as e: logger.warning(f"Connection attempt {attempt + 1} failed: {e}") if attempt < max_retries - 1: time.sleep(2) else: raise Exception(f"Failed to connect after {max_retries} attempts") @safe_robot_operation def monitored_movement(self, x, y, z, duration=1.0): """Perform movement with safety monitoring.""" # Validate movement parameters if not (-50 <= x <= 50 and -30 <= y <= 30 and 20 <= z <= 80): raise ValueError(f"Movement parameters out of safe range: x={x}, y={y}, z={z}") # Perform movement with monitoring logger.info(f"Moving to position: ({x}, {y}, {z})") self.reachy.head.look_at(x=x, y=y, z=z, duration=duration) # Verify movement completed successfully time.sleep(duration + 0.5) logger.info("Movement completed successfully") @safe_robot_operation def safe_voice_output(self, text): """Speak text with error handling.""" if not isinstance(text, str) or len(text) == 0: raise ValueError("Invalid text for voice output") # Limit text length to prevent extremely long speech if len(text) > 200: text = text[:197] + "..." logger.info(f"Speaking: {text}") self.reachy.voice.say(text) def health_check(self): """Perform comprehensive robot health check.""" health_status = { 'connection': False, 'head_movement': False, 'voice': False, 'camera': False, 'microphones': False } try: # Test connection if self.reachy: health_status['connection'] = True # Test head movement self.reachy.head.look_at(x=0, y=0, z=50, duration=0.5) health_status['head_movement'] = True # Test voice self.reachy.voice.say("System check") health_status['voice'] = True # Test camera frame = self.reachy.camera.capture_frame() if frame is not None: health_status['camera'] = True # Test microphones audio = self.reachy.microphones.record(duration=0.1) if audio is not None: health_status['microphones'] = True except Exception as e: logger.error(f"Health check failed: {e}") # Report health status healthy_systems = sum(health_status.values()) total_systems = len(health_status) logger.info(f"Health check: {healthy_systems}/{total_systems} systems operational") for system, status in health_status.items(): status_text = "✓" if status else "✗" logger.info(f" {system}: {status_text}") return health_status # Usage example robot = RobustRobot() # Perform health check robot.health_check() # Safe operations robot.monitored_movement(x=20, y=10, z=50) robot.safe_voice_output("Hello, I'm operating safely!")