物联网设备控制小程序：从使用到设备接入

这个小程序是给 50 到 60 人规模的小团队使用的设备控制工具。它的目标不是做复杂的大平台，而是让团队成员能稳定地登录、绑定设备、查看状态、下发命令，并把常用操作沉淀成任务或模板。

查看完整接入示例

小程序能做什么

微信一键登录，首次使用访问码绑定身份，后续直接登录。
管理员维护访问码、团队人数上限和允许名单。
用户手动绑定设备，可选择个人可见或团队共享。
查看设备在线状态、最近上报时间和状态数据。
下发设备命令，例如开关、模式切换、参数设置和请求链接。
创建手动任务、任务模板和自动任务。
自动任务支持两种触发表达：定时触发和钩子触发。

为什么不是只支持 ESP32

ESP32-S3 只是当前固件示例。真正的接入协议是 HTTP JSON，所以树莓派、Linux 网关、Windows 上位机、其他联网 MCU，甚至一段 Python 脚本都可以接入。

设备只要能完成三件事就行：

定期上报状态。
主动拉取待执行命令。
执行后回写成功或失败。

这种设计适合弱服务器，因为服务器不需要和每台设备保持长连接，小程序也不需要高频直连设备。

设备接入流程

1. 上报状态

设备定期请求：

1
POST https://api.niuhaoyu.xyz/api/device/{deviceId}/status

请求体示例：

1
{
2
  "online": true,
3
  "status": {
4
    "power": "on",
5
    "temperature": 26.5,
6
    "mode": "eco"
7
  }
8
}

2. 拉取命令

设备轮询：

1
GET https://api.niuhaoyu.xyz/api/device/{deviceId}/commands/next

如果有命令，服务器会返回类似数据：

1
{
2
  "id": "cmd-xxx",
3
  "command": {
4
    "type": "setPower",
5
    "payload": {
6
      "power": "off"
7
    }
8
  }
9
}

3. 回写结果

执行完成后请求：

1
POST https://api.niuhaoyu.xyz/api/device/{deviceId}/commands/{commandId}

成功示例：

1
{
2
  "status": "success",
3
  "result": {
4
    "power": "off"
5
  }
6
}

失败示例：

1
{
2
  "status": "failed",
3
  "message": "unsupported command"
4
}

任务系统怎么理解

小程序里只需要理解两类任务：

手动任务：用户点击后立即执行。
自动任务：保存成规则，由定时或钩子触发。

模板不是任务类型，它只是常用操作的保存方式。比如“打开电源 10 秒后关闭”可以保存成模板，需要时一键执行；“每天 08<00> 打开设备”则应该保存成自动任务。

常用命令类型

类型	用途	Payload 示例
`setPower`	开关	`{"power":"on"}`
`setMode`	模式切换	`{"mode":"eco"}`
`setValue`	参数设置	`{"key":"threshold","value":50}`
`httpRequest`	让设备请求一个链接	`{"method":"GET","url":"https://example.com/action"}`

完整设备示例

下面示例可以在电脑、树莓派或普通 Linux 网关上运行。真实项目里不要把设备 token 写进公开仓库，应该从环境变量或本地安全配置读取。

1
import json
2
import os
3
import time
4
import urllib.request
5

6
API_BASE = os.getenv("IOT_API_BASE", "https://api.niuhaoyu.xyz/api")
7
DEVICE_ID = os.getenv("IOT_DEVICE_ID", "demo-python-001")
8
DEVICE_TOKEN = os.getenv("IOT_DEVICE_TOKEN", "replace-with-device-token")
9

10

11
def request_json(path, method="GET", payload=None):
12
    data = None
13
    if payload is not None:
14
        data = json.dumps(payload).encode("utf-8")
15
    req = urllib.request.Request(
16
        API_BASE + path,
17
        data=data,
18
        method=method,
19
        headers={
20
            "Content-Type": "application/json",
21
            "Authorization": "Bearer " + DEVICE_TOKEN,
22
        },
23
    )
24
    with urllib.request.urlopen(req, timeout=8) as resp:
25
        return json.loads(resp.read().decode("utf-8"))
26

27

28
def report_status(state):
29
    return request_json(
30
        "/device/%s/status" % DEVICE_ID,
31
        method="POST",
32
        payload={
33
            "online": True,
34
            "status": state,
35
        },
36
    )
37

38

39
def next_command():
40
    result = request_json("/device/%s/commands/next" % DEVICE_ID)
41
    return (result.get("data") or {}).get("command")
42

43

44
def finish_command(command_id, status, result=None, message=""):
45
    payload = {"status": status}
46
    if result is not None:
47
        payload["result"] = result
48
    if message:
49
        payload["message"] = message
50
    return request_json(
51
        "/device/%s/commands/%s" % (DEVICE_ID, command_id),
52
        method="POST",
53
        payload=payload,
54
    )
55

56

57
def execute(command, state):
58
    command_type = command.get("type")
59
    payload = command.get("payload") or {}
60

61
    if command_type == "setPower":
62
        state["power"] = payload.get("power", "off")
63
        return True, state, ""
64

65
    if command_type == "setMode":
66
        state["mode"] = payload.get("mode", "normal")
67
        return True, state, ""
68

69
    if command_type == "setValue":
70
        key = payload.get("key", "value")
71
        state[key] = payload.get("value")
72
        return True, state, ""
73

74
    return False, state, "unsupported command: %s" % command_type
75

76

77
def main():
78
    state = {
79
        "power": "off",
80
        "mode": "normal",
81
        "temperature": 26.5,
82
    }
83
    while True:
84
        report_status(state)
85
        item = next_command()
86
        if item:
87
            command_id = item.get("id")
88
            command = item.get("command") or {}
89
            ok, state, message = execute(command, state)
90
            if ok:
91
                finish_command(command_id, "success", result=state)
92
            else:
93
                finish_command(command_id, "failed", message=message)
94
        time.sleep(3)
95

96

97
if __name__ == "__main__":
98
    main()

小程序使用建议

首次使用时先完成微信登录和访问码绑定。添加设备时，如果只是自己调试，选择“个人可见”；如果团队都要用，选择“团队共享”。设备上线后，先在设备详情页确认状态能刷新，再创建任务。

手动任务适合立即控制。自动任务适合规则化场景，例如每天固定时间执行，或者后续由设备告警、传感器阈值、外部 HTTP 事件触发。

niuniu的博客