#!/usr/bin/env python3 """ Lightweight Smart Home Controller - Docker Edition A simple Flask app to control TP-Link switches and WebOS TVs """ from flask import Flask, render_template, request, jsonify import asyncio import json import socket import struct import websockets import ssl import requests from contextlib import asynccontextmanager import logging import consul import json from threading import Lock import time import os import sys # Configure logging for container logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.StreamHandler(sys.stdout) ] ) app = Flask(__name__) app.secret_key = os.environ.get('SECRET_KEY', 'your-secret-key-change-this') logger = logging.getLogger(__name__) # Consul configuration CONSUL_HOST = 'consul.service.dc1.consul' CONSUL_PORT = 8500 CONSUL_BASE_KEY = 'MiniHass/' class ConsulConfigManager: """Manage configuration using Consul KV store""" def __init__(self, host=CONSUL_HOST, port=CONSUL_PORT, base_key=CONSUL_BASE_KEY): self.client = consul.Consul(host=host, port=port) self.base_key = base_key def get(self, key): """Get value from Consul""" _, data = self.client.kv.get(f"{self.base_key}{key}") return data['Value'].decode() if data else None def put(self, key, value): """Store value in Consul""" return self.client.kv.put(f"{self.base_key}{key}", value) def get_json(self, key): """Get JSON value from Consul""" value = self.get(key) return json.loads(value) if value else None def put_json(self, key, value): """Store JSON value in Consul""" return self.put(key, json.dumps(value)) # Initialize Consul config manager consul_config = ConsulConfigManager() # Load configuration from Consul config = consul_config.get_json('config') or { 'tplink_ip': os.environ.get('TPLINK_IP', '192.168.1.100'), 'tv_ip': os.environ.get('TV_IP', '192.168.1.101'), 'tv_mac': os.environ.get('TV_MAC', 'AA:BB:CC:DD:EE:FF') } # Config file path for persistence CONFIG_DIR = '/app/config' CONFIG_FILE = os.path.join(CONFIG_DIR, 'config.json') # Device states cache device_states = { 'tplink': False, 'tv': False, 'last_update': 0 } # Thread lock for state updates state_lock = Lock() class TPLinkDevice: """Control TP-Link Kasa devices using the local protocol""" @staticmethod def encrypt(string): """Encrypt command for TP-Link protocol""" key = 171 result = struct.pack('>I', len(string)) for char in string: a = key ^ ord(char) key = a result += bytes([a]) return result @staticmethod def decrypt(data): """Decrypt response from TP-Link protocol""" key = 171 result = "" for byte in data: a = key ^ byte key = byte result += chr(a) return result @staticmethod def send_command(ip, command, port=9999, timeout=5): """Send command to TP-Link device""" try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(timeout) sock.connect((ip, port)) sock.send(TPLinkDevice.encrypt(json.dumps(command))) # Receive response data = sock.recv(2048) sock.close() # Skip the first 4 bytes (length header) and decrypt response = TPLinkDevice.decrypt(data[4:]) return json.loads(response) except Exception as e: logger.error(f"TP-Link command error for {ip}: {e}") return None @staticmethod def get_info(ip): """Get device information""" command = {"system": {"get_sysinfo": {}}} return TPLinkDevice.send_command(ip, command) @staticmethod def turn_on(ip): """Turn on the device""" command = {"system": {"set_relay_state": {"state": 1}}} return TPLinkDevice.send_command(ip, command) @staticmethod def turn_off(ip): """Turn off the device""" command = {"system": {"set_relay_state": {"state": 0}}} return TPLinkDevice.send_command(ip, command) class WebOSTV: """Control WebOS TV using aiowebostv library""" def __init__(self, ip): self.ip = ip self.consul_key = f'tv_credentials/{ip.replace(".", "_")}' def load_client_key(self): """Load client key from Consul""" return consul_config.get(self.consul_key) def save_client_key(self, key): """Save client key to Consul""" consul_config.put(self.consul_key, key) async def _execute_command(self, command): """Execute TV command with automatic connection handling""" from aiowebostv import WebOsClient try: client_key = self.load_client_key() async with WebOsClient(self.ip, client_key) as client: # If we have a new client key after connection, save it if client.client_key and client.client_key != client_key: self.save_client_key(client.client_key) if command == "turn_off": await client.turn_off() return True elif command == "turn_on": await client.turn_on() return True elif command == "get_power": return client.power_state == "on" except Exception as e: logger.error(f"WebOS TV error: {e}") return None async def turn_screen_off(self): """Turn off TV screen""" return await self._execute_command("turn_off") async def turn_screen_on(self): """Turn on TV screen""" return await self._execute_command("turn_on") async def get_power_state(self): """Get TV power state""" return await self._execute_command("get_power") def update_device_state(device, state): """Thread-safe device state update""" with state_lock: device_states[device] = state device_states['last_update'] = time.time() @app.route('/') def index(): """Serve the main interface""" return render_template('index.html') @app.route('/health') def health_check(): """Health check endpoint for container monitoring""" # Check Consul connectivity consul_ok = True try: # Try to get a key to verify connectivity consul_config.get('healthcheck') except Exception as e: logger.error(f"Consul connection error: {e}") consul_ok = False return jsonify({ 'status': 'healthy', 'timestamp': time.time(), 'config': {k: v for k, v in config.items() if k != 'tv_mac'}, # Don't expose MAC 'services': { 'consul_connected': consul_ok } }) @app.route('/debug') def debug_info(): """Debug information endpoint""" import platform import sys debug_data = { 'flask': { 'debug_mode': app.debug, 'testing': app.testing, 'version': flask.__version__ if 'flask' in globals() else 'unknown' }, 'system': { 'python_version': sys.version, 'platform': platform.platform(), 'hostname': socket.gethostname() }, 'app': { 'working_directory': os.getcwd(), 'template_dir_exists': os.path.exists('templates'), 'index_template_exists': os.path.exists('templates/index.html'), 'config_dir_exists': os.path.exists(CONFIG_DIR), 'config': config }, 'files': { 'templates': [f for f in os.listdir('templates')] if os.path.exists('templates') else [], 'app_dir': [f for f in os.listdir('.') if not f.startswith('.')] }, 'environment': { 'FLASK_DEBUG': os.environ.get('FLASK_DEBUG', 'not set'), 'FLASK_ENV': os.environ.get('FLASK_ENV', 'not set'), 'PYTHONUNBUFFERED': os.environ.get('PYTHONUNBUFFERED', 'not set') } } if request.args.get('format') == 'json': return jsonify(debug_data) else: # Return as HTML for easy browser viewing html = "

Debug Information

" + json.dumps(debug_data, indent=2, default=str) + "
" return html @app.route('/api/config', methods=['GET', 'POST']) def handle_config(): """Handle configuration updates""" global config if request.method == 'POST': data = request.get_json() config.update(data) # Save to Consul consul_config.put_json('config', config) logger.info("Configuration saved to Consul") return jsonify({'status': 'success', 'message': 'Configuration updated and saved to Consul'}) return jsonify(config) @app.route('/api/status') def get_status(): """Get current device states""" return jsonify(device_states) @app.route('/api/tplink/') def control_tplink(action): """Control TP-Link switch""" ip = config.get('tplink_ip') if not ip: return jsonify({'error': 'TP-Link IP not configured'}), 400 try: if action == 'on': result = TPLinkDevice.turn_on(ip) if result and 'system' in result: update_device_state('tplink', True) return jsonify({'status': 'success', 'state': True}) elif action == 'off': result = TPLinkDevice.turn_off(ip) if result and 'system' in result: update_device_state('tplink', False) return jsonify({'status': 'success', 'state': False}) elif action == 'status': result = TPLinkDevice.get_info(ip) if result and 'system' in result: state = result['system']['get_sysinfo']['relay_state'] == 1 update_device_state('tplink', state) return jsonify({'status': 'success', 'state': state}) return jsonify({'error': 'Command failed'}), 500 except Exception as e: logger.error(f"TP-Link control error: {e}") return jsonify({'error': str(e)}), 500 @app.route('/api/tv/') def control_tv(action): """Control WebOS TV""" ip = config.get('tv_ip') if not ip: return jsonify({'error': 'TV IP not configured'}), 400 try: tv = WebOSTV(ip) async def run_command(): if action == 'screen_on': result = await tv.turn_screen_on() if result: update_device_state('tv', True) return {'status': 'success', 'state': True} else: return {'error': 'Failed to turn screen on'} elif action == 'screen_off': result = await tv.turn_screen_off() if result: update_device_state('tv', False) return {'status': 'success', 'state': False} else: return {'error': 'Failed to turn screen off'} elif action == 'status': result = await tv.get_power_state() if result is not None: update_device_state('tv', result) return {'status': 'success', 'state': result} else: return {'error': 'Failed to get power state'} return {'error': 'Invalid action'} # Run async function loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) result = loop.run_until_complete(run_command()) loop.close() if 'error' in result: return jsonify(result), 500 return jsonify(result) except Exception as e: logger.error(f"WebOS TV control error: {e}") return jsonify({'error': str(e)}), 500 # HTML template (same as before but stored in templates/index.html) HTML_TEMPLATE = ''' Smart Home Controller

šŸ  Smart Home

šŸ’” TP-Link Switch
šŸ“ŗ LG WebOS TV Screen
OFF

Device Configuration

šŸ³ Running in Docker Container
Configuration persists in mounted volume
''' def create_template_file(): """Create the HTML template file""" template_dir = 'templates' if not os.path.exists(template_dir): os.makedirs(template_dir) template_path = os.path.join(template_dir, 'index.html') with open(template_path, 'w') as f: f.write(HTML_TEMPLATE) if __name__ == '__main__': # Configuration already loaded from Consul during initialization # Create template file create_template_file() print("šŸ  Smart Home Controller - Docker Edition") print("=" * 50) print(f"šŸ“” TP-Link IP: {config['tplink_ip']}") print(f"šŸ“ŗ TV IP: {config['tv_ip']}") print("šŸŒ Web interface: http://localhost:5000") print("ā¤ļø Health check: http://localhost:5000/health") print("\nšŸ³ Container Features:") print(" ā€¢ Persistent configuration and TV credentials storage") print(" ā€¢ Health checks for monitoring") print(" ā€¢ Host network access for device discovery") print(" ā€¢ Automatic container restart") print("\nšŸ”§ First time setup:") print(" 1. On first TV control, accept the pairing prompt on your TV") print(" 2. TV credentials will be saved automatically for future use") print("\nšŸš€ Starting Flask server...") # Enable debug mode based on environment debug_mode = os.environ.get('FLASK_DEBUG', '0') == '1' print(f"šŸš€ Starting Flask server on 0.0.0.0:5000 (debug={debug_mode})") app.run(host='0.0.0.0', port=5000, debug=debug_mode, use_reloader=False)