#本程序用于启用fastApi,用于与摄像头的数据交互建立连接 from fastapi import FastAPI, File, UploadFile, Form, WebSocket, WebSocketDisconnect from fastapi.responses import JSONResponse 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" Camera_Config_File = "/opt/rknn-yolov11/.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: while True: data = await websocket.receive_text() try: msg = json.loads(data) distance = msg.get("distance") ts = msg.get("ts") print(f"📨 收到距离: {distance}m, 时间戳: {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}") # ⭐ GPIO状态控制逻辑 with gpio_state_lock: current_time = time.time() # 读取配置文件中的危险距离阈值 danger_distance = 3.0 # 默认值 try: with open(GPIO_CONFIG_FILE, 'r', encoding='utf-8') as f: lines = f.readlines() for line in lines: if line.strip().startswith('MAX_DISTANCE'): danger_distance = float(line.strip().split(':')[1]) break except Exception as e: print(f"[Config] ⚠️ 读取危险距离失败，使用默认值3.0m: {e}") print(f"[Config] 📏 危险距离阈值: {danger_distance}m, 当前距离: {distance}m") # 判断是否在危险区域内 if distance is not None and distance < danger_distance: # 人在危险区域内，更新时间戳 last_person_time = current_time print(f"[GPIO] ⚠️ 危险！距离 {distance:.2f}m < {danger_distance:.2f}m") # 如果当前是高电平，切换到低电平 if current_gpio_state == 'HIGH': print(f"[GPIO] 🔻 当前为高电平，准备切换到低电平") set_gpio_low() else: print(f"[GPIO] ⚡ 当前已是低电平，保持状态") else: # 人在安全区域，不触发GPIO print(f"[GPIO] ✅ 安全距离 {distance:.2f}m >= {danger_distance:.2f}m，不触发GPIO") # # 发送响应给客户端 # await websocket.send_json({ # "status": "success", # "distance": distance, # "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 客户端断开连接") 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)})