如何高效使用Python-miio：5个实战场景完整指南

【免费下载链接】python-miio Python library & console tool for controlling Xiaomi smart appliances 项目地址: https://gitcode.com/gh_mirrors/py/python-miio

Python-miio是一个强大的开源工具，让你能够通过Python代码或命令行控制小米智能设备。无论是扫地机器人、空气净化器还是智能灯泡，这个库都提供了完整的miIO和MIoT协议支持，将小米生态设备转化为可编程的智能终端。在本文中，我们将通过5个实际应用场景，深入探索Python-miio的核心功能和使用技巧。

场景一：家庭环境自动化控制

想象一下，你希望根据室内空气质量自动调节空气净化器的工作模式。Python-miio让你能够轻松实现这样的智能场景。

解决方案：空气质量监控与自动调节

首先，你需要获取设备的IP地址和Token。如果使用小米官方App，可以通过miio-extract-tokens工具导出所有设备的连接信息：

# 安装提取工具
pip install python-miio

# 提取设备Token
miio-extract-tokens

对于空气净化器控制，Python-miio提供了专门的模块。以小米空气净化器为例：

from miio import AirPurifier

# 初始化设备连接
purifier = AirPurifier("192.168.1.100", "your_device_token")

# 获取当前状态
status = purifier.status()
print(f"PM2.5浓度: {status.aqi} μg/m³")
print(f"滤芯剩余寿命: {status.filter_life_remaining}%")
print(f"当前模式: {status.mode}")

# 根据PM2.5浓度自动调节
if status.aqi > 75:
    purifier.set_mode('favorite')  # 切换到最爱模式
    purifier.set_favorite_level(10)  # 最大风量
elif status.aqi > 35:
    purifier.set_mode('auto')  # 自动模式
else:
    purifier.set_mode('silent')  # 静音模式

相关设备支持位于miio/integrations/zhimi/airpurifier/，这里包含了空气净化器的完整实现。

场景二：智能照明系统编程

智能照明系统可以根据时间、天气或用户活动自动调整亮度和色温，创造舒适的居住环境。

解决方案：动态照明控制

Yeelight智能灯是小米生态中广泛使用的照明设备。Python-miio提供了完整的控制接口：

from miio import Yeelight

# 连接Yeelight设备
light = Yeelight("192.168.1.101", "your_device_token")

# 日出模拟：逐渐增加亮度和色温
import time
from datetime import datetime

def sunrise_simulation():
    current_hour = datetime.now().hour
    
    if 6 <= current_hour < 8:  # 早晨6-8点
        for brightness in range(10, 80, 5):
            light.set_brightness(brightness)
            light.set_color_temp(6500 - (brightness * 50))  # 色温逐渐变暖
            time.sleep(60)  # 每分钟调整一次
    
    elif current_hour >= 18:  # 晚上6点后
        light.set_brightness(30)
        light.set_color_temp(2700)  # 暖黄色光
        light.set_power("on")

# 设置定时任务
import schedule

schedule.every().day.at("06:00").do(sunrise_simulation)
schedule.every().day.at("22:00").do(lambda: light.set_power("off"))

照明设备的具体实现在miio/integrations/yeelight/目录中，包含了各种Yeelight型号的支持。

场景三：扫地机器人智能调度

现代家庭中，扫地机器人需要根据家庭活动模式智能规划清扫时间，避免打扰用户。

解决方案：基于活动模式的清扫计划

Roborock扫地机器人支持丰富的控制选项：

from miio import Vacuum

# 初始化扫地机器人
vacuum = Vacuum("192.168.1.102", "your_device_token")

def smart_cleaning_schedule():
    """根据家庭活动模式智能调度清扫"""
    import datetime
    
    now = datetime.datetime.now()
    weekday = now.weekday()
    hour = now.hour
    
    # 工作日白天无人在家时清扫
    if 0 <= weekday <= 4:  # 周一到周五
        if 9 <= hour <= 17:  # 工作时间
            if not vacuum.status().is_on:
                print("开始日常清扫")
                vacuum.start()
    
    # 周末避开休息时间
    else:
        if 10 <= hour <= 12 or 14 <= hour <= 17:
            if not vacuum.status().is_on:
                print("开始周末清扫")
                vacuum.start()
                vacuum.set_fan_speed(60)  # 中等吸力

# 获取清扫统计数据
def get_cleaning_stats():
    stats = vacuum.clean_history()
    print(f"总清扫面积: {stats.total_area} m²")
    print(f"总清扫时间: {stats.total_duration} 分钟")
    print(f"清扫次数: {stats.count}")

扫地机器人的完整功能实现在miio/integrations/roborock/vacuum/目录中，包括地图管理、禁区设置等高级功能。

场景四：多设备协同工作

智能家居的真正价值在于设备之间的协同工作。Python-miio让你能够创建复杂的设备联动场景。

解决方案：设备联动与状态同步

from miio import Device, AirPurifier, Yeelight
import threading

class SmartHomeOrchestrator:
    def __init__(self):
        self.devices = {}
        self.initialize_devices()
    
    def initialize_devices(self):
        """初始化所有设备"""
        # 空气净化器
        self.devices['purifier'] = AirPurifier("192.168.1.100", "token1")
        
        # 智能灯
        self.devices['living_room_light'] = Yeelight("192.168.1.101", "token2")
        self.devices['bedroom_light'] = Yeelight("192.168.1.102", "token3")
        
        # 扫地机器人
        self.devices['vacuum'] = Vacuum("192.168.1.103", "token4")
    
    def good_night_scene(self):
        """晚安场景：关闭所有灯光，开启空气净化器静音模式"""
        print("启动晚安场景...")
        
        # 关闭所有灯光
        for name, device in self.devices.items():
            if 'light' in name:
                device.set_power("off")
        
        # 设置空气净化器为睡眠模式
        self.devices['purifier'].set_mode('silent')
        
        # 确保扫地机器人返回充电
        if self.devices['vacuum'].status().state_code == 6:  # 清扫中
            self.devices['vacuum'].home()
    
    def morning_wakeup_scene(self):
        """早晨唤醒场景：逐渐亮灯，关闭空气净化器"""
        print("启动早晨唤醒场景...")
        
        # 逐渐增加卧室灯光亮度
        bedroom_light = self.devices['bedroom_light']
        for brightness in range(10, 80, 5):
            bedroom_light.set_brightness(brightness)
            time.sleep(30)  # 每30秒增加亮度
        
        # 关闭空气净化器
        self.devices['purifier'].set_power("off")

场景五：故障诊断与设备监控

设备连接问题或异常状态是智能家居系统的常见挑战。Python-miio提供了完善的诊断工具。

解决方案：系统化故障排查

import logging
from miio import Device
from miio.exceptions import DeviceException

class DeviceMonitor:
    def __init__(self):
        self.logger = logging.getLogger(__name__)
        logging.basicConfig(level=logging.INFO)
    
    def check_device_health(self, ip, token, device_type):
        """检查设备健康状况"""
        try:
            # 尝试连接设备
            if device_type == 'airpurifier':
                from miio import AirPurifier
                device = AirPurifier(ip, token)
            elif device_type == 'vacuum':
                from miio import Vacuum
                device = Vacuum(ip, token)
            else:
                device = Device(ip, token)
            
            # 获取设备信息
            info = device.info()
            status = device.status() if hasattr(device, 'status') else None
            
            health_report = {
                'ip': ip,
                'type': device_type,
                'online': True,
                'model': info.model if info else 'Unknown',
                'firmware': info.firmware_version if info else 'Unknown',
                'status': str(status) if status else 'N/A'
            }
            
            return health_report
            
        except DeviceException as e:
            self.logger.error(f"设备 {ip} 连接失败: {str(e)}")
            return {
                'ip': ip,
                'type': device_type,
                'online': False,
                'error': str(e)
            }
    
    def diagnose_common_issues(self):
        """诊断常见问题"""
        common_issues = {
            'connection_timeout': '检查设备IP和网络连接',
            'token_error': '确认Token是否正确，区分大小写',
            'device_not_found': '设备可能离线或IP地址已变更',
            'command_not_supported': '设备型号可能不受支持'
        }
        
        # 使用原始命令测试设备响应
        device = Device("192.168.1.100", "test_token")
        try:
            response = device.send("miIO.info", [])
            print(f"设备响应: {response}")
        except Exception as e:
            error_type = type(e).__name__
            suggestion = common_issues.get(error_type.lower(), '请查看官方文档')
            print(f"问题类型: {error_type}")
            print(f"建议: {suggestion}")

进阶技巧：自定义设备扩展

当遇到Python-miio尚未支持的设备时，你可以通过扩展库来添加支持。

最佳实践：创建自定义设备类

from miio import Device, DeviceException
from miio.device import DeviceStatus

class CustomDevice(Device):
    def __init__(self, ip: str, token: str, start_id: int = 0):
        super().__init__(ip, token, start_id)
    
    def get_custom_property(self, property_name):
        """获取自定义属性"""
        try:
            response = self.send("get_prop", [property_name])
            return response[0] if response else None
        except DeviceException as e:
            print(f"获取属性失败: {e}")
            return None
    
    def set_custom_mode(self, mode_value):
        """设置自定义模式"""
        try:
            return self.send("set_mode", [mode_value])
        except DeviceException as e:
            print(f"设置模式失败: {e}")
            return False

class CustomDeviceStatus(DeviceStatus):
    """自定义设备状态容器"""
    def __init__(self, data):
        self.data = data
    
    @property
    def custom_property(self):
        return self.data.get("custom_prop", "unknown")
    
    @property
    def is_online(self):
        return self.data.get("online", False) == "true"
    
    def __str__(self):
        return f"CustomDeviceStatus(custom_prop={self.custom_property}, online={self.is_online})"

# 使用自定义设备
custom_device = CustomDevice("192.168.1.105", "your_token")
status_data = custom_device.send("get_status", [])
status = CustomDeviceStatus(status_data)
print(f"设备状态: {status}")

协议实现深度解析

Python-miio的核心在于其双协议支持。了解这些协议的工作原理能帮助你更好地使用这个库。

miIO协议基础

miIO协议是小米早期设备使用的通信协议，位于miio/protocol/目录。该协议使用JSON-RPC over UDP进行设备通信：

# miIO协议通信示例
from miio import miioprotocol

class MiIOProtocolHandler:
    def __init__(self, ip, token):
        self.protocol = miioprotocol.MiIOProtocol(ip, token)
    
    def send_command(self, method, params):
        """发送miIO命令"""
        return self.protocol.send(method, params)
    
    def discover_devices(self):
        """发现局域网内miIO设备"""
        import socket
        import json
        
        # 广播发现消息
        message = json.dumps({"cmd": "whois"}).encode()
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
        sock.sendto(message, ('255.255.255.255', 54321))
        
        # 接收响应
        sock.settimeout(5)
        devices = []
        try:
            while True:
                data, addr = sock.recvfrom(1024)
                devices.append({
                    'ip': addr[0],
                    'response': json.loads(data.decode())
                })
        except socket.timeout:
            pass
        
        return devices

MIoT协议优势

MIoT（小米物联网）协议是新一代的统一协议，提供了更标准化的设备控制接口。相关实现在miio/miot_device.py中：

from miio import MiotDevice

class MIoTDeviceController:
    def __init__(self, ip, token, mapping):
        """
        MIoT设备控制器
        :param mapping: 设备映射配置，定义属性和操作
        """
        self.device = MiotDevice(ip, token, mapping)
    
    def get_all_properties(self):
        """获取设备所有属性"""
        properties = []
        for prop in self.device.mapping.get("properties", []):
            try:
                value = self.device.get_property_by(prop["siid"], prop["piid"])
                properties.append({
                    "name": prop.get("description", "Unknown"),
                    "value": value,
                    "unit": prop.get("unit", "")
                })
            except Exception as e:
                print(f"获取属性失败 {prop}: {e}")
        
        return properties
    
    def execute_action(self, siid, aiid, params=None):
        """执行MIoT操作"""
        return self.device.send("action", {
            "siid": siid,
            "aiid": aiid,
            "in": params or []
        })

性能优化与最佳实践

连接池管理

对于需要频繁与设备通信的应用，连接池能显著提升性能：

import threading
from queue import Queue
from miio import Device

class DeviceConnectionPool:
    def __init__(self, ip, token, pool_size=5):
        self.ip = ip
        self.token = token
        self.pool_size = pool_size
        self._pool = Queue(maxsize=pool_size)
        self._lock = threading.Lock()
        self._initialize_pool()
    
    def _initialize_pool(self):
        """初始化连接池"""
        for _ in range(self.pool_size):
            device = Device(self.ip, self.token)
            self._pool.put(device)
    
    def get_connection(self):
        """从池中获取连接"""
        return self._pool.get()
    
    def return_connection(self, device):
        """归还连接到池中"""
        self._pool.put(device)
    
    def execute_with_connection(self, func):
        """使用连接执行函数"""
        device = self.get_connection()
        try:
            return func(device)
        finally:
            self.return_connection(device)

# 使用连接池
pool = DeviceConnectionPool("192.168.1.100", "token")

def get_device_info(device):
    return device.info()

# 并发获取设备信息
with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(
        pool.execute_with_connection, 
        get_device_info
    ) for _ in range(3)]
    
    results = [f.result() for f in futures]

错误处理与重试机制

import time
from functools import wraps
from miio.exceptions import DeviceException

def retry_on_failure(max_retries=3, delay=1):
    """失败重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except DeviceException as e:
                    last_exception = e
                    if attempt < max_retries - 1:
                        time.sleep(delay * (attempt + 1))
                        continue
                    else:
                        raise last_exception
            return wrapper
        return decorator

class RobustDeviceController:
    def __init__(self, device):
        self.device = device
    
    @retry_on_failure(max_retries=3, delay=2)
    def safe_send_command(self, method, params):
        """安全发送命令，自动重试"""
        return self.device.send(method, params)
    
    def batch_commands(self, commands):
        """批量执行命令，带错误恢复"""
        results = []
        for method, params in commands:
            try:
                result = self.safe_send_command(method, params)
                results.append((method, "success", result))
            except DeviceException as e:
                results.append((method, "failed", str(e)))
                # 记录错误但继续执行其他命令
                print(f"命令 {method} 执行失败: {e}")
        
        return results

总结

Python-miio为小米智能设备控制提供了强大而灵活的工具集。通过本文的5个实战场景，你已经掌握了从基础设备控制到高级自动化场景的实现方法。无论是简单的设备状态查询，还是复杂的多设备联动，Python-miio都能帮助你构建智能家居解决方案。

记住这些关键点：

1. 正确获取Token是成功连接设备的第一步
2. 理解设备类型有助于选择合适的控制模块
3. 错误处理机制能提升系统稳定性
4. 性能优化对于大规模部署至关重要

随着小米生态的不断发展，Python-miio社区也在持续更新和完善。如果你遇到了尚未支持的设备，不妨参考现有实现，为开源项目贡献代码，共同完善这个优秀的工具库。

现在，开始你的智能家居编程之旅吧！从简单的灯光控制到复杂的家庭自动化系统，Python-miio都能为你提供强大的支持。

【免费下载链接】python-miio Python library & console tool for controlling Xiaomi smart appliances 项目地址: https://gitcode.com/gh_mirrors/py/python-miio