Quella/RkApp
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

finnal

Browse Source
This commit is contained in:
Quella777
2025-10-31 13:32:51 +08:00
parent e4ae201289
commit b807b0f1b4
10 changed files with 4799 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
FastApi/__pycache__/fastApi.cpython-38.pyc
View File

Binary file not shown.

144
FastApi/fastApi.py
View File

@@ -2,21 +2,123 @@
from fastapi import FastAPI, File, UploadFile, Form, WebSocket, WebSocketDisconnect
from fastapi.responses import JSONResponse
import os, shutil, subprocess, json
import os, shutil, subprocess, json, time, threading, asyncio
app = FastAPI()
VIDEO_SAVE_PATH = "/mnt/save/video"
IMAGE_SAVE_PATH = "/mnt/save/warning"
MODBUS_BIN_PATH = "/home/orangepi/RKApp/GPIOSignal/bin/sendGpioSignal"
GPIO_CONFIG_FILE = "/home/orangepi/RKApp/InitAuth/conf/.env"
os.makedirs(VIDEO_SAVE_PATH, exist_ok=True)
os.makedirs(IMAGE_SAVE_PATH, exist_ok=True)
# ============================== GPIO 状态控制器 ============================== #
# 全局状态变量
current_gpio_state = 'HIGH' # 当前GPIO状态 ('HIGH' 或 'LOW')
last_person_time = 0.0 # 最后一次检测到人的时间
gpio_state_lock = threading.Lock() # 线程锁
GPIO_DELAY_SECONDS = 2.0 # 人离开后的延迟时间（秒）
def update_gpio_config(value: str):
"""
更新GPIO配置文件中的 outPutMode 值
Args:
value: 'true' 或 'false'
"""
try:
with open(GPIO_CONFIG_FILE, 'r', encoding='utf-8') as f:
lines = f.readlines()
modified = False
for i, line in enumerate(lines):
if line.strip().startswith('outPutMode'):
lines[i] = f"outPutMode:{value}\n"
modified = True
break
if not modified:
lines.append(f"outPutMode:{value}\n")
with open(GPIO_CONFIG_FILE, 'w', encoding='utf-8') as f:
f.writelines(lines)
print(f"[GPIO] 配置文件已更新: outPutMode={value}")
return True
except Exception as e:
print(f"[GPIO ERROR] 更新配置文件失败: {e}")
return False
def call_gpio_program():
"""调用GPIO控制程序"""
try:
signal = "echo 'orangepi' | sudo " + MODBUS_BIN_PATH
result = subprocess.run([signal], shell=True, capture_output=True, text=True, timeout=5)
if result.returncode == 0:
print(f"[GPIO] GPIO程序调用成功")
return True
else:
print(f"[GPIO ERROR] GPIO程序调用失败: {result.stderr}")
return False
except Exception as e:
print(f"[GPIO ERROR] 调用GPIO程序时发生异常: {e}")
return False
def set_gpio_low():
"""设置GPIO为低电平"""
global current_gpio_state
if update_gpio_config('false') and call_gpio_program():
current_gpio_state = 'LOW'
print(f"[GPIO] ✅ 已切换到低电平 (检测到人)")
return True
return False
def set_gpio_high():
"""设置GPIO为高电平"""
global current_gpio_state
if update_gpio_config('true') and call_gpio_program():
current_gpio_state = 'HIGH'
print(f"[GPIO] ✅ 已恢复高电平 (延迟{GPIO_DELAY_SECONDS}秒)")
return True
return False
def gpio_monitor_task():
"""
GPIO监控后台任务
定期检查是否需要恢复高电平
"""
global current_gpio_state, last_person_time
print("[GPIO] 🚀 GPIO监控线程已启动")
while True:
time.sleep(0.5) # 每0.5秒检查一次
with gpio_state_lock:
current_time = time.time()
time_since_last_person = current_time - last_person_time
# 如果当前是低电平，且距离上次检测到人已超过延迟时间
if (current_gpio_state == 'LOW' and
last_person_time > 0 and
time_since_last_person >= GPIO_DELAY_SECONDS):
print(f"[GPIO] ⏰ 距上次检测: {time_since_last_person:.1f}秒，准备恢复高电平")
set_gpio_high()
# 启动GPIO监控线程
gpio_monitor_thread = threading.Thread(target=gpio_monitor_task, daemon=True)
gpio_monitor_thread.start()
@app.websocket("/ws/distance")
@app.websocket("/ws/distance/")
async def websocket_distance(websocket: WebSocket):
global current_gpio_state, last_person_time
await websocket.accept()
print("✅ WebSocket 客户端已连接")
try:
@@ -26,7 +128,7 @@ async def websocket_distance(websocket: WebSocket):
msg = json.loads(data)
distance = msg.get("distance")
ts = msg.get("ts")
print(f"收到距离: {distance}, 时间戳: {ts}")
print(f"📨 收到距离: {distance}m, 时间戳: {ts}")
# # 写入日志，单独try，避免异常影响后续流程
# try:
@@ -36,33 +138,35 @@ async def websocket_distance(websocket: WebSocket):
# f.write(log_line)
# except Exception as log_e:
# print(f"日志写入失败: {log_e}")
#调用 Modbus 可执行文件
# print(f"调用 Modbus 程序，传递距离: {distance}")
signal = "echo 'orangepi' | sudo "+ MODBUS_BIN_PATH
result = subprocess.run(
[signal],
# shell=True,ocapture_output=True, text=True, timeout=0
shell=True
)
# print("signal:", signal)
# print(f"Modbus 程序返回: {result.stdout.strip()}, 错误: {result.stderr.strip()}")
if result.returncode != 0:
# 发送错误信息给客户端
await websocket.send_json({
"error": f"modbus 程序执行失败: {result.stderr.strip()}"
})
continue
# ⭐ GPIO状态控制逻辑
with gpio_state_lock:
current_time = time.time()
# 更新最后检测到人的时间
last_person_time = current_time
print(f"[GPIO] 🔄 更新最后检测时间")
# 如果当前是高电平，切换到低电平
if current_gpio_state == 'HIGH':
print(f"[GPIO] 🔻 当前为高电平，准备切换到低电平")
set_gpio_low()
else:
print(f"[GPIO] ⚡ 当前已是低电平，保持状态")
# 正常发送结果
# 发送响应给客户端
await websocket.send_json({
"status": "success",
"distance": distance,
"modbus_output": result.stdout.strip()
"gpio_state": current_gpio_state,
"message": f"已检测到人，GPIO状态: {current_gpio_state}"
})
except json.JSONDecodeError:
await websocket.send_text("invalid JSON")
except Exception as e:
await websocket.send_text(f"server error: {e}")
print(f"❌ 处理消息时出错: {e}")
except WebSocketDisconnect:
print("⚠️ WebSocket 客户端断开连接")

121
FastApi/fastApi.py.bak Normal file
View File

@@ -0,0 +1,121 @@
#本程序用于启用fastApi,用于与摄像头的数据交互建立连接
from fastapi import FastAPI, File, UploadFile, Form, WebSocket, WebSocketDisconnect
from fastapi.responses import JSONResponse
import os, shutil, subprocess, json
app = FastAPI()
VIDEO_SAVE_PATH = "/mnt/save/video"
IMAGE_SAVE_PATH = "/mnt/save/warning"
MODBUS_BIN_PATH = "/home/orangepi/RKApp/GPIOSignal/bin/sendGpioSignal"
os.makedirs(VIDEO_SAVE_PATH, exist_ok=True)
os.makedirs(IMAGE_SAVE_PATH, exist_ok=True)
@app.websocket("/ws/distance")
@app.websocket("/ws/distance/")
async def websocket_distance(websocket: WebSocket):
await websocket.accept()
print("✅ WebSocket 客户端已连接")
try:
while True:
data = await websocket.receive_text()
try:
msg = json.loads(data)
distance = msg.get("distance")
ts = msg.get("ts")
print(f"收到距离: {distance}, 时间戳: {ts}")
# # 写入日志，单独try，避免异常影响后续流程
# try:
# now = datetime.now()
# log_line = f"{now.strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]} distance={distance} ts={ts}\n"
# with open("/home/orangepi/Opencv/time.txt", "a") as f:
# f.write(log_line)
# except Exception as log_e:
# print(f"日志写入失败: {log_e}")
#调用 Modbus 可执行文件
# print(f"调用 Modbus 程序，传递距离: {distance}")
signal = "echo 'orangepi' | sudo "+ MODBUS_BIN_PATH
result = subprocess.run(
[signal],
# shell=True,ocapture_output=True, text=True, timeout=0
shell=True
)
# print("signal:", signal)
# print(f"Modbus 程序返回: {result.stdout.strip()}, 错误: {result.stderr.strip()}")
if result.returncode != 0:
# 发送错误信息给客户端
await websocket.send_json({
"error": f"modbus 程序执行失败: {result.stderr.strip()}"
})
continue
# 正常发送结果
await websocket.send_json({
"distance": distance,
"modbus_output": result.stdout.strip()
})
except json.JSONDecodeError:
await websocket.send_text("invalid JSON")
except Exception as e:
await websocket.send_text(f"server error: {e}")
except WebSocketDisconnect:
print("⚠️ WebSocket 客户端断开连接")
except Exception as e:
print("❌ WebSocket 处理出错:", e)
@app.post("/upload_video/")
async def upload_video(
video: UploadFile = File(..., description="上传的视频（.mp4）")
):
if not video.filename.lower().endswith('.mp4'):
return JSONResponse(status_code=400, content={"error": "视频必须为.mp4格式"})
video_path = os.path.join(VIDEO_SAVE_PATH, video.filename)
with open(video_path, "wb") as vid_file:
shutil.copyfileobj(video.file, vid_file)
return {"video_saved_to": video_path}
@app.post("/upload_image/")
async def upload_image(
image: UploadFile = File(..., description="上传的图片（.jpg）")
):
if not image.filename.lower().endswith('.jpg'):
return JSONResponse(status_code=400, content={"error": "图片必须为.jpg格式"})
image_path = os.path.join(IMAGE_SAVE_PATH, image.filename)
with open(image_path, "wb") as img_file:
shutil.copyfileobj(image.file, img_file)
return {"image_saved_to": image_path}
@app.post("/upload_distance/")
async def upload_distance(
distance: float = Form(..., description="距离")
):
# 调用本地 modbus 程序，将距离作为参数传递
try:
#写入日志精确到毫秒
# now = datetime.now()
# log_line = f"{now.strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]} distance={distance} ts={ts}\n"
# with open("/home/orangepi/Opencv/time.txt", "a") as f:
# f.write(log_line)
result = subprocess.run(
[MODBUS_BIN_PATH, str(int(distance))],
capture_output=True,
text=True,
timeout=5
)
if result.returncode != 0:
return JSONResponse(status_code=500, content={"error": f"modbus程序执行失败: {result.stderr}"})
return {
"distance": distance,
"modbus_output": result.stdout.strip()
}
except Exception as e:
return JSONResponse(status_code=500, content={"error": str(e)})

BIN
GPIOSignal/bin/sendGpioSignal
View File

Binary file not shown.

12
GPIOSignal/src/sendGpioSignal.cpp
View File

@@ -19,7 +19,7 @@ Parmas:
argv[2]: 设置引脚为高/低电平
*/
const string SetFile = "/home/orangepi/InitAuth/conf/.env";
const string SetFile = "/home/orangepi/RKApp/InitAuth/conf/.env";
// 初始化GPIO引脚
int InitGpio(int GPIO_Pin1, int GPIO_Pin2);
@@ -59,9 +59,6 @@ int main(int argc, char *argv[])
this_thread::sleep_for(chrono::milliseconds(100));
WriteGpio(GPIO_Pin1, !value);
WriteGpio(GPIO_Pin2, !value);
return 0;
}
@@ -70,14 +67,14 @@ bool GetOutValue(int &value)
{
bool flag = true;
// 读取文件
ReadFile *rf =new ReadFile(SetFile);
if (rf->Open() == false)
ReadFile rf(SetFile);
if (rf.Open() == false)
{
cerr << "读取文件失败" << endl;
flag = false;
}
auto str = rf->ReadLines();
auto str = rf.ReadLines();
for (auto &ii : str)
{
if (ii.find("outPutMode") != string::npos)
@@ -86,6 +83,7 @@ bool GetOutValue(int &value)
}
}
rf.Close();
return flag;
}

113
GPIOSignal/src/sendGpioSignal.cpp.bak Normal file
View File

@@ -0,0 +1,113 @@
/*
本程序用于读取配置文件
根据配置文件发送高低电平
发送引脚固定:7,8
*/
#include <iostream>
#include <thread>
#include <wiringPi.h>
#include "/home/orangepi/RKApp/softWareInit/NetraLib/include/Netra.hpp"
using namespace std;
using namespace QCL;
/*
Parmas:
argv[1]: GPIO引脚编号
argv[2]: 设置引脚为高/低电平
*/
const string SetFile = "/home/orangepi/InitAuth/conf/.env";
// 初始化GPIO引脚
int InitGpio(int GPIO_Pin1, int GPIO_Pin2);
// 写入GPIO引脚
void WriteGpio(int GPIO_Pin, int value);
// 获取输出模式
bool GetOutValue(int &value);
int main(int argc, char *argv[])
{
int GPIO_Pin1 = 7;
int GPIO_Pin2 = 8;
int value = 0;
cout << "[sendGpioSignal] 启动，读取配置: " << SetFile << endl;
if (GetOutValue(value) == false)
{
cerr << "[sendGpioSignal] 未读取到 outPutMode，程序退出" << endl;
return -1;
}
cout << "[sendGpioSignal] 读取到 outPutMode=" << (value == 1 ? "true" : "false") << endl;
// 初始化GPIO引脚
if (InitGpio(GPIO_Pin1, GPIO_Pin2) != 0)
{
cout << "Error: Failed to initialize GPIO pin " << endl;
return 1;
}
// 写入GPIO引脚
cout << "[sendGpioSignal] 设置 GPIO(" << GPIO_Pin1 << "," << GPIO_Pin2 << ") 为 "
<< (value == 1 ? "HIGH" : "LOW") << endl;
WriteGpio(GPIO_Pin1, value);
WriteGpio(GPIO_Pin2, value);
cout << "[sendGpioSignal] 完成" << endl;
this_thread::sleep_for(chrono::milliseconds(100));
WriteGpio(GPIO_Pin1, !value);
WriteGpio(GPIO_Pin2, !value);
return 0;
}
// 获取输出模式
bool GetOutValue(int &value)
{
bool flag = true;
// 读取文件
ReadFile *rf =new ReadFile(SetFile);
if (rf->Open() == false)
{
cerr << "读取文件失败" << endl;
flag = false;
}
auto str = rf->ReadLines();
for (auto &ii : str)
{
if (ii.find("outPutMode") != string::npos)
{
value = (ii.substr(string("outPutMode:").size()) == "true" ? 1 : 0);
}
}
return flag;
}
// 初始化GPIO引脚
int InitGpio(int GPIO_Pin1, int GPIO_Pin2)
{
// 使用物理引脚编号，确保与实际排针一致
if (wiringPiSetupPhys() != 0)
{
return -1; // 初始化失败
}
pinMode(GPIO_Pin1, OUTPUT);
pinMode(GPIO_Pin2, OUTPUT);
digitalWrite(GPIO_Pin1, LOW); // 默认设置为低电平
digitalWrite(GPIO_Pin2, LOW); // 默认设置为低电平
return 0; // 初始化成功
}
// 写入GPIO引脚
void WriteGpio(int GPIO_Pin, int value)
{
digitalWrite(GPIO_Pin, value == 1 ? HIGH : LOW);
}

2013
GPIOSignal/src/yolov11_stereo_distance_ws.py Normal file
View File

File diff suppressed because it is too large Load Diff

4
InitAuth/conf/.env
View File

@@ -4,5 +4,5 @@ UUID:null
#以下配置用于获取云端发送秘钥
ServerPwd:"17227ca72f30f8bca8158bb78f25b43e"
#以下配置存储报警输出高低电平
outSignal:low
#以下配置存储GPIO输出高低电平--状态机
outPutMode:false

2421
PyApp/upload.log
View File

File diff suppressed because it is too large Load Diff

BIN
softWareInit/bin/wifi
View File

Binary file not shown.