[米尔-全志T536工控板] [原创] 基于T536的IoT平台模拟器

楼主

JerryZhen

6 主题
7 帖子
429 积分

身份：LV2 初级技术员

E币：391

发消息

 339

 0

发表于 2025-3-24 18:29:38

显示全部楼层

本帖最后由 JerryZhen 于 2025-3-24 18:33 编辑

物联网平台模拟器作为连接物理世界与数字系统的关键桥梁，既降低了技术验证门槛，又加速了产品迭代周期。本文基于T536实现一个轻量级的IoT平台模拟器，为后续构建真实物联网平台提供了可靠的技术蓝图和架构参考。它的核心设计思想是使用发布/订阅（PubSub）模式，结合多线程、数据类（dataclass）和枚举类型（Enum），模拟IoT设备的数据采集、告警处理和云端控制等核心功能。

1. 架构概述：基于事件驱动的松耦合系统

模拟器采用事件驱动架构，各组件之间通过PubSub机制进行通信，实现了松耦合。这意味着各个组件可以独立开发、部署和修改，而无需了解其他组件的具体实现细节。

组件划分：

VirtualDevice (虚拟设备): 模拟IoT设备，周期性地生成传感器数据，并订阅命令通道，响应云端发来的指令。
AlertEngine (告警引擎): 监听设备发送的遥测数据，根据预设规则检测异常情况，并发布告警事件。
CloudService (云端服务): 模拟云平台，负责收集设备数据、处理告警，并提供设备控制接口。
DeviceManager (设备管理器): 管理虚拟设备的生命周期，包括添加、移除设备。

通信方式：

组件之间通过pubsub库进行消息发布与订阅。

架构示意图：

二、核心代模块解析

1. 设备模拟器（VirtualDevice）

代码片段：

class VirtualDevice(threading.Thread):

    SENSOR_RANGES = {  # 传感器数值范围定义

        SensorType.TEMPERATURE: (20, 40),

        SensorType.HUMIDITY: (30, 80)

    }



    def run(self):

        while not self._stop_event.is_set():

            # 生成数据并发布到MQTT主题

            data = TelemetryData(...)

            pub.sendMessage(f"devices/{self.device_id}/telemetry", data=data)

复制代码

设计要点：

线程继承：每个设备独立线程运行（threading.Thread）
事件循环：_stop_event控制线程生命周期
数据随机化：通过random.uniform模拟真实传感器波动
双通道通信：

上行通道：telemetry发送传感器数据

下行通道：command接收控制指令

2. 消息总线设计（PubSub模式）

通信流程图：

关键代码：

# 设备端订阅指令通道

pub.subscribe(self._handle_command, f"devices/{device_id}/command")



# 告警引擎订阅数据通道

pub.subscribe(self._check_temperature, "devices/DEV001/telemetry")

复制代码

模式优势：

松耦合：新增订阅者无需修改发布者代码
动态路由：通过主题名实现消息定向广播
可扩展性：轻松添加新设备或订阅组件

3. 云端服务（CloudService）

代码结构图：

关键机制：

命令队列缓冲：使用Queue实现生产-消费模式
线程池处理：独立工作线程处理命令下发
双重订阅：

接收设备数据（telemetry）

接收告警通知（alerts）

4. 设备管理器（DeviceManager）

类结构：

class DeviceManager:

    def __init__(self):

        self.devices = {}  # 设备ID到实例的映射

        self._lock = threading.Lock()  # 线程安全锁



    def add_device(self, device_id, sensor_type):

        with self._lock:  # 保证线程安全

            device = VirtualDevice(...)

            device.start()

复制代码

生命周期管理:

三、关键设计模式应用

1. 观察者模式（Observer Pattern）

# 设备数据发布

pub.sendMessage("devices/DEV001/telemetry", data=data)



# 多个订阅者同时接收

class AlertEngine:

    pub.subscribe(...)

    

class CloudService:

    pub.subscribe(...)

复制代码

2. 生产者-消费者模式# 云端命令队列实现

self._command_queue = Queue()  # 生产者写入

self._worker = threading.Thread(target=self._process_commands)  # 消费者读取

复制代码

3. 策略模式（传感器配置）

SENSOR_RANGES = {

    SensorType.TEMPERATURE: (20, 40),  # 温度策略

    SensorType.HUMIDITY: (30, 80)      # 湿度策略

}

复制代码

四、完整代码及运行结果

from pubsub import pub

import threading

import random

import time

from dataclasses import dataclass, field

from queue import Queue

from enum import Enum

import logging

from typing import Dict, Tuple



# -------------------------- 配置日志 --------------------------

logging.basicConfig(

    level=logging.INFO,

    format="%(asctime)s - %(levelname)s - %(message)s",

    datefmt="%Y-%m-%d %H:%M:%S"

)

logger = logging.getLogger("IoTPlatform")



# -------------------------- 枚举类型定义 --------------------------

class SensorType(str, Enum):

    TEMPERATURE = "temperature"

    HUMIDITY = "humidity"



class CommandType(str, Enum):

    REBOOT = "reboot"

    UPDATE_CONFIG = "update_config"



# -------------------------- IoT领域模型定义 --------------------------

@dataclass

class TelemetryData:

    device_id: str

    sensor_type: SensorType

    value: float

    timestamp: str = field(

        default_factory=lambda: time.strftime("%Y-%m-%d %H:%M:%S")

    )



@dataclass

class Command:

    device_id: str

    cmd_type: CommandType

    params: dict



# -------------------------- 设备模拟器 --------------------------

class VirtualDevice(threading.Thread):

    # 传感器配置：类型 -> (最小值, 最大值)

    SENSOR_RANGES: Dict[SensorType, Tuple[float, float]] = {

        SensorType.TEMPERATURE: (20, 40),

        SensorType.HUMIDITY: (30, 80)

    }



    def __init__(self, device_id: str, sensor_type: SensorType):

        super().__init__(daemon=True)

        self.device_id = device_id

        self.sensor_type = sensor_type

        self._stop_event = threading.Event()

        self._lock = threading.Lock()



    def run(self):

        """模拟设备周期性发送传感器数据"""

        pub.subscribe(self._handle_command, f"devices/{self.device_id}/command")

        

        try:

            while not self._stop_event.is_set():

                min_val, max_val = self.SENSOR_RANGES[self.sensor_type]

                data = TelemetryData(

                    device_id=self.device_id,

                    sensor_type=self.sensor_type,

                    value=random.uniform(min_val, max_val)

                )

                # print("data=", data)

                pub.sendMessage(f"devices/{self.device_id}/telemetry", data=data)

                self._stop_event.wait(3)  # 可中断的等待

        except Exception as e:

            logger.error(f"设备 {self.device_id} 运行异常: {str(e)}")



    def _handle_command(self, cmd: Command):

        """处理来自云端的指令"""

        with self._lock:

            logger.info(f"[设备 {self.device_id}] 执行 {cmd.cmd_type} 命令, 参数: {cmd.params}")



    def stop(self):

        """安全停止设备"""

        self._stop_event.set()



# -------------------------- 告警引擎 --------------------------

# -------------------------- 告警引擎 --------------------------

class AlertEngine:

    def __init__(self):

        # 添加调试日志验证订阅状态

        print("\033[36m[系统初始化] AlertEngine 已启动，正在监听设备数据...\033[0m")

        pub.subscribe(self._check_temperature, "devices/DEV001/telemetry")

        

    def _check_temperature(self, data: TelemetryData):

        # 添加调试日志

        print(f"\033[36m[DEBUG] 收到 {data.device_id} 的温度数据: {data.value}℃\033[0m")

        if data.sensor_type == "temperature" and data.value > 25:

            pub.sendMessage("alerts/high_temperature", 

                msg=f"\033[31m设备 {data.device_id} 温度超标! 当前值: {data.value}℃\033[0m")



# -------------------------- 增强型云端服务 --------------------------

class CloudService:

    def __init__(self):

        pub.subscribe(self._log_telemetry, "devices/DEV001/telemetry")

        pub.subscribe(self._handle_alerts, "alerts/high_temperature")

        self._command_queue = Queue()

        self._worker = threading.Thread(target=self._process_commands, daemon=True)

        self._worker.start()



    def _log_telemetry(self, data: TelemetryData):

        """数据持久化到云端数据库"""

        logger.info(f"[cloud 数据采集] {data.device_id} {data.sensor_type.value}={data.value:.2f}")



    def _handle_alerts(self, msg):

        """处理所有告警事件"""

        logger.warning(f"[cloud 告警处理] {msg}")



    def send_command(self, device_id: str, cmd_type: CommandType, params: dict):

        """异步发送设备控制指令"""

        self._command_queue.put(Command(device_id, cmd_type, params))



    def _process_commands(self):

        """后台线程处理命令队列"""

        while True:

            cmd = self._command_queue.get()

            try:

                pub.sendMessage(f"devices/{cmd.device_id}/command", cmd=cmd)

                logger.info(f"[cloud 指令发送] {cmd.device_id} {cmd.cmd_type.value}")

            except Exception as e:

                logger.error(f"命令发送失败: {str(e)}")



# -------------------------- 线程安全设备管理 --------------------------

class DeviceManager:

    def __init__(self):

        self.devices: Dict[str, VirtualDevice] = {}

        self._lock = threading.Lock()



    def add_device(self, device_id: str, sensor_type: SensorType):

        """动态添加新设备"""

        with self._lock:

            if device_id not in self.devices:

                device = VirtualDevice(device_id, sensor_type)

                device.start()

                self.devices[device_id] = device

                logger.info(f"设备 {device_id} 已上线")



    def remove_device(self, device_id: str):

        """安全移除设备"""

        with self._lock:

            if device_id in self.devices:

                self.devices[device_id].stop()

                del self.devices[device_id]

                logger.info(f"设备 {device_id} 已下线")



# -------------------------- 主程序 --------------------------

if __name__ == "__main__":

    # 初始化核心组件

    alert_engine = AlertEngine()

    cloud = CloudService()

    device_manager = DeviceManager()



    



    # 创建初始设备

    device_manager.add_device("DEV001", SensorType.TEMPERATURE)

    device_manager.add_device("DEV002", SensorType.HUMIDITY)



    # 动态设备管理演示

    threading.Timer(8, lambda: device_manager.add_device("DEV003", SensorType.TEMPERATURE)).start()

    threading.Timer(5, lambda: cloud.send_command("DEV001", CommandType.REBOOT, {"force": True})).start()



    # 运行演示

    try:

        time.sleep(60)

    finally:

        device_manager.remove_device("DEV001")

        device_manager.remove_device("DEV002")

复制代码

五、总结

本文实现了IoT平台模拟器，它展示了如何使用PubSub模式、多线程、数据类和枚举类型构建一个可扩展、灵活和易于维护的系统。通过分析代码的设计思想和关键组件，可以更好地理解IoT平台的核心概念和技术，为开发更复杂的IoT应用打下基础。该代码也体现了面向对象编程的思想，将系统划分为多个独立的组件，每个组件负责特定的功能，并通过接口进行交互，降低了系统的复杂性。希望本文能够帮助读者更好地理解物联网设备的工作原理及其背后的设计思想。

写原创有奖励！2025面包板原创奖励正在进行中

 举报



 

  0