4. HTTP协议接入

系统架构与通信流程

代码实现

以下代码包含两个文件：cloud_server.py（云平台模拟）与device.py（设备端）。服务器维护一个命令队列，设备上报数据时若队列中有命令则随响应返回；设备通过GET主动拉取命令。

# cloud_server.py - 云平台模拟（HTTP服务器）
from http.server import HTTPServer, BaseHTTPRequestHandler
import json, threading, time

# 内存数据存储
device_data = {}       # 设备上报的遥测数据
command_queue = []     # 待下发的命令队列
lock = threading.Lock()

class CloudHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        """处理设备数据上报"""
        if self.path != '/api/telemetry':
            self.send_error(404)
            return
        length = int(self.headers.get('Content-Length', 0))
        body = json.loads(self.rfile.read(length).decode('utf-8'))
        
        with lock:
            device_data.update(body)  # 存储遥测数据
            # 从队列取一条命令返回给设备（若无命令则返回空）
            cmd = command_queue.pop(0) if command_queue else None
        
        # 响应中包含当前待执行命令
        response = {"status": "ok", "command": cmd}
        self.send_response(200)
        self.send_header('Content-Type', 'application/json')
        self.end_headers()
        self.wfile.write(json.dumps(response).encode('utf-8'))
        print(f"[云平台] 收到上报: {body}, 下发命令: {cmd}")
    
    def do_GET(self):
        """处理设备命令拉取"""
        if self.path != '/api/commands':
            self.send_error(404)
            return
        with lock:
            cmd = command_queue.pop(0) if command_queue else None
        response = {"command": cmd}
        self.send_response(200)
        self.send_header('Content-Type', 'application/json')
        self.end_headers()
        self.wfile.write(json.dumps(response).encode('utf-8'))
        print(f"[云平台] 命令拉取, 返回: {cmd}")
    
    def log_message(self, format, *args):
        pass  # 关闭默认日志

def start_server(port=8000):
    """启动云平台服务器（后台线程运行）"""
    server = HTTPServer(('127.0.0.1', port), CloudHandler)
    t = threading.Thread(target=server.serve_forever, daemon=True)
    t.start()
    print(f"[云平台] 服务器已启动: http://127.0.0.1:{port}")
    return server

def push_command(cmd):
    """向命令队列添加一条命令（模拟平台下发）"""
    with lock:
        command_queue.append(cmd)
    print(f"[云平台] 新命令入队: {cmd}")

if __name__ == '__main__':
    server = start_server()
    # 模拟平台主动下发命令
    time.sleep(1)
    push_command({"cmd": "relay_on", "param": 1})
    push_command({"cmd": "relay_off", "param": 0})
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        server.shutdown()

# device.py - 设备端（HTTP客户端）
import urllib.request
import json, time, random

SERVER_URL = "http://127.0.0.1:8000"

def read_sensor():
    """模拟读取温湿度传感器"""
    return {
        "temp": round(random.uniform(20.0, 30.0), 1),
        "hum": round(random.uniform(40.0, 70.0), 1)
    }

def report_data(payload):
    """上报数据并获取响应中的命令"""
    req = urllib.request.Request(
        f"{SERVER_URL}/api/telemetry",
        data=json.dumps(payload).encode('utf-8'),
        headers={'Content-Type': 'application/json'},
        method='POST'
    )
    with urllib.request.urlopen(req, timeout=5) as resp:
        return json.loads(resp.read().decode('utf-8'))

def fetch_command():
    """主动拉取命令"""
    req = urllib.request.Request(f"{SERVER_URL}/api/commands", method='GET')
    with urllib.request.urlopen(req, timeout=5) as resp:
        return json.loads(resp.read().decode('utf-8'))

def execute_command(cmd):
    """执行下发的命令"""
    if cmd is None:
        return
    action = cmd.get("cmd", "")
    if action == "relay_on":
        print(f"[设备] 执行: 继电器开启 (参数={cmd.get('param')})")
    elif action == "relay_off":
        print(f"[设备] 执行: 继电器关闭 (参数={cmd.get('param')})")
    else:
        print(f"[设备] 未知命令: {cmd}")

def run_device(cycles=3):
    """设备主循环：上报数据 → 拉取命令 → 执行"""
    for i in range(cycles):
        print(f"\n===== 第 {i+1} 轮 =====")
        # 1. 读取传感器并上报
        sensor_data = read_sensor()
        print(f"[设备] 上报数据: {sensor_data}")
        resp = report_data(sensor_data)
        print(f"[设备] 上报响应: {resp}")
        
        # 2. 若上报响应中带命令则执行
        if resp.get("command"):
            execute_command(resp["command"])
        
        # 3. 主动拉取命令（处理上报时未返回的命令）
        cmd = fetch_command()
        if cmd.get("command"):
            print(f"[设备] 拉取到命令: {cmd['command']}")
            execute_command(cmd["command"])
        else:
            print("[设备] 无待执行命令")
        
        time.sleep(1)  # 间隔1秒

if __name__ == '__main__':
    run_device(cycles=3)

运行与验证

依赖：Python 3.6+（仅标准库，无第三方包）。

运行步骤：

1. 终端1启动云平台：python cloud_server.py，看到“服务器已启动”即就绪；
2. 终端2运行设备：python device.py，观察设备上报与命令执行日志。

验证方法：

• 云平台启动后自动向队列压入relay_on和relay_off两条命令；
• 设备每轮上报后，云平台响应中应携带一条命令（第一轮返回relay_on，第二轮返回relay_off），设备打印对应执行日志；
• 第三轮上报时命令队列已空，响应中command为null，设备拉取命令也返回空，打印“无待执行命令”；
• 云平台终端同步打印收到的上报数据和下发的命令，与设备端日志相互印证。

判定标准：设备端依次出现“继电器开启”和“继电器关闭”的执行日志，且云平台日志显示对应命令已下发，即判定数据上报与命令下发链路验证通过。