jiuxing_analysis/battery_swap_station

换电站 MQTT 协议解析器

该软件包提供了一个基于 Python 的换电站 MQTT 协议实现，符合 V2.6.2 规范。

主要功能

• 完整实现所有消息类型（状态类、事件类、请求/响应类）
• 支持加密功能（RSA/AES）
• 全面的消息字段验证
• 简单易用的消息创建和解析 API
• 完整的测试覆盖

安装方法

pip install -e .

快速开始

下面是创建和解析站点状态消息的简单示例：

from mqtt_protocol import StationStateMessage

# 创建站点状态消息
state_msg = StationStateMessage.create(
    state=1,  # 运营状态
    smoke="1,1,1,1",  # 所有烟感正常
    fire=1,  # 消防系统正常
    temp=25,  # 温度
    humid=60,  # 湿度
    totalElect=1000.5  # 总电量
)

# 转换为 JSON
json_data = state_msg.to_json()

# 从 JSON 解析
parsed_msg = StationStateMessage.from_json(json_data)

消息类型

状态类消息

• 换电站属性信息
• 换电站状态
• 机器人状态
• 换电车辆状态
• 换电过程实时状态
• 电池状态
• 充电机状态

事件类消息

• 充电事件记录
• 换电事件记录
• 故障告警事件记录

请求/响应类消息

• 换电启动请求/响应
• 充电启动请求/响应
• 记录查询请求/响应
• 费率模型查询请求/响应

项目结构

mqtt_protocol/
├── README.md            # 项目说明文档
├── requirements.txt     # 项目依赖
├── setup.py            # 安装配置
├── docs/               # 文档目录
├── examples/           # 示例代码
├── tests/              # 测试用例
└── mqtt_protocol/      # 主代码目录
    ├── base_message.py     # 基础消息类
    ├── state_messages.py   # 状态类消息
    ├── event_messages.py   # 事件类消息
    ├── request_response.py # 请求响应类消息
    ├── encryption_handler.py # 加密处理
    └── utils/              # 工具类目录

运行测试

使用 pytest 运行测试：

pytest tests/

示例代码

查看 examples/ 目录中的使用示例：

python examples/state_example.py    # 状态消息示例
python examples/event_example.py    # 事件消息示例
python examples/request_example.py  # 请求响应示例

协议文档

详细的协议规范请参见 docs/protocol_spec.md。

系统要求

• Python 3.7+
• pycryptodome>=3.15.0（加密库）
• pytest>=7.0.0（测试用）
• paho-mqtt>=1.6.1（MQTT客户端）
• python-dateutil>=2.8.2（日期处理）

主要功能模块说明

1. 基础消息模块 (base_message.py)

• 定义了所有消息的基础类
• 包含消息头部处理
• 提供 JSON 序列化和反序列化功能

2. 状态消息模块 (state_messages.py)

• 实现各类状态信息的处理
• 包括站点状态、设备状态等
• 支持实时状态更新

3. 事件消息模块 (event_messages.py)

• 处理充电、换电等事件记录
• 支持事件确认机制
• 包含事件重试逻辑

4. 请求响应模块 (request_response.py)

• 实现所有请求响应类消息
• 支持同步和异步请求处理
• 包含超时和重试机制

5. 加密模块 (encryption_handler.py)

• 实现 RSA 和 AES 加密
• 提供密钥管理功能
• 支持安全通信

使用注意事项

1. 在使用前确保正确配置 MQTT 服务器连接信息
2. 注意处理所有可能的异常情况
3. 建议在生产环境中启用加密功能
4. 定期检查并处理消息重试队列

许可证

MIT 许可证