update
This commit is contained in:
xyj
parent
3449a3df36
commit
3791368e67
19
MQTT.py
19
MQTT.py
|
|
@ -1,25 +1,16 @@
|
|||
import datetime
|
||||
import time
|
||||
|
||||
import paho.mqtt.client as mqtt
|
||||
|
||||
from tool import *
|
||||
|
||||
from config import broker, port, subscribe_topic, publish_topic, username, password, publish_pwd
|
||||
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('error', payload='msg must be supplied', qos=0)
|
||||
return False
|
||||
# TODO 增加安全性
|
||||
if 'pwd' not in origin_data:
|
||||
client.publish('error', payload='pwd must be supplied', qos=0)
|
||||
return False
|
||||
# 判断密码是否正确
|
||||
if publish_pwd != origin_data['pwd']:
|
||||
client.publish('error', payload='pwd is not corrected', qos=0)
|
||||
client.publish(info_topic, payload=BaseResponse(code=404, msg="msg must be supplied"), qos=0)
|
||||
return False
|
||||
return True
|
||||
|
||||
|
|
@ -40,14 +31,12 @@ class MQTTClient:
|
|||
def on_connect(self, client, userdata, flags, rc):
|
||||
if rc == 0:
|
||||
self.client.subscribe(self.topic)
|
||||
client.publish('success', payload='成功订阅' + subscribe_topic + ',time=' + str(datetime.datetime.now()),
|
||||
client.publish(info_topic, payload=BaseResponse(code=200, msg='成功订阅' + self.topic),
|
||||
qos=0)
|
||||
|
||||
def on_message(self, client, userdata, msg):
|
||||
if not valid(msg, client):
|
||||
client.publish('error', payload='验证失败', qos=0)
|
||||
return
|
||||
client.publish('success', payload='验证通过', qos=0)
|
||||
try:
|
||||
origin_data = json.loads(msg.payload.decode('utf-8'))
|
||||
data = origin_data["msg"]
|
||||
|
|
@ -77,7 +66,7 @@ class MQTTClient:
|
|||
get_status(client)
|
||||
else:
|
||||
# 错误类型
|
||||
client.publish('error', payload='No Such Type', qos=0)
|
||||
client.publish(info_topic, payload=BaseResponse(code=404, msg='No Such Msg Type'), qos=0)
|
||||
except Exception as e:
|
||||
pass
|
||||
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@
|
|||
subscribe_topic=lihai_lot_walnutpi_dev_1
|
||||
# 发布消息的主题
|
||||
publish_topic=camera_1
|
||||
info_topic=info_dev_1
|
||||
username=lihai_lot_land_1
|
||||
password=123456
|
||||
####################################
|
||||
|
|
@ -14,6 +15,7 @@ password=123456
|
|||
subscribe_topic=lihai_lot_walnutpi_dev_2
|
||||
# 发布消息的主题
|
||||
publish_topic=camera_2
|
||||
info_topic=info_dev_2
|
||||
username=lihai_lot_land_1
|
||||
password=123456
|
||||
####################################
|
||||
|
|
@ -23,6 +25,7 @@ password=123456
|
|||
subscribe_topic=lihai_lot_walnutpi_dev_3
|
||||
# 发布消息的主题
|
||||
publish_topic=camera_3
|
||||
info_topic=info_dev_3
|
||||
username=lihai_lot_land_1
|
||||
password=123456
|
||||
####################################
|
||||
|
|
@ -32,6 +35,7 @@ password=123456
|
|||
subscribe_topic=lihai_lot_walnutpi_dev_4
|
||||
# 发布消息的主题
|
||||
publish_topic=camera_4
|
||||
info_topic=info_dev_4
|
||||
username=lihai_lot_land_1
|
||||
password=123456
|
||||
####################################
|
||||
|
|
@ -41,6 +45,7 @@ password=123456
|
|||
subscribe_topic=lihai_lot_walnutpi_dev_5
|
||||
# 发布消息的主题
|
||||
publish_topic=camera_5
|
||||
info_topic=info_dev_5
|
||||
username=lihai_lot_land_1
|
||||
password=123456
|
||||
####################################
|
||||
|
|
@ -50,5 +55,6 @@ password=123456
|
|||
subscribe_topic=lihai_lot_walnutpi_dev_6
|
||||
# 发布消息的主题
|
||||
publish_topic=camera_6
|
||||
info_topic=info_dev_6
|
||||
username=lihai_lot_land_1
|
||||
password=123456
|
||||
|
|
@ -4,6 +4,4 @@ host=mqtt.lihaink.cn
|
|||
port=1883
|
||||
[record]
|
||||
post_record_list_url=https://ceshi-iot.lihaink.cn/api/index/file_list
|
||||
post_record_url=https://ceshi-iot.lihaink.cn/api/index/upload
|
||||
[publish_pwd]
|
||||
value=123456
|
||||
post_record_url=https://ceshi-iot.lihaink.cn/api/index/upload
|
||||
69
config.py
69
config.py
|
|
@ -2,40 +2,43 @@ import configparser
|
|||
|
||||
import subprocess
|
||||
|
||||
p = subprocess.Popen(['cat /home/pi/device_name'], shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||
out, err = p.communicate()
|
||||
# 设备名称，必须要有
|
||||
device_name = out.decode('utf-8').strip()
|
||||
try:
|
||||
p = subprocess.Popen(['cat /home/pi/device_name'], shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||
out, err = p.communicate()
|
||||
# 设备名称，必须要有
|
||||
device_name = out.decode('utf-8').strip()
|
||||
|
||||
# 读取配置
|
||||
config = configparser.ConfigParser()
|
||||
# 读取公共配置
|
||||
config.read('conf/main/common.conf')
|
||||
# 域名
|
||||
broker = config.get("broker", "host")
|
||||
# 端口，这里必须是int类型
|
||||
port = config.getint("broker", "port")
|
||||
# 录像地址
|
||||
post_record_list_url = config.get("record", "post_record_list_url")
|
||||
post_record_url = config.get("record", "post_record_url")
|
||||
# 发布消息的认证pwd
|
||||
publish_pwd = config.get("publish_pwd", "value")
|
||||
# 读取配置
|
||||
config = configparser.ConfigParser()
|
||||
# 读取公共配置
|
||||
config.read('conf/main/common.conf')
|
||||
# 域名
|
||||
broker = config.get("broker", "host")
|
||||
# 端口，这里必须是int类型
|
||||
port = config.getint("broker", "port")
|
||||
# 录像地址
|
||||
post_record_list_url = config.get("record", "post_record_list_url")
|
||||
post_record_url = config.get("record", "post_record_url")
|
||||
|
||||
# 读取设备配置
|
||||
config.read('conf/device/device.conf')
|
||||
# 订阅的主题
|
||||
subscribe_topic = config.get(device_name, "subscribe_topic")
|
||||
# 发布数据的主题
|
||||
publish_topic = config.get(device_name, "publish_topic")
|
||||
# 发布信息的主题
|
||||
info_topic = config.get(device_name, "info_topic")
|
||||
# 用户
|
||||
username = config.get(device_name, "username")
|
||||
# 密码
|
||||
password = config.get(device_name, "password")
|
||||
|
||||
# 读取设备配置
|
||||
config.read('conf/device/device.conf')
|
||||
# 订阅的主题
|
||||
subscribe_topic = config.get(device_name, "subscribe_topic")
|
||||
# 发布的主题
|
||||
publish_topic = config.get(device_name, "publish_topic")
|
||||
# 用户
|
||||
username = config.get(device_name, "username")
|
||||
# 密码
|
||||
password = config.get(device_name, "password")
|
||||
# 特殊配置
|
||||
config.read('conf/zhanguan/topic.conf')
|
||||
zhanguan_device_name = config.get("device", "name")
|
||||
|
||||
# 特殊配置
|
||||
config.read('conf/zhanguan/topic.conf')
|
||||
zhanguan_device_name = config.get("device", "name")
|
||||
|
||||
# tool配置
|
||||
mp4_path = '/home/pi/mp4'
|
||||
# tool配置
|
||||
mp4_path = '/home/pi/mp4'
|
||||
except Exception as e:
|
||||
# print(e)
|
||||
pass
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@ import time
|
|||
import paho.mqtt.client as mqtt
|
||||
|
||||
from tool import *
|
||||
from config import broker, port, subscribe_topic, username, password
|
||||
from config import broker, port, subscribe_topic, username, password, info_topic
|
||||
|
||||
times = 6
|
||||
|
||||
|
|
@ -24,7 +24,7 @@ class DataUploadClient:
|
|||
def on_connect(self, client, userdata, flags, rc):
|
||||
global times
|
||||
times = 6
|
||||
self.client.publish('success', payload=subscribe_topic + ':连接成功，执行数据推送和本地存储' + str(time.time()),
|
||||
self.client.publish(info_topic, payload=subscribe_topic + ':连接成功，执行数据推送和本地存储' + str(time.time()),
|
||||
qos=0)
|
||||
subprocess.Popen(['/usr/bin/bash /home/pi/lot_manager/bash/start_data_upload.sh'], shell=True)
|
||||
|
||||
|
|
|
|||
38
tool.py
38
tool.py
|
|
@ -2,9 +2,18 @@ import json
|
|||
import os
|
||||
import subprocess
|
||||
|
||||
import pydantic
|
||||
import requests
|
||||
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")
|
||||
|
||||
from config import mp4_path, post_record_list_url, post_record_url
|
||||
|
||||
def exception_handler(func):
|
||||
def wrapper(*args, **kwargs):
|
||||
|
|
@ -14,6 +23,8 @@ def exception_handler(func):
|
|||
print(f"函数{func.__name__}中发生了异常：{e}")
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
def push_stream(client):
|
||||
p = subprocess.Popen(['/bin/bash /home/pi/lot_manager/bash/start_push_stream.sh'],
|
||||
shell=True,
|
||||
|
|
@ -21,7 +32,7 @@ def push_stream(client):
|
|||
stderr=subprocess.PIPE)
|
||||
out, err = p.communicate()
|
||||
output = out.decode('utf-8').strip()
|
||||
client.publish('success', payload=json.dumps(output, ensure_ascii=False), qos=0)
|
||||
client.publish(info_topic, payload=BaseResponse(code=200, msg=json.dumps(output, ensure_ascii=False)), qos=0)
|
||||
|
||||
|
||||
def close_stream(client):
|
||||
|
|
@ -31,13 +42,13 @@ def close_stream(client):
|
|||
stderr=subprocess.PIPE)
|
||||
out, err = p.communicate()
|
||||
output = out.decode('utf-8').strip()
|
||||
client.publish('success', payload=json.dumps(output, ensure_ascii=False), qos=0)
|
||||
client.publish(info_topic, payload=BaseResponse(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('error', payload='data must be supplied', qos=0)
|
||||
client.publish(info_topic, payload='data must be supplied', qos=0)
|
||||
return
|
||||
cmd = origin_data["data"]
|
||||
if cmd in ["supervisorctl stop __mqtt__",
|
||||
|
|
@ -45,13 +56,13 @@ def exec_sh(msg, client):
|
|||
"supervisorctl stop all"]:
|
||||
return
|
||||
if cmd == "supervisorctl reload":
|
||||
client.publish('success', payload='reloading', qos=0)
|
||||
client.publish(info_topic, payload='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('success', payload=json.dumps(output, ensure_ascii=False), qos=0)
|
||||
client.publish(info_topic, payload=BaseResponse(code=200, msg=json.dumps(output, ensure_ascii=False)), qos=0)
|
||||
|
||||
|
||||
def get_status(client):
|
||||
|
|
@ -61,7 +72,7 @@ def get_status(client):
|
|||
stderr=subprocess.PIPE)
|
||||
out, err = p.communicate()
|
||||
output = out.decode('utf-8').strip()
|
||||
client.publish('success', payload=json.dumps(output, ensure_ascii=False), qos=0)
|
||||
client.publish(info_topic, payload=BaseResponse(code=200, msg=json.dumps(output, ensure_ascii=False)), qos=0)
|
||||
|
||||
|
||||
def update(client):
|
||||
|
|
@ -71,11 +82,11 @@ def update(client):
|
|||
stderr=subprocess.PIPE)
|
||||
out, err = p.communicate()
|
||||
output = out.decode('utf-8').strip()
|
||||
client.publish('success', payload=json.dumps(output, ensure_ascii=False), qos=0)
|
||||
client.publish(info_topic, payload=BaseResponse(code=200, msg=json.dumps(output, ensure_ascii=False)), qos=0)
|
||||
|
||||
|
||||
def reload(client):
|
||||
client.publish('success', payload="reloading", qos=0)
|
||||
client.publish(info_topic, payload="reloading", qos=0)
|
||||
subprocess.Popen(['supervisorctl reload'], shell=True)
|
||||
|
||||
|
||||
|
|
@ -84,18 +95,15 @@ def get_list_record(client):
|
|||
"data": os.listdir(mp4_path)
|
||||
}
|
||||
r = requests.post(post_record_list_url, json=data)
|
||||
client.publish('success', payload=str(r), qos=0)
|
||||
client.publish(info_topic, payload=BaseResponse(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('error', payload='没有该文件', qos=0)
|
||||
client.publish(info_topic, payload=BaseResponse(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('success', payload=str(r), qos=0)
|
||||
|
||||
|
||||
|
||||
client.publish(info_topic, payload=BaseResponse(code=200, msg=str(r)), qos=0)
|
||||
|
|
|
|||
Loading…
Reference in New Issue