怎样二次开发智能空调控制器来实现远程定时任务设置_解决方案

CATALOG

芯步的智能空调控制器开放了标准HTTP接口，支持远程控制与定时任务设置。以下方案将从接口调用原理、签名计算、核心功能封装到定时任务调度进行全面阐述，帮助开发者快速实现远程定时任务功能。

一、技术背景与准备

要实现远程定时任务，核心逻辑是：上层应用根据预设的时间规则，自动化调用芯步开放平台的HTTP接口。由于芯步的设备支持直连控制，无需经过第三方云中转，因此响应速度快且稳定性高。

在开始开发前，你需要准备以下三个关键凭证：

AppID（应用ID）：标识你的应用身份。
AppSecret（应用密钥）：用于加密签名，保障通信安全。
Device ID（设备ID）：即智能空调控制器的唯一标识，可通过控制台获取。

二、接口调用核心：签名与空调指令

所有二次开发都基于HTTP请求。为了安全，芯步使用了动态签名机制，每次请求都需要计算一个 sign 参数。

1. 签名算法（Python示例）

根据官方文档，签名的生成规则为：sign = md5( md5(AppSecret) + ts )，其中 ts 是Unix时间戳（秒）。

2. 空调控制指令集

芯步的智能空调遥控器通过红外或红外波码学习控制空调。其 order 参数为JSON字符串，核心指令如下

功能描述	Order JSON 示例	说明
开机/关机	`{"power": 1}` 或 `{"power": 0}`	1=开机，0=关机
设置模式	`{"mode": "cool"}`	cool(制冷)/heat(制热)/auto(自动)/dry(除湿)/wind(送风)
设置温度	`{"temp": 24}`	温度范围通常为16-30℃，数字代表摄氏度
设置风速	`{"windspeed": "auto"}`	auto(自动)/low(低风)/medium(中风)/high(高风)

三、解决方案架构：实现远程定时任务

实现定时任务通常有两种架构模式，根据你的业务场景（是简单的周期性任务，还是复杂的用户自定义任务）选择：

方案A：业务后端轮询（适用于中小规模、固定规则）

如果你的需求是“工作日下午6点自动开空调”这类固定规则，可以在你的服务器上通过定时任务（如Linux Crontab、Windows Task Scheduler或Quartz）调用接口。

逻辑流程：

你的服务器维护一个任务列表。
时间到达既定时刻（如18:00），触发执行函数。
函数计算签名，向 api.thingboot.com/{AppID}/device/control/ 发起POST请求。

优势：实现简单，无需管理复杂的云端定时队列。劣势：如果服务器重启或任务较多，管理较乱。

方案B：依赖数据库与调度器（适用于大规模、用户自定义）

在SaaS平台中，用户需要自定义“几点几分开关空调”。此时需要引入任务调度框架（如APScheduler, Celery Beat, ）来处理大量动态任务。

核心数据库设计：你需要设计一张 timing_tasks 表来存储用户配置的定时任务。

字段名	数据类型	说明
`id`	int	主键
`device_id`	varchar	关联的空调控制器ID
`cron_exp`	varchar	时间规则（如 `0 30 7 * * ?` 表示每天早上7:30）
`order_json`	text	存储具体的指令，如 `{"power":1,"temp":24,"mode":"cool"}`
`is_enabled`	bool	是否启用

业务执行逻辑：

调度器每秒扫描数据库，找出 next_run_time <= now() 的任务。
取出该条任务对应的 device_id 和 order_json。
调用芯步API执行控制。
更新该任务的 next_run_time（计算下次执行时间），实现循环。
异常处理：如果API返回设备离线，可以设置重试机制（如每隔5分钟重试一次，持续30分钟），因为定时任务往往对时效性要求不是纳秒级的。

四、实战：二次开发代码示例

这里以 Python + Flask 为例，展示如何封装一个智能空调的定时任务核心模块。

1. 封装设备控制类

首先封装一个 ACController 类，专门负责与芯步的硬件“对话”。

import requests
import hashlib
import time
import json

class YoyoACController:
    def __init__(self, app_id, app_secret):
        self.app_id = app_id
        self.app_secret = app_secret
        self.base_url = f"https://api.thingboot.com/{app_id}/device/control/"

def _generate_sign(self):
        ts = int(time.time())
        md5_secret = hashlib.md5(self.app_secret.encode()).hexdigest()
        sign_str = md5_secret + str(ts)
        sign = hashlib.md5(sign_str.encode()).hexdigest()
        return sign, ts

def send_command(self, device_id, command):
        """
        向指定设备发送指令
        device_id: 设备ID
        command: dict, 例如 {"power":1}
        """
        sign, ts = self._generate_sign()
        url = f"{self.base_url}?sign={sign}&ts={ts}"
        
        payload = {
            "device": device_id,
            "order": json.dumps(command)  # 注意:指令需要转为JSON字符串
        }
        
        headers = {"Content-Type": "application/json"}
        
        try:
            response = requests.post(url, json=payload, headers=headers, timeout=10)
            # 实际项目中应解析response.json()判断code是否为200
            if response.status_code == 200:
                print(f"指令发送成功: {command}")
                return True
            else:
                print(f"指令发送失败: {response.text}")
                return False
        except Exception as e:
            print(f"网络错误: {e}")
            return False

2. 定时任务调度逻辑

利用 schedule 库（或APScheduler）来模拟定时检查。

import schedule
import time
import threading

class TaskScheduler:
    def __init__(self, controller):
        self.controller = controller
        # 模拟数据库中的定时任务列表
        self.tasks = [
            {"id": 1, "device_id": "AC10001", "time": "22:00", "command": {"power": 0}, "desc": "每晚10点自动关机"},
            {"id": 2, "device_id": "AC10001", "time": "07:30", "command": {"power": 1, "temp": 24}, "desc": "每天早上7点半开机并设置24度"}
        ]

def job_wrapper(self, task):
        """执行任务的包装函数"""
        print(f"执行定时任务: {task['desc']}")
        self.controller.send_command(task['device_id'], task['command'])

def run_forever(self):
        """将数据库任务动态注册到调度器中"""
        for task in self.tasks:
            # 设定每天具体时间执行
            getattr(schedule.every(), 'day').at(task['time']).do(self.job_wrapper, task)
        
        while True:
            schedule.run_pending()
            time.sleep(1)

# 启动示例
if __name__ == "__main__":
    # 填入你在芯步控制台获取的凭证
    my_controller = YoyoACController("YOUR_APP_ID", "YOUR_APP_SECRET")
    scheduler = TaskScheduler(my_controller)
    
    # 启动定时监听线程
    # scheduler.run_forever() # 生产环境使用更健壮的APScheduler
    print("定时服务已启动...")

五、进阶优化

状态同步机制：目前提供的API主要是下行控制。为了确保定时任务生效（例如空调是否真的打开了），结合芯步提供的上行消息接收地址（Webhook）。设备状态变化（如开机、关机）会主动推送到你配置的URL，你可以据此更新数据库中的设备状态，避免“指令发了但红外没对准导致没执行”的状态误差。
局域网直连（可选）：芯步支持局域网控制。如果你的定时任务服务器和空调控制器在同一个局域网内，可以考虑开启局域网控制功能，这样即使外网断开，定时任务依然可以执行，且延迟更低。
分布式锁：如果你部署了多个服务实例来做高可用，需要注意定时任务的防重复执行。使用Redis分布式锁（如Redisson），确保同一个定时任务在同一个时间点只被一个实例执行一次。

通过上述方案，开发者只需关注业务层的日历规则（Crontab表达式），将底层的硬件通信完全交由芯步的标准HTTP接口处理，即可快速、稳定地实现智能空调的远程定时任务功能。