test

MQTT.py

@@ -25,12 +25,12 @@ def on_message(client, userdata, msg):
         close_stream()
     elif data == "exec":
         #  执行命令
-        exec_sh()
+        exec_sh(msg)
     elif data == "update":
         #  git更新项目和配置文件
         update()
     else:
-        client.publish('conn_error', payload='No Such Type', qos=0)
+        client.publish('error', payload='No Such Type', qos=0)
 
 
 if __name__ == '__main__':

api.py

@@ -8,92 +8,18 @@ from db.repository import add_kb_to_db, get_kb_detail, get_kb_detail_by_time, de
 
 from fastapi import FastAPI
 
-app = FastAPI()
 
-
+def add(data: LOT_DATA):
-@app.post("/add/lot_data/{device_name}")
-async def add(data: LOT_DATA, device_name: str):
     try:
-        add_kb_to_db(data, device_name)
+        add_kb_to_db(data)
         return BaseResponse()
     except Exception as e:
         return BaseResponse(code=500, msg=e)
 
 
-@app.get("/get/all/lot_data")
+def get_data():
-async def get_data():
     try:
         data = get_kb_detail()
         return BaseResponse(data=data)
     except Exception as e:
         return BaseResponse(code=404, msg=e)
-
-
-@app.get("/get/data/by/time")
-async def get_data(start_time, end_time):
-    try:
-        data = get_kb_detail_by_time(start_time, end_time)
-        return BaseResponse(data=data)
-    except Exception as e:
-        return BaseResponse(code=404, msg=e)
-
-
-@app.get("/delete/data/by/time")
-async def delete_data(start_time, end_time):
-    try:
-        data = delete_kb_detail_by_time(start_time, end_time)
-        return BaseResponse(data=data)
-    except Exception as e:
-        return BaseResponse(code=404, msg=e)
-
-
-@app.get("/create/db")
-async def create():
-    from db.base import Base, engine
-    try:
-        Base.metadata.create_all(bind=engine)
-        return BaseResponse()
-    except Exception as e:
-        return BaseResponse(code=404, msg=e)
-
-
-@app.get("/start")
-def device_test():
-    ## 不管有没有都进行一个杀死
-    # os.popen()
-    # stop_cmd = r'kill -9 `lsof -t -i:554`'
-    # os.system(stop_cmd)
-    start_cmd = r'nohup ffmpeg -rtsp_transport tcp -re -i rtsp://admin:123456@192.168.0.226:554/mpeg4 -c:v copy -c:a aac -preset ultrafast -r 20 -flvflags no_duration_filesize -f rtsp -rtsp_transport tcp  rtsp://127.0.0.1/live/test > app.out 2>&1 & echo $! > app.pid'
-    p = subprocess.Popen(start_cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-    out, err = p.communicate()
-    return_code = p.returncode
-    if return_code == 0:
-        return BaseResponse(code=return_code, msg=out,
-                            data="http://192.168.1.27/live/test.live.mp4?secret=gqig2yFKkDpIMic1uWZY1L5MsIo0eflm")
-    else:
-        return BaseResponse(code=return_code, msg=err)
-
-
-@app.get("/stop")
-async def device_test2():
-    ## 不管有没有都进行一个杀死
-    cmd = 'lsof -t -i:554'
-    p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-    # 获取执行结果
-    out, err = p.communicate()
-    # 获取返回值
-    return_code = p.returncode
-    if out == b'':
-        print("失败")
-    else:
-        pid = out
-        cmd = f'kill -9 {pid}'
-        print(pid)
-        os.system(cmd)
-    if return_code == 0:
-        return BaseResponse(code=return_code, msg=out)
-    return BaseResponse(code=return_code, msg=err)
-
-
-if __name__ == '__main__':
-    uvicorn.run(app, host="127.0.0.1", port=8000)

conf/mqtt.conf

@@ -1,10 +0,0 @@
-[program:mqtt]
-directory=/home/lihai//PycharmProjects/pythonProject/lot_manager
-command=/home/lihai//anaconda3/envs/chatchat/bin/python MQTT.py
-user=lihai
-autostart=true
-autorestart=true
-redirect_stderr=true
-stopsignal=TERM
-stopasgroup=True
-priority=1

conf/push_stream.conf

@@ -2,7 +2,7 @@
 directory=/home/lihai/PycharmProjects/pythonProject/lot_manager
 command=/usr/bin/ffmpeg -rtsp_transport tcp -re -i rtsp://admin:123456@192.168.0.123:554/mpeg4 -c:v copy -c:a aac -preset ultrafast -r 20 -flvflags no_duration_filesize -f rtsp -rtsp_transport tcp rstp://127.0.0.1:554/live/test
 user=lihai
-autostart=true
+autostart=false
 autorestart=true
 redirect_stderr=true
 stopsignal=TERM

conf/test2.conf

@@ -1,10 +0,0 @@
-[program:test2]
-directory=/home/lihai/PycharmProjects/pythonProject/lot_manager
-command=/home/lihai/anaconda3/envs/chatchat/bin/python test.py
-user=lihai
-autostart=false
-autorestart=true
-restart_times=3
-redirect_stderr=true
-stopsignal=TERM
-stopasgroup=True

db/models/log_data_model.py

@@ -4,6 +4,7 @@ from db.base import Base
 
 
 class LOT_DATA(BaseModel):
+    create_time: int = Field(None, description='创建时间(时间戳) ')
     wind_speed: float = Field(None, description='风速：(0到30)m/s ')
     wind_direction: float = Field(None, description='风向:0～360°')
     ambient_temperature: float = Field(None, description='环境温度：℃')
@@ -27,7 +28,6 @@ class LOT_DATA_MODEL(Base):
     """
     __tablename__ = 'LOT_DATA_MODEL'
     id = Column(Integer, primary_key=True, autoincrement=True, comment='ID')
-    device_name = Column(String(50), comment='设备编号', default="0")
     create_time = Column(Integer, comment="创建时间")
     wind_speed = Column(Float, comment='风速：(0到30)m/s ')
     wind_direction = Column(Float, comment='风向:0～360°')

db/repository/lot_data_repository.py

@@ -5,9 +5,9 @@ from db.session import with_session
 
 
 @with_session
-def add_kb_to_db(session, data: LOT_DATA, device_name: str):
+def add_kb_to_db(session, data: LOT_DATA):
     # 创建知识库实例
-    kb = LOT_DATA_MODEL(**data.__dict__, device_name=device_name, create_time=int(time.time()))
+    kb = LOT_DATA_MODEL(**data.__dict__, create_time=int(time.time()))
     session.add(kb)
 
 

exec_sh.sh

@@ -1 +0,0 @@
-#!/bin/bash

reload_conf.sh

@@ -1,6 +0,0 @@
-#!/bin/bash
-#复制配置文件
-set -e
-cp -r reload_conf/*.conf /etc/supervisor/conf.d/
-# 重启所有配置
-supervisorctl reload

stop_push.sh

@@ -1,4 +1,4 @@
 #!/bin/bash
 set -e
 
-supervisorctl stop test
+supervisorctl stop push_stream

tool.py

@@ -1,18 +1,19 @@
+import json
 import subprocess
 
 
 def push_stream():
-    print("启动推流")
+    subprocess.Popen(['/bin/bash start_push.sh'], shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-    subprocess.Popen(['start_push.sh'], shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
 
 
 def close_stream():
-    subprocess.Popen(['stop_push.sh'], shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    subprocess.Popen(['/bin/bash stop_push.sh'], shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
 
 
-def exec_sh():
+def exec_sh(msg):
-    subprocess.Popen(['exec_sh.sh'], shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    cmd = json.loads(msg.payload.decode('utf-8'))["cmd"]
+    subprocess.Popen([cmd], shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
 
 
 def update():
-    subprocess.Popen(['update.sh'], shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    subprocess.Popen(['/bin/bash update.sh'], shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

update.sh

@@ -5,4 +5,4 @@ git pull origin master
 # 配置文件复制到supervisor管理
 cp -r conf/*.conf /etc/supervisor/conf.d/
 # 更新配置文件
-supervisorctl reread
+supervisorctl update