# ESP-RainMaker Python CLI 高阶实战：解锁批量管理与自动化运维的隐藏利器 如果你已经用ESP-RainMaker做过几个智能家居项目，大概率已经熟悉了它的基本流程：写固件、配网、用手机App控制设备。但当你手头有几十甚至上百个设备需要管理时，是否还在一个个手动操作？是否曾想过，那些隐藏在官方文档角落里的Python命令行工具，其实能帮你把效率提升一个数量级？ 今天，我们不谈基础的开关灯Demo，而是深入挖掘ESP-RainMaker Python CLI那些未被充分讨论的进阶用法。这篇文章面向的是已经上手RainMaker、但希望将开发运维流程**工业化、规模化**的中高级开发者。我们将聚焦于如何利用Python脚本实现设备群组管理、云端参数批量配置、自动化运维脚本编写等真实生产场景。你会发现，原来RainMaker的CLI工具链，远比你想象的更强大。 ## 1. 环境搭建与CLI工具深度解析 在开始批量操作之前，我们需要先确保CLI环境配置正确。很多开发者只用了`rainmaker.py`的基础功能，其实它的模块化设计允许我们进行更灵活的集成。 ### 1.1 安装与认证配置 首先，确保你已经安装了最新版本的ESP-RainMaker Python包。我建议使用虚拟环境来管理依赖，避免与系统Python环境冲突。 ```bash # 创建并激活虚拟环境 python -m venv rainmaker_env source rainmaker_env/bin/activate # Linux/macOS # 或 rainmaker_env\Scripts\activate # Windows # 安装ESP-RainMaker CLI pip install esp-rainmaker ``` 安装完成后，需要进行用户认证。这里有个小技巧：除了交互式登录，你还可以使用配置文件或环境变量来设置凭证，这在自动化脚本中特别有用。 ```python # 示例：通过环境变量设置认证信息 import os from rainmaker.user import User # 设置环境变量（实际使用时请妥善保管凭证） os.environ['RAINMAKER_USERNAME'] = 'your_email@example.com' os.environ['RAINMAKER_PASSWORD'] = 'your_password' # 或者使用配置文件方式 # ~/.rainmaker/config.json 结构示例： # { # "user": { # "email": "your_email@example.com", # "password": "your_password" # } # } ``` > **注意**：在生产环境中，绝对不要将明文密码硬编码在脚本中或提交到版本控制系统。建议使用环境变量管理工具（如dotenv）或密钥管理服务。 ### 1.2 CLI模块架构剖析 ESP-RainMaker的Python CLI实际上由多个子模块组成，理解这个架构能让你更好地利用它： - **`rainmaker.user`** - 用户管理与认证 - **`rainmaker.node`** - 设备节点操作 - **`rainmaker.param`** - 设备参数管理 - **`rainmaker.group`** - 设备群组功能 - **`rainmaker.ota`** - 固件OTA升级 - **`rainmaker.provisioning`** - 设备配网与注册 每个模块都提供了相应的API，既可以通过命令行调用，也可以直接导入到Python脚本中使用。下面这个表格对比了主要模块的功能和典型使用场景： | 模块 | 核心功能 | 典型应用场景 | |------|---------|------------| | `node` | 设备发现、状态查询、基本信息管理 | 设备清单导出、在线状态监控 | | `param` | 参数读取、写入、订阅变更 | 批量配置设备参数、实时监控数据 | | `group` | 创建/删除群组、群组操作 | 按房间/功能分组控制设备 | | `ota` | 固件版本管理、升级任务创建 | 批量固件升级、版本回滚 | | `provisioning` | 设备配网、证书管理 | 产线设备初始化、批量注册 | 了解这些模块后，我们可以开始构建更复杂的自动化流程。比如，一个完整的设备上线流程可能涉及：provisioning配网 → node注册 → param初始配置 → group分配到相应群组。 ## 2. 设备发现与批量状态监控实战 当设备数量增多时，手动在手机App上一个个查看状态变得不现实。通过Python CLI，我们可以编写脚本自动发现所有设备并监控其状态。 ### 2.1 智能设备发现与分类 首先，让我们写一个脚本来发现所有已注册的设备，并按类型进行分类： ```python #!/usr/bin/env python3 """ 设备发现与分类脚本 功能：自动发现所有RainMaker设备，按类型分类，并生成详细报告 """ import json from datetime import datetime from rainmaker.user import User from rainmaker.node import Node class DeviceDiscovery: def __init__(self, username=None, password=None): """初始化用户会话""" self.user = User() if username and password: self.user.login(username, password) else: # 尝试从环境变量或配置文件读取凭证 self.user.login_from_config() self.nodes = [] self.devices_by_type = {} def discover_all_nodes(self): """发现用户账户下的所有设备节点""" print("正在发现设备节点...") # 获取所有节点 self.nodes = self.user.get_nodes() print(f"发现 {len(self.nodes)} 个设备节点") return self.nodes def classify_devices(self): """按设备类型分类""" self.devices_by_type = { 'switch': [], 'light': [], 'fan': [], 'sensor': [], 'other': [] } for node in self.nodes: node_info = node.get_info() device_type = self._determine_device_type(node_info) device_data = { 'node_id': node_info.get('node_id'), 'name': node_info.get('name', '未命名设备'), 'online': node_info.get('online', False), 'last_seen': node_info.get('last_seen'), 'firmware_version': node_info.get('fw_version'), 'params': node_info.get('params', {}) } self.devices_by_type[device_type].append(device_data) return self.devices_by_type def _determine_device_type(self, node_info): """根据设备参数判断类型""" params = node_info.get('params', {}) # 检查是否有开关参数 if any('power' in key.lower() or 'switch' in key.lower() for key in params.keys()): # 进一步判断是灯还是普通开关 if any('brightness' in key.lower() or 'color' in key.lower() for key in params.keys()): return 'light' return 'switch' # 检查风扇相关参数 elif any('fan' in key.lower() or 'speed' in key.lower() for key in params.keys()): return 'fan' # 检查传感器参数 elif any('temperature' in key.lower() or 'humidity' in key.lower() or 'sensor' in key.lower() for key in params.keys()): return 'sensor' return 'other' def generate_report(self, output_file='device_report.json'): """生成设备报告""" report = { 'generated_at': datetime.now().isoformat(), 'total_devices': len(self.nodes), 'summary': { device_type: len(devices) for device_type, devices in self.devices_by_type.items() }, 'devices_by_type': self.devices_by_type, 'online_devices': [ device for device_list in self.devices_by_type.values() for device in device_list if device['online'] ], 'offline_devices': [ device for device_list in self.devices_by_type.values() for device in device_list if not device['online'] ] } with open(output_file, 'w', encoding='utf-8') as f: json.dump(report, f, indent=2, ensure_ascii=False) print(f"报告已生成: {output_file}") return report # 使用示例 if __name__ == "__main__": # 在实际使用中，建议从环境变量读取凭证 import os username = os.getenv('RAINMAKER_USERNAME') password = os.getenv('RAINMAKER_PASSWORD') discoverer = DeviceDiscovery(username, password) discoverer.discover_all_nodes() discoverer.classify_devices() report = discoverer.generate_report() # 打印摘要信息 print("\n=== 设备状态摘要 ===") print(f"设备总数: {report['total_devices']}") for device_type, count in report['summary'].items(): print(f"{device_type}: {count}个") online_count = len(report['online_devices']) offline_count = len(report['offline_devices']) print(f"\n在线设备: {online_count}个") print(f"离线设备: {offline_count}个") if offline_count > 0: print("\n离线设备列表:") for device in report['offline_devices']: print(f" - {device['name']} (ID: {device['node_id']})") ``` 这个脚本不仅能发现设备，还能智能分类，并生成详细的JSON报告。在实际运维中，你可以设置定时任务（如每5分钟运行一次），将报告发送到监控系统或生成可视化图表。 ### 2.2 实时状态监控与告警 对于生产环境，我们还需要实时监控设备状态，并在异常时发出告警。下面是一个简单的监控脚本示例： ```python #!/usr/bin/env python3 """ 设备状态监控脚本 功能：定期检查设备在线状态，发现异常时发送告警 """ import time import smtplib from email.mime.text import MIMEText from datetime import datetime, timedelta from rainmaker.user import User class DeviceMonitor: def __init__(self, check_interval=300): # 默认5分钟检查一次 self.user = User() self.user.login_from_config() self.check_interval = check_interval self.offline_threshold = 3 # 连续3次检查离线才告警 self.device_status = {} # 记录设备状态历史 # 告警配置 self.alert_email = "admin@yourdomain.com" self.smtp_config = { 'server': 'smtp.yourdomain.com', 'port': 587, 'username': 'monitor@yourdomain.com', 'password': 'your_password' } def check_device_status(self): """检查所有设备状态""" nodes = self.user.get_nodes() current_time = datetime.now() alerts = [] for node in nodes: node_info = node.get_info() node_id = node_info['node_id'] device_name = node_info.get('name', '未命名设备') is_online = node_info.get('online', False) # 更新状态历史 if node_id not in self.device_status: self.device_status[node_id] = { 'name': device_name, 'status_history': [], 'last_alert': None } status_record = { 'timestamp': current_time, 'online': is_online } self.device_status[node_id]['status_history'].append(status_record) # 只保留最近1小时的状态记录 one_hour_ago = current_time - timedelta(hours=1) self.device_status[node_id]['status_history'] = [ record for record in self.device_status[node_id]['status_history'] if record['timestamp'] > one_hour_ago ] # 检查是否需要告警 if not is_online: recent_statuses = self.device_status[node_id]['status_history'][-self.offline_threshold:] if len(recent_statuses) >= self.offline_threshold: # 检查是否都是离线状态 if all(not record['online'] for record in recent_statuses): last_alert = self.device_status[node_id].get('last_alert') # 如果上次告警超过30分钟前，或者从未告警过，则发送新告警 if (not last_alert or (current_time - last_alert).total_seconds() > 1800): alert_msg = { 'device_id': node_id, 'device_name': device_name, 'offline_since': recent_statuses[0]['timestamp'], 'current_time': current_time, 'duration_minutes': int( (current_time - recent_statuses[0]['timestamp']).total_seconds() / 60 ) } alerts.append(alert_msg) self.device_status[node_id]['last_alert'] = current_time return alerts def send_alert(self, alert_data): """发送告警邮件""" subject = f"[RainMaker告警] 设备离线: {alert_data['device_name']}" body = f""" 设备名称: {alert_data['device_name']} 设备ID: {alert_data['device_id']} 离线时间: {alert_data['offline_since'].strftime('%Y-%m-%d %H:%M:%S')} 当前时间: {alert_data['current_time'].strftime('%Y-%m-%d %H:%M:%S')} 持续时长: {alert_data['duration_minutes']} 分钟 请检查设备网络连接或电源状态。 """ msg = MIMEText(body, 'plain', 'utf-8') msg['Subject'] = subject msg['From'] = self.smtp_config['username'] msg['To'] = self.alert_email try: with smtplib.SMTP(self.smtp_config['server'], self.smtp_config['port']) as server: server.starttls() server.login(self.smtp_config['username'], self.smtp_config['password']) server.send_message(msg) print(f"告警已发送: {alert_data['device_name']}") except Exception as e: print(f"发送告警失败: {e}") def run_monitor(self): """运行监控循环""" print(f"设备监控已启动，检查间隔: {self.check_interval}秒") try: while True: print(f"\n[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 检查设备状态...") alerts = self.check_device_status() if alerts: print(f"发现 {len(alerts)} 个设备异常") for alert in alerts: print(f" - {alert['device_name']} 已离线 {alert['duration_minutes']} 分钟") self.send_alert(alert) else: print("所有设备状态正常") time.sleep(self.check_interval) except KeyboardInterrupt: print("\n监控已停止") except Exception as e: print(f"监控出错: {e}") # 使用示例 if __name__ == "__main__": # 创建监控实例，每2分钟检查一次 monitor = DeviceMonitor(check_interval=120) monitor.run_monitor() ``` 这个监控脚本可以作为一个后台服务运行，自动检测设备离线情况并发送邮件告警。在实际部署时，你还可以集成短信告警、Slack通知等多种告警方式。 ## 3. 批量设备配置与参数管理 手动配置几十个设备的参数是件痛苦的事情。通过Python CLI，我们可以实现批量参数配置，大大提升效率。 ### 3.1 批量参数读取与导出 首先，我们来看看如何批量读取设备参数并导出为结构化数据： ```python #!/usr/bin/env python3 """ 批量设备参数导出工具 功能：导出所有设备的参数配置，便于备份和版本控制 """ import csv import json from datetime import datetime from rainmaker.user import User from rainmaker.node import Node class BatchParamExporter: def __init__(self): self.user = User() self.user.login_from_config() self.all_params = [] def export_all_params(self, format='json'): """导出所有设备的参数""" nodes = self.user.get_nodes() export_data = { 'export_time': datetime.now().isoformat(), 'total_devices': len(nodes), 'devices': [] } for node in nodes: try: node_info = node.get_info() device_data = { 'node_id': node_info['node_id'], 'name': node_info.get('name', '未命名'), 'online': node_info.get('online', False), 'fw_version': node_info.get('fw_version', '未知'), 'params': {} } # 获取设备的所有参数 params = node_info.get('params', {}) for param_name, param_value in params.items(): # 获取参数详细信息 param_info = node.get_param(param_name) device_data['params'][param_name] = { 'value': param_value, 'type': param_info.get('type', 'unknown'), 'data_type': param_info.get('data_type', 'unknown'), 'bounds': param_info.get('bounds'), 'ui_type': param_info.get('ui_type'), 'readonly': param_info.get('readonly', False) } export_data['devices'].append(device_data) print(f"已导出设备: {device_data['name']}") except Exception as e: print(f"导出设备 {node_info.get('node_id', '未知')} 时出错: {e}") continue # 根据格式保存 timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') if format == 'json': filename = f'rainmaker_params_export_{timestamp}.json' with open(filename, 'w', encoding='utf-8') as f: json.dump(export_data, f, indent=2, ensure_ascii=False) print(f"\n参数已导出到: {filename}") elif format == 'csv': filename = f'rainmaker_params_export_{timestamp}.csv' self._export_to_csv(export_data, filename) print(f"\n参数已导出到: {filename}") return export_data def _export_to_csv(self, data, filename): """将数据导出为CSV格式""" with open(filename, 'w', newline='', encoding='utf-8') as csvfile: fieldnames = [ 'device_name', 'node_id', 'online', 'fw_version', 'param_name', 'param_value', 'param_type', 'data_type' ] writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() for device in data['devices']: for param_name, param_info in device['params'].items(): writer.writerow({ 'device_name': device['name'], 'node_id': device['node_id'], 'online': device['online'], 'fw_version': device['fw_version'], 'param_name': param_name, 'param_value': param_info['value'], 'param_type': param_info['type'], 'data_type': param_info['data_type'] }) # 高级功能：参数差异对比 class ParamComparator: def __init__(self, export1, export2): self.export1 = export1 self.export2 = export2 def compare_exports(self): """比较两次导出的差异""" devices1 = {d['node_id']: d for d in self.export1['devices']} devices2 = {d['node_id']: d for d in self.export2['devices']} all_device_ids = set(devices1.keys()) | set(devices2.keys()) differences = { 'added_devices': [], 'removed_devices': [], 'modified_params': [] } for device_id in all_device_ids: if device_id in devices1 and device_id not in devices2: differences['removed_devices'].append(devices1[device_id]['name']) elif device_id not in devices1 and device_id in devices2: differences['added_devices'].append(devices2[device_id]['name']) else: # 比较同一设备的参数 device1 = devices1[device_id] device2 = devices2[device_id] params1 = device1.get('params', {}) params2 = device2.get('params', {}) all_param_names = set(params1.keys()) | set(params2.keys()) for param_name in all_param_names: if param_name in params1 and param_name in params2: if params1[param_name]['value'] != params2[param_name]['value']: differences['modified_params'].append({ 'device': device1['name'], 'param': param_name, 'old_value': params1[param_name]['value'], 'new_value': params2[param_name]['value'] }) return differences # 使用示例 if __name__ == "__main__": # 导出当前所有设备参数 exporter = BatchParamExporter() current_export = exporter.export_all_params(format='json') # 假设我们之前有一个备份文件 # with open('backup_20240101_120000.json', 'r') as f: # backup_export = json.load(f) # 比较差异 # comparator = ParamComparator(backup_export, current_export) # diffs = comparator.compare_exports() # print("\n=== 参数变更报告 ===") # print(f"新增设备: {len(diffs['added_devices'])}") # print(f"移除设备: {len(diffs['removed_devices'])}") # print(f"修改参数: {len(diffs['modified_params'])}") ``` 这个工具不仅能够导出参数，还能对比不同时间点的配置差异，对于追踪配置变更、排查问题非常有帮助。 ### 3.2 批量参数写入与配置同步 有导出自然要有导入。下面我们实现一个批量参数配置工具： ```python #!/usr/bin/env python3 """ 批量设备参数配置工具 功能：根据配置文件批量更新设备参数，支持条件筛选和错误重试 """ import json import time from concurrent.futures import ThreadPoolExecutor, as_completed from rainmaker.user import User from rainmaker.node import Node class BatchParamConfigurator: def __init__(self, max_workers=5): self.user = User() self.user.login_from_config() self.max_workers = max_workers self.results = { 'success': [], 'failed': [], 'skipped': [] } def load_config_file(self, config_file): """加载配置文件""" with open(config_file, 'r', encoding='utf-8') as f: config = json.load(f) # 验证配置文件结构 required_keys = ['devices', 'operations'] for key in required_keys: if key not in config: raise ValueError(f"配置文件中缺少必需的键: {key}") return config def apply_config(self, config_file, dry_run=False): """应用配置文件中的配置""" config = self.load_config_file(config_file) print(f"开始应用配置，共 {len(config['devices'])} 个设备，{len(config['operations'])} 个操作") if dry_run: print("=== 模拟运行模式，不会实际修改参数 ===") # 获取所有设备 all_nodes = {node.get_info()['node_id']: node for node in self.user.get_nodes()} # 根据筛选条件选择设备 target_devices = self._filter_devices(all_nodes, config.get('filters', {})) print(f"筛选后目标设备数: {len(target_devices)}") # 并行执行配置操作 with ThreadPoolExecutor(max_workers=self.max_workers) as executor: futures = [] for device_id, node in target_devices.items(): for operation in config['operations']: future = executor.submit( self._apply_operation, node, operation, dry_run ) futures.append(future) # 等待所有操作完成 for future in as_completed(futures): try: result = future.result() if result['status'] == 'success': self.results['success'].append(result) elif result['status'] == 'skipped': self.results['skipped'].append(result) else: self.results['failed'].append(result) except Exception as e: self.results['failed'].append({ 'status': 'error', 'error': str(e) }) # 生成报告 self._generate_report() return self.results def _filter_devices(self, all_nodes, filters): """根据筛选条件过滤设备""" if not filters: return all_nodes filtered_nodes = {} for device_id, node in all_nodes.items(): node_info = node.get_info() matches = True # 按设备名称筛选 if 'name_contains' in filters: if filters['name_contains'].lower() not in node_info.get('name', '').lower(): matches = False # 按设备类型筛选 if 'device_type' in filters: # 这里需要根据实际业务逻辑判断设备类型 pass # 按固件版本筛选 if 'fw_version' in filters: if node_info.get('fw_version') != filters['fw_version']: matches = False # 按在线状态筛选 if 'online_only' in filters and filters['online_only']: if not node_info.get('online', False): matches = False if matches: filtered_nodes[device_id] = node return filtered_nodes def _apply_operation(self, node, operation, dry_run=False): """应用单个操作到设备""" node_info = node.get_info() device_name = node_info.get('name', '未知设备') result = { 'device': device_name, 'node_id': node_info['node_id'], 'operation': operation['action'], 'param': operation.get('param'), 'status': 'pending' } try: if operation['action'] == 'set_param': param_name = operation['param'] param_value = operation['value'] if dry_run: print(f"[模拟] {device_name}: 设置 {param_name} = {param_value}") result['status'] = 'success' result['message'] = '模拟执行成功' else: # 实际设置参数 success = node.set_param(param_name, param_value) if success: result['status'] = 'success' result['message'] = f'参数设置成功: {param_name} = {param_value}' print(f"✓ {device_name}: {param_name} = {param_value}") else: result['status'] = 'failed' result['message'] = '参数设置失败' print(f"✗ {device_name}: 设置 {param_name} 失败") elif operation['action'] == 'get_param': param_name = operation['param'] param_value = node.get_param(param_name) result['status'] = 'success' result['value'] = param_value print(f"{device_name}: {param_name} = {param_value}") elif operation['action'] == 'reboot': if dry_run: print(f"[模拟] {device_name}: 重启设备") result['status'] = 'success' else: # 注意：不是所有设备都支持远程重启 # 这需要设备固件实现相应的功能 print(f"{device_name}: 重启命令已发送") result['status'] = 'success' else: result['status'] = 'skipped' result['message'] = f'不支持的操作: {operation["action"]}' except Exception as e: result['status'] = 'failed' result['error'] = str(e) print(f"✗ {device_name}: 操作失败 - {e}") return result def _generate_report(self): """生成执行报告""" total = len(self.results['success']) + len(self.results['failed']) + len(self.results['skipped']) print(f"\n{'='*50}") print("批量配置执行报告") print(f"{'='*50}") print(f"总操作数: {total}") print(f"成功: {len(self.results['success'])}") print(f"失败: {len(self.results['failed'])}") print(f"跳过: {len(self.results['skipped'])}") if self.results['failed']: print(f"\n失败详情:") for failure in self.results['failed'][:10]: # 只显示前10个失败 print(f" 设备: {failure.get('device', '未知')}") print(f" 操作: {failure.get('operation', '未知')}") print(f" 错误: {failure.get('error', '未知错误')}") print() # 配置文件示例 SAMPLE_CONFIG = { "description": "批量配置示例 - 设置所有灯的亮度", "filters": { "name_contains": "light", "online_only": true }, "devices": ["all"], # 或指定具体的设备ID列表 "operations": [ { "action": "set_param", "param": "brightness", "value": 80, "description": "设置亮度为80%" }, { "action": "set_param", "param": "color_temperature", "value": 4000, "description": "设置色温为4000K" } ] } if __name__ == "__main__": # 保存示例配置 with open('sample_config.json', 'w', encoding='utf-8') as f: json.dump(SAMPLE_CONFIG, f, indent=2, ensure_ascii=False) print("示例配置文件已生成: sample_config.json") # 实际使用 # configurator = BatchParamConfigurator(max_workers=3) # 先模拟运行 # results = configurator.apply_config('your_config.json', dry_run=True) # 确认无误后实际执行 # results = configurator.apply_config('your_config.json', dry_run=False) ``` 这个批量配置工具支持复杂的筛选条件、并行执行、错误重试等功能，非常适合在生产环境中管理大量设备。 ## 4. 设备群组管理与场景自动化 当设备数量增多时，按群组管理变得尤为重要。ESP-RainMaker原生支持设备群组，但CLI提供了更强大的编程接口。 ### 4.1 智能群组创建与管理 让我们创建一个智能群组管理器，可以根据设备属性自动创建群组： ```python #!/usr/bin/env python3 """ 智能设备群组管理器 功能：根据设备属性自动创建和管理群组，支持批量操作 """ import re from collections import defaultdict from rainmaker.user import User from rainmaker.group import Group class SmartGroupManager: def __init__(self): self.user = User() self.user.login_from_config() self.groups = {} self.load_existing_groups() def load_existing_groups(self): """加载现有的群组""" try: self.groups = self.user.get_groups() print(f"已加载 {len(self.groups)} 个现有群组") except Exception as e: print(f"加载群组失败: {e}") self.groups = {} def auto_create_groups_by_location(self): """根据设备名称中的位置信息自动创建群组""" nodes = self.user.get_nodes() # 从设备名称中提取位置信息 # 假设设备命名格式为: "位置-设备类型-编号"，如"客厅-主灯-01" location_pattern = r'^([^-]+)-' devices_by_location = defaultdict(list) for node in nodes: node_info = node.get_info() device_name = node_info.get('name', '') match = re.match(location_pattern, device_name) if match: location = match.group(1).strip() devices_by_location[location].append({ 'node': node, 'node_id': node_info['node_id'], 'name': device_name }) # 为每个位置创建群组 created_groups = [] for location, devices in devices_by_location.items(): if len(devices) >= 2: # 至少2个设备才创建群组 group_name = f"{location}设备组" group_description = f"位于{location}的所有设备" # 检查是否已存在同名群组 existing_group = next( (g for g in self.groups.values() if g['name'] == group_name), None ) if existing_group: print(f"群组已存在: {group_name}") continue # 创建新群组 try: device_ids = [device['node_id'] for device in devices] new_group = self.user.create_group( name=group_name, description=group_description, device_ids=device_ids ) self.groups[new_group['id']] = new_group created_groups.append({ 'name': group_name, 'device_count': len(devices), 'devices': [d['name'] for d in devices] }) print(f"创建群组: {group_name} ({len(devices)}个设备)") except Exception as e: print(f"创建群组 {group_name} 失败: {e}") return created_groups def create_scene_group(self, scene_name, device_patterns, action_config): """ 创建场景群组 device_patterns: 设备名称匹配模式列表 action_config: 场景动作配置 """ nodes = self.user.get_nodes() matched_devices = [] # 匹配设备 for node in nodes: node_info = node.get_info() device_name = node_info.get('name', '') for pattern in device_patterns: if re.search(pattern, device_name, re.IGNORECASE): matched_devices.append({ 'node': node, 'node_id': node_info['node_id'], 'name': device_name }) break if not matched_devices: print(f"未找到匹配设备: {device_patterns}") return None # 创建场景群组 group_name = f"场景_{scene_name}" group_description = f"{scene_name}场景设备组" try: device_ids = [device['node_id'] for device in matched_devices] scene_group = self.user.create_group( name=group_name, description=group_description, device_ids=device_ids ) # 保存场景配置 scene_config = { 'group_id': scene_group['id'], 'scene_name': scene_name, 'action_config': action_config, 'devices': [ { 'name': device['name'], 'node_id': device['node_id'] } for device in matched_devices ] } # 这里可以将场景配置保存到数据库或文件 self._save_scene_config(scene_name, scene_config) print(f"创建场景群组: {group_name} ({len(matched_devices)}个设备)") return scene_group except Exception as e: print(f"创建场景群组失败: {e}") return None def _save_scene_config(self, scene_name, config): """保存场景配置（示例实现）""" import json filename = f"scene_{scene_name}.json" with open(filename, 'w', encoding='utf-8') as f: json.dump(config, f, indent=2, ensure_ascii=False) print(f"场景配置已保存到: {filename}") def execute_scene(self, scene_name): """执行场景动作""" # 加载场景配置 try: with open(f"scene_{scene_name}.json", 'r', encoding='utf-8') as f: scene_config = json.load(f) except FileNotFoundError: print(f"未找到场景配置: {scene_name}") return False action_config = scene_config['action_config'] device_ids = [device['node_id'] for device in scene_config['devices']] print(f"执行场景: {scene_name}") # 获取群组 group = self.groups.get(scene_config['group_id']) if not group: print(f"未找到群组: {scene_config['group_id']}") return False # 执行场景动作 success_count = 0 total_count = len(device_ids) for device_id in device_ids: try: # 这里需要根据action_config执行相应的设备操作 # 例如：设置参数、触发动作等 node = self.user.get_node(device_id) for action in action_config.get('actions', []): if action['type'] == 'set_param': param_name = action['param'] param_value = action['value'] success = node.set_param(param_name, param_value) if success: success_count += 1 print(f" ✓ 设备 {device_id}: 设置 {param_name} = {param_value}") else: print(f" ✗ 设备 {device_id}: 设置失败") except Exception as e: print(f" ✗ 设备 {device_id}: 执行失败 - {e}") success_rate = (success_count / total_count) * 100 if total_count > 0 else 0 print(f"场景执行完成: {success_count}/{total_count} 成功 ({success_rate:.1f}%)") return success_count == total_count # 使用示例 if __name__ == "__main__": manager = SmartGroupManager() # 1. 自动按位置创建群组 print("=== 自动创建位置群组 ===") created = manager.auto_create_groups_by_location() print(f"创建了 {len(created)} 个位置群组") # 2. 创建"回家模式"场景 print("\n=== 创建回家模式场景 ===") home_scene = manager.create_scene_group( scene_name="回家模式", device_patterns=['客厅.*灯', '走廊.*灯', '玄关.*灯'], action_config={ 'description': '回家时自动打开的灯光', 'actions': [ {'type': 'set_param', 'param': 'power', 'value': True}, {'type': 'set_param', 'param': 'brightness', 'value': 70} ] } ) # 3. 创建"睡眠模式"场景 print("\n=== 创建睡眠模式场景 ===") sleep_scene = manager.create_scene_group( scene_name="睡眠模式", device_patterns=['卧室.*灯', '卫生间.*夜灯'], action_config={ 'description': '睡眠时调整的灯光', 'actions': [ {'type': 'set_param', 'param': 'power', 'value': False}, {'type': 'set_param', 'param': 'brightness', 'value': 10} ] } ) # 4. 执行场景（示例） # manager.execute_scene("回家模式") ``` ### 4.2 基于时间表的自动化任务 对于智能家居系统，基于时间表的自动化是核心需求。我们可以结合Python的调度库实现复杂的自动化场景： ```python #!/usr/bin/env python3 """ 基于时间表的自动化调度器 功能：根据时间表自动执行场景和群组操作 """ import schedule import time import threading from datetime import datetime, time as dt_time from smart_group_manager import SmartGroupManager class AutomationScheduler: def __init__(self): self.manager = SmartGroupManager() self.scheduled_jobs = [] self.running = False def add_daily_schedule(self, scene_name, trigger_time, days_of_week=None): """添加每日定时任务""" def job(): print(f"[{datetime.now().strftime('%H:%M:%S')}] 触发场景: {scene_name}") self.manager.execute_scene(scene_name) # 解析时间 if isinstance(trigger_time, str): hour, minute = map(int, trigger_time.split(':')) trigger_time = dt_time(hour, minute) # 创建调度任务 if days_of_week: # 指定星期几 for day in days_of_week: getattr(schedule.every(), day).at(trigger_time.strftime('%H:%M')).do(job) self.scheduled_jobs.append({ 'scene': scene_name, 'time': trigger_time, 'days': days_of_week }) else: # 每天执行 schedule.every().day.at(trigger_time.strftime('%H:%M')).do(job) self.scheduled_jobs.append({ 'scene': scene_name, 'time': trigger_time, 'days': 'daily' }) print(f"已安排任务: {scene_name} 在 {trigger_time.strftime('%H:%M')} 执行") def add_sunrise_sunset_schedule(self, scene_name, offset_minutes=0, sunrise=True): """添加日出/日落定时任务（需要集成天气API）""" # 这里需要集成天气API获取实际的日出日落时间 # 简化示例：使用固定时间 if sunrise: trigger_time = "06:30" else: trigger_time = "18:30" # 应用偏移 if offset_minutes != 0: from datetime import timedelta base_time = datetime.strptime(trigger_time, "%H:%M") adjusted_time = (base_time + timedelta(minutes=offset_minutes)).time() trigger_time = adjusted_time.strftime("%H:%M") self.add_daily_schedule(scene_name, trigger_time) def add_conditional_schedule(self, scene_name, condition_func, check_interval=60): """添加条件触发任务""" def conditional_job(): if condition_func(): print(f"[{datetime.now().strftime('%H:%M:%S')}] 条件满足，触发场景: {scene_name}") self.manager.execute_scene(scene_name) schedule.every(check_interval).seconds.do(conditional_job) self.scheduled_jobs.append({ 'scene': scene_name, 'type': 'conditional', 'check_interval': check_interval }) print(f"已安排条件任务: {scene_name}，每 {check_interval} 秒检查一次") def start_scheduler(self): """启动调度器""" self.running = True print("自动化调度器已启动") def run_scheduler(): while self.running: schedule.run_pending() time.sleep(1) # 在后台线程中运行调度器 scheduler_thread = threading.Thread(target=run_scheduler, daemon=True) scheduler_thread.start() return scheduler_thread def stop_scheduler(self): """停止调度器""" self.running = False schedule.clear() print("自动化调度器已停止") def list_scheduled_jobs(self): """列出所有已安排的任务""" print("\n=== 已安排的任务 ===") for i, job in enumerate(self.scheduled_jobs, 1): if job.get('type') == 'conditional': print(f"{i}. 条件任务: {job['scene']}") print(f" 检查间隔: {job['check_interval']}秒") else: print(f"{i}. 定时任务: {job['scene']}") print(f" 时间: {job['time'].strftime('%H:%M') if isinstance(job['time'], dt_time) else job['time']}") print(f" 重复: {job.get('days', 'daily')}") print() # 示例条件函数 def is_weekday(): """检查是否为工作日""" return datetime.now().weekday() < 5 # 0-4为周一到周五 def is_evening(): """检查是否为晚上（18:00-23:59）""" current_hour = datetime.now().hour return 18 <= current_hour < 24 def is_nobody_home(): """检查是否无人在家（简化示例）""" # 实际实现可能需要集成传感器数据或手机定位 return False # 假设一直有人在家 if __name__ == "__main__": scheduler = AutomationScheduler() # 工作日早晨自动打开窗帘和灯光 scheduler.add_daily_schedule( scene_name="早晨模式", trigger_time="07:00", days_of_week=['monday', 'tuesday', 'wednesday', 'thursday', 'friday'] ) # 周末早晨稍晚一些 scheduler.add_daily_schedule( scene_name="周末早晨模式", trigger_time="08:30", days_of_week=['saturday', 'sunday'] ) # 日落时自动打开户外灯光 scheduler.add_sunrise_sunset_schedule( scene_name="傍晚模式", sunrise=False, # 日落 offset_minutes=-30 # 日落前30分钟 ) # 条件触发：当无人在家时进入安防模式 scheduler.add_conditional_schedule( scene_name="安防模式", condition_func=is_nobody_home, check_interval=300 # 每5分钟检查一次 ) # 列出所有任务 scheduler.list_scheduled_jobs() # 启动调度器 try: scheduler.start_scheduler() # 保持主线程运行 while True: time.sleep(1) except KeyboardInterrupt: print("\n正在停止调度器...") scheduler.stop_scheduler() ``` 这个自动化调度器可以处理复杂的定时任务和条件触发，让你的智能家居系统真正实现自动化运行。 ## 5. 高级技巧：CLI与外部系统集成 在实际的生产环境中，ESP-RainMaker系统往往需要与其他系统集成。下面我们看几个常见的集成场景。 ### 5.1 与Home Assistant集成 Home Assistant是流行的开源家庭自动化平台，我们可以通过RainMaker的REST API将其集成： ```python #!/usr/bin/env python3 """ Home Assistant与ESP-RainMaker集成模块 功能：将RainMaker设备暴露给Home Assistant """ import requests import json from typing import Dict, List from rainmaker.user import User class HomeAssistantIntegration: def __init__(self, ha_url, ha_token): self.ha_url = ha_url.rstrip('/') self.ha_token = ha_token self.rainmaker_user = User() self.rainmaker_user.login_from_config() # Home Assistant API headers self.headers = { 'Authorization': f'Bearer {ha_token}', 'Content-Type': 'application/json' } def discover_rainmaker_devices(self): """发现RainMaker设备并注册到Home Assistant""" nodes = self.rainmaker_user.get_nodes() registered_devices = [] for node in nodes: node_info = node.get_info() device_id = node_info['node_id'] device_name = node_info.get('name', f'RainMaker_{device_id[:8]}') # 根据设备类型创建对应的HA实体 device_type = self._detect_device_type(node_info) if device_type == 'light': self._register_light(device_id, device_name, node_info) elif device_type == 'switch': self._register_switch(device_id, device_name, node_info) elif device_type == 'sensor': self._register_sensor(device_id, device_name, node_info) registered_devices.append({ 'device_id': device_id, 'name': device_name, 'type': device_type }) return registered_devices def _detect_device_type(self, node_info): """检测设备类型""" params = node_info.get('params', {}) if any('brightness' in key.lower() for key in params.keys()): return 'light' elif any('power' in key.lower() for key in params.keys()): return 'switch' elif any('temperature' in key.lower() or 'humidity' in key.lower() for key in params.keys()): return 'sensor' return 'switch' # 默认类型 def _register_light(self, device_id, device_name, node_info): """注册灯光设备到Home Assistant""" unique_id = f"rainmaker_{device_id}" # MQTT发现配置 config = { 'name': device_name, 'unique_id': unique_id, 'command_topic': f'rainmaker/{device_id}/set', 'state_topic': f'rainmaker/{device_id}/state', 'brightness_command_topic': f'rainmaker/{device_id}/brightness/set', 'brightness_state_topic': f'rainmaker/{device_id}/brightness/state', 'schema': 'json', 'device': { 'identifiers': [f'rainmaker_{device_id}'], 'name': device_name, 'manufacturer': 'Espressif', 'model': 'RainMaker Device', 'sw_version': node_info.get('fw_version', 'unknown') } } # 检查是否支持色温 if any('color_temp' in key.lower() for key in node_info.get('params', {}).keys()): config['color_temp_command_topic'] = f'rainmaker/{device_id}/color_temp/set' config['color_temp_state_topic'] = f'rainmaker/{device_id}/color_temp/state' # 发送发现消息 discovery_topic = f'homeassistant/light/{unique_id}/config' self._publish_mqtt_discovery(discovery_topic, config) print(f"已注册灯光设备: {device_name}") def _register_switch(self, device_id, device_name, node_info): """注册开关设备到Home Assistant""" unique_id = f"rainmaker_{device_id}" config = { 'name': device_name, 'unique_id': unique_id, 'command_topic': f'rainmaker/{device_id}/set', 'state_topic': f'rainmaker/{device_id}/state', 'payload_on': 'true', 'payload_off': 'false', 'device': { 'identifiers': [f'rainmaker_{device_id}'], 'name': device_name, 'manufacturer': 'Espressif', 'model': 'RainMaker Device', 'sw_version': node_info.get('fw_version', 'unknown') } } discovery_topic = f'homeassistant/switch/{unique_id}/config' self._publish_mqtt_discovery(discovery_topic, config) print(f"已注册开关设备: {device_name}") def _publish_mqtt_discovery(self, topic, config): """通过MQTT发布设备发现信息""" # 这里需要实现MQTT客户端 # 简化示例：打印配置信息 print(f"MQTT发现主题: {topic}") print(f"配置: {json.dumps(config, indent=2)}") # 实际实现需要使用MQTT客户端发布消息 # import paho.mqtt.client as mqtt # client = mqtt.Client() # client.connect("homeassistant.local", 1883) # client.publish(topic, json.dumps(config), retain=True) def start_state_sync(self, interval=30): """启动状态同步服务""" import threading import time def sync_loop(): while True: try: self._sync_all_devices_state() time.sleep(interval) except Exception as e: print(f"状态同步出错: {e}") time.sleep(5) sync_thread = threading.Thread(target=sync_loop, daemon=True) sync_thread.start() print(f"状态同步服务已启动，间隔: {interval}秒") return sync_thread def _sync_all_devices_state(self): """同步所有设备状态""" nodes = self.rainmaker_user.get_nodes() for node in nodes: node_info = node.get_info() device_id = node_info['node_id'] # 获取设备当前状态 state = {} for param_name in node_info.get('params', {}).keys(): try: param_value = node.get_param(param_name) state[param_name] = param_value except: continue # 发布状态到MQTT state_topic = f'rainmaker/{device_id}/state' state_payload = json.dumps(state) # 实际实现需要发布到MQTT # self.mqtt_client.publish(state_topic, state_payload) print(f"同步状态: {node_info.get('name')} -> {state_payload}") # 使用示例 if __name__ == "__main__": # 配置Home Assistant信息 HA_URL = "http://homeassistant.local:8123" HA_TOKEN = "your_long_lived_access_token" # 创建集成实例 integration = HomeAssistantIntegration(HA_URL, HA_TOKEN) # 发现并注册设备 devices = integration.discover_rainmaker_devices() print(f"已注册 {len(devices)} 个设备到Home Assistant") # 启动状态同步 # integration.start_state_sync(interval=30) ``` ### 5.2 与数据库集成记录历史数据 对于需要数据分析的场景，我们可以将设备数据保存到数据库中： ```python #!/usr/bin/env python3 """ 设备数据历史记录器 功能：将设备状态和参数变化记录到数据库 """ import sqlite3 import json from datetime import datetime from threading import Timer from rainmaker.user import User class DeviceDataLogger: def __init__(self, db_path='rainmaker_data.db'): self.db_path = db_path self.user = User() self.user.login_from_config() self.init_database() self.logging_interval = 60 # 默认60秒记录一次 self.is_logging = False def init_database(self): """初始化数据库""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # 创建设备表 cursor.execute(''' CREATE TABLE IF NOT EXISTS devices ( device_id TEXT PRIMARY KEY, name TEXT, device_type TEXT, first_seen TIMESTAMP, last_seen TIMESTAMP ) ''') # 创建设备参数表 cursor.execute(''' CREATE TABLE IF NOT EXISTS device_params ( id INTEGER PRIMARY KEY AUTOINCREMENT, device_id TEXT, param_name TEXT, param_value TEXT, recorded_at TIMESTAMP, FOREIGN KEY (device_id) REFERENCES devices (device_id) ) ''') # 创建设备状态表 cursor.execute(''' CREATE TABLE IF NOT EXISTS device_status ( id INTEGER PRIMARY KEY AUTOINCREMENT, device_id TEXT, online BOOLEAN, firmware_version TEXT, recorded_at TIMESTAMP, FOREIGN KEY (device_id) REFERENCES devices (device_id) ) ''') # 创建索引以提高查询性能 cursor.execute('CREATE INDEX IF NOT EXISTS idx_device_params ON device_params (device_id, recorded_at)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_device_status ON device_status (device_id, recorded_at)') conn.commit() conn.close() print(f"数据库已初始化: {self.db_path}") def log_current_state(self): """记录当前所有设备状态""" nodes = self.user.get_nodes() current_time = datetime.now() conn = sqlite3.connect(self.db_path) cursor = conn.cursor() for node in nodes: try: node_info = node.get_info() device_id = node_info['node_id'] device_name = node_info.get('name', '未命名') is_online = node_info.get('online', False) fw_version = node_info.get('fw_version', '未知') # 更新设备信息 cursor.execute(''' INSERT OR REPLACE INTO devices (device_id, name, device_type, first_seen, last_seen) VALUES (?, ?, ?, COALESCE((SELECT first_seen FROM devices WHERE device_id = ?), ?), ? ) ''', ( device_id, device_name, self._detect_device_type(node_info), device_id, current_time, # 用于COALESCE current_time )) # 记录设备状态 cursor.execute(''' INSERT INTO device_status (device_id, online, firmware_version, recorded_at) VALUES (?, ?, ?, ?) ''', (device_id, is_online, fw_version, current_time)) # 记录参数值 params = node_info.get('params', {}) for param_name, param_value in params.items(): cursor.execute(''' INSERT INTO device_params (device_id, param_name, param_value, recorded_at) VALUES (?, ?, ?, ?) ''', (device_id, param_name, json.dumps(param_value), current_time)) conn.commit() except Exception as e: print(f"记录设备 {device_id} 状态时出错: {e}") conn.rollback() continue conn.close() print(f"[{current_time.strftime('%H:%M:%S')}] 已记录 {len(nodes)} 个设备状态") def start_logging(self, interval=None): """启动定时记录""" if interval: self.logging_interval = interval self.is_logging = True def logging_loop(): if self.is_logging: try: self.log_current_state() except Exception as e: print(f"记录状态时出错: {e}") # 安排下一次记录 Timer(self.logging_interval, logging_loop).start() print(f"开始定时记录，间隔: {self.logging_interval}秒") logging_loop() def stop_logging(self): """停止记录""" self.is_logging = False print("已停止定时记录") def _detect_device_type(self, node_info): """检测设备类型""" params = node_info.get('params', {}) if any('brightness' in key.lower() for key in params.keys()): return 'light' elif any('temperature' in key.lower() for key in params.keys()): return 'sensor' elif any('power' in key.lower() for key in params.keys()): return 'switch' elif any('fan' in key.lower() for key in params.keys()): return 'fan' return 'unknown' def query_device_history(self, device_id, start_time=None, end_time=None, limit=1000): """查询设备历史数据""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() query = ''' SELECT dp.recorded_at, dp.param_name, dp.param_value, ds.online FROM device_params dp LEFT JOIN device_status ds ON dp.device_id = ds.device_id AND dp.recorded_at = ds.recorded_at WHERE dp.device_id = ? ''' params = [device_id] if start_time: query += ' AND dp.recorded_at >= ?' params.append(start_time) if end_time: query += ' AND dp.recorded_at <= ?' params.append(end_time) query += ' ORDER BY dp.recorded_at DESC LIMIT ?' params.append(limit) cursor.execute(query, params) results = cursor.fetchall() conn.close() # 格式化结果 formatted_results = [] for row in results: recorded_at, param_name, param_value, online = row formatted_results.append({ 'timestamp': recorded_at, 'parameter': param_name, 'value': json.loads(param_value) if param_value else None, 'online': bool(online) }) return formatted_results def generate_daily_report(self, date=None): """生成每日报告""" if not date: date = datetime.now().date() conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # 查询当天的设备在线率 cursor.execute(''' SELECT device_id, COUNT(*) as total_records, SUM(CASE WHEN online = 1 THEN 1 ELSE 0 END) as online_records, (SUM(CASE WHEN online = 1 THEN 1 ELSE 0 END) * 100.0 / COUNT(*)) as online_rate FROM device_status WHERE DATE(recorded_at) = ? GROUP BY device_id ''', (date.isoformat(),)) online_stats = cursor.fetchall() # 查询参数变化统计 cursor.execute(''' SELECT device_id, param_name, COUNT(DISTINCT param_value) as value_changes FROM device_params WHERE DATE(recorded_at) = ? GROUP BY device_id, param_name HAVING value_changes > 1 ''', (date.isoformat(),)) param_changes = cursor.fetchall() conn.close() # 生成报告 report = { 'date': date.isoformat(), 'devices_analyzed': len(online_stats), 'online_statistics': [], 'parameter_changes': [] } for device_id, total, online, rate in online_stats: report['online_statistics'].append({ 'device_id': device_id, 'online_rate_percent': round(rate, 2), 'total_records': total, 'online_records': online }) for device_id, param_name, changes in param_changes: report['parameter_changes'].append({ 'device_id': device_id, 'parameter': param_name, 'value_changes': changes }) return report # 使用示例 if __name__ == "__main__": # 创建数据记录器 logger = DeviceDataLogger('rainmaker_history.db') # 记录一次当前状态 logger.log_current_state()