当前位置：首页 > news >正文

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

news 2026/4/18 20:52:37

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

【免费下载链接】python-miioPython 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