Smart Home Integration: Ollama IoT Device Control Made Simple

# Download and install Ollama
curl -fsSL https://ollama.ai/install.sh | sh

# Pull the recommended model for IoT control
ollama pull llama2:13b

# Verify installation
ollama list

# requirements.txt
ollama==0.1.8
paho-mqtt==1.6.1
requests==2.31.0
python-socketio==5.8.0
aiohttp==3.8.5

pip install -r requirements.txt

{
  "devices": {
    "living_room_lights": {
      "type": "philips_hue",
      "ip": "192.168.1.100",
      "api_key": "your_hue_api_key",
      "capabilities": ["on", "off", "brightness", "color"]
    },
    "thermostat": {
      "type": "nest",
      "ip": "192.168.1.101", 
      "api_key": "your_nest_api_key",
      "capabilities": ["temperature", "mode", "schedule"]
    },
    "security_camera": {
      "type": "hikvision",
      "ip": "192.168.1.102",
      "username": "admin",
      "password": "your_password",
      "capabilities": ["record", "snapshot", "pan", "tilt"]
    }
  }
}

import ollama
import json
import re

class IoTCommandProcessor:
    def __init__(self, model_name="llama2:13b"):
        self.model = model_name
        self.device_registry = self.load_devices()
        
    def load_devices(self):
        """Load device registry from configuration file"""
        with open('device_registry.json', 'r') as f:
            return json.load(f)
    
    def process_command(self, user_input):
        """Convert natural language to structured device commands"""
        
        # Create context-aware prompt for Ollama
        prompt = f"""
        Parse this smart home command into JSON format:
        Command: "{user_input}"
        
        Available devices: {list(self.device_registry['devices'].keys())}
        
        Response format:
        {{
            "device": "device_name",
            "action": "action_type", 
            "parameters": {{"key": "value"}}
        }}
        
        Examples:
        "Turn on living room lights" → {{"device": "living_room_lights", "action": "on", "parameters": {{}}}}
        "Set temperature to 72 degrees" → {{"device": "thermostat", "action": "temperature", "parameters": {{"value": 72}}}}
        """
        
        # Send prompt to Ollama
        response = ollama.generate(
            model=self.model,
            prompt=prompt,
            options={'temperature': 0.1}  # Low temperature for consistent parsing
        )
        
        # Extract JSON from response
        json_match = re.search(r'\{.*\}', response['response'], re.DOTALL)
        if json_match:
            try:
                return json.loads(json_match.group())
            except json.JSONDecodeError:
                return {"error": "Failed to parse command"}
        
        return {"error": "No valid command found"}

def process_complex_command(self, user_input):
    """Handle commands affecting multiple devices"""
    
    scene_commands = {
        "movie time": [
            {"device": "living_room_lights", "action": "brightness", "parameters": {"value": 10}},
            {"device": "tv", "action": "on", "parameters": {}},
            {"device": "thermostat", "action": "temperature", "parameters": {"value": 68}}
        ],
        "goodnight": [
            {"device": "all_lights", "action": "off", "parameters": {}},
            {"device": "security_camera", "action": "record", "parameters": {"duration": "8h"}},
            {"device": "thermostat", "action": "temperature", "parameters": {"value": 65}}
        ]
    }
    
    # Check for predefined scenes
    for scene, commands in scene_commands.items():
        if scene.lower() in user_input.lower():
            return {"scene": scene, "commands": commands}
    
    # Use Ollama for complex parsing
    prompt = f"""
    Parse this multi-device command: "{user_input}"
    
    Return JSON array of device commands:
    [
        {{"device": "name", "action": "type", "parameters": {{}}}},
        {{"device": "name2", "action": "type2", "parameters": {{}}}}
    ]
    """
    
    response = ollama.generate(model=self.model, prompt=prompt)
    # Parse response logic here...

import requests
import paho.mqtt.client as mqtt
from typing import Dict, Any

class UniversalDeviceController:
    def __init__(self, device_registry):
        self.devices = device_registry['devices']
        self.mqtt_client = mqtt.Client()
        self.setup_mqtt()
    
    def setup_mqtt(self):
        """Initialize MQTT connection for Zigbee/Z-Wave devices"""
        self.mqtt_client.connect("localhost", 1883, 60)
        self.mqtt_client.loop_start()
    
    def execute_command(self, command: Dict[str, Any]):
        """Execute parsed command on target device"""
        device_name = command.get('device')
        action = command.get('action')
        parameters = command.get('parameters', {})
        
        if device_name not in self.devices:
            return {"error": f"Device {device_name} not found"}
        
        device_config = self.devices[device_name]
        device_type = device_config['type']
        
        # Route to appropriate handler
        if device_type == 'philips_hue':
            return self.control_hue_device(device_config, action, parameters)
        elif device_type == 'nest':
            return self.control_nest_device(device_config, action, parameters)
        elif device_type == 'mqtt':
            return self.control_mqtt_device(device_config, action, parameters)
        else:
            return {"error": f"Unsupported device type: {device_type}"}
    
    def control_hue_device(self, config, action, params):
        """Control Philips Hue lights via REST API"""
        base_url = f"http://{config['ip']}/api/{config['api_key']}/lights"
        
        # Map actions to Hue API commands
        hue_commands = {
            'on': {'on': True},
            'off': {'on': False},
            'brightness': {'bri': params.get('value', 128)},
            'color': {'hue': params.get('hue', 0), 'sat': params.get('saturation', 255)}
        }
        
        if action not in hue_commands:
            return {"error": f"Unsupported Hue action: {action}"}
        
        # Send command to all lights or specific light
        light_id = params.get('light_id', '1')
        url = f"{base_url}/{light_id}/state"
        
        try:
            response = requests.put(url, json=hue_commands[action], timeout=5)
            return {"success": True, "response": response.json()}
        except requests.RequestException as e:
            return {"error": f"Hue communication failed: {str(e)}"}
    
    def control_mqtt_device(self, config, action, params):
        """Control MQTT-enabled devices"""
        topic = f"{config['topic_prefix']}/{action}"
        payload = json.dumps(params)
        
        try:
            self.mqtt_client.publish(topic, payload)
            return {"success": True, "topic": topic, "payload": payload}
        except Exception as e:
            return {"error": f"MQTT publish failed: {str(e)}"}

def get_device_status(self, device_name):
    """Retrieve current device status"""
    if device_name not in self.devices:
        return {"error": "Device not found"}
    
    device_config = self.devices[device_name]
    
    try:
        if device_config['type'] == 'philips_hue':
            url = f"http://{device_config['ip']}/api/{device_config['api_key']}/lights"
            response = requests.get(url, timeout=3)
            return {"status": "online", "data": response.json()}
        
        # Add other device type status checks
        
    except requests.RequestException:
        return {"status": "offline", "error": "Device unreachable"}

def health_check_all_devices(self):
    """Monitor all device connectivity"""
    health_report = {}
    
    for device_name in self.devices.keys():
        status = self.get_device_status(device_name)
        health_report[device_name] = status
    
    return health_report

import speech_recognition as sr
import pyttsx3

class VoiceInterface:
    def __init__(self, command_processor):
        self.recognizer = sr.Recognizer()
        self.microphone = sr.Microphone()
        self.tts_engine = pyttsx3.init()
        self.processor = command_processor
        
        # Calibrate for ambient noise
        with self.microphone as source:
            self.recognizer.adjust_for_ambient_noise(source)
    
    def listen_for_commands(self):
        """Continuous voice command listening"""
        print("Voice interface ready. Say 'Hey Ollama' to activate...")
        
        while True:
            try:
                with self.microphone as source:
                    # Listen for wake phrase
                    audio = self.recognizer.listen(source, timeout=1, phrase_time_limit=3)
                
                command = self.recognizer.recognize_google(audio).lower()
                
                if "hey ollama" in command:
                    self.process_voice_command()
                    
            except sr.WaitTimeoutError:
                pass  # Continue listening
            except sr.UnknownValueError:
                pass  # Ignore unclear audio
    
    def process_voice_command(self):
        """Process voice command after wake phrase"""
        self.speak("Ready for command")
        
        try:
            with self.microphone as source:
                audio = self.recognizer.listen(source, timeout=5, phrase_time_limit=10)
            
            command = self.recognizer.recognize_google(audio)
            print(f"Voice command: {command}")
            
            # Process through Ollama
            result = self.processor.process_command(command)
            
            if 'error' in result:
                self.speak(f"Sorry, {result['error']}")
            else:
                self.speak(f"Executing {result['action']} on {result['device']}")
                
        except sr.UnknownValueError:
            self.speak("I didn't understand that command")
        except sr.RequestError as e:
            self.speak("Speech recognition service unavailable")
    
    def speak(self, text):
        """Convert text to speech"""
        self.tts_engine.say(text)
        self.tts_engine.runAndWait()

from flask import Flask, render_template, request, jsonify
import json

app = Flask(__name__)

@app.route('/')
def dashboard():
    """Main dashboard page"""
    return render_template('dashboard.html')

@app.route('/api/command', methods=['POST'])
def api_command():
    """API endpoint for text commands"""
    data = request.get_json()
    command = data.get('command', '')
    
    if not command:
        return jsonify({"error": "No command provided"}), 400
    
    # Process through Ollama
    processor = IoTCommandProcessor()
    result = processor.process_command(command)
    
    if 'error' not in result:
        # Execute on device
        controller = UniversalDeviceController(processor.device_registry)
        execution_result = controller.execute_command(result)
        result.update(execution_result)
    
    return jsonify(result)

@app.route('/api/devices', methods=['GET'])
def api_devices():
    """Get all device status"""
    controller = UniversalDeviceController({})
    health_report = controller.health_check_all_devices()
    return jsonify(health_report)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

import schedule
import time
from datetime import datetime, timedelta

class AutomationScheduler:
    def __init__(self, processor, controller):
        self.processor = processor
        self.controller = controller
        self.scheduled_tasks = {}
    
    def add_schedule_from_text(self, schedule_text):
        """Parse natural language scheduling commands"""
        
        prompt = f"""
        Parse this scheduling command into structured format:
        Command: "{schedule_text}"
        
        Return JSON:
        {{
            "trigger": "time|sunrise|sunset|event",
            "condition": "daily|weekly|once|when_condition",
            "time": "HH:MM or offset",
            "actions": [
                {{"device": "name", "action": "type", "parameters": {{}}}}
            ]
        }}
        
        Examples:
        "Turn on lights every day at 6 PM" → time trigger, daily condition
        "Start coffee maker 30 minutes before my first meeting" → event trigger
        """
        
        response = ollama.generate(
            model=self.processor.model,
            prompt=prompt,
            options={'temperature': 0.1}
        )
        
        # Parse and schedule the task
        try:
            schedule_data = json.loads(response['response'])
            return self.create_schedule(schedule_data)
        except json.JSONDecodeError:
            return {"error": "Failed to parse schedule command"}
    
    def create_schedule(self, schedule_data):
        """Create actual scheduled task"""
        trigger = schedule_data['trigger']
        condition = schedule_data['condition']
        time_spec = schedule_data['time']
        actions = schedule_data['actions']
        
        task_id = f"task_{len(self.scheduled_tasks)}"
        
        if trigger == 'time' and condition == 'daily':
            schedule.every().day.at(time_spec).do(
                self.execute_scheduled_actions, actions
            )
        elif trigger == 'time' and condition == 'weekly':
            # Parse day from time_spec or use default
            schedule.every().monday.at(time_spec).do(
                self.execute_scheduled_actions, actions
            )
        
        self.scheduled_tasks[task_id] = schedule_data
        return {"success": True, "task_id": task_id}
    
    def execute_scheduled_actions(self, actions):
        """Execute list of scheduled actions"""
        results = []
        
        for action in actions:
            result = self.controller.execute_command(action)
            results.append(result)
            time.sleep(0.5)  # Prevent device overload
        
        return results
    
    def run_scheduler(self):
        """Main scheduler loop"""
        while True:
            schedule.run_pending()
            time.sleep(60)  # Check every minute

class ContextualAwareness:
    def __init__(self, processor):
        self.processor = processor
        self.user_locations = {}
        self.environmental_data = {}
    
    def update_user_location(self, user_id, location):
        """Track user location for presence-based automation"""
        self.user_locations[user_id] = {
            'location': location,
            'timestamp': datetime.now(),
            'last_location': self.user_locations.get(user_id, {}).get('location')
        }
        
        # Trigger location-based automations
        self.handle_location_change(user_id, location)
    
    def handle_location_change(self, user_id, new_location):
        """Execute automations based on location changes"""
        
        # Arriving home
        if new_location == 'home':
            arrival_commands = [
                {"device": "entry_lights", "action": "on", "parameters": {}},
                {"device": "thermostat", "action": "temperature", "parameters": {"value": 72}},
                {"device": "security_system", "action": "disarm", "parameters": {}}
            ]
            
            for command in arrival_commands:
                self.processor.controller.execute_command(command)
        
        # Leaving home
        elif self.user_locations[user_id]['last_location'] == 'home':
            departure_commands = [
                {"device": "all_lights", "action": "off", "parameters": {}},
                {"device": "thermostat", "action": "temperature", "parameters": {"value": 65}},
                {"device": "security_system", "action": "arm", "parameters": {}}
            ]
            
            for command in departure_commands:
                self.processor.controller.execute_command(command)
    
    def get_environmental_context(self):
        """Gather environmental data for intelligent decisions"""
        import requests
        
        try:
            # Weather API call (replace with your API key)
            weather_url = "http://api.openweathermap.org/data/2.5/weather"
            params = {
                'q': 'your_city',
                'appid': 'your_api_key',
                'units': 'imperial'
            }
            
            response = requests.get(weather_url, params=params, timeout=5)
            weather_data = response.json()
            
            self.environmental_data = {
                'temperature': weather_data['main']['temp'],
                'humidity': weather_data['main']['humidity'],
                'weather': weather_data['weather'][0]['main'],
                'sunrise': weather_data['sys']['sunrise'],
                'sunset': weather_data['sys']['sunset']
            }
            
            return self.environmental_data
            
        except Exception as e:
            return {"error": f"Failed to get weather data: {str(e)}"}

import hashlib
import jwt
import secrets
from functools import wraps

class SecurityManager:
    def __init__(self):
        self.secret_key = secrets.token_urlsafe(32)
        self.authorized_users = {}
        self.device_tokens = {}
    
    def create_user_token(self, username, password):
        """Generate JWT token for authenticated users"""
        password_hash = hashlib.sha256(password.encode()).hexdigest()
        
        if username in self.authorized_users:
            stored_hash = self.authorized_users[username]['password_hash']
            if password_hash == stored_hash:
                token = jwt.encode({
                    'username': username,
                    'exp': datetime.utcnow() + timedelta(hours=24)
                }, self.secret_key, algorithm='HS256')
                
                return {"token": token, "expires": "24 hours"}
        
        return {"error": "Invalid credentials"}
    
    def verify_token(self, token):
        """Verify JWT token validity"""
        try:
            payload = jwt.decode(token, self.secret_key, algorithms=['HS256'])
            return {"valid": True, "username": payload['username']}
        except jwt.ExpiredSignatureError:
            return {"error": "Token expired"}
        except jwt.InvalidTokenError:
            return {"error": "Invalid token"}
    
    def require_auth(self, f):
        """Decorator for API endpoint authentication"""
        @wraps(f)
        def decorated_function(*args, **kwargs):
            token = request.headers.get('Authorization', '').replace('Bearer ', '')
            
            if not token:
                return jsonify({"error": "No token provided"}), 401
            
            verification = self.verify_token(token)
            if 'error' in verification:
                return jsonify(verification), 401
            
            return f(*args, **kwargs)
        return decorated_function

from cryptography.fernet import Fernet
import base64

class SecureDeviceCommunication:
    def __init__(self):
        self.encryption_key = Fernet.generate_key()
        self.cipher_suite = Fernet(self.encryption_key)
    
    def encrypt_command(self, command_data):
        """Encrypt command before sending to device"""
        command_json = json.dumps(command_data)
        encrypted_data = self.cipher_suite.encrypt(command_json.encode())
        return base64.b64encode(encrypted_data).decode()
    
    def decrypt_response(self, encrypted_response):
        """Decrypt device response"""
        try:
            encrypted_data = base64.b64decode(encrypted_response.encode())
            decrypted_data = self.cipher_suite.decrypt(encrypted_data)
            return json.loads(decrypted_data.decode())
        except Exception as e:
            return {"error": f"Decryption failed: {str(e)}"}
    
    def secure_device_handshake(self, device_ip, device_id):
        """Establish secure connection with IoT device"""
        handshake_data = {
            'device_id': device_id,
            'timestamp': datetime.utcnow().isoformat(),
            'challenge': secrets.token_urlsafe(16)
        }
        
        # Send encrypted handshake
        encrypted_handshake = self.encrypt_command(handshake_data)
        
        try:
            response = requests.post(
                f"http://{device_ip}/secure/handshake",
                json={"data": encrypted_handshake},
                timeout=5
            )
            
            if response.status_code == 200:
                return {"success": True, "session_established": True}
            else:
                return {"error": "Handshake failed"}
                
        except requests.RequestException as e:
            return {"error": f"Connection failed: {str(e)}"}

import redis
import pickle
from datetime import timedelta

class PerformanceOptimizer:
    def __init__(self):
        self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
        self.command_cache = {}
        self.device_status_cache = {}
    
    def cache_command_result(self, command_hash, result, ttl=300):
        """Cache processed command results"""
        cache_key = f"command:{command_hash}"
        serialized_result = pickle.dumps(result)
        self.redis_client.setex(cache_key, ttl, serialized_result)
    
    def get_cached_command(self, command_hash):
        """Retrieve cached command result"""
        cache_key = f"command:{command_hash}"
        cached_data = self.redis_client.get(cache_key)
        
        if cached_data:
            return pickle.loads(cached_data)
        return None
    
    def optimize_command_processing(self, user_input):
        """Use caching to speed up repeated commands"""
        
        # Create hash of command for caching
        command_hash = hashlib.md5(user_input.encode()).hexdigest()
        
        # Check cache first
        cached_result = self.get_cached_command(command_hash)
        if cached_result:
            return cached_result
        
        # Process command if not cached
        processor = IoTCommandProcessor()
        result = processor.process_command(user_input)
        
        # Cache the result
        self.cache_command_result(command_hash, result)
        
        return result
    
    def batch_device_commands(self, commands):
        """Execute multiple device commands efficiently"""
        grouped_commands = {}
        
        # Group commands by device
        for command in commands:
            device = command['device']
            if device not in grouped_commands:
                grouped_commands[device] = []
            grouped_commands[device].append(command)
        
        # Execute commands in parallel
        import concurrent.futures
        
        with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
            futures = []
            
            for device, device_commands in grouped_commands.items():
                future = executor.submit(self.execute_device_batch, device, device_commands)
                futures.append(future)
            
            # Collect results
            results = []
            for future in concurrent.futures.as_completed(futures):
                results.extend(future.result())
        
        return results
    
    def execute_device_batch(self, device, commands):
        """Execute batched commands for single device"""
        controller = UniversalDeviceController({})
        results = []
        
        for command in commands:
            result = controller.execute_command(command)
            results.append(result)
            time.sleep(0.1)  # Small delay between commands
        
        return results

import logging
import traceback
from datetime import datetime

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('ollama_iot.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger('OllamaIoT')

class DiagnosticSystem:
    def __init__(self):
        self.error_log = []
        self.performance_metrics = {}
    
    def log_command_processing(self, user_input, processing_time, result):
        """Log command processing metrics"""
        log_entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'input': user_input,
            'processing_time_ms': processing_time * 1000,
            'success': 'error' not in result,
            'result': result
        }
        
        logger.info(f"Command processed: {user_input} ({processing_time*1000:.2f}ms)")
        
        if 'error' in result:
            logger.error(f"Command failed: {result['error']}")
            self.error_log.append(log_entry)
    
    def diagnose_device_connectivity(self, device_name):
        """Comprehensive device connectivity diagnosis"""
        diagnostics = {
            'device': device_name,
            'timestamp': datetime.utcnow().isoformat(),
            'tests': {}
        }
        
        controller = UniversalDeviceController({})
        
        # Test 1: Ping test
        try:
            import subprocess
            device_config = controller.devices[device_name]
            device_ip = device_config['ip']
            
            ping_result = subprocess.run(['ping', '-c', '1', device_ip], 
                                       capture_output=True, text=True, timeout=5)
            diagnostics['tests']['ping'] = ping_result.returncode == 0
            
        except Exception as e:
            diagnostics['tests']['ping'] = False
            logger.error(f"Ping test failed: {str(e)}")
        
        # Test 2: API endpoint test
        try:
            status = controller.get_device_status(device_name)
            diagnostics['tests']['api_response'] = 'error' not in status
            
        except Exception as e:
            diagnostics['tests']['api_response'] = False
            logger.error(f"API test failed: {str(e)}")
        
        # Test 3: Command execution test
        try:
            test_command = {"device": device_name, "action": "status", "parameters": {}}
            result = controller.execute_command(test_command)
            diagnostics['tests']['command_execution'] = 'error' not in result
            
        except Exception as e:
            diagnostics['tests']['command_execution'] = False
            logger.error(f"Command test failed: {str(e)}")
        
        return diagnostics
    
    def generate_health_report(self):
        """Generate comprehensive system health report"""
        report = {
            'timestamp': datetime.utcnow().isoformat(),
            'ollama_status': self.check_ollama_health(),
            'device_status': {},
            'recent_errors': self.error_log[-10:],  # Last 10 errors
            'performance_summary': self.get_performance_summary()
        }
        
        # Check all devices
        controller = UniversalDeviceController({})
        for device_name in controller.devices.keys():
            report['device_status'][device_name] = self.diagnose_device_connectivity(device_name)
        
        return report
    
    def check_ollama_health(self):
        """Verify Ollama service status"""
        try:
            response = ollama.list()
            return {
                'status': 'healthy',
                'models': response.get('models', []),
                'model_count': len(response.get('models', []))
            }
        except Exception as e:
            return {
                'status': 'unhealthy',
                'error': str(e)
            }