米家API深度解析：Python智能家居编程控制的架构设计与实战应用

【免费下载链接】mijia-api 米家API，使用Python控制米家设备 项目地址: https://gitcode.com/gh_mirrors/mi/mijia-api

在智能家居日益普及的今天，传统的APP控制方式已难以满足开发者对灵活性和扩展性的需求。智能家居编程控制的需求催生了专业的技术解决方案，而米家API正是这样一个为Python开发者量身打造的设备控制SDK。本文将深入解析这一自动化编程接口的设计理念、技术架构和实战应用，帮助您掌握智能家居开发的核心技术。

1. 项目定位与核心价值

1.1 为什么需要编程控制米家设备？

传统的小米米家APP虽然功能完善，但在以下场景中显得力不从心：

• 批量设备管理：需要同时控制多个设备的复杂场景
• 自动化调度：基于时间、传感器数据或外部事件的智能联动
• 系统集成：将智能家居功能嵌入到其他应用或平台中
• 数据分析：收集设备使用数据进行分析和优化

米家API正是为解决这些痛点而生，提供了完整的Python自动化解决方案，让开发者能够以编程方式访问和控制所有米家设备。

1.2 核心功能矩阵

功能模块	技术特性	应用场景
扫码认证	OAuth2.0协议，Token自动刷新	安全接入，长期可用
设备抽象	面向对象封装，属性式访问	简化开发，提高效率
MCP集成	支持LLM直接控制	AI助手集成，自然语言交互
批量操作	并发控制，性能优化	大规模设备管理
数据统计	能耗分析，使用统计	能源管理，智能优化

2. 架构设计与技术特色

2.1 分层架构设计

米家API采用清晰的三层架构设计，确保了系统的可扩展性和维护性：

# 架构层次示例
┌─────────────────────────────┐
│    应用层 (Application)      │
│  • CLI命令行工具             │
│  • MCP Server                │
│  • Agent Skill               │
└─────────────┬───────────────┘
┌─────────────┴───────────────┐
│    业务层 (Business)         │
│  • mijiaDevice封装类         │
│  • 设备规格解析              │
│  • 批量操作管理              │
└─────────────┬───────────────┘
┌─────────────┴───────────────┐
│    数据层 (Data)             │
│  • API通信协议               │
│  • 加密解密模块              │
│  • 数据持久化存储            │
└─────────────────────────────┘

2.2 技术创新点

设备规格动态解析：项目通过在线获取设备规格信息，无需硬编码设备属性，实现了对新设备的自动支持。这种设计使得设备管理框架能够适应不断增长的米家设备生态。

MCP协议集成：作为首批支持Model Context Protocol的智能家居SDK，米家API让LLM能够直接控制设备，实现了自然语言编程的智能家居控制。

认证机制优化：采用扫码登录和Token自动刷新机制，既保证了安全性，又提供了良好的用户体验。

3. 核心功能深度解析

3.1 设备控制的高级封装

mijiaDevice类是整个项目的核心，它提供了直观的设备控制接口：

from mijiaAPI import mijiaAPI, mijiaDevice

# 初始化API连接
api = mijiaAPI()
api.login()

# 通过设备名称初始化设备对象
lamp = mijiaDevice(api, dev_name='智能台灯')

# 属性式操作 - 直观的设备控制
lamp.on = True                    # 开启设备
lamp.brightness = 75              # 设置亮度为75%
lamp.color_temperature = 4500     # 设置色温为4500K

# 获取设备状态
current_status = {
    'power': lamp.on,
    'brightness': lamp.brightness,
    'color_temp': lamp.color_temperature
}
print(f"设备状态: {current_status}")

# 查看设备支持的所有功能
print(lamp)  # 显示设备属性和动作列表

3.2 MCP Server：AI助手的智能家居接口

MCP（Model Context Protocol）集成是项目的亮点之一，它让LLM能够直接控制智能家居设备：

# MCP配置示例
{
  "mcpServers": {
    "mijia-api": {
      "command": "uvx",
      "args": ["mijiaAPI", "mcp", "-p", "/path/to/auth.json"]
    }
  }
}

# LLM可以直接调用的工具包括：
# • login - 发起二维码登录
# • list_devices - 列出所有设备
# • get_device_properties - 获取设备属性
# • set_device_property - 设置设备属性
# • run_device_action - 执行设备动作
# • run_scene - 运行场景

3.3 批量操作与性能优化

对于需要控制多个设备的场景，项目提供了高效的批量操作接口：

# 批量获取设备属性
devices = api.get_devices_list()
bulk_properties = []

for device in devices[:5]:  # 控制前5个设备
    if device['model'] == 'yeelink.light.lamp4':
        bulk_properties.extend([
            {"did": device['did'], "siid": 2, "piid": 2},  # 亮度
            {"did": device['did'], "siid": 2, "piid": 3},  # 色温
            {"did": device['did'], "siid": 2, "piid": 1}   # 开关状态
        ])

# 单次API调用获取所有属性
results = api.get_devices_prop(bulk_properties)
print(f"批量获取完成，共{len(results)}个属性值")

# 智能设备管理器类
class SmartHomeManager:
    def __init__(self, max_workers=3):
        self.api = mijiaAPI()
        self.api.login()
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
    
    def batch_control(self, device_names, operation, **kwargs):
        """批量控制多个设备"""
        futures = []
        for name in device_names:
            future = self.executor.submit(
                self._control_device, name, operation, **kwargs
            )
            futures.append(future)
        
        results = []
        for future in as_completed(futures):
            try:
                results.append(future.result())
            except Exception as e:
                logger.error(f"设备控制失败: {e}")
        
        return results

4. 实战应用场景

4.1 家庭自动化系统

基于米家API，我们可以构建完整的家庭自动化系统：

class HomeAutomationSystem:
    def __init__(self):
        self.api = mijiaAPI()
        self.api.login()
        self.devices = {}
        self.scenes = {}
        self._initialize_system()
    
    def _initialize_system(self):
        """初始化系统：发现设备并加载场景"""
        # 获取所有设备
        all_devices = self.api.get_devices_list()
        for device in all_devices:
            self.devices[device['did']] = {
                'name': device['name'],
                'model': device['model'],
                'type': self._detect_device_type(device['model'])
            }
        
        # 获取场景配置
        self.scenes = self.api.get_scenes_list()
    
    def execute_routine(self, routine_name, conditions):
        """执行自动化例程"""
        if routine_name == 'morning_wakeup':
            self._morning_wakeup_routine()
        elif routine_name == 'evening_relax':
            self._evening_relax_routine()
        elif routine_name == 'away_mode':
            self._away_mode_routine()
    
    def _morning_wakeup_routine(self):
        """早晨唤醒例程"""
        # 逐渐增加灯光亮度
        for brightness in range(10, 101, 10):
            self._set_all_lights_brightness(brightness)
            time.sleep(30)  # 每30秒增加10%亮度
        
        # 开启咖啡机
        self._control_device('厨房咖啡机', 'on', True)
        
        # 播放天气预报
        self._run_speaker_command("播放今天的天气预报")

4.2 能耗监控与分析

class EnergyMonitor:
    def __init__(self):
        self.api = mijiaAPI()
        self.api.login()
        self.energy_data = {}
    
    def collect_daily_consumption(self, days=30):
        """收集设备能耗数据"""
        end_time = int(time.time())
        start_time = end_time - days * 24 * 3600
        
        for device in self.api.get_devices_list():
            if self._is_power_device(device['model']):
                stats = self.api.get_statistics({
                    "did": device['did'],
                    "key": self._get_energy_key(device['model']),
                    "data_type": "stat_day_v3",
                    "limit": days,
                    "time_start": start_time,
                    "time_end": end_time,
                })
                
                if stats:
                    self.energy_data[device['did']] = {
                        'name': device['name'],
                        'daily_avg': self._calculate_daily_average(stats),
                        'peak_hours': self._analyze_peak_usage(stats),
                        'total_consumption': sum(float(item['value']) for item in stats)
                    }
    
    def generate_energy_report(self):
        """生成能耗分析报告"""
        report = "# 智能家居能耗分析报告\n\n"
        
        total_consumption = sum(
            data['total_consumption'] 
            for data in self.energy_data.values()
        )
        
        report += f"## 总能耗: {total_consumption:.2f} kWh\n\n"
        report += "## 设备能耗排名\n\n"
        
        sorted_devices = sorted(
            self.energy_data.items(),
            key=lambda x: x[1]['total_consumption'],
            reverse=True
        )
        
        for did, data in sorted_devices:
            percentage = (data['total_consumption'] / total_consumption * 100) if total_consumption > 0 else 0
            report += f"### {data['name']}\n"
            report += f"- 总耗电: {data['total_consumption']:.2f} kWh\n"
            report += f"- 日均耗电: {data['daily_avg']:.2f} kWh\n"
            report += f"- 占比: {percentage:.1f}%\n"
            report += f"- 高峰时段: {data['peak_hours']}\n\n"
        
        return report

4.3 安防监控系统

class SecuritySystem:
    def __init__(self):
        self.api = mijiaAPI()
        self.api.login()
        self.sensors = []
        self.alerts_enabled = True
        self._setup_security_sensors()
    
    def _setup_security_sensors(self):
        """设置安防传感器"""
        devices = self.api.get_devices_list()
        for device in devices:
            if self._is_security_sensor(device['model']):
                sensor = mijiaDevice(self.api, did=device['did'])
                self.sensors.append({
                    'device': sensor,
                    'name': device['name'],
                    'last_status': None,
                    'alert_threshold': self._get_threshold(device['model'])
                })
    
    def monitor_security(self, check_interval=60):
        """持续监控安防状态"""
        while True:
            for sensor_info in self.sensors:
                current_status = sensor_info['device'].get('status')
                
                if (sensor_info['last_status'] is not None and 
                    current_status != sensor_info['last_status']):
                    
                    self._handle_status_change(
                        sensor_info['name'],
                        sensor_info['last_status'],
                        current_status
                    )
                
                sensor_info['last_status'] = current_status
            
            time.sleep(check_interval)
    
    def _handle_status_change(self, sensor_name, old_status, new_status):
        """处理传感器状态变化"""
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        message = f"[{timestamp}] 传感器 {sensor_name} 状态变化: {old_status} -> {new_status}"
        
        logger.info(message)
        
        if self.alerts_enabled and self._is_security_breach(old_status, new_status):
            # 触发警报
            self._trigger_alarm()
            self._send_notification(f"安全警报: {message}")
            self._activate_emergency_lights()

5. 性能调优与扩展

5.1 连接池与缓存优化

import threading
from functools import lru_cache
from datetime import datetime, timedelta

class OptimizedMijiaAPI:
    def __init__(self, auth_data_path=None, max_connections=10):
        self._api_pool = []
        self._lock = threading.Lock()
        self.max_connections = max_connections
        self.auth_data_path = auth_data_path
        
        # 设备信息缓存
        self._device_cache = {}
        self._cache_expiry = {}
        self.cache_ttl = 300  # 5分钟缓存
        
        # 连接池初始化
        self._init_connection_pool()
    
    def _init_connection_pool(self):
        """初始化连接池"""
        for _ in range(self.max_connections):
            api_instance = mijiaAPI(self.auth_data_path)
            if api_instance.available:
                self._api_pool.append(api_instance)
    
    @lru_cache(maxsize=100)
    def get_cached_device_info(self, model):
        """缓存设备规格信息"""
        cache_key = f"device_info_{model}"
        
        if (cache_key in self._device_cache and 
            cache_key in self._cache_expiry and
            datetime.now() < self._cache_expiry[cache_key]):
            return self._device_cache[cache_key]
        
        # 从API获取设备信息
        device_info = get_device_info(model)
        
        # 更新缓存
        self._device_cache[cache_key] = device_info
        self._cache_expiry[cache_key] = datetime.now() + timedelta(seconds=self.cache_ttl)
        
        return device_info
    
    def get_api_instance(self):
        """从连接池获取API实例"""
        with self._lock:
            if not self._api_pool:
                # 池为空，创建新连接
                new_api = mijiaAPI(self.auth_data_path)
                return new_api
            
            # 使用最近最少使用的连接
            api_instance = self._api_pool.pop(0)
            self._api_pool.append(api_instance)  # 移到末尾
            return api_instance

5.2 异步操作支持

import asyncio
import aiohttp
from concurrent.futures import ThreadPoolExecutor

class AsyncMijiaDevice:
    def __init__(self, api, dev_name=None, did=None):
        self.api = api
        self.dev_name = dev_name
        self.did = did
        self._executor = ThreadPoolExecutor(max_workers=5)
        self._session = None
    
    async def __aenter__(self):
        self._session = aiohttp.ClientSession()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self._session:
            await self._session.close()
    
    async def async_get(self, property_name):
        """异步获取设备属性"""
        loop = asyncio.get_event_loop()
        
        # 在线程池中执行同步API调用
        result = await loop.run_in_executor(
            self._executor,
            lambda: self._get_property_sync(property_name)
        )
        return result
    
    async def async_set(self, property_name, value):
        """异步设置设备属性"""
        loop = asyncio.get_event_loop()
        
        result = await loop.run_in_executor(
            self._executor,
            lambda: self._set_property_sync(property_name, value)
        )
        return result
    
    async def async_batch_control(self, operations):
        """异步批量控制"""
        tasks = []
        
        for op in operations:
            if op['type'] == 'get':
                task = self.async_get(op['property'])
            elif op['type'] == 'set':
                task = self.async_set(op['property'], op['value'])
            else:
                continue
            
            tasks.append(task)
        
        # 并发执行所有操作
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

5.3 自定义设备扩展

from abc import ABC, abstractmethod
from typing import Dict, Any, List

class BaseDevice(ABC):
    """设备基类，定义统一接口"""
    
    def __init__(self, api, device_info: Dict[str, Any]):
        self.api = api
        self.device_info = device_info
        self.did = device_info['did']
        self.name = device_info['name']
        self.model = device_info['model']
    
    @abstractmethod
    def get_status(self) -> Dict[str, Any]:
        """获取设备状态"""
        pass
    
    @abstractmethod
    def set_property(self, prop_name: str, value: Any) -> bool:
        """设置设备属性"""
        pass
    
    @abstractmethod
    def execute_action(self, action_name: str, **kwargs) -> Any:
        """执行设备动作"""
        pass

class LightDevice(BaseDevice):
    """灯光设备专用类"""
    
    def __init__(self, api, device_info):
        super().__init__(api, device_info)
        self._device = mijiaDevice(api, did=device_info['did'])
    
    def get_status(self) -> Dict[str, Any]:
        """获取灯光状态"""
        try:
            return {
                'power': self._device.get('on'),
                'brightness': self._device.get('brightness'),
                'color_temp': self._device.get('color_temperature'),
                'mode': self._device.get('mode') if hasattr(self._device, 'mode') else None
            }
        except Exception as e:
            logger.error(f"获取灯光状态失败: {e}")
            return {}
    
    def set_property(self, prop_name: str, value: Any) -> bool:
        """设置灯光属性"""
        try:
            if prop_name == 'brightness' and isinstance(value, int):
                # 确保亮度在有效范围内
                value = max(1, min(100, value))
            
            self._device.set(prop_name, value)
            return True
        except Exception as e:
            logger.error(f"设置灯光属性失败: {e}")
            return False
    
    def execute_action(self, action_name: str, **kwargs) -> Any:
        """执行灯光动作"""
        if action_name == 'toggle':
            current_state = self._device.get('on')
            self._device.set('on', not current_state)
            return not current_state
        elif action_name == 'dim_to':
            brightness = kwargs.get('brightness', 50)
            self._device.set('brightness', brightness)
            return brightness
        else:
            raise ValueError(f"不支持的灯光动作: {action_name}")

class DeviceFactory:
    """设备工厂类，根据设备类型创建对应的设备对象"""
    
    @staticmethod
    def create_device(api, device_info: Dict[str, Any]) -> BaseDevice:
        """根据设备型号创建对应的设备对象"""
        model = device_info['model']
        
        # 灯光设备
        if model.startswith('yeelink.light.') or model.startswith('philips.light.'):
            return LightDevice(api, device_info)
        
        # 插座设备
        elif model.startswith('chuangmi.plug.') or model.startswith('lumi.plug.'):
            return PlugDevice(api, device_info)
        
        # 传感器设备
        elif model.startswith('lumi.sensor_') or model.startswith('xiaomi.sensor.'):
            return SensorDevice(api, device_info)
        
        # 默认设备
        else:
            return GenericDevice(api, device_info)

6. 生态系统与社区

6.1 社区贡献指南

米家API作为开源项目，欢迎社区成员的贡献：

问题报告与功能建议

• 在项目仓库提交详细的Issue描述
• 提供复现步骤和期望行为
• 附上相关日志和错误信息

代码贡献流程

1. Fork项目仓库到个人账户
2. 创建功能分支 (git checkout -b feature/amazing-feature)
3. 提交更改 (git commit -m 'Add amazing feature')
4. 推送到分支 (git push origin feature/amazing-feature)
5. 创建Pull Request

文档改进

• 完善API文档和示例代码
• 添加使用教程和最佳实践
• 翻译多语言文档

6.2 集成生态

米家API可以与多种系统和技术栈集成：

Home Assistant集成

# Home Assistant自定义组件示例
class MijiaLight(LightEntity):
    """Home Assistant中的米家灯光实体"""
    
    def __init__(self, device):
        self._device = device
        self._name = device.name
        self._state = None
        self._brightness = None
        self._color_temp = None
    
    @property
    def name(self):
        return self._name
    
    @property
    def is_on(self):
        return self._state
    
    @property
    def brightness(self):
        return self._brightness
    
    @property
    def color_temp(self):
        return self._color_temp
    
    async def async_turn_on(self, **kwargs):
        """打开灯光"""
        if 'brightness' in kwargs:
            await self._device.async_set('brightness', kwargs['brightness'])
        if 'color_temp' in kwargs:
            await self._device.async_set('color_temperature', kwargs['color_temp'])
        
        await self._device.async_set('on', True)
        await self.async_update_ha_state()
    
    async def async_turn_off(self, **kwargs):
        """关闭灯光"""
        await self._device.async_set('on', False)
        await self.async_update_ha_state()

Web API服务

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI(title="米家API服务")

class DeviceControlRequest(BaseModel):
    device_name: str
    action: str
    value: Any = None

@app.post("/api/devices/control")
async def control_device(request: DeviceControlRequest):
    """设备控制API端点"""
    try:
        device = mijiaDevice(api, dev_name=request.device_name)
        
        if request.action == "get":
            result = device.get(request.value) if request.value else None
        elif request.action == "set" and request.value is not None:
            device.set(request.action, request.value)
            result = {"status": "success"}
        else:
            raise HTTPException(status_code=400, detail="Invalid action")
        
        return {"success": True, "data": result}
    
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/api/devices")
async def list_devices():
    """获取设备列表"""
    devices = api.get_devices_list()
    return {"success": True, "data": devices}

6.3 最佳实践与性能建议

认证管理最佳实践

# 使用环境变量管理认证路径
import os
from pathlib import Path

class SecureAuthManager:
    def __init__(self):
        # 从环境变量读取认证路径
        auth_path = os.getenv('MIJIA_AUTH_PATH')
        
        if not auth_path:
            # 默认路径，但可以加密存储
            default_path = Path.home() / ".config" / "mijia-api" / "auth.json"
            auth_path = self._ensure_secure_storage(default_path)
        
        self.api = mijiaAPI(auth_path)
    
    def _ensure_secure_storage(self, path: Path) -> str:
        """确保认证文件安全存储"""
        # 创建目录并设置权限
        path.parent.mkdir(parents=True, exist_ok=True)
        path.parent.chmod(0o700)  # 仅所有者可访问
        
        # 如果文件不存在，提示用户登录
        if not path.exists():
            print("请使用米家APP扫描二维码登录...")
            self.api.login()
        
        return str(path)
    
    def rotate_credentials(self):
        """定期轮换认证凭证"""
        # 检查凭证有效期
        if not self.api.available:
            print("凭证已过期，需要重新登录...")
            self.api.login()

错误处理与重试机制

import time
from functools import wraps
from typing import Callable, TypeVar, Any

T = TypeVar('T')

def retry_with_backoff(
    max_retries: int = 3,
    initial_delay: float = 1.0,
    max_delay: float = 10.0,
    exponential_base: float = 2.0
):
    """带指数退避的重试装饰器"""
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @wraps(func)
        def wrapper(*args, **kwargs) -> T:
            delay = initial_delay
            last_exception = None
            
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except (APIError, DeviceActionError) as e:
                    last_exception = e
                    
                    if attempt == max_retries - 1:
                        raise last_exception
                    
                    # 指数退避
                    time.sleep(delay)
                    delay = min(delay * exponential_base, max_delay)
                    
                    logger.warning(
                        f"操作失败，第{attempt + 1}次重试，等待{delay:.1f}秒"
                    )
            
            raise last_exception
        
        return wrapper
    return decorator

class RobustDeviceController:
    """带有重试机制的设备控制器"""
    
    def __init__(self, api):
        self.api = api
    
    @retry_with_backoff(max_retries=3, initial_delay=1.0)
    def robust_get_device_property(self, device_name: str, property_name: str):
        """带重试的设备属性获取"""
        device = mijiaDevice(self.api, dev_name=device_name)
        return device.get(property_name)
    
    @retry_with_backoff(max_retries=3, initial_delay=1.0)
    def robust_set_device_property(self, device_name: str, property_name: str, value: Any):
        """带重试的设备属性设置"""
        device = mijiaDevice(self.api, dev_name=device_name)
        device.set(property_name, value)
        return True

6.4 未来发展方向

米家API项目在以下方向有广阔的发展空间：

设备类型扩展

• 支持更多新型米家设备
• 添加设备特定的高级功能封装
• 优化设备发现和识别机制

协议增强

• 支持更多通信协议（如MQTT、WebSocket）
• 实现本地网络控制，减少云端依赖
• 添加设备状态实时推送功能

生态系统建设

• 开发更多语言绑定（JavaScript、Go、Rust等）
• 创建可视化配置工具
• 构建插件市场，支持第三方扩展

性能优化

• 实现连接池和请求合并
• 添加数据缓存和预加载
• 优化大规模设备管理性能

通过本文的深度解析，我们可以看到米家API不仅仅是一个简单的设备控制库，而是一个完整的智能家居开发库解决方案。它提供了从基础设备控制到高级自动化、从命令行工具到AI集成的完整技术栈。无论是个人开发者构建智能家居项目，还是企业级应用集成，米家API都能提供强大而灵活的技术支持。

随着智能家居技术的不断发展，Python自动化在家庭智能化领域的重要性日益凸显。米家API作为连接Python开发者与小米智能家居生态的桥梁，将持续演进，为开发者提供更强大、更易用的设备控制SDK，推动智能家居技术的普及和创新。

【免费下载链接】mijia-api 米家API，使用Python控制米家设备 项目地址: https://gitcode.com/gh_mirrors/mi/mijia-api