143 lines
4.5 KiB
Python
143 lines
4.5 KiB
Python
# usb_camera_app.py
|
||
# 菜品识别：USB 相机实时视频流（UI 复用 ui.classifier_ui）
|
||
|
||
import os
|
||
import sys
|
||
import threading
|
||
import time
|
||
|
||
import cv2
|
||
|
||
if hasattr(sys, 'frozen'):
|
||
os.environ['PYTHONIOENCODING'] = 'utf-8'
|
||
|
||
from ui.classifier_ui import BaseDishApp, run_app
|
||
|
||
|
||
def _open_usb_capture(index=0):
|
||
"""Windows 优先 DirectShow，避免默认 MSMF 打开要几十秒。"""
|
||
candidates = []
|
||
if sys.platform.startswith('win'):
|
||
candidates.append(('DSHOW', cv2.CAP_DSHOW))
|
||
candidates.append(('MSMF', cv2.CAP_MSMF))
|
||
candidates.append(('ANY', cv2.CAP_ANY))
|
||
|
||
last_err = None
|
||
for name, backend in candidates:
|
||
cap = None
|
||
try:
|
||
t0 = time.perf_counter()
|
||
cap = cv2.VideoCapture(index, backend)
|
||
if not cap.isOpened():
|
||
cap.release()
|
||
continue
|
||
cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
|
||
cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640)
|
||
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)
|
||
ok, frame = cap.read()
|
||
elapsed = time.perf_counter() - t0
|
||
if ok and frame is not None and frame.size > 0:
|
||
print(f'[OK] USB camera opened via {name} in {elapsed:.2f}s')
|
||
return cap
|
||
cap.release()
|
||
except Exception as exc:
|
||
last_err = exc
|
||
try:
|
||
if cap is not None:
|
||
cap.release()
|
||
except Exception:
|
||
pass
|
||
if last_err:
|
||
print(f'[WARN] USB open attempts failed: {last_err}')
|
||
return None
|
||
|
||
|
||
class ImageClassifierDemo(BaseDishApp):
|
||
window_title = "菜品识别系统"
|
||
app_title = "菜品识别系统"
|
||
edition_label = "USB 相机版"
|
||
camera_display_name = "USB相机"
|
||
|
||
def __init__(self, root):
|
||
self.cap = None
|
||
self._frame_lock = threading.Lock()
|
||
self._cam_ready = threading.Event()
|
||
self._cam_ok = False
|
||
# 与 YOLO/UI 初始化并行打开相机，缩短出画面时间
|
||
threading.Thread(target=self._connect_camera_worker, daemon=True).start()
|
||
super().__init__(root)
|
||
|
||
def _connect_camera_worker(self):
|
||
try:
|
||
cap = _open_usb_capture(0)
|
||
if cap is not None:
|
||
self.cap = cap
|
||
self._cam_ok = True
|
||
else:
|
||
self.cap = None
|
||
self._cam_ok = False
|
||
except Exception as e:
|
||
self.cap = None
|
||
self._cam_ok = False
|
||
print(f"USB相机连接失败：{e}")
|
||
finally:
|
||
self._cam_ready.set()
|
||
|
||
def auto_connect_camera(self):
|
||
def finish():
|
||
self._cam_ready.wait(timeout=90)
|
||
if self._cam_ok and self.cap is not None and self.cap.isOpened():
|
||
self.camera_connected = True
|
||
self.root.after(0, self._update_camera_ui_success)
|
||
else:
|
||
self.camera_connected = False
|
||
self.root.after(0, self._update_camera_ui_fail)
|
||
|
||
threading.Thread(target=finish, daemon=True).start()
|
||
|
||
def grab_preview_frame(self):
|
||
if not self.camera_connected or self.cap is None:
|
||
return None
|
||
with self._frame_lock:
|
||
try:
|
||
self.cap.grab()
|
||
ret, frame = self.cap.retrieve()
|
||
if not ret or frame is None or frame.size == 0:
|
||
ret, frame = self.cap.read()
|
||
if ret and frame is not None and frame.size > 0:
|
||
return frame
|
||
except Exception as e:
|
||
print(f"USB取帧失败: {e}")
|
||
return None
|
||
|
||
def take_photo(self):
|
||
if not self.camera_connected or self.cap is None:
|
||
return -1, None
|
||
os.makedirs("images", exist_ok=True)
|
||
capture_time = time.strftime("%Y%m%d_%H%M%S")
|
||
save_path = os.path.abspath(f'images/capture_{capture_time}.jpg')
|
||
try:
|
||
frame = self.grab_preview_frame()
|
||
if frame is None:
|
||
return -1, None
|
||
if cv2.imwrite(save_path, frame):
|
||
return 0, save_path
|
||
except Exception as e:
|
||
print(f"USB拍照失败: {e}")
|
||
return -1, None
|
||
|
||
def close_camera(self):
|
||
if self.cap is not None:
|
||
self.cap.release()
|
||
self.cap = None
|
||
self.camera_connected = False
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
from service.media_server import ensure_media_server
|
||
ensure_media_server()
|
||
except Exception as exc:
|
||
print(f'[WARN] media server start failed: {exc}')
|
||
run_app(ImageClassifierDemo)
|