190 lines
6.5 KiB
Python
190 lines
6.5 KiB
Python
# hik_camera_app.py
|
||
# 菜品识别：海康相机 / 本地 videos 回放（UI 复用 ui.classifier_ui）
|
||
|
||
import os
|
||
import sys
|
||
import threading
|
||
import time
|
||
|
||
import cv2
|
||
|
||
if hasattr(sys, 'frozen'):
|
||
os.environ['PYTHONIOENCODING'] = 'utf-8'
|
||
|
||
from dao.settings import DISH_USE_VIDEO_FOLDER, DISH_VIDEO_SOURCE_DIR
|
||
from ui.classifier_ui import BaseDishApp, run_app
|
||
|
||
CAMERA_AVAILABLE = False
|
||
HikCameraCapture = None
|
||
|
||
if not DISH_USE_VIDEO_FOLDER:
|
||
dll_path = r'C:\Program Files (x86)\Common Files\MVS\Runtime\Win64_x64\MvCameraControl.dll'
|
||
if not os.path.exists(dll_path):
|
||
raise FileNotFoundError(f"找不到海康 DLL：{dll_path}")
|
||
try:
|
||
from ctypes import WinDLL
|
||
MvCamCtrldll = WinDLL(dll_path)
|
||
print(f"[OK] loaded DLL: {dll_path}")
|
||
except Exception as e:
|
||
raise RuntimeError(f"加载海康 DLL 失败：{e}")
|
||
try:
|
||
from camera.hik_camera_tool import HikCameraCapture
|
||
CAMERA_AVAILABLE = True
|
||
except ImportError:
|
||
CAMERA_AVAILABLE = False
|
||
|
||
print("[WARN] hik camera tool not found")
|
||
else:
|
||
print(f"[INFO] 视频回放模式：读取目录 {os.path.abspath(DISH_VIDEO_SOURCE_DIR)}")
|
||
|
||
|
||
class ImageClassifierDemo(BaseDishApp):
|
||
window_title = "菜品识别系统"
|
||
app_title = "菜品识别系统"
|
||
edition_label = "视频回放版" if DISH_USE_VIDEO_FOLDER else "海康相机版"
|
||
camera_display_name = "视频回放" if DISH_USE_VIDEO_FOLDER else "海康相机"
|
||
|
||
def __init__(self, root):
|
||
self.camera = None
|
||
self._video_source = None
|
||
self._frame_lock = threading.Lock()
|
||
super().__init__(root)
|
||
|
||
def auto_connect_camera(self):
|
||
if DISH_USE_VIDEO_FOLDER:
|
||
self._connect_video_folder()
|
||
return
|
||
if not CAMERA_AVAILABLE:
|
||
self.root.after(0, self._update_camera_ui_fail)
|
||
return
|
||
|
||
def connect_thread():
|
||
try:
|
||
self.camera = HikCameraCapture()
|
||
devices = self.camera.enum_devices()
|
||
if devices:
|
||
self.camera.connect_camera(0)
|
||
self.camera.start_grabbing()
|
||
self.camera_connected = True
|
||
self.root.after(0, self._update_camera_ui_success)
|
||
return
|
||
self.camera = None
|
||
except Exception as e:
|
||
self.camera = None
|
||
print(f"海康相机连接失败：{e}")
|
||
self.root.after(0, self._update_camera_ui_fail)
|
||
|
||
threading.Thread(target=connect_thread, daemon=True).start()
|
||
|
||
def _connect_video_folder(self):
|
||
def connect_thread():
|
||
try:
|
||
from service.video_folder_source import VideoFolderSource
|
||
src = VideoFolderSource(DISH_VIDEO_SOURCE_DIR)
|
||
if src.open():
|
||
self._video_source = src
|
||
self.camera_connected = True
|
||
self.root.after(0, self._update_camera_ui_success)
|
||
return
|
||
self._video_source = None
|
||
except Exception as e:
|
||
self._video_source = None
|
||
print(f"视频目录打开失败：{e}")
|
||
self.root.after(0, self._update_camera_ui_fail)
|
||
|
||
threading.Thread(target=connect_thread, daemon=True).start()
|
||
|
||
def grab_preview_frame(self):
|
||
"""内存取 BGR，供流畅预览（不落盘）。"""
|
||
if DISH_USE_VIDEO_FOLDER:
|
||
if not self.camera_connected or self._video_source is None:
|
||
return None
|
||
with self._frame_lock:
|
||
try:
|
||
return self._video_source.read_bgr()
|
||
except Exception as e:
|
||
print(f"视频取帧失败: {e}")
|
||
return None
|
||
if not self.camera_connected or self.camera is None:
|
||
return None
|
||
with self._frame_lock:
|
||
try:
|
||
return self.camera.grab_bgr_frame(timeout_ms=200)
|
||
except Exception as e:
|
||
print(f"海康取帧失败: {e}")
|
||
return None
|
||
|
||
def get_video_progress(self):
|
||
if not DISH_USE_VIDEO_FOLDER or self._video_source is None:
|
||
return None
|
||
try:
|
||
return self._video_source.get_progress()
|
||
except Exception:
|
||
return None
|
||
|
||
def seek_video_ratio(self, ratio):
|
||
if not DISH_USE_VIDEO_FOLDER or self._video_source is None:
|
||
return False
|
||
with self._frame_lock:
|
||
try:
|
||
return bool(self._video_source.seek_ratio(ratio))
|
||
except Exception as e:
|
||
print(f"视频跳转失败: {e}")
|
||
return False
|
||
|
||
def take_live_frame(self):
|
||
"""识别用：优先复用预览缓存帧；不落盘（有订单时由 _save_order_image 写入）。"""
|
||
frame = None
|
||
with getattr(self, "_live_lock", self._frame_lock):
|
||
latest = getattr(self, "_latest_frame", None)
|
||
if latest is not None:
|
||
try:
|
||
frame = latest.copy()
|
||
except Exception:
|
||
frame = None
|
||
if frame is None:
|
||
frame = self.grab_preview_frame()
|
||
if frame is None:
|
||
return -1, None, None
|
||
return 0, None, frame
|
||
|
||
def take_photo(self):
|
||
os.makedirs("images", exist_ok=True)
|
||
capture_time = time.strftime("%Y%m%d_%H%M%S")
|
||
save_path = f'images/capture_{capture_time}.jpg'
|
||
frame = self.grab_preview_frame()
|
||
if frame is None:
|
||
return -1, None
|
||
try:
|
||
if cv2.imwrite(save_path, frame):
|
||
return 0, save_path
|
||
except Exception as e:
|
||
print(f"拍照失败: {e}")
|
||
return -1, None
|
||
|
||
def close_camera(self):
|
||
if self._video_source is not None:
|
||
try:
|
||
self._video_source.close()
|
||
except Exception as e:
|
||
print(f"关闭视频源失败: {e}")
|
||
finally:
|
||
self._video_source = None
|
||
if self.camera is not None:
|
||
try:
|
||
self.camera.finalize()
|
||
except Exception as e:
|
||
print(f"释放海康相机失败: {e}")
|
||
finally:
|
||
self.camera = 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)
|