From a6db80feca9b9d66cae7b8e526eb17b54cb038f4 Mon Sep 17 00:00:00 2001 From: xyj <10908227994@qq.com> Date: Mon, 18 Dec 2023 11:06:04 +0800 Subject: [PATCH] update --- MQTT.py | 7 ++++--- data_upload.py | 4 ++-- tool.py | 29 +++++++++++++++++------------ 3 files changed, 23 insertions(+), 17 deletions(-) diff --git a/MQTT.py b/MQTT.py index a1f7496..c4f215e 100644 --- a/MQTT.py +++ b/MQTT.py @@ -1,3 +1,4 @@ +import json import time import paho.mqtt.client as mqtt @@ -10,7 +11,7 @@ from config import broker, port, subscribe_topic, username, password, info_topic def valid(msg, client): origin_data = json.loads(msg.payload.decode('utf-8')) if 'msg' not in origin_data: - client.publish(info_topic, payload=BaseResponse(code=404, msg="msg must be supplied"), qos=0) + client.publish(info_topic, payload=publish_payload(code=404, msg="msg must be supplied"), qos=0) return False return True @@ -31,7 +32,7 @@ class MQTTClient: def on_connect(self, client, userdata, flags, rc): if rc == 0: self.client.subscribe(self.topic) - client.publish(info_topic, payload=BaseResponse(code=200, msg='成功订阅' + self.topic), + client.publish(info_topic, payload=publish_payload(code=200, msg='成功订阅' + self.topic), qos=0) def on_message(self, client, userdata, msg): @@ -66,7 +67,7 @@ class MQTTClient: get_status(client) else: # 错误类型 - client.publish(info_topic, payload=BaseResponse(code=404, msg='No Such Msg Type'), qos=0) + client.publish(info_topic, payload=publish_payload(code=404, msg='No Such Msg Type'), qos=0) except Exception as e: pass diff --git a/data_upload.py b/data_upload.py index 4dc15f3..f9ed9d0 100644 --- a/data_upload.py +++ b/data_upload.py @@ -1,3 +1,4 @@ +import json import time import paho.mqtt.client as mqtt @@ -24,8 +25,7 @@ class DataUploadClient: def on_connect(self, client, userdata, flags, rc): global times times = 6 - self.client.publish(info_topic, payload=subscribe_topic + ':连接成功，执行数据推送和本地存储' + str(time.time()), - qos=0) + self.client.publish(info_topic, payload=json.dumps(subscribe_topic + ':连接成功，执行数据推送和本地存储'), qos=0) subprocess.Popen(['/usr/bin/bash /home/pi/lot_manager/bash/start_data_upload.sh'], shell=True) def on_disconnect(self, client, userdata, rc): diff --git a/tool.py b/tool.py index 4862dab..405012a 100755 --- a/tool.py +++ b/tool.py @@ -9,12 +9,16 @@ from pydantic import BaseModel from config import mp4_path, post_record_list_url, post_record_url, info_topic -# 统一返回 class BaseResponse(BaseModel): code: int = pydantic.Field(200, description="MQTT Return Status Code") msg: str = pydantic.Field("success", description="MQTT Status Message") +# 统一返回 +def publish_payload(code, msg): + return json.dumps(BaseResponse(code=code, msg=msg), ensure_ascii=False) + + def exception_handler(func): def wrapper(*args, **kwargs): try: @@ -32,7 +36,7 @@ def push_stream(client): stderr=subprocess.PIPE) out, err = p.communicate() output = out.decode('utf-8').strip() - client.publish(info_topic, payload=BaseResponse(code=200, msg=json.dumps(output, ensure_ascii=False)), qos=0) + client.publish(info_topic, payload=publish_payload(200, json.dumps(output, ensure_ascii=False)), qos=0) def close_stream(client): @@ -42,13 +46,13 @@ def close_stream(client): stderr=subprocess.PIPE) out, err = p.communicate() output = out.decode('utf-8').strip() - client.publish(info_topic, payload=BaseResponse(code=200, msg=json.dumps(output, ensure_ascii=False)), qos=0) + client.publish(info_topic, payload=publish_payload(code=200, msg=json.dumps(output, ensure_ascii=False)), qos=0) def exec_sh(msg, client): origin_data = json.loads(msg.payload.decode('utf-8')) if 'data' not in origin_data: - client.publish(info_topic, payload='data must be supplied', qos=0) + client.publish(info_topic, payload=publish_payload(code=404, msg='data must be supplied'), qos=0) return cmd = origin_data["data"] if cmd in ["supervisorctl stop __mqtt__", @@ -56,13 +60,13 @@ def exec_sh(msg, client): "supervisorctl stop all"]: return if cmd == "supervisorctl reload": - client.publish(info_topic, payload='reloading', qos=0) + client.publish(info_topic, payload=publish_payload(code=200, msg='reloading'), qos=0) subprocess.Popen([cmd], shell=True) return p = subprocess.Popen([cmd], shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) out, err = p.communicate() output = out.decode('utf-8').strip() - client.publish(info_topic, payload=BaseResponse(code=200, msg=json.dumps(output, ensure_ascii=False)), qos=0) + client.publish(info_topic, payload=publish_payload(code=200, msg=json.dumps(output, ensure_ascii=False)), qos=0) def get_status(client): @@ -72,7 +76,7 @@ def get_status(client): stderr=subprocess.PIPE) out, err = p.communicate() output = out.decode('utf-8').strip() - client.publish(info_topic, payload=BaseResponse(code=200, msg=json.dumps(output, ensure_ascii=False)), qos=0) + client.publish(info_topic, payload=publish_payload(code=200, msg=json.dumps(output, ensure_ascii=False)), qos=0) def update(client): @@ -82,11 +86,12 @@ def update(client): stderr=subprocess.PIPE) out, err = p.communicate() output = out.decode('utf-8').strip() - client.publish(info_topic, payload=BaseResponse(code=200, msg=json.dumps(output, ensure_ascii=False)), qos=0) + client.publish(info_topic, payload=publish_payload(code=200, msg=json.dumps(output, ensure_ascii=False)), + qos=0) def reload(client): - client.publish(info_topic, payload="reloading", qos=0) + client.publish(info_topic, payload=publish_payload(200, "reloading"), qos=0) subprocess.Popen(['supervisorctl reload'], shell=True) @@ -95,15 +100,15 @@ def get_list_record(client): "data": os.listdir(mp4_path) } r = requests.post(post_record_list_url, json=data) - client.publish(info_topic, payload=BaseResponse(code=200, msg=str(r)), qos=0) + client.publish(info_topic, payload=publish_payload(code=200, msg=str(r)), qos=0) def get_record(msg, client): filename = json.loads(msg.payload.decode('utf-8'))["data"] if filename is None or filename == '': - client.publish(info_topic, payload=BaseResponse(code=404, msg='没有该文件'), qos=0) + client.publish(info_topic, payload=publish_payload(code=404, msg='没有该文件'), qos=0) return files = {filename: open(os.path.join(mp4_path, filename), 'rb'), "Content-Type": "application/octet-stream"} r = requests.post(post_record_url, files=files) - client.publish(info_topic, payload=BaseResponse(code=200, msg=str(r)), qos=0) + client.publish(info_topic, payload=publish_payload(code=200, msg=str(r)), qos=0)