Reachy Mini Home Automation Integration

import paho.mqtt.client as mqtt import requests import json import time from reachy_sdk import ReachySDK class SmartHomeIntegrator: def __init__(self, reachy): self.reachy = reachy self.mqtt_client = mqtt.Client() self.device_registry = {} self.automation_rules = [] self.current_environment = {} self.setup_connections() def setup_connections(self): """Initialize connections to smart home platforms""" # MQTT setup for Home Assistant, OpenHAB, etc. self.mqtt_client.on_connect = self.on_mqtt_connect self.mqtt_client.on_message = self.on_mqtt_message try: self.mqtt_client.connect("localhost", 1883, 60) self.mqtt_client.loop_start() print("MQTT connection established") except Exception as e: print(f"MQTT connection failed: {e}") # Register common smart home devices self.register_devices() def register_devices(self): """Register and discover smart home devices""" # Example device registry - adapt to your specific devices self.device_registry = { 'living_room_lights': { 'type': 'light', 'topic': 'homeassistant/light/living_room/set', 'status_topic': 'homeassistant/light/living_room/state', 'capabilities': ['brightness', 'color', 'on_off'] }, 'thermostat': { 'type': 'climate', 'topic': 'homeassistant/climate/main/set', 'status_topic': 'homeassistant/climate/main/state', 'capabilities': ['temperature', 'mode', 'fan_speed'] }, 'security_system': { 'type': 'alarm', 'topic': 'homeassistant/alarm/security/set', 'status_topic': 'homeassistant/alarm/security/state', 'capabilities': ['arm', 'disarm', 'status'] }, 'smart_door_lock': { 'type': 'lock', 'topic': 'homeassistant/lock/front_door/set', 'status_topic': 'homeassistant/lock/front_door/state', 'capabilities': ['lock', 'unlock', 'status'] } } # Subscribe to status updates for device_id, device_info in self.device_registry.items(): if 'status_topic' in device_info: self.mqtt_client.subscribe(device_info['status_topic']) def on_mqtt_connect(self, client, userdata, flags, rc): """Callback for successful MQTT connection""" print(f"Connected to MQTT broker with result code {rc}") def on_mqtt_message(self, client, userdata, msg): """Handle incoming MQTT messages""" try: topic = msg.topic payload = json.loads(msg.payload.decode()) # Update device status in registry for device_id, device_info in self.device_registry.items(): if device_info.get('status_topic') == topic: self.current_environment[device_id] = payload self.process_device_update(device_id, payload) break except Exception as e: print(f"Error processing MQTT message: {e}") def process_device_update(self, device_id, status): """Process device status updates and trigger robot responses""" print(f"Device update - {device_id}: {status}") # Example: React to motion sensor if device_id == 'motion_sensor' and status.get('motion') == 'detected': self.greet_person_detected() # Example: React to door opening elif device_id == 'smart_door_lock' and status.get('state') == 'unlocked': self.welcome_home_sequence() # Example: React to security alarm elif device_id == 'security_system' and status.get('state') == 'triggered': self.security_alert_response() def control_device(self, device_id, command, parameters=None): """Send control commands to smart home devices""" if device_id not in self.device_registry: print(f"Unknown device: {device_id}") return False device_info = self.device_registry[device_id] topic = device_info['topic'] # Construct command payload if device_info['type'] == 'light': payload = self.build_light_command(command, parameters) elif device_info['type'] == 'climate': payload = self.build_climate_command(command, parameters) elif device_info['type'] == 'lock': payload = self.build_lock_command(command) else: payload = {"command": command} # Send command via MQTT try: self.mqtt_client.publish(topic, json.dumps(payload)) print(f"Command sent to {device_id}: {command}") # Provide physical feedback self.robot_acknowledge_command(device_id, command) return True except Exception as e: print(f"Failed to control device {device_id}: {e}") return False

import re from datetime import datetime, timedelta class VoiceHomeController: def __init__(self, smart_home_integrator, reachy): self.smart_home = smart_home_integrator self.reachy = reachy self.command_patterns = self.setup_command_patterns() self.context = {"last_room": "living_room", "last_device": None} def setup_command_patterns(self): """Define patterns for voice command recognition""" return { 'lights': { 'on': [r'turn on.*lights?', r'lights? on', r'brighten'], 'off': [r'turn off.*lights?', r'lights? off', r'dim.*lights?'], 'brightness': [r'set.*brightness.*(\d+)', r'dim.*to.*(\d+)'], 'color': [r'change.*color.*to.*(\w+)', r'make.*lights?.*(\w+)'] }, 'temperature': { 'set': [r'set.*temperature.*to.*(\d+)', r'make it.*(\d+).*degrees'], 'increase': [r'increase.*temperature', r'warmer', r'heat up'], 'decrease': [r'decrease.*temperature', r'cooler', r'cool down'] }, 'security': { 'arm': [r'arm.*security', r'enable.*alarm', r'secure.*house'], 'disarm': [r'disarm.*security', r'disable.*alarm', r'unlock.*house'] }, 'entertainment': { 'play': [r'play.*music', r'start.*playlist', r'turn on.*tv'], 'stop': [r'stop.*music', r'pause', r'turn off.*tv'], 'volume': [r'volume.*(\d+)', r'louder', r'quieter'] } } def process_voice_command(self, user_speech): """Process natural language voice commands""" command_lower = user_speech.lower() # Extract room context if mentioned room = self.extract_room_context(command_lower) if room: self.context['last_room'] = room # Process different command categories for category, commands in self.command_patterns.items(): for action, patterns in commands.items(): for pattern in patterns: match = re.search(pattern, command_lower) if match: return self.execute_home_command( category, action, match, room or self.context['last_room'] ) return "I didn't understand that command. Could you try rephrasing?" def extract_room_context(self, command_text): """Extract room context from voice command""" rooms = ['living room', 'bedroom', 'kitchen', 'bathroom', 'office', 'dining room'] for room in rooms: if room in command_text: return room.replace(' ', '_') return None def execute_home_command(self, category, action, match, room): """Execute the interpreted home automation command""" try: if category == 'lights': return self.control_lights(action, match, room) elif category == 'temperature': return self.control_temperature(action, match) elif category == 'security': return self.control_security(action) elif category == 'entertainment': return self.control_entertainment(action, match, room) except Exception as e: print(f"Error executing command: {e}") return "Sorry, I encountered an error executing that command." def control_lights(self, action, match, room): """Control lighting systems""" device_id = f"{room}_lights" if action == 'on': self.smart_home.control_device(device_id, 'turn_on') self.reachy.head.look_up() # Look toward lights return f"Turning on {room} lights" elif action == 'off': self.smart_home.control_device(device_id, 'turn_off') self.reachy.head.look_down() # Look down when dimming return f"Turning off {room} lights" elif action == 'brightness': brightness = int(match.group(1)) self.smart_home.control_device( device_id, 'set_brightness', {'brightness': brightness} ) # Gesture to indicate brightness level head_angle = (brightness / 100) * 0.3 # Scale to robot range self.reachy.head.look_at(0, head_angle, 0, duration=1.0) return f"Setting {room} lights to {brightness}% brightness" elif action == 'color': color = match.group(1) self.smart_home.control_device( device_id, 'set_color', {'color': color} ) return f"Changing {room} lights to {color}" def control_temperature(self, action, match): """Control climate systems""" if action == 'set': temperature = int(match.group(1)) self.smart_home.control_device( 'thermostat', 'set_temperature', {'temperature': temperature} ) # Gesture to indicate temperature if temperature > 72: self.reachy.head.look_at(0.2, 0.1, 0, duration=1.0) # Warm gesture else: self.reachy.head.look_at(-0.2, -0.1, 0, duration=1.0) # Cool gesture return f"Setting temperature to {temperature} degrees" elif action == 'increase': self.smart_home.control_device('thermostat', 'increase_temperature') self.reachy.head.look_at(0.1, 0.2, 0, duration=1.0) return "Increasing temperature" elif action == 'decrease': self.smart_home.control_device('thermostat', 'decrease_temperature') self.reachy.head.look_at(-0.1, -0.2, 0, duration=1.0) return "Decreasing temperature"

from datetime import datetime, time import threading import schedule class AutomationRoutineManager: def __init__(self, smart_home_integrator, reachy): self.smart_home = smart_home_integrator self.reachy = reachy self.routines = {} self.active_routines = set() self.setup_default_routines() self.start_scheduler() def setup_default_routines(self): """Define common home automation routines""" self.routines = { 'morning_wake_up': { 'name': 'Morning Wake Up', 'trigger': 'schedule', 'schedule_time': '07:00', 'conditions': [], 'actions': [ {'device': 'bedroom_lights', 'action': 'fade_on', 'brightness': 30}, {'device': 'thermostat', 'action': 'set_temperature', 'temperature': 72}, {'device': 'coffee_maker', 'action': 'start_brewing'}, {'robot_action': 'morning_greeting'} ], 'enabled': True }, 'leaving_home': { 'name': 'Leaving Home', 'trigger': 'voice_command', 'conditions': [ {'type': 'security_check', 'requirement': 'all_doors_locked'} ], 'actions': [ {'device': 'all_lights', 'action': 'turn_off'}, {'device': 'thermostat', 'action': 'set_away_mode'}, {'device': 'security_system', 'action': 'arm_away'}, {'robot_action': 'farewell_gesture'} ], 'enabled': True }, 'arriving_home': { 'name': 'Arriving Home', 'trigger': 'door_unlock', 'conditions': [ {'type': 'time_range', 'start': '17:00', 'end': '23:00'} ], 'actions': [ {'device': 'entryway_lights', 'action': 'turn_on'}, {'device': 'living_room_lights', 'action': 'fade_on', 'brightness': 60}, {'device': 'thermostat', 'action': 'set_home_mode'}, {'device': 'security_system', 'action': 'disarm'}, {'robot_action': 'welcome_home'} ], 'enabled': True }, 'bedtime_routine': { 'name': 'Bedtime Routine', 'trigger': 'voice_command', 'conditions': [ {'type': 'time_after', 'time': '21:00'} ], 'actions': [ {'device': 'all_lights', 'action': 'dim_to', 'brightness': 10}, {'device': 'thermostat', 'action': 'set_sleep_temperature', 'temperature': 68}, {'device': 'entertainment_system', 'action': 'turn_off'}, {'robot_action': 'goodnight_routine'}, {'delay': 300}, # 5-minute delay {'device': 'all_lights', 'action': 'turn_off'} ], 'enabled': True }, 'security_alert': { 'name': 'Security Alert Response', 'trigger': 'security_breach', 'conditions': [], 'actions': [ {'device': 'all_lights', 'action': 'flash_red'}, {'device': 'security_cameras', 'action': 'start_recording'}, {'robot_action': 'security_patrol_mode'}, {'notification': 'send_alert', 'message': 'Security breach detected'} ], 'enabled': True, 'priority': 'high' } } def execute_routine(self, routine_name, context=None): """Execute a specific automation routine""" if routine_name not in self.routines: print(f"Unknown routine: {routine_name}") return False routine = self.routines[routine_name] if not routine.get('enabled', True): print(f"Routine {routine_name} is disabled") return False # Check conditions if not self.check_routine_conditions(routine.get('conditions', []), context): print(f"Conditions not met for routine: {routine_name}") return False print(f"Executing routine: {routine['name']}") self.active_routines.add(routine_name) # Execute actions success = self.execute_routine_actions(routine['actions']) self.active_routines.discard(routine_name) return success def check_routine_conditions(self, conditions, context): """Check if routine conditions are satisfied""" for condition in conditions: if not self.evaluate_condition(condition, context): return False return True def evaluate_condition(self, condition, context): """Evaluate a single condition""" condition_type = condition['type'] if condition_type == 'time_range': current_time = datetime.now().time() start_time = time.fromisoformat(condition['start']) end_time = time.fromisoformat(condition['end']) return start_time <= current_time <= end_time elif condition_type == 'time_after': current_time = datetime.now().time() check_time = time.fromisoformat(condition['time']) return current_time >= check_time elif condition_type == 'security_check': if condition['requirement'] == 'all_doors_locked': return self.check_all_doors_locked() elif condition_type == 'device_state': device_id = condition['device'] expected_state = condition['state'] current_state = self.smart_home.current_environment.get(device_id, {}) return current_state.get('state') == expected_state return True def execute_routine_actions(self, actions): """Execute a sequence of routine actions""" for action in actions: try: if 'device' in action: device_id = action['device'] action_type = action['action'] parameters = {k: v for k, v in action.items() if k not in ['device', 'action']} if device_id == 'all_lights': self.control_all_lights(action_type, parameters) else: self.smart_home.control_device(device_id, action_type, parameters) elif 'robot_action' in action: self.execute_robot_action(action['robot_action']) elif 'delay' in action: time.sleep(action['delay']) elif 'notification' in action: self.send_notification(action) except Exception as e: print(f"Error executing routine action: {e}") return False return True def execute_robot_action(self, robot_action): """Execute robot-specific actions during routines""" if robot_action == 'morning_greeting': self.reachy.head.look_at(0.3, 0.1, 0.2, duration=2.0) # Could add voice greeting here elif robot_action == 'farewell_gesture': # Wave goodbye self.reachy.head.look_at(-0.2, 0.0, 0.1, duration=1.0) time.sleep(0.5) self.reachy.head.look_at(0.2, 0.0, 0.1, duration=1.0) elif robot_action == 'welcome_home': # Enthusiastic greeting self.reachy.head.look_at(0.0, 0.2, 0.0, duration=1.0) time.sleep(0.5) self.reachy.head.look_at(0.0, -0.1, 0.0, duration=1.0) elif robot_action == 'goodnight_routine': # Peaceful goodnight gesture self.reachy.head.look_at(0.0, 0.0, 0.0, duration=3.0) elif robot_action == 'security_patrol_mode': # Alert posture and scanning self.reachy.head.look_at(0.3, 0.0, 0.2, duration=0.5) time.sleep(0.3) self.reachy.head.look_at(-0.3, 0.0, 0.2, duration=0.5) def start_scheduler(self): """Start the routine scheduler""" # Schedule time-based routines for routine_name, routine in self.routines.items(): if routine.get('trigger') == 'schedule': schedule_time = routine.get('schedule_time') if schedule_time: schedule.every().day.at(schedule_time).do( self.execute_routine, routine_name ) # Start scheduler thread def run_scheduler(): while True: schedule.run_pending() time.sleep(60) # Check every minute scheduler_thread = threading.Thread(target=run_scheduler, daemon=True) scheduler_thread.start() print("Automation scheduler started")

class EnvironmentalMonitor: def __init__(self, smart_home_integrator, reachy): self.smart_home = smart_home_integrator self.reachy = reachy self.sensor_data = {} self.thresholds = self.setup_thresholds() self.monitoring_active = True self.start_monitoring() def setup_thresholds(self): """Define environmental thresholds for automation""" return { 'temperature': {'min': 68, 'max': 78, 'optimal': 72}, 'humidity': {'min': 40, 'max': 60, 'optimal': 45}, 'air_quality': {'good': 50, 'moderate': 100, 'unhealthy': 150}, 'light_level': {'dim': 100, 'comfortable': 300, 'bright': 800}, 'noise_level': {'quiet': 40, 'normal': 60, 'loud': 80} } def process_sensor_data(self, sensor_type, value, location='general'): """Process incoming sensor data and trigger responses""" self.sensor_data[f"{sensor_type}_{location}"] = { 'value': value, 'timestamp': time.time(), 'location': location } # Check for threshold violations and respond self.check_environmental_thresholds(sensor_type, value, location) def check_environmental_thresholds(self, sensor_type, value, location): """Check sensor values against thresholds and respond""" thresholds = self.thresholds.get(sensor_type) if not thresholds: return if sensor_type == 'temperature': if value < thresholds['min']: self.respond_to_cold_temperature(value, location) elif value > thresholds['max']: self.respond_to_hot_temperature(value, location) elif sensor_type == 'humidity': if value < thresholds['min']: self.respond_to_low_humidity(value, location) elif value > thresholds['max']: self.respond_to_high_humidity(value, location) elif sensor_type == 'air_quality': if value > thresholds['unhealthy']: self.respond_to_poor_air_quality(value, location) elif sensor_type == 'light_level': if value < thresholds['dim']: self.respond_to_low_light(value, location) def respond_to_cold_temperature(self, temperature, location): """Respond to cold temperature detection""" print(f"Cold temperature detected: {temperature}°F in {location}") # Increase thermostat self.smart_home.control_device('thermostat', 'increase_temperature') # Robot gesture indicating cold self.reachy.head.look_at(-0.1, -0.2, 0, duration=1.0) # Could trigger additional actions like closing smart vents def respond_to_hot_temperature(self, temperature, location): """Respond to hot temperature detection""" print(f"Hot temperature detected: {temperature}°F in {location}") # Decrease thermostat and increase fan speed self.smart_home.control_device('thermostat', 'decrease_temperature') self.smart_home.control_device('ceiling_fans', 'turn_on') # Robot gesture indicating heat self.reachy.head.look_at(0.1, 0.2, 0, duration=1.0) def respond_to_poor_air_quality(self, aqi_value, location): """Respond to poor air quality""" print(f"Poor air quality detected: AQI {aqi_value} in {location}") # Turn on air purifiers self.smart_home.control_device('air_purifier', 'turn_on') self.smart_home.control_device('air_purifier', 'set_speed', {'speed': 'high'}) # Close smart windows if open self.smart_home.control_device('smart_windows', 'close') # Alert gesture self.reachy.head.look_at(0, 0.1, 0.2, duration=0.5) time.sleep(0.3) self.reachy.head.look_at(0, -0.1, 0.2, duration=0.5)