增加读取多个FsuCode,用于模拟多个不同的设备,配置文件里增加多个fsucode配置,详见config4.json
parent
a65572183c
commit
67dc39a5f0
|
@ -1,7 +1,7 @@
|
|||
{
|
||||
"client_id": "gw_simulator_888",
|
||||
"broker_ip": "172.16.100.23",
|
||||
"port": 8764,
|
||||
"broker_ip": "127.0.0.1",
|
||||
"port": 1883,
|
||||
"username": "hjemq-c2",
|
||||
"passwd": "Hj57471000",
|
||||
"publish_topic": "GateWayPublicTopic_Server",
|
||||
|
@ -10,6 +10,7 @@
|
|||
"publish_qos": 0,
|
||||
"subscribe_qos": 0,
|
||||
"subcriber_topic": "ServerPublicTopic_GateWay",
|
||||
"FsuCodes": ["11010110100111","11010110100112","11010110100113","11010110100114"],
|
||||
"publish_data": {
|
||||
"FsuCode": "11010110100999",
|
||||
"type": "gateway-data",
|
||||
|
|
|
@ -79,6 +79,7 @@ class MqttConfigure:
|
|||
self.subscriber_topic = ''
|
||||
self.client_id = f'python-mqtt-{random.randint(0, 1000)}'
|
||||
self.max_publish_count = 0
|
||||
self.fsuCodes = []
|
||||
|
||||
|
||||
class PublishData:
|
||||
|
@ -102,6 +103,8 @@ def read_configure(mqtt_conf=None, publish_data=None):
|
|||
mqtt_conf.subscribe_qos = json_data['subscribe_qos']
|
||||
mqtt_conf.subscriber_topic = json_data['subcriber_topic']
|
||||
mqtt_conf.client_id = json_data['client_id']
|
||||
# 2025.03.11 增加读取多个FsuCode,用于模拟多个不同的设备
|
||||
mqtt_conf.fsuCodes = json_data['FsuCodes']
|
||||
publish_data.publish_data = json_data['publish_data']
|
||||
except Exception as e:
|
||||
dump_exception_msg(e)
|
||||
|
@ -142,9 +145,12 @@ def sub_connect_mqtt():
|
|||
|
||||
|
||||
# 根据配置信息构建发布的报文消息
|
||||
def build_publish_msg():
|
||||
# 2025.03.11 增加函数参数,处理制定的fsucode
|
||||
def build_publish_msg(fsu_code=None):
|
||||
pd = publish_data.publish_data
|
||||
msg = {"FsuCode": pd["FsuCode"], "type": pd["type"], "IdCodeContent": [],
|
||||
# msg = {"FsuCode": pd["FsuCode"], "type": pd["type"], "IdCodeContent": [],
|
||||
# "TimeStamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}
|
||||
msg = {"FsuCode": fsu_code, "type": pd["type"], "IdCodeContent": [],
|
||||
"TimeStamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}
|
||||
for one in pd["IdCodeContent"]:
|
||||
idcode = one["OID"]
|
||||
|
@ -170,15 +176,19 @@ def publish(client):
|
|||
while True:
|
||||
if mqtt_conf.max_publish_count != 0 and msg_count > mqtt_conf.max_publish_count:
|
||||
break
|
||||
msg = build_publish_msg()
|
||||
result = client.publish(topic=mqtt_conf.publish_topic, payload=msg, qos=mqtt_conf.publish_qos)
|
||||
# result: [0, 1]
|
||||
status = result[0]
|
||||
msg_count += 1
|
||||
if status == 0:
|
||||
logger.info(f"Send `#{msg_count}` message to topic `{mqtt_conf.publish_topic}`: `{msg}` ")
|
||||
else:
|
||||
logger.error(f"Failed to send `#{msg_count}` message to topic {mqtt_conf.publish_topic}")
|
||||
|
||||
# 2025.03.11 增加循环处理多个FsuCode,用于模拟多个不同的设备
|
||||
for fsu_code in mqtt_conf.fsuCodes:
|
||||
logger.info("Processing fsuCode: " + fsu_code)
|
||||
msg = build_publish_msg(fsu_code)
|
||||
result = client.publish(topic=mqtt_conf.publish_topic, payload=msg, qos=mqtt_conf.publish_qos)
|
||||
# result: [0, 1]
|
||||
status = result[0]
|
||||
msg_count += 1
|
||||
if status == 0:
|
||||
logger.info(f"Send `#{msg_count}` message to topic `{mqtt_conf.publish_topic}`: `{msg}` ")
|
||||
else:
|
||||
logger.error(f"Failed to send `#{msg_count}` message to topic {mqtt_conf.publish_topic}")
|
||||
time.sleep(mqtt_conf.publish_interval/1000)
|
||||
|
||||
|
Loading…
Reference in New Issue