zhaoziyi/jiuxing_analysis
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

玖行换电站协议

Browse Source
This commit is contained in:
MATRIX\29620 2025-03-07 10:54:05 +08:00
parent 5f53e0edff
commit 68b599a660
28 changed files with 312 additions and 704 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

164
battery_swap_station/README.md
View File

@ -1,150 +1,20 @@
# 换电站 MQTT 协议解析器
# Swap Station Project
A PyCharm project for connecting to a swap station via MQTT on Windows.
该软件包提供了一个基于 Python 的换电站 MQTT 协议实现，符合 V2.6.2 规范。
## Structure
- `src/common/`: Shared configurations and encryption utilities.
- `src/key_management/`: Handles key exchange and AES key storage.
- `src/state_receiver/`: Receives state information messages.
- `src/event_receiver/`: Receives event record messages.
- `data/`: Stores AES keys (aes_key.bin, aes_iv.bin).
- `logs/`: Optional log files.
## 主要功能
## Setup
1. Install EMQX and configure authentication.
2. Run `pip install -r requirements.txt` in the PyCharm terminal.
3. Replace the platform public key in `common.py`.
4. Run `key_management.py` first, then `state_receiver.py` and `event_receiver.py`.
- 完整实现所有消息类型（状态类、事件类、请求/响应类）
- 支持加密功能（RSA/AES）
- 全面的消息字段验证
- 简单易用的消息创建和解析 API
- 完整的测试覆盖
## 安装方法
```bash
pip install -e .
```
## 快速开始
下面是创建和解析站点状态消息的简单示例：
```python
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 运行测试：
```bash
pytest tests/
```
## 示例代码
查看 `examples/` 目录中的使用示例：
```bash
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 许可证
## Usage
- Use PyCharm's "Run Configurations" to execute each module.
- Simulate messages with MQTTX or a test script.

0
battery_swap_station/config/__init__.py
View File

0
battery_swap_station/config/config.json
View File

0
battery_swap_station/config/logging_config.json
View File

BIN
battery_swap_station/docs/玖行充换电云平台与站控系统数据接口协议（MQTT）V2.6.2.pdf
View File

Binary file not shown.

0
battery_swap_station/examples/__init__.py
View File

79
battery_swap_station/examples/event_example.py
View File

@ -1,79 +0,0 @@
"""
Examples of using event messages
"""
from datetime import datetime
from mqtt_protocol import ChargeRecord, SwapRecord
from mqtt_protocol.event_messages import RateDetail, ChargeStartType, SwapType, SwapMode
def charge_record_example():
# Create rate details
rate_details = [
RateDetail(rateType=1, startTime="2024-02-24 10:00:00",
stopTime="2024-02-24 12:00:00", elect=50.5),
RateDetail(rateType=2, startTime="2024-02-24 12:00:00",
stopTime="2024-02-24 14:00:00", elect=45.8)
]
# Create a charge record
charge_record = ChargeRecord.create(
equipNo="CHG001",
orderSn="CR202402240001",
gunNo="1",
startTime="2024-02-24 10:00:00",
stopTime="2024-02-24 14:00:00",
startSOC=20.5,
endSOC=85.3,
chgQty=96.3,
startMeter=1000.0,
endMeter=1096.3,
plateNo="京A12345",
vin="LSVAA12345G123456",
stopReason=1,
sharpElect=50.5,
peakElect=45.8,
flatElect=0.0,
valleyElect=0.0,
rateModelID="RATE001",
detailsList=rate_details,
swapSn="",
chgSwitch=0,
startType=ChargeStartType.STATION_AUTO
)
print("Charge Record Example:")
print(charge_record.to_json())
def swap_record_example():
# Create a swap record
swap_record = SwapRecord(
equipNo="ROBOT001",
orderSn="SR202402240001",
startTime="2024-02-24 15:00:00",
stopTime="2024-02-24 15:10:00",
vin="LSVAA12345G123456",
rfidCode="123456789012345678901234",
plateNo="京A12345",
newBatID="BAT001",
newCabinetNo=1,
newBatSoc=95.5,
oldBatID="BAT002",
oldCabinetNo=2,
oldBatSoc=15.5,
swapType=SwapType.BATTERY_SWAP,
swapMode=SwapMode.FULL_AUTO,
swapStartType=1,
lane=1,
userID="USER001",
stationMode=1
)
print("\nSwap Record Example:")
print(swap_record.to_json())
if __name__ == "__main__":
charge_record_example()
swap_record_example()

103
battery_swap_station/examples/request_example.py
View File

@ -1,103 +0,0 @@
"""
Examples of using request and response messages
"""
from mqtt_protocol import (
SwapStartRequest,
SwapStartResponse,
ChargeStartRequest,
ChargeStartResponse
)
from mqtt_protocol.request_response import (
RateModelQuery,
RateModelResponse,
RateInfo,
RateTimeSegment
)
def swap_request_example():
# Create a swap start request
swap_req = SwapStartRequest(
vin="LSVAA12345G123456",
rfid_code="123456789012345678901234",
plate_no="京A12345",
lane=1,
user_id="USER001"
)
print("Swap Start Request Example:")
print(swap_req.to_json())
# Create a response
swap_resp = SwapStartResponse(
vin="LSVAA12345G123456",
rfid_code="123456789012345678901234",
plate_no="京A12345",
result=1, # Success
order_sn="SR202402240001"
)
print("\nSwap Start Response Example:")
print(swap_resp.to_json())
def charge_request_example():
# Create a charge start request
charge_req = ChargeStartRequest(
chg_id="CHG001",
gun_no=1
)
print("\nCharge Start Request Example:")
print(charge_req.to_json())
# Create a response
charge_resp = ChargeStartResponse(
result=1, # Success
chg_id="CHG001",
gun_no=1,
order_sn="CR202402240001"
)
print("\nCharge Start Response Example:")
print(charge_resp.to_json())
def rate_model_example():
# Create rate model query
rate_query = RateModelQuery("RATE001")
print("\nRate Model Query Example:")
print(rate_query.to_json())
# Create rate info list
rate_list = [
RateInfo(rateType=1, electPrice=1.2, servicePrice=0.5),
RateInfo(rateType=2, electPrice=1.0, servicePrice=0.4),
RateInfo(rateType=3, electPrice=0.8, servicePrice=0.3),
RateInfo(rateType=4, electPrice=0.6, servicePrice=0.2)
]
# Create rate time segments
time_segments = [
RateTimeSegment(rateType=1, index=1,
startTime="10:00:00", stopTime="12:00:00"),
RateTimeSegment(rateType=2, index=2,
startTime="12:00:00", stopTime="14:00:00")
]
# Create response
rate_resp = RateModelResponse(
rate_model_id="RATE001",
rate_list=rate_list,
rate_details_list=time_segments
)
print("\nRate Model Response Example:")
print(rate_resp.to_json())
if __name__ == "__main__":
swap_request_example()
charge_request_example()
rate_model_example()

68
battery_swap_station/examples/state_example.py
View File

@ -1,68 +0,0 @@
"""
Constants used throughout the MQTT protocol implementation
"""
# Protocol version
PROTOCOL_VERSION = "V2.6.2"
# Topic related constants
TOPIC_PREFIX = "HCMS" # Heavy truck Charging Management System
TOPIC_SEPARATOR = "/"
# Message directions
M2S = "M2S" # Master to Slave (Cloud Platform to Station)
S2M = "S2M" # Slave to Master (Station to Cloud Platform)
# Message types
STATE = "state"
EVENT = "event"
CONFIRM = "confirm"
REQUEST = "request"
RESPONSE = "response"
KEEPALIVE = "keepalive"
ENCRYPT_KEY_REQ = "encryptKeyReq"
ENCRYPT_KEY_RESP = "encryptKeyResp"
# Station modes
STATION_MODE_OPERATION = 1
STATION_MODE_DEBUG = 2
STATION_MODE_MAINTENANCE = 3
# Device states
DEVICE_STATE_UNKNOWN = 0
DEVICE_STATE_NORMAL = 1
DEVICE_STATE_ALARM = 2
# Charging states
CHARGING_STATE_STANDBY = 1
CHARGING_STATE_CHARGING = 2
CHARGING_STATE_COMPLETE = 3
CHARGING_STATE_FAULT = 4
CHARGING_STATE_OFFLINE = 5
# Swap states
SWAP_STATE_NOT_STARTED = 1
SWAP_STATE_STARTED = 2
SWAP_STATE_PAUSED = 3
SWAP_STATE_RESUMED = 4
SWAP_STATE_TERMINATED = 5
SWAP_STATE_COMPLETED = 6
# Rate types
RATE_TYPE_SHARP = 1
RATE_TYPE_PEAK = 2
RATE_TYPE_FLAT = 3
RATE_TYPE_VALLEY = 4
# Lane types
LANE_TYPE_SINGLE = 1
LANE_TYPE_DOUBLE = 2
# Error codes
ERROR_SUCCESS = 1
ERROR_FAILURE = 2
# Timeouts (in seconds)
KEEPALIVE_INTERVAL = 30
EVENT_RETRY_INTERVAL = 15
COMMUNICATION_TIMEOUT = 90 # 3 * KEEPALIVE_INTERVAL

40
battery_swap_station/mqtt/__init__.py
View File

@ -1,40 +0,0 @@
"""
MQTT Protocol Parser for Battery Swap Station
"""
__version__ = "1.0.0"
from .base_message import BaseMessage, MessageHeader, MessageType, MessageDirection
from .state_messages import (
StationInfo,
StationStateMessage,
RobotStateMessage
)
from .event_messages import (
ChargeRecord,
SwapRecord
)
from .request_response import (
SwapStartRequest,
SwapStartResponse,
ChargeStartRequest,
ChargeStartResponse
)
from .encryption_handler import EncryptionHandler
__all__ = [
'BaseMessage',
'MessageHeader',
'MessageType',
'MessageDirection',
'StationInfo',
'StationStateMessage',
'RobotStateMessage',
'ChargeRecord',
'SwapRecord',
'SwapStartRequest',
'SwapStartResponse',
'ChargeStartRequest',
'ChargeStartResponse',
'EncryptionHandler'
]

0
battery_swap_station/mqtt/base_message.py
View File

0
battery_swap_station/mqtt/encryption_handler.py
View File

0
battery_swap_station/mqtt/event_messages.py
View File

0
battery_swap_station/mqtt/request_response.py
View File

0
battery_swap_station/mqtt/state_messages.py
View File

0
battery_swap_station/mqtt/utils/__init__.py
View File

68
battery_swap_station/mqtt/utils/constants.py
View File

@ -1,68 +0,0 @@
"""
Constants used throughout the MQTT protocol implementation
"""
# Protocol version
PROTOCOL_VERSION = "V2.6.2"
# Topic related constants
TOPIC_PREFIX = "HCMS" # Heavy truck Charging Management System
TOPIC_SEPARATOR = "/"
# Message directions
M2S = "M2S" # Master to Slave (Cloud Platform to Station)
S2M = "S2M" # Slave to Master (Station to Cloud Platform)
# Message types
STATE = "state"
EVENT = "event"
CONFIRM = "confirm"
REQUEST = "request"
RESPONSE = "response"
KEEPALIVE = "keepalive"
ENCRYPT_KEY_REQ = "encryptKeyReq"
ENCRYPT_KEY_RESP = "encryptKeyResp"
# Station modes
STATION_MODE_OPERATION = 1
STATION_MODE_DEBUG = 2
STATION_MODE_MAINTENANCE = 3
# Device states
DEVICE_STATE_UNKNOWN = 0
DEVICE_STATE_NORMAL = 1
DEVICE_STATE_ALARM = 2
# Charging states
CHARGING_STATE_STANDBY = 1
CHARGING_STATE_CHARGING = 2
CHARGING_STATE_COMPLETE = 3
CHARGING_STATE_FAULT = 4
CHARGING_STATE_OFFLINE = 5
# Swap states
SWAP_STATE_NOT_STARTED = 1
SWAP_STATE_STARTED = 2
SWAP_STATE_PAUSED = 3
SWAP_STATE_RESUMED = 4
SWAP_STATE_TERMINATED = 5
SWAP_STATE_COMPLETED = 6
# Rate types
RATE_TYPE_SHARP = 1
RATE_TYPE_PEAK = 2
RATE_TYPE_FLAT = 3
RATE_TYPE_VALLEY = 4
# Lane types
LANE_TYPE_SINGLE = 1
LANE_TYPE_DOUBLE = 2
# Error codes
ERROR_SUCCESS = 1
ERROR_FAILURE = 2
# Timeouts (in seconds)
KEEPALIVE_INTERVAL = 30
EVENT_RETRY_INTERVAL = 15
COMMUNICATION_TIMEOUT = 90 # 3 * KEEPALIVE_INTERVAL

68
battery_swap_station/mqtt/utils/validators.py
View File

@ -1,68 +0,0 @@
"""
Constants used throughout the MQTT protocol implementation
"""
# Protocol version
PROTOCOL_VERSION = "V2.6.2"
# Topic related constants
TOPIC_PREFIX = "HCMS" # Heavy truck Charging Management System
TOPIC_SEPARATOR = "/"
# Message directions
M2S = "M2S" # Master to Slave (Cloud Platform to Station)
S2M = "S2M" # Slave to Master (Station to Cloud Platform)
# Message types
STATE = "state"
EVENT = "event"
CONFIRM = "confirm"
REQUEST = "request"
RESPONSE = "response"
KEEPALIVE = "keepalive"
ENCRYPT_KEY_REQ = "encryptKeyReq"
ENCRYPT_KEY_RESP = "encryptKeyResp"
# Station modes
STATION_MODE_OPERATION = 1
STATION_MODE_DEBUG = 2
STATION_MODE_MAINTENANCE = 3
# Device states
DEVICE_STATE_UNKNOWN = 0
DEVICE_STATE_NORMAL = 1
DEVICE_STATE_ALARM = 2
# Charging states
CHARGING_STATE_STANDBY = 1
CHARGING_STATE_CHARGING = 2
CHARGING_STATE_COMPLETE = 3
CHARGING_STATE_FAULT = 4
CHARGING_STATE_OFFLINE = 5
# Swap states
SWAP_STATE_NOT_STARTED = 1
SWAP_STATE_STARTED = 2
SWAP_STATE_PAUSED = 3
SWAP_STATE_RESUMED = 4
SWAP_STATE_TERMINATED = 5
SWAP_STATE_COMPLETED = 6
# Rate types
RATE_TYPE_SHARP = 1
RATE_TYPE_PEAK = 2
RATE_TYPE_FLAT = 3
RATE_TYPE_VALLEY = 4
# Lane types
LANE_TYPE_SINGLE = 1
LANE_TYPE_DOUBLE = 2
# Error codes
ERROR_SUCCESS = 1
ERROR_FAILURE = 2
# Timeouts (in seconds)
KEEPALIVE_INTERVAL = 30
EVENT_RETRY_INTERVAL = 15
COMMUNICATION_TIMEOUT = 90 # 3 * KEEPALIVE_INTERVAL

9
battery_swap_station/requirements.txt
View File

@ -1,6 +1,3 @@
pycryptodome>=3.15.0
pytest>=7.0.0
paho-mqtt>=1.6.1
dataclasses>=0.6
typing-extensions>=4.0.0
python-dateutil>=2.8.2
# requirements.txt
paho-mqtt==1.6.1
cryptography==41.0.7

15
battery_swap_station/setup.py
View File

@ -1,15 +0,0 @@
from setuptools import setup, find_packages
setup(
name="mqtt_protocol",
version="1.0.0",
packages=find_packages(),
install_requires=[
"pycryptodome>=3.15.0",
],
author="Your Name",
author_email="your.email@example.com",
description="MQTT Protocol Parser for Battery Swap Station",
long_description=open("README.md").read(),
long_description_content_type="text/markdown",
)

79
battery_swap_station/src/event.py Normal file
View File

@ -0,0 +1,79 @@
import paho.mqtt.client as mqtt
import json
from datetime import datetime
# MQTT Broker 配置
BROKER = "192.168.8.139" # 替换为你的 MQTT Broker 地址
PORT = 1883
SITE_CODE = "124127"
TOPIC_EVENT = f"HCMS/{SITE_CODE}/S2M/event"
TOPIC_KEEPALIVE_IN = f"HCMS/{SITE_CODE}/S2M/keepalive"
TOPIC_KEEPALIVE_OUT = f"HCMS/{SITE_CODE}/M2S/keepalive"
TOPIC_CONFIRM = f"HCMS/{SITE_CODE}/M2S/confirm"
# 连接回调
def on_connect(client, userdata, flags, rc):
print(f"Connected to MQTT Broker with result code {rc}")
client.subscribe([(TOPIC_EVENT, 0), (TOPIC_KEEPALIVE_IN, 0)])
print(f"Subscribed to {TOPIC_EVENT} and {TOPIC_KEEPALIVE_IN}")
# 消息处理回调
def on_message(client, userdata, msg):
payload = msg.payload.decode()
topic = msg.topic
print(f"Received: {topic} - {payload}")
if topic == TOPIC_KEEPALIVE_IN:
# 处理心跳报文
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
client.publish(TOPIC_KEEPALIVE_OUT, current_time, qos=0)
print(f"Sent keepalive reply: {TOPIC_KEEPALIVE_OUT} - {current_time}")
elif topic == TOPIC_EVENT:
try:
data = json.loads(payload)
header = data.get("header", {})
data_body = data.get("dataBody", {})
function = header.get("function", "unknown")
index = header.get("index", 0)
order_sn = data_body.get("orderSn", "unknown")
# 构造确认报文
confirm_payload = {
"header": {
"version": "V2.0",
"timeStamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"index": index,
"function": f"{function}Conf"
},
"dataBody": {
"orderSn": order_sn,
"result": 1
}
}
# 发送确认
client.publish(TOPIC_CONFIRM, json.dumps(confirm_payload), qos=0)
print(f"Sent confirmation: {TOPIC_CONFIRM} - {json.dumps(confirm_payload)}")
except json.JSONDecodeError:
print("Failed to parse JSON payload")
except Exception as e:
print(f"Error processing message: {e}")
# 主程序
def main():
client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, PORT, 60)
client.loop_forever()
if __name__ == "__main__":
main()

136
battery_swap_station/src/model.py Normal file
View File

@ -0,0 +1,136 @@
import paho.mqtt.client as mqtt
import json
import requests
from datetime import datetime
BROKER = "192.168.8.139"
PORT = 1883
SITE_CODE = "124127"
TOPIC_REQUEST = f"HCMS/{SITE_CODE}/M2S/request"
TOPIC_RESPONSE = f"HCMS/{SITE_CODE}/S2M/response"
# HTTP 接口配置（示例）
RATE_API_URL = "http://example.com/api/rate_model" # 替换为实际接口地址
def fetch_rate_model_from_api():
try:
response = requests.get(RATE_API_URL)
response.raise_for_status()
data = response.json()
return {
"rateModelID": data.get("rateModelID", "default_id"),
"rateList": data.get("rateList", []),
"rateDetailsList": data.get("rateDetailsList", [])
}
except requests.RequestException as e:
print(f"Failed to fetch rate model from API: {e}")
return None
# 连接回调
def on_connect(client, userdata, flags, rc):
if rc == 0:
print("Connected to MQTT Broker successfully")
client.subscribe(TOPIC_RESPONSE)
print(f"Subscribed to {TOPIC_RESPONSE}")
else:
print(f"Connection failed with code {rc}")
# 消息处理回调
def on_message(client, userdata, msg):
payload = msg.payload.decode()
topic = msg.topic
print(f"Received: {topic} - {payload}")
try:
data = json.loads(payload)
header = data.get("header", {})
data_body = data.get("dataBody", {})
function = header.get("function", "unknown")
if function == "rateModeSyncResp":
result = data_body.get("result", 2)
reason = data_body.get("reason", "No reason provided")
print(f"Rate Mode Sync Response: Result={result}, Reason={reason}")
elif function == "checkRateModelResp":
rate_model_id = data_body.get("rateModelID", "unknown")
rate_list = data_body.get("rateList", [])
rate_details_list = data_body.get("rateDetailsList", [])
print(f"Rate Model Query Response: ID={rate_model_id}")
print(f"Rate List: {json.dumps(rate_list, indent=2)}")
print(f"Rate Details List: {json.dumps(rate_details_list, indent=2)}")
except json.JSONDecodeError:
print("Failed to parse JSON payload")
# 发送费率模型同步请求
def send_rate_model_sync(client, rate_model_id, rate_list, rate_details_list):
request_payload = {
"header": {
"version": "V2.0",
"timeStamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"index": 1,
"function": "rateModeSyncReq"
},
"dataBody": {
"rateModelID": rate_model_id,
"rateList": rate_list,
"rateDetailsList": rate_details_list
}
}
client.publish(TOPIC_REQUEST, json.dumps(request_payload), qos=0)
print(f"Sent rate model sync request: {json.dumps(request_payload)}")
# 发送费率模型查询请求
def send_rate_model_query(client, rate_model_id):
request_payload = {
"header": {
"version": "V2.0",
"timeStamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"index": 2,
"function": "checkRateModelReq"
},
"dataBody": {
"rateModelID": rate_model_id
}
}
client.publish(TOPIC_REQUEST, json.dumps(request_payload), qos=0)
print(f"Sent rate model query request: {json.dumps(request_payload)}")
# 主程序
def main():
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.on_connect = on_connect
client.on_message = on_message
try:
client.connect(BROKER, PORT, 60)
# 从 HTTP 接口获取费率数据
rate_data = fetch_rate_model_from_api()
if rate_data:
send_rate_model_sync(client, rate_data["rateModelID"],
rate_data["rateList"], rate_data["rateDetailsList"])
send_rate_model_query(client, rate_data["rateModelID"])
else:
print("Using default rate data due to API failure")
rate_model_id = "default_id"
rate_list = [{"rateType": 1, "electPrice": 1.0, "servicePrice": 0.3}]
rate_details_list = [{"rateType": 1, "startTime": "2025-03-05 00:00:00", "stopTime": "2025-03-05 23:59:59"}]
send_rate_model_sync(client, rate_model_id, rate_list, rate_details_list)
send_rate_model_query(client, rate_model_id)
client.loop_forever()
except Exception as e:
print(f"Failed to connect to {BROKER}:{PORT}: {e}")
if __name__ == "__main__":
main()

76
battery_swap_station/src/state.py Normal file
View File

@ -0,0 +1,76 @@
import paho.mqtt.client as mqtt
import json
from datetime import datetime
# MQTT Broker 配置
BROKER = "192.168.8.139" # 替换为你的 MQTT Broker 地址
PORT = 1883
SITE_CODE = "124127"
TOPIC_STATE = f"HCMS/{SITE_CODE}/S2M/state"
TOPIC_KEEPALIVE_IN = f"HCMS/{SITE_CODE}/S2M/keepalive"
TOPIC_KEEPALIVE_OUT = f"HCMS/{SITE_CODE}/M2S/keepalive"
TOPIC_REQUEST = f"HCMS/{SITE_CODE}/M2S/request"
# 连接回调
def on_connect(client, userdata, flags, rc):
print(f"Connected to MQTT Broker with result code {rc}")
client.subscribe([(TOPIC_STATE, 0), (TOPIC_KEEPALIVE_IN, 0)])
print(f"Subscribed to {TOPIC_STATE} and {TOPIC_KEEPALIVE_IN}")
# 消息处理回调
def on_message(client, userdata, msg):
payload = msg.payload.decode()
topic = msg.topic
print(f"Received: {topic} - {payload}")
if topic == TOPIC_KEEPALIVE_IN:
# 处理心跳报文
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
client.publish(TOPIC_KEEPALIVE_OUT, current_time, qos=0)
print(f"Sent keepalive reply: {TOPIC_KEEPALIVE_OUT} - {current_time}")
elif topic == TOPIC_STATE:
try:
data = json.loads(payload)
header = data.get("header", {})
function = header.get("function", "unknown")
timestamp = header.get("timeStamp", "unknown")
index = header.get("index", 0)
print(f"State - Function: {function}, TimeStamp: {timestamp}, Index: {index}")
# 状态类报文无需回复，此处仅记录
except json.JSONDecodeError:
print("Failed to parse JSON payload")
# 发送状态召唤请求（可选）
def send_state_call(client, function):
request_payload = {
"header": {
"version": "V2.0",
"timeStamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"index": 1, # 可递增管理
"function": f"{function}Call"
}
}
client.publish(TOPIC_REQUEST, json.dumps(request_payload), qos=0)
print(f"Sent state call request: {function}Call")
# 主程序
def main():
client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, PORT, 60)
# 示例：发送 swapState 召唤请求（可选）
# send_state_call(client, "swapState")
client.loop_forever()
if __name__ == "__main__":
main()

0
battery_swap_station/tests/__init__.py
View File

0
battery_swap_station/tests/test_event.py
View File

0
battery_swap_station/tests/test_request.py
View File

109
battery_swap_station/tests/test_state.py
View File

@ -1,109 +0,0 @@
"""
Tests for state messages
"""
import pytest
from mqtt_protocol import (
StationStateMessage,
StationInfo,
RobotStateMessage
)
from mqtt_protocol.utils.constants import (
STATION_MODE_OPERATION,
DEVICE_STATE_NORMAL,
DEVICE_STATE_ALARM
)
def test_station_state_message():
# Test creation
state_msg = StationStateMessage.create(
state=STATION_MODE_OPERATION,
smoke="1,1,1,1",
fire=DEVICE_STATE_NORMAL,
temp=25,
humid=60,
totalElect=1000.5
)
# Test serialization
json_data = state_msg.to_json()
assert isinstance(json_data, str)
# Test deserialization
parsed_msg = StationStateMessage.from_json(json_data)
assert parsed_msg.state == STATION_MODE_OPERATION
assert parsed_msg.temp == 25
assert parsed_msg.humid == 60
assert parsed_msg.totalElect == 1000.5
def test_station_info():
info = StationInfo(
name="Test Station",
stationID="100001",
version="V2.6.2",
startDate="2024-02-24",
robotNum=2,
robotID="1234567890123456,1234567890123457",
chgNum=8,
chgID="CHG001,CHG002,CHG003,CHG004,CHG005,CHG006,CHG007,CHG008",
laneType=1,
rateModelID="RATE001",
chgVersion="V1.0"
)
assert info.name == "Test Station"
assert info.stationID == "100001"
assert info.robotNum == 2
assert info.chgNum == 8
def test_station_state_validation():
# Test invalid state
with pytest.raises(ValueError):
StationStateMessage.create(
state=999, # Invalid state
smoke="1,1,1,1",
fire=DEVICE_STATE_NORMAL,
temp=25,
humid=60,
totalElect=1000.5
)
# Test invalid smoke detector format
with pytest.raises(ValueError):
StationStateMessage.create(
state=STATION_MODE_OPERATION,
smoke="invalid",
fire=DEVICE_STATE_NORMAL,
temp=25,
humid=60,
totalElect=1000.5
)
def test_robot_state():
robot_msg = RobotStateMessage(
robotID="1234567890123456",
state=1,
fault=None,
runMode=1
)
assert robot_msg.robotID == "1234567890123456"
assert robot_msg.state == 1
assert robot_msg.fault is None
assert robot_msg.runMode == 1
def test_robot_state_with_fault():
robot_msg = RobotStateMessage(
robotID="1234567890123456",
state=3, # Fault state
fault="ERROR001,ERROR002",
runMode=1
)
assert robot_msg.state == 3
assert robot_msg.fault == "ERROR001,ERROR002"

2
charging_pile_proxy/README.md
View File

@ -31,7 +31,7 @@ server = ChargingPileProxyServer(
## 运行
```bash
python main.py
python
```
## 日志