物联网设备模拟器
基于MQTT协议的物联网设备模拟器，用于开发和测试物联网平台。支持多种设备类型的模拟，包括传感器和执行器设备。
功能特性
- 多种设备类型：支持温度传感器、湿度传感器、气压传感器、智能开关、智能灯泡、运动传感器等
- MQTT通信：基于MQTT协议进行设备通信，支持QoS、保留消息、通配符主题等高级特性
- 设备管理：支持批量设备创建、分组管理、状态监控等功能
- 灵活配置：通过YAML配置文件轻松配置设备参数和MQTT连接
- 完善日志：支持多级别日志记录，文件轮转，便于调试和监控
- 扩展性强：基于面向对象设计，易于添加新的设备类型
- 命令行界面：提供友好的命令行接口，支持多种运行模式
项目结构
物联网设备模拟器/
├── main.py # 主程序入口
├── config.yaml # 配置文件
├── requirements.txt # Python依赖包
├── README.md # 项目说明文档
├── src/ # 源代码目录
│ ├── __init__.py
│ ├── mqtt_client.py # MQTT客户端封装
│ ├── device_base.py # 设备基类和接口
│ ├── devices.py # 具体设备实现
│ ├── device_manager.py # 设备管理器
│ └── config_manager.py # 配置和日志管理
└── examples/ # 示例脚本
├── basic_usage.py # 基本使用示例
└── mqtt_test.py # MQTT连接测试
快速开始
1. 环境要求
- Python 3.7+
- MQTT服务器（如Mosquitto、EMQX等）
2. 安装依赖
# 克隆或下载项目到本地
cd 物联网设备模拟器
# 安装Python依赖
pip install -r requirements.txt
3. 配置MQTT服务器
使用Mosquitto（推荐）
Windows:
# 下载并安装Mosquitto
# 从 https://mosquitto.org/download/ 下载Windows版本
# 启动Mosquitto服务器
mosquitto -v
Linux/macOS:
# 安装Mosquitto
sudo apt-get install mosquitto mosquitto-clients # Ubuntu/Debian
brew install mosquitto # macOS
# 启动Mosquitto服务器
mosquitto -v
使用Docker
# 运行Mosquitto容器
docker run -it -p 1883:1883 eclipse-mosquitto:latest
4. 测试MQTT连接
在运行主程序之前，建议先测试MQTT连接：
# 测试MQTT连接
python examples/mqtt_test.py
# 测试特定功能
python examples/mqtt_test.py --test connection
python examples/mqtt_test.py --test pubsub
# 使用自定义MQTT服务器
python examples/mqtt_test.py -b your-mqtt-broker -p 1883
5. 运行模拟器
# 使用默认配置运行
python main.py
# 使用自定义配置文件
python main.py -c custom_config.yaml
# 仅验证配置文件
python main.py --validate
# 显示当前配置
python main.py --show-config
配置说明
配置文件结构
config.yaml 文件包含以下主要配置项：
# MQTT服务器配置
mqtt:
broker: "localhost" # MQTT服务器地址
port: 1883 # MQTT服务器端口
username: "" # 用户名（可选）
password: "" # 密码（可选）
keepalive: 60 # 心跳间隔（秒）
qos: 1 # 默认QoS级别
clean_session: true # 清理会话
# 设备配置
devices:
temperature_sensors: # 温度传感器
count: 3 # 设备数量
topic_prefix: "sensors/temperature" # 主题前缀
publish_interval: 5 # 发布间隔（秒）
min_value: -10 # 最小值
max_value: 50 # 最大值
humidity_sensors: # 湿度传感器
count: 2
topic_prefix: "sensors/humidity"
publish_interval: 10
min_value: 0
max_value: 100
smart_switches: # 智能开关
count: 5
topic_prefix: "devices/switch"
publish_interval: 30
smart_lights: # 智能灯泡
count: 4
topic_prefix: "devices/light"
publish_interval: 15
# 日志配置
logging:
level: "INFO" # 日志级别
format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
file: "device_simulator.log" # 日志文件
max_size: 10485760 # 最大文件大小（字节）
backup_count: 5 # 备份文件数量
# 模拟器配置
simulator:
auto_start: true # 自动启动
random_seed: null # 随机种子（可选）
data_variation: 0.1 # 数据变化幅度
设备类型说明
| 设备类型 |
说明 |
数据类型 |
支持命令 |
temperature_sensors |
温度传感器 |
温度值（°C） |
无 |
humidity_sensors |
湿度传感器 |
湿度值（%） |
无 |
pressure_sensors |
气压传感器 |
气压值（hPa） |
无 |
smart_switches |
智能开关 |
开关状态 |
turn_on, turn_off, toggle |
smart_lights |
智能灯泡 |
亮度、颜色、状态 |
turn_on, turn_off, set_brightness, set_color |
motion_sensors |
运动传感器 |
运动检测状态 |
无 |
使用示例
基本使用
from src.device_manager import DeviceManager
from src.devices import TemperatureSensor, SmartSwitch
# 创建设备管理器
manager = DeviceManager(broker="localhost", port=1883)
# 创建设备
temp_sensor = TemperatureSensor("temp_001", "sensors/temperature", 5)
smart_switch = SmartSwitch("switch_001", "devices/switch", 30)
# 添加设备
manager.add_device(temp_sensor)
manager.add_device(smart_switch)
# 连接并启动
manager.connect()
manager.start_all_devices()
# 发送命令
manager.send_command_to_device("switch_001", {"action": "turn_on"})
运行示例脚本
# 基本使用示例
python examples/basic_usage.py basic
# 高级使用示例
python examples/basic_usage.py advanced
# 交互式示例
python examples/basic_usage.py interactive
MQTT主题结构
数据发布主题
设备会定期向以下主题发布数据：
{topic_prefix}/{device_id}/data
例如：
sensors/temperature/temp_001/datadevices/switch/switch_001/data
状态发布主题
设备状态变化时会发布到：
{topic_prefix}/{device_id}/status
命令订阅主题
设备会订阅以下主题接收命令：
{topic_prefix}/{device_id}/command
消息格式
数据消息
{
"device_id": "temp_001",
"device_type": "TemperatureSensor",
"timestamp": 1640995200.123,
"data": {
"temperature": 23.5,
"unit": "°C"
}
}
状态消息
{
"device_id": "switch_001",
"device_type": "SmartSwitch",
"timestamp": 1640995200.123,
"status": "online",
"state": {
"power": true
}
}
命令消息
{
"action": "turn_on",
"parameters": {
"brightness": 80
}
}
扩展开发
添加新设备类型
- 继承
SensorDevice 或 ActuatorDevice 基类
- 实现
generate_data() 方法
- 如果是执行器，实现
handle_command() 方法
from src.device_base import SensorDevice
class CustomSensor(SensorDevice):
def __init__(self, device_id: str, topic_prefix: str, publish_interval: int):
super().__init__(device_id, topic_prefix, publish_interval)
def generate_data(self) -> dict:
return {
"custom_value": self._generate_random_value(0, 100),
"unit": "custom_unit"
}
自定义MQTT处理
from src.mqtt_client import MQTTClient
def custom_message_handler(topic: str, payload: str, qos: int):
print(f"收到消息: {topic} -> {payload}")
client = MQTTClient("custom_client")
client.set_on_message_callback(custom_message_handler)
client.connect()
client.subscribe("custom/topic", qos=1)
故障排除
常见问题
-
MQTT连接失败
- 检查MQTT服务器是否运行
- 验证服务器地址和端口
- 检查用户名密码（如果需要）
-
设备无法启动
-
消息收发异常
调试技巧
-
启用调试日志
logging:
level: "DEBUG"
-
使用MQTT测试工具
# 订阅所有消息
mosquitto_sub -h localhost -t "#" -v
# 发布测试消息
mosquitto_pub -h localhost -t "test/topic" -m "test message"
-
检查日志文件
tail -f device_simulator.log
性能优化
大规模设备模拟
- 调整发布间隔：增加
publish_interval 减少消息频率
- 使用QoS 0：对于不重要的数据使用QoS 0
- 批量操作：使用设备分组功能进行批量控制
- 资源监控：监控CPU和内存使用情况
网络优化
- 消息压缩：对于大量数据考虑消息压缩
- 连接池：复用MQTT连接
- 本地缓存：缓存频繁访问的数据
许可证
本项目采用 MIT 许可证。详见 LICENSE 文件。
贡献
欢迎提交 Issue 和 Pull Request 来改进这个项目。
联系方式
开源仓库: https://github.com/xsxz01/iot-device-emulator
如有问题或建议，请通过以下方式联系：
- 提交 GitHub Issue
- 发送邮件至项目维护者
注意：这是一个用于开发和测试的模拟器，不建议在生产环境中使用。在生产环境中，请使用真实的物联网设备。
/2 
粤公网安备 44522102000125 增值电信业务经营许可证 粤B2-20192173
发表于 
主题置顶卡
沉默卡
变色卡
楼主
扫一扫，关注易界动态