99 lines
3.1 KiB
Python
99 lines
3.1 KiB
Python
"""
|
|
二维码识别模块 - 使用 OpenCV 识别二维码获取 serialNumber
|
|
"""
|
|
import cv2
|
|
import time
|
|
import logging
|
|
import numpy as np
|
|
from typing import Optional, Tuple
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# 尝试导入二维码识别库
|
|
try:
|
|
from pyzbar.pyzbar import decode as qr_decode
|
|
PYZBAR_AVAILABLE = True
|
|
except ImportError:
|
|
PYZBAR_AVAILABLE = False
|
|
logger.warning("pyzbar 未安装,尝试用 OpenCV 内置 QRCodeDetector")
|
|
|
|
|
|
class QRScanner:
|
|
"""二维码扫描器"""
|
|
|
|
def __init__(self, device_index: int = 0):
|
|
self.device_index = device_index
|
|
self._cap: Optional[cv2.VideoCapture] = None
|
|
self._qr_detector = cv2.QRCodeDetector() # OpenCV 内置二维码检测器
|
|
|
|
def open(self) -> bool:
|
|
"""打开摄像头"""
|
|
try:
|
|
# 强制 V4L2 后端,获取标准彩色格式(与 test/server.py 一致)
|
|
self._cap = cv2.VideoCapture(self.device_index, cv2.CAP_V4L2)
|
|
if self._cap.isOpened():
|
|
logger.info(f"摄像头 {self.device_index} 已打开 (V4L2)")
|
|
return True
|
|
else:
|
|
# fallback: 不指定后端
|
|
self._cap = cv2.VideoCapture(self.device_index)
|
|
if self._cap.isOpened():
|
|
logger.info(f"摄像头 {self.device_index} 已打开 (默认后端)")
|
|
return True
|
|
logger.error(f"无法打开摄像头 {self.device_index}")
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"摄像头打开失败: {e}")
|
|
return False
|
|
|
|
def close(self):
|
|
if self._cap:
|
|
self._cap.release()
|
|
self._cap = None
|
|
|
|
def read_frame(self) -> Optional[np.ndarray]:
|
|
"""读取一帧"""
|
|
if not self._cap or not self._cap.isOpened():
|
|
return None
|
|
ret, frame = self._cap.read()
|
|
if not ret:
|
|
return None
|
|
return frame
|
|
|
|
def detect_qr(self, frame: np.ndarray) -> Optional[str]:
|
|
"""从图像帧中检测二维码"""
|
|
if frame is None:
|
|
return None
|
|
try:
|
|
# OpenCV 内置二维码检测
|
|
data, vertices, _ = self._qr_detector.detectAndDecode(frame)
|
|
if data and len(data) > 0:
|
|
return data.strip()
|
|
except Exception as e:
|
|
logger.debug(f"二维码检测失败: {e}")
|
|
return None
|
|
|
|
def scan_once(self) -> Optional[str]:
|
|
"""扫描一次(读取一帧并检测)"""
|
|
frame = self.read_frame()
|
|
return self.detect_qr(frame)
|
|
|
|
def scan_with_retry(self, max_attempts: int = 5, interval: float = 0.5) -> Optional[str]:
|
|
"""多次扫描直到成功或达到最大次数"""
|
|
for i in range(max_attempts):
|
|
result = self.scan_once()
|
|
if result:
|
|
return result
|
|
time.sleep(interval)
|
|
return None
|
|
|
|
def get_preview_frame(self) -> Optional[np.ndarray]:
|
|
"""获取预览帧(用于界面显示)"""
|
|
return self.read_frame()
|
|
|
|
def __enter__(self):
|
|
self.open()
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
self.close() |