update
This commit is contained in:
parent
3791368e67
commit
a6db80feca
7
MQTT.py
7
MQTT.py
|
@ -1,3 +1,4 @@
|
||||||
|
import json
|
||||||
import time
|
import time
|
||||||
|
|
||||||
import paho.mqtt.client as mqtt
|
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):
|
def valid(msg, client):
|
||||||
origin_data = json.loads(msg.payload.decode('utf-8'))
|
origin_data = json.loads(msg.payload.decode('utf-8'))
|
||||||
if 'msg' not in origin_data:
|
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 False
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
@ -31,7 +32,7 @@ class MQTTClient:
|
||||||
def on_connect(self, client, userdata, flags, rc):
|
def on_connect(self, client, userdata, flags, rc):
|
||||||
if rc == 0:
|
if rc == 0:
|
||||||
self.client.subscribe(self.topic)
|
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)
|
qos=0)
|
||||||
|
|
||||||
def on_message(self, client, userdata, msg):
|
def on_message(self, client, userdata, msg):
|
||||||
|
@ -66,7 +67,7 @@ class MQTTClient:
|
||||||
get_status(client)
|
get_status(client)
|
||||||
else:
|
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:
|
except Exception as e:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
|
@ -1,3 +1,4 @@
|
||||||
|
import json
|
||||||
import time
|
import time
|
||||||
|
|
||||||
import paho.mqtt.client as mqtt
|
import paho.mqtt.client as mqtt
|
||||||
|
@ -24,8 +25,7 @@ class DataUploadClient:
|
||||||
def on_connect(self, client, userdata, flags, rc):
|
def on_connect(self, client, userdata, flags, rc):
|
||||||
global times
|
global times
|
||||||
times = 6
|
times = 6
|
||||||
self.client.publish(info_topic, payload=subscribe_topic + ':连接成功,执行数据推送和本地存储' + str(time.time()),
|
self.client.publish(info_topic, payload=json.dumps(subscribe_topic + ':连接成功,执行数据推送和本地存储'), qos=0)
|
||||||
qos=0)
|
|
||||||
subprocess.Popen(['/usr/bin/bash /home/pi/lot_manager/bash/start_data_upload.sh'], shell=True)
|
subprocess.Popen(['/usr/bin/bash /home/pi/lot_manager/bash/start_data_upload.sh'], shell=True)
|
||||||
|
|
||||||
def on_disconnect(self, client, userdata, rc):
|
def on_disconnect(self, client, userdata, rc):
|
||||||
|
|
29
tool.py
29
tool.py
|
@ -9,12 +9,16 @@ from pydantic import BaseModel
|
||||||
from config import mp4_path, post_record_list_url, post_record_url, info_topic
|
from config import mp4_path, post_record_list_url, post_record_url, info_topic
|
||||||
|
|
||||||
|
|
||||||
# 统一返回
|
|
||||||
class BaseResponse(BaseModel):
|
class BaseResponse(BaseModel):
|
||||||
code: int = pydantic.Field(200, description="MQTT Return Status Code")
|
code: int = pydantic.Field(200, description="MQTT Return Status Code")
|
||||||
msg: str = pydantic.Field("success", description="MQTT Status Message")
|
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 exception_handler(func):
|
||||||
def wrapper(*args, **kwargs):
|
def wrapper(*args, **kwargs):
|
||||||
try:
|
try:
|
||||||
|
@ -32,7 +36,7 @@ def push_stream(client):
|
||||||
stderr=subprocess.PIPE)
|
stderr=subprocess.PIPE)
|
||||||
out, err = p.communicate()
|
out, err = p.communicate()
|
||||||
output = out.decode('utf-8').strip()
|
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):
|
def close_stream(client):
|
||||||
|
@ -42,13 +46,13 @@ def close_stream(client):
|
||||||
stderr=subprocess.PIPE)
|
stderr=subprocess.PIPE)
|
||||||
out, err = p.communicate()
|
out, err = p.communicate()
|
||||||
output = out.decode('utf-8').strip()
|
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):
|
def exec_sh(msg, client):
|
||||||
origin_data = json.loads(msg.payload.decode('utf-8'))
|
origin_data = json.loads(msg.payload.decode('utf-8'))
|
||||||
if 'data' not in origin_data:
|
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
|
return
|
||||||
cmd = origin_data["data"]
|
cmd = origin_data["data"]
|
||||||
if cmd in ["supervisorctl stop __mqtt__",
|
if cmd in ["supervisorctl stop __mqtt__",
|
||||||
|
@ -56,13 +60,13 @@ def exec_sh(msg, client):
|
||||||
"supervisorctl stop all"]:
|
"supervisorctl stop all"]:
|
||||||
return
|
return
|
||||||
if cmd == "supervisorctl reload":
|
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)
|
subprocess.Popen([cmd], shell=True)
|
||||||
return
|
return
|
||||||
p = subprocess.Popen([cmd], shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
p = subprocess.Popen([cmd], shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||||
out, err = p.communicate()
|
out, err = p.communicate()
|
||||||
output = out.decode('utf-8').strip()
|
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):
|
def get_status(client):
|
||||||
|
@ -72,7 +76,7 @@ def get_status(client):
|
||||||
stderr=subprocess.PIPE)
|
stderr=subprocess.PIPE)
|
||||||
out, err = p.communicate()
|
out, err = p.communicate()
|
||||||
output = out.decode('utf-8').strip()
|
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):
|
def update(client):
|
||||||
|
@ -82,11 +86,12 @@ def update(client):
|
||||||
stderr=subprocess.PIPE)
|
stderr=subprocess.PIPE)
|
||||||
out, err = p.communicate()
|
out, err = p.communicate()
|
||||||
output = out.decode('utf-8').strip()
|
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):
|
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)
|
subprocess.Popen(['supervisorctl reload'], shell=True)
|
||||||
|
|
||||||
|
|
||||||
|
@ -95,15 +100,15 @@ def get_list_record(client):
|
||||||
"data": os.listdir(mp4_path)
|
"data": os.listdir(mp4_path)
|
||||||
}
|
}
|
||||||
r = requests.post(post_record_list_url, json=data)
|
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):
|
def get_record(msg, client):
|
||||||
filename = json.loads(msg.payload.decode('utf-8'))["data"]
|
filename = json.loads(msg.payload.decode('utf-8'))["data"]
|
||||||
if filename is None or filename == '':
|
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
|
return
|
||||||
|
|
||||||
files = {filename: open(os.path.join(mp4_path, filename), 'rb'), "Content-Type": "application/octet-stream"}
|
files = {filename: open(os.path.join(mp4_path, filename), 'rb'), "Content-Type": "application/octet-stream"}
|
||||||
r = requests.post(post_record_url, files=files)
|
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)
|
||||||
|
|
Loading…
Reference in New Issue