lot_manager/tool.py

87 lines
3.1 KiB
Python
Executable File

import configparser
import json
import os
import subprocess
import requests
mp4_path = '/home/pi/mp4'
config = configparser.ConfigParser()
config.read('/home/pi/lot_manager/conf/main/config.conf')
def push_stream(client):
p = subprocess.Popen(['/bin/bash /home/pi/lot_manager/bash/start_push_stream.sh'],
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)
def close_stream(client):
p = subprocess.Popen(['/bin/bash /home/pi/lot_manager/bash/stop_push_stream.sh'],
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)
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)
return
cmd = origin_data["data"]
if cmd == "supervisorctl stop __mqtt__" or cmd == "supervisorctl restart __mqtt__":
return
if cmd == "supervisorctl reload":
client.publish('success', 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)
def get_status(client):
p = subprocess.Popen(['supervisorctl status'],
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)
def update(client):
p = subprocess.Popen(['/bin/bash /home/pi/lot_manager/git_update.sh'],
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)
def get_list_record():
data = {
"data": os.listdir(mp4_path)
}
post_record_list_url = config.get("record", "post_record_list_url")
requests.post(post_record_list_url, json=data)
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)
return
post_record_url = config.get("record", "post_record_url")
files = {filename: open(os.path.join(mp4_path, filename), 'rb'), "Content-Type": "application/octet-stream"}
requests.post(post_record_url, files=files)