2. 设备影子状态同步

任务目标

实现一个设备影子（Device Shadow）的最小可运行闭环：设备端上报实际状态（reported），云端保存；应用端可设置期望状态（desired），设备端拉取并执行后，将执行结果同步回云端，最终使 reported 与 desired 收敛一致。本案例聚焦影子状态同步的核心逻辑，不涉及具体云厂商SDK，用Python模拟云端存储与设备端行为。

核心实现

import json
import time
from typing import Dict, Any

# ---------- 云端影子存储（模拟） ----------
class ShadowStore:
    def __init__(self):
        self.state = {
            "reported": {},   # 设备实际状态
            "desired": {}     # 应用期望状态
        }
        self.version = 0

    def update_reported(self, reported: Dict[str, Any]) -> Dict[str, Any]:
        """设备上报状态，合并到 reported"""
        self.state["reported"].update(reported)
        self.version += 1
        return self._snapshot()

    def update_desired(self, desired: Dict[str, Any]) -> Dict[str, Any]:
        """应用设置期望状态，合并到 desired"""
        self.state["desired"].update(desired)
        self.version += 1
        return self._snapshot()

    def get_delta(self) -> Dict[str, Any]:
        """计算 desired 与 reported 的差异（需要同步的字段）"""
        delta = {}
        for key, desired_val in self.state["desired"].items():
            if self.state["reported"].get(key) != desired_val:
                delta[key] = desired_val
        return delta

    def _snapshot(self) -> Dict[str, Any]:
        return {
            "state": self.state,
            "version": self.version
        }

# ---------- 设备端 ----------
class Device:
    def __init__(self, device_id: str, shadow: ShadowStore):
        self.device_id = device_id
        self.shadow = shadow
        # 设备本地实际状态（初始为空）
        self.local_state: Dict[str, Any] = {}

    def report_current_state(self):
        """设备上报当前实际状态"""
        return self.shadow.update_reported(self.local_state)

    def sync_from_shadow(self):
        """拉取影子 delta，执行期望状态，并回传执行结果"""
        delta = self.shadow.get_delta()
        if not delta:
            print(f"[{self.device_id}] 无待同步字段，状态已一致")
            return self.shadow._snapshot()

        print(f"[{self.device_id}] 从影子获取期望变更: {delta}")
        # 模拟执行期望状态（实际场景中可能是控制硬件）
        for key, value in delta.items():
            self.local_state[key] = value
            print(f"[{self.device_id}] 执行: 设置 {key} = {value}")

        # 执行完成后，将新状态上报为 reported
        return self.report_current_state()

# ---------- 最小闭环演示 ----------
if __name__ == "__main__":
    shadow = ShadowStore()
    device = Device("sensor-001", shadow)

    # 1. 设备首次上线，上报初始状态
    device.local_state = {"power": "off", "brightness": 0}
    snap = device.report_current_state()
    print(f"设备上线上报: {json.dumps(snap, ensure_ascii=False)}")

    # 2. 应用端设置期望状态
    snap = shadow.update_desired({"power": "on", "brightness": 80})
    print(f"应用设置期望: {json.dumps(snap, ensure_ascii=False)}")

    # 3. 设备端同步影子，执行变更并回传
    snap = device.sync_from_shadow()
    print(f"同步后云端状态: {json.dumps(snap, ensure_ascii=False)}")

    # 4. 再次同步，验证无差异
    device.sync_from_shadow()

运行与验证

运行方式：将上述代码保存为 shadow_sync.py，直接执行 python shadow_sync.py（Python 3.6+，无第三方依赖）。

预期输出（确定性结果）：

设备上线上报: {"state": {"reported": {"power": "off", "brightness": 0}, "desired": {}}, "version": 1}
应用设置期望: {"state": {"reported": {"power": "off", "brightness": 0}, "desired": {"power": "on", "brightness": 80}}, "version": 2}
[sensor-001] 从影子获取期望变更: {"power": "on", "brightness": 80}
[sensor-001] 执行: 设置 power = on
[sensor-001] 执行: 设置 brightness = 80
同步后云端状态: {"state": {"reported": {"power": "on", "brightness": 80}, "desired": {"power": "on", "brightness": 80}}, "version": 3}
[sensor-001] 无待同步字段，状态已一致

通过判定标准：程序无异常退出；第3步执行后，云端 reported 与 desired 中 power 和 brightness 字段值完全一致；第4步输出“无待同步字段”，证明状态收敛。

关键说明

• 影子同步机制：设备端每次同步时通过 get_delta() 对比 desired 与 reported，只拉取差异字段执行，避免全量覆盖。
• 版本号：version 字段模拟云端影子版本控制，每次状态变更自增，实际工程中用于解决并发冲突。
• 扩展接入：若对接真实云平台（如AWS IoT Shadow、阿里云设备影子），只需将 ShadowStore 的读写操作替换为对应云SDK的 get_thing_shadow/update_thing_shadow 接口，设备端同步逻辑保持不变。