工业视觉 OCR 与瓶盖缺陷检测结合 IoTDA 智慧物联应用实践
·
1 工效组合展示
2 具身智能工业视觉获取与分析应用
2.1 工具软件安装
2.2 工业视觉环境构建与应用
2.3 昇腾工业视觉终端应用展现
3 具身智能工业视觉识别快速编码实现
3.1 OCR识别
3.1.1 IMA-DS借力
给出python虚拟环境OCR文字识别的编码及其运行过程，包括汉字，windows环境，非COnDA，使用本机连接的网络摄像机，windows只能使用CPU，整个代码需要用一个文件表示，我的python版本是3.9.10，摄像头使用的是usb连接，我现在需要调用海康机器人的工业相机，请帮我在程序中再加上可以调用MVS数据库的程序，请你也适配一下设备
3.1.2 运行环境构造
3.1.3编码实现
# -*- coding: utf-8 -*-
"""
海康机器人工业相机(MVS SDK) 实时中文 OCR 文字识别 —— 单文件版
环境: Windows 10/11 + Python 3.9.10(64位) + venv(非conda) + CPU
相机: 海康机器人工业相机（GigE 网口 / USB3 均可）
按键: q=退出 s=截图 c=立即识别
"""
import os
# 必须在 import paddleocr 之前禁用 oneDNN，否则 paddlepaddle 3.3.x CPU 会报错
os.environ["PADDLE_PDX_ENABLE_MKLDNN_BYDEFAULT"] = "False"
import sys
import time
import argparse
import cv2
import numpy as np
from ctypes import *
from PIL import Image, ImageDraw, ImageFont
if sys.platform == "win32":
try:
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
sys.stderr.reconfigure(encoding="utf-8", errors="replace")
except Exception:
pass
# ==================== 配置区 ====================
# 海康 MVS 的 Python SDK 目录（MvImport）。
# 推荐：把 MvImport 里的 .py 文件复制到本文件同目录，这里留空即可；
# 或填写完整路径，例如：
# MVS_IMPORT_PATH = r"C:\Program Files (x86)\MVS\Development\Samples\Python\MvImport"
MVS_IMPORT_PATH = ""
CAMERA_INDEX = 0 # 相机索引（0=第一台）
RESIZE_MAX_SIDE = 1280 # 图像最长边超过该值则等比缩小（提速；0=不缩放）
RECOGNIZE_INTERVAL = 1.0 # 识别间隔(秒)
CONF_THRESHOLD = 0.5 # 置信度阈值
SHOW_WINDOW = True
SAVE_LOG = True
LOG_FILE = "ocr_result.log"
FONT_CANDIDATES = [r"C:\Windows\Fonts\msyh.ttc",
r"C:\Windows\Fonts\simhei.ttf",
r"C:\Windows\Fonts\simsun.ttc"]
FONT_SIZE = 20
# ================================================
if MVS_IMPORT_PATH and os.path.isdir(MVS_IMPORT_PATH):
sys.path.insert(0, MVS_IMPORT_PATH)
try:
from MvCameraControl_class import *
except ImportError:
print("[错误] 找不到海康 MVS 的 Python SDK（MvCameraControl_class.py）")
print("请把 MVS 安装目录下 MvImport 文件夹里的 .py 文件复制到本程序同目录，")
print("或在本文件配置区 MVS_IMPORT_PATH 填写 MvImport 完整路径。")
sys.exit(1)
def log(msg):
line = "[{}] {}".format(time.strftime("%Y-%m-%d %H:%M:%S"), msg)
print(line, flush=True)
if SAVE_LOG:
try:
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(line + "\n")
except OSError:
pass
# 像素格式编码（GigE Vision 国际标准，海康 MVS 遵循）
PF_MONO8 = 0x01080001 # 黑白
PF_RGB8 = 0x02180014 # 彩色 RGB
PF_BAYER_GR8 = 0x01080008 # 彩色 Bayer
PF_BAYER_RG8 = 0x01080009
PF_BAYER_GB8 = 0x0108000A
PF_BAYER_BG8 = 0x0108000B
class HKCamera(object):
"""海康机器人工业相机封装（MVS SDK）"""
def __init__(self, index=0):
self.index = index
self.cam = None
self.payload_size = 0
self.data_buf = None
self.frame_info = None
def connect(self):
self.release()
try:
device_list = MV_CC_DEVICE_INFO_LIST()
ret = MvCamera.MV_CC_EnumDevices(MV_GIGE_DEVICE | MV_USB_DEVICE, device_list)
if ret != 0:
log("枚举设备失败，错误码 0x{:X}".format(ret))
return False
if device_list.nDeviceNum == 0:
log("未找到海康工业相机：请检查连接/驱动，或 MVS 客户端是否占用")
return False
if self.index >= device_list.nDeviceNum:
log("相机索引 {} 超出范围（共 {} 台）".format(self.index, device_list.nDeviceNum))
return False
st_dev = cast(device_list.pDeviceInfo[self.index],
POINTER(MV_CC_DEVICE_INFO)).contents
self.cam = MvCamera()
ret = self.cam.MV_CC_CreateHandle(st_dev)
if ret != 0:
log("创建句柄失败，错误码 0x{:X}".format(ret))
self.release()
return False
ret = self.cam.MV_CC_OpenDevice(MV_ACCESS_Exclusive, 0)
if ret != 0:
log("打开设备失败，错误码 0x{:X}。请确认 MVS 客户端已关闭".format(ret))
self.release()
return False
# 关闭触发模式 -> 连续采集（关键，否则可能取不到流）
self.cam.MV_CC_SetEnumValue("TriggerMode", MV_TRIGGER_MODE_OFF)
st_param = MVCC_INTVALUE()
memset(byref(st_param), 0, sizeof(st_param))
ret = self.cam.MV_CC_GetIntValue("PayloadSize", st_param)
if ret != 0:
log("获取 PayloadSize 失败，错误码 0x{:X}".format(ret))
self.release()
return False
self.payload_size = st_param.nCurValue
ret = self.cam.MV_CC_StartGrabbing()
if ret != 0:
log("开始取流失败，错误码 0x{:X}".format(ret))
self.release()
return False
self.data_buf = (c_ubyte * self.payload_size)()
self.frame_info = MV_FRAME_OUT_INFO_EX()
log("海康相机打开成功：第 {} 台，PayloadSize={}".format(self.index, self.payload_size))
return True
except Exception as e:
log("连接相机异常: {}".format(e))
self.release()
return False
def read(self):
if self.cam is None:
return False, None
try:
memset(byref(self.frame_info), 0, sizeof(self.frame_info))
ret = self.cam.MV_CC_GetOneFrameTimeout(
byref(self.data_buf), self.payload_size, self.frame_info, 1000)
if ret != 0:
return False, None
frame = self._to_bgr()
return frame is not None, frame
except Exception:
return False, None
def _to_bgr(self):
w = self.frame_info.nWidth
h = self.frame_info.nHeight
if w <= 0 or h <= 0:
return None
pt = self.frame_info.enPixelType
n = w * h
try:
if pt == PF_MONO8:
img = np.frombuffer(self.data_buf, dtype=np.uint8, count=n).reshape(h, w)
return cv2.cvtColor(img, cv2.COLOR_GRAY2BGR)
elif pt == PF_RGB8:
img = np.frombuffer(self.data_buf, dtype=np.uint8, count=n * 3).reshape(h, w, 3)
return cv2.cvtColor(img, cv2.COLOR_RGB2BGR)
elif pt == PF_BAYER_RG8:
img = np.frombuffer(self.data_buf, dtype=np.uint8, count=n).reshape(h, w)
return cv2.cvtColor(img, cv2.COLOR_BayerRG2BGR)
elif pt == PF_BAYER_GB8:
img = np.frombuffer(self.data_buf, dtype=np.uint8, count=n).reshape(h, w)
return cv2.cvtColor(img, cv2.COLOR_BayerGB2BGR)
elif pt == PF_BAYER_GR8:
img = np.frombuffer(self.data_buf, dtype=np.uint8, count=n).reshape(h, w)
return cv2.cvtColor(img, cv2.COLOR_BayerGR2BGR)
elif pt == PF_BAYER_BG8:
img = np.frombuffer(self.data_buf, dtype=np.uint8, count=n).reshape(h, w)
return cv2.cvtColor(img, cv2.COLOR_BayerBG2BGR)
else:
log("不支持的像素格式 0x{:X}，请用 MVS 客户端把 PixelFormat 改为 Mono8 或 RGB8".format(pt))
return None
except Exception as e:
log("图像转换失败: {}".format(e))
return None
def release(self):
if self.cam is not None:
for fn in ("MV_CC_StopGrabbing", "MV_CC_CloseDevice", "MV_CC_DestroyHandle"):
try:
getattr(self.cam, fn)()
except Exception:
pass
self.cam = None
def resize_if_needed(frame):
if RESIZE_MAX_SIDE <= 0:
return frame
h, w = frame.shape[:2]
longest = max(h, w)
if longest > RESIZE_MAX_SIDE:
scale = RESIZE_MAX_SIDE / float(longest)
frame = cv2.resize(frame, (int(w * scale), int(h * scale)), interpolation=cv2.INTER_AREA)
return frame
def init_ocr():
from paddleocr import PaddleOCR
log("正在初始化 PaddleOCR 中文模型(首次运行自动下载，请稍候)...")
ocr = PaddleOCR(
text_detection_model_name="PP-OCRv4_mobile_det",
text_recognition_model_name="PP-OCRv4_mobile_rec",
use_doc_orientation_classify=False,
use_doc_unwarping=False,
use_textline_orientation=True,
)
log("OCR 模型就绪")
return ocr
def parse_result(result):
items = []
if not result:
return items
res = result[0]
texts = res.get("rec_texts")
polys = res.get("rec_polys")
scores = res.get("rec_scores")
if texts is None:
return items
for i, txt in enumerate(texts):
score = float(scores[i]) if scores is not None and i < len(scores) else 1.0
poly = polys[i] if polys is not None and i < len(polys) else None
items.append((txt, poly, score))
return items
def draw_results(frame, items, font):
if not items:
return frame
pil_img = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
draw = ImageDraw.Draw(pil_img)
for txt, poly, score in items:
color = (255, 0, 0)
label = "{} {:.2f}".format(txt, score)
if poly is not None and len(poly) >= 4:
pts = [(int(float(p[0])), int(float(p[1]))) for p in poly]
draw.polygon(pts, outline=color, width=2)
x0 = min(p[0] for p in pts)
y0 = min(p[1] for p in pts)
else:
x0, y0 = 10, 30
if font is not None:
draw.text((x0, max(0, y0 - FONT_SIZE - 4)), label, fill=color, font=font)
return cv2.cvtColor(np.array(pil_img), cv2.COLOR_RGB2BGR)
def load_font():
try:
for p in FONT_CANDIDATES:
if os.path.exists(p):
return ImageFont.truetype(p, FONT_SIZE)
log("警告: 未找到中文字体，画面只画框")
return None
except Exception as e:
log("警告: 中文字体加载失败({})".format(e))
return None
def main():
p = argparse.ArgumentParser()
p.add_argument("--camera", type=int, default=CAMERA_INDEX, help="相机索引(0/1/2...)")
p.add_argument("--interval", type=float, default=RECOGNIZE_INTERVAL)
p.add_argument("--threshold", type=float, default=CONF_THRESHOLD)
p.add_argument("--no-window", action="store_true")
args = p.parse_args()
show_window = (not args.no_window) and SHOW_WINDOW
font = load_font()
ocr = init_ocr()
cam = HKCamera(args.camera)
if not cam.connect():
log("相机打开失败，程序退出。请关闭 MVS 客户端后重试，或换 --camera 1")
return
log("开始识别: 间隔{}s / 阈值{}".format(args.interval, args.threshold))
last_ocr_time = 0.0
last_texts = []
fps_start, fps_frames = time.time(), 0
while True:
ok, frame = cam.read()
if not ok:
log("相机取流中断，重连中...")
if not cam.connect():
time.sleep(3)
continue
continue
frame = resize_if_needed(frame)
if show_window:
key = cv2.waitKey(1) & 0xFF
else:
key = -1
time.sleep(0.01)
do_ocr = (time.time() - last_ocr_time >= args.interval) or key == ord("c")
if do_ocr:
last_ocr_time = time.time()
try:
items = parse_result(ocr.predict(frame))
items = [it for it in items if it[2] >= args.threshold]
if items:
last_texts = items
for txt, _, sc in items:
log("识别: {} (置信度 {:.3f})".format(txt, sc))
else:
last_texts = []
log("本帧未识别到文字")
except Exception as e:
log("OCR 出错: {}".format(e))
frame = draw_results(frame, last_texts, font)
fps_frames += 1
if time.time() - fps_start >= 1.0:
fps = fps_frames / (time.time() - fps_start)
cv2.putText(frame, "FPS:{:.1f}".format(fps), (10, 30),
cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 0), 2)
fps_start, fps_frames = time.time(), 0
if show_window:
cv2.imshow("HIKROBOT OCR (q=退出 s=截图 c=识别)", frame)
if key == ord("q"):
break
elif key == ord("s"):
fn = "snapshot_{}.jpg".format(time.strftime("%Y%m%d_%H%M%S"))
cv2.imwrite(fn, frame)
log("已保存截图: {}".format(fn))
cam.release()
cv2.destroyAllWindows()
log("程序结束")
if __name__ == "__main__":
main()
3.1.4 运行测试
3.2 缺陷检查（瓶盖有无）
3.2.1 IMA-DS借力
现在我需要重新弄一份python虚拟环境可以检查矿泉水瓶盖有无缺陷检查的编码及其运行过程，Windows环境，只用CPU运行，非conda，使用本机USB连接的海康工业相机（MV系列），编码需要用一个完整文件，我的摄像头离瓶盖位置较近，是10cm左右，而且我所需要的瓶盖可能会有毛糙，还有一部分颜色检测范围会出现错误
3.2.2 运行环境构造
3.2.3编码实现
# -*- coding: utf-8 -*-
"""
海康工业相机(MV系列) 矿泉水瓶盖缺陷检测 —— 单文件版
环境: Windows 10/11 + Python 3.9.10(64位) + venv(非conda) + CPU
依赖: opencv-python / pillow / 海康 MVS SDK（无需 PaddlePaddle）
按键: q=退出 s=截图 d=显示/隐藏颜色分割图(调阈值用)
"""
import os
import sys
import time
import argparse
import cv2
import numpy as np
from ctypes import *
from PIL import Image, ImageDraw, ImageFont
if sys.platform == "win32":
try:
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
sys.stderr.reconfigure(encoding="utf-8", errors="replace")
except Exception:
pass
# ==================== 相机配置区 ====================
MVS_IMPORT_PATH = "" # MvImport 复制到同目录则留空，否则填完整路径
CAMERA_INDEX = 0 # 相机索引（0=第一台）
RESIZE_MAX_SIDE = 1280 # 图像最长边超过该值则等比缩小（0=不缩放）
# ==================================================
# ==================== 检测配置区（按实际调整）====================
ROI = None # 瓶盖区域 (x,y,宽,高)，None=整个画面
# 瓶盖颜色 HSV 范围（默认蓝色瓶盖，务必按实际颜色调整）
# 调法：运行后按 d 键看分割图，调到瓶盖白色、背景黑色为止
CAP_COLOR_LOWER = (100, 60, 60)
CAP_COLOR_UPPER = (130, 255, 255)
MIN_AREA = 5000 # 瓶盖最小面积(像素)
MAX_AREA = 200000 # 瓶盖最大面积(像素)
MIN_FILL_RATIO = 0.80 # 最小填充率=面积/外接圆面积，正常圆形≈0.9+
CENTER_OFFSET = 40 # 圆心允许偏移(像素)
# =============================================================
DETECT_INTERVAL = 0.2
SAVE_LOG = True
LOG_FILE = "cap_inspection.log"
FONT_CANDIDATES = [r"C:\Windows\Fonts\msyh.ttc",
r"C:\Windows\Fonts\simhei.ttf",
r"C:\Windows\Fonts\simsun.ttc"]
FONT_SIZE = 22
if MVS_IMPORT_PATH and os.path.isdir(MVS_IMPORT_PATH):
sys.path.insert(0, MVS_IMPORT_PATH)
try:
from MvCameraControl_class import *
except ImportError:
print("[错误] 找不到海康 MVS 的 Python SDK（MvCameraControl_class.py）")
print("请把 MVS 安装目录下 MvImport 文件夹里的 .py 文件复制到本程序同目录，")
print("或在本文件 MVS_IMPORT_PATH 填写 MvImport 完整路径。")
sys.exit(1)
def log(msg):
line = "[{}] {}".format(time.strftime("%Y-%m-%d %H:%M:%S"), msg)
print(line, flush=True)
if SAVE_LOG:
try:
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(line + "\n")
except OSError:
pass
PF_MONO8 = 0x01080001
PF_RGB8 = 0x02180014
PF_BAYER_GR8 = 0x01080008
PF_BAYER_RG8 = 0x01080009
PF_BAYER_GB8 = 0x0108000A
PF_BAYER_BG8 = 0x0108000B
class HKCamera(object):
"""海康机器人工业相机封装（MVS SDK）"""
def __init__(self, index=0):
self.index = index
self.cam = None
self.payload_size = 0
self.data_buf = None
self.frame_info = None
def connect(self):
self.release()
try:
device_list = MV_CC_DEVICE_INFO_LIST()
ret = MvCamera.MV_CC_EnumDevices(MV_GIGE_DEVICE | MV_USB_DEVICE, device_list)
if ret != 0:
log("枚举设备失败，错误码 0x{:X}".format(ret))
return False
if device_list.nDeviceNum == 0:
log("未找到海康工业相机：请检查连接/驱动，或 MVS 客户端是否占用")
return False
if self.index >= device_list.nDeviceNum:
log("相机索引 {} 超出范围（共 {} 台）".format(self.index, device_list.nDeviceNum))
return False
st_dev = cast(device_list.pDeviceInfo[self.index],
POINTER(MV_CC_DEVICE_INFO)).contents
self.cam = MvCamera()
ret = self.cam.MV_CC_CreateHandle(st_dev)
if ret != 0:
log("创建句柄失败，错误码 0x{:X}".format(ret))
self.release()
return False
ret = self.cam.MV_CC_OpenDevice(MV_ACCESS_Exclusive, 0)
if ret != 0:
log("打开设备失败，错误码 0x{:X}。请确认 MVS 客户端已关闭".format(ret))
self.release()
return False
self.cam.MV_CC_SetEnumValue("TriggerMode", MV_TRIGGER_MODE_OFF)
st_param = MVCC_INTVALUE()
memset(byref(st_param), 0, sizeof(st_param))
ret = self.cam.MV_CC_GetIntValue("PayloadSize", st_param)
if ret != 0:
log("获取 PayloadSize 失败，错误码 0x{:X}".format(ret))
self.release()
return False
self.payload_size = st_param.nCurValue
ret = self.cam.MV_CC_StartGrabbing()
if ret != 0:
log("开始取流失败，错误码 0x{:X}".format(ret))
self.release()
return False
self.data_buf = (c_ubyte * self.payload_size)()
self.frame_info = MV_FRAME_OUT_INFO_EX()
log("海康相机打开成功：第 {} 台".format(self.index))
return True
except Exception as e:
log("连接相机异常: {}".format(e))
self.release()
return False
def read(self):
if self.cam is None:
return False, None
try:
memset(byref(self.frame_info), 0, sizeof(self.frame_info))
ret = self.cam.MV_CC_GetOneFrameTimeout(
byref(self.data_buf), self.payload_size, self.frame_info, 1000)
if ret != 0:
return False, None
frame = self._to_bgr()
return frame is not None, frame
except Exception:
return False, None
def _to_bgr(self):
w = self.frame_info.nWidth
h = self.frame_info.nHeight
if w <= 0 or h <= 0:
return None
pt = self.frame_info.enPixelType
n = w * h
try:
if pt == PF_MONO8:
img = np.frombuffer(self.data_buf, dtype=np.uint8, count=n).reshape(h, w)
return cv2.cvtColor(img, cv2.COLOR_GRAY2BGR)
elif pt == PF_RGB8:
img = np.frombuffer(self.data_buf, dtype=np.uint8, count=n * 3).reshape(h, w, 3)
return cv2.cvtColor(img, cv2.COLOR_RGB2BGR)
elif pt == PF_BAYER_RG8:
img = np.frombuffer(self.data_buf, dtype=np.uint8, count=n).reshape(h, w)
return cv2.cvtColor(img, cv2.COLOR_BayerRG2BGR)
elif pt == PF_BAYER_GB8:
img = np.frombuffer(self.data_buf, dtype=np.uint8, count=n).reshape(h, w)
return cv2.cvtColor(img, cv2.COLOR_BayerGB2BGR)
elif pt == PF_BAYER_GR8:
img = np.frombuffer(self.data_buf, dtype=np.uint8, count=n).reshape(h, w)
return cv2.cvtColor(img, cv2.COLOR_BayerGR2BGR)
elif pt == PF_BAYER_BG8:
img = np.frombuffer(self.data_buf, dtype=np.uint8, count=n).reshape(h, w)
return cv2.cvtColor(img, cv2.COLOR_BayerBG2BGR)
else:
log("不支持的像素格式 0x{:X}，请用 MVS 客户端改为 Mono8 或 RGB8".format(pt))
return None
except Exception as e:
log("图像转换失败: {}".format(e))
return None
def release(self):
if self.cam is not None:
for fn in ("MV_CC_StopGrabbing", "MV_CC_CloseDevice", "MV_CC_DestroyHandle"):
try:
getattr(self.cam, fn)()
except Exception:
pass
self.cam = None
def resize_if_needed(frame):
if RESIZE_MAX_SIDE <= 0:
return frame
h, w = frame.shape[:2]
longest = max(h, w)
if longest > RESIZE_MAX_SIDE:
scale = RESIZE_MAX_SIDE / float(longest)
frame = cv2.resize(frame, (int(w * scale), int(h * scale)), interpolation=cv2.INTER_AREA)
return frame
def detect_cap(roi):
"""检测瓶盖，返回 (是否OK, 缺陷列表, 检测信息, 分割掩码图)。"""
h, w = roi.shape[:2]
info = None
# 1) 高斯模糊：平滑颜色噪声和表面毛糙，减少反光/纹理造成的分割碎块
blurred = cv2.GaussianBlur(roi, (7, 7), 0)
# 2) HSV 颜色分割
hsv = cv2.cvtColor(blurred, cv2.COLOR_BGR2HSV)
mask = cv2.inRange(hsv, CAP_COLOR_LOWER, CAP_COLOR_UPPER)
# 3) 形态学：较大结构元平滑毛糙边缘、填小孔、去小噪点
kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (9, 9))
mask = cv2.morphologyEx(mask, cv2.MORPH_OPEN, kernel, iterations=1)
mask = cv2.morphologyEx(mask, cv2.MORPH_CLOSE, kernel, iterations=3)
# 4) 找轮廓
contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
if not contours:
return False, ["未检测到瓶盖(可能缺盖)"], info, mask
# 5) 取最大轮廓
largest = max(contours, key=cv2.contourArea)
area = cv2.contourArea(largest)
# 6) 填充率 = 面积 / 外接圆面积（对毛糙不敏感）
(cx, cy), radius = cv2.minEnclosingCircle(largest)
cx, cy, radius = int(cx), int(cy), int(radius)
enclosing_area = np.pi * radius * radius
fill_ratio = area / enclosing_area if enclosing_area > 0 else 0.0
info = {"cx": cx, "cy": cy, "radius": radius,
"area": int(area), "fill_ratio": round(fill_ratio, 3)}
# 7) 缺陷判定
defects = []
if area < MIN_AREA:
defects.append("面积过小(缺盖/残缺)")
elif area > MAX_AREA:
defects.append("面积过大(异常)")
if fill_ratio < MIN_FILL_RATIO:
defects.append("破损/缺口")
if ROI is not None:
offset = np.hypot(cx - w / 2.0, cy - h / 2.0)
info["offset"] = round(offset, 1)
if offset > CENTER_OFFSET:
defects.append("位置偏移/歪斜")
return len(defects) == 0, defects, info, mask
def draw_chinese(frame, text, pos, color, font):
pil = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
d = ImageDraw.Draw(pil)
if font is not None:
d.text(pos, text, fill=(color[2], color[1], color[0]), font=font)
return cv2.cvtColor(np.array(pil), cv2.COLOR_RGB2BGR)
def load_font():
try:
for p in FONT_CANDIDATES:
if os.path.exists(p):
return ImageFont.truetype(p, FONT_SIZE)
log("警告: 未找到中文字体，画面中文标注将不显示")
return None
except Exception as e:
log("警告: 中文字体加载失败({})".format(e))
return None
def main():
p = argparse.ArgumentParser()
p.add_argument("--camera", type=int, default=CAMERA_INDEX)
p.add_argument("--interval", type=float, default=DETECT_INTERVAL)
p.add_argument("--no-window", action="store_true")
args = p.parse_args()
show_window = not args.no_window
font = load_font()
cam = HKCamera(args.camera)
if not cam.connect():
log("相机打开失败，程序退出。请关闭 MVS 客户端后重试，或换 --camera 1")
return
log("开始瓶盖缺陷检测：间隔 {}s".format(args.interval))
last_detect = 0.0
last_state = None
last_defects = []
show_mask = False
fps_start, fps_frames = time.time(), 0
while True:
ok, frame = cam.read()
if not ok:
log("相机取流中断，重连中...")
if not cam.connect():
time.sleep(3)
continue
continue
frame = resize_if_needed(frame)
H, W = frame.shape[:2]
if ROI is not None:
x, y, rw, rh = ROI
x = max(0, min(x, W - 1)); y = max(0, min(y, H - 1))
rw = min(rw, W - x); rh = min(rh, H - y)
roi = frame[y:y + rh, x:x + rw]
else:
x, y, rw, rh = 0, 0, W, H
roi = frame
if show_window:
key = cv2.waitKey(1) & 0xFF
else:
key = -1
time.sleep(0.01)
if key == ord("d"):
show_mask = not show_mask
do_detect = (time.time() - last_detect >= args.interval)
if do_detect:
last_detect = time.time()
is_ok, defects, info, mask = detect_cap(roi)
last_defects = defects
state = "OK" if is_ok else "NG"
if state != last_state:
last_state = state
if is_ok:
log("结果: OK 面积={} 填充率={}".format(info["area"], info["fill_ratio"]) if info else "结果: OK")
else:
log("结果: NG 缺陷: {}".format("、".join(defects)))
else:
is_ok = (last_state == "OK")
info = None
mask = None
if show_mask and mask is not None:
display = cv2.cvtColor(mask, cv2.COLOR_GRAY2BGR)
else:
display = frame
if ROI is not None:
cv2.rectangle(display, (x, y), (x + rw, y + rh), (255, 200, 0), 2)
color = (0, 255, 0) if is_ok else (0, 0, 255)
status_text = "OK" if is_ok else "NG: " + "、".join(last_defects)
if font is not None:
display = draw_chinese(display, status_text, (10, 10), color, font)
else:
cv2.putText(display, status_text, (10, 35),
cv2.FONT_HERSHEY_SIMPLEX, 1.0, color, 2)
if not show_mask and info is not None:
cv2.circle(display, (x + info["cx"], y + info["cy"]), info["radius"], color, 3)
cv2.circle(display, (x + info["cx"], y + info["cy"]), 3, color, -1)
cv2.putText(display, "area:{} fill:{:.2f}".format(info["area"], info["fill_ratio"]),
(x + info["cx"] - info["radius"], max(10, y + info["cy"] - info["radius"] - 10)),
cv2.FONT_HERSHEY_SIMPLEX, 0.6, color, 2)
fps_frames += 1
if time.time() - fps_start >= 1.0:
fps = fps_frames / (time.time() - fps_start)
cv2.putText(display, "FPS:{:.1f}".format(fps), (W - 140, 30),
cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 255), 2)
fps_start, fps_frames = time.time(), 0
if show_window:
cv2.imshow("Cap Defect Inspection (q=退出 s=截图 d=分割图)", display)
if key == ord("q"):
break
elif key == ord("s"):
fn = "cap_{}.jpg".format(time.strftime("%Y%m%d_%H%M%S"))
cv2.imwrite(fn, display)
log("已保存截图: {}".format(fn))
cam.release()
cv2.destroyAllWindows()
log("程序结束")
if __name__ == "__main__":
main()
3.2.4 运行测试
3.2.5 运行中可能出现的问题及处理方法
在实验中可能会出现绿色识别范围不全，识别范围错误的问题其修改过后的代码为：
# -*- coding: utf-8 -*-
"""
海康工业相机(MV系列) 矿泉水瓶盖缺陷检测 —— 单文件版
========================================================
运行环境：Windows 10/11 + Python 3.9.10(64位) + venv(非conda) + CPU
相机：海康机器人工业相机（USB3 或 GigE，通过 MVS SDK 调用）
依赖：opencv-python / pillow / 海康 MVS SDK（无需 PaddlePaddle）
检测原理（传统图像处理，CPU 实时）：
1. 默认用灰度 Otsu 自动阈值分割瓶盖（瓶盖亮、背景暗，红/绿/白盖通用）
2. 提取瓶盖轮廓，计算面积 / 填充率 / 圆心位置
3. 根据阈值判断缺陷：缺盖、破损/缺口、面积异常、位置偏移
4. 可选 HSV 颜色分割模式（DETECT_MODE="color"）
【使用前必做】
1. 安装海康 MVS，把 MvImport 文件夹里的 .py 文件复制到本文件同目录
（或配置 MVS_IMPORT_PATH 为 MvImport 完整路径）
2. 运行本程序前关闭 MVS 客户端
按键：q=退出 s=截图 d=显示/隐藏分割图
"""
import os
import sys
import time
import argparse
import cv2
import numpy as np
from ctypes import *
from PIL import Image, ImageDraw, ImageFont
if sys.platform == "win32":
try:
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
sys.stderr.reconfigure(encoding="utf-8", errors="replace")
except Exception:
pass
# ==================== 相机配置区 ====================
MVS_IMPORT_PATH = "" # MvImport 复制到同目录则留空，否则填完整路径
CAMERA_INDEX = 0 # 相机索引（0=第一台）
RESIZE_MAX_SIDE = 1280 # 图像最长边超过该值则等比缩小（0=不缩放）
# ==================================================
# ==================== 检测配置区（按实际调整）====================
# 检测模式：
# "gray" = 灰度Otsu自动阈值（推荐！瓶盖亮、背景暗即可，红/绿/白盖全部通用，免调颜色）
# "color" = HSV颜色分割（仅当灰度分割不适合时才用，需按颜色调范围）
DETECT_MODE = "gray"
ROI = None # 瓶盖区域 (x,y,宽,高)，None=整个画面
# --- 灰度模式参数（DETECT_MODE="gray" 时生效）---
BRIGHT_CAP = True # True=瓶盖比背景亮（你的现场就是这种）；False=瓶盖比背景暗
# --- 颜色分割模式参数（DETECT_MODE="color" 时生效）---
CAP_COLOR_LOWER = (100, 60, 60)
CAP_COLOR_UPPER = (130, 255, 255)
MIN_AREA = 5000 # 瓶盖最小面积(像素)
MAX_AREA = 200000 # 瓶盖最大面积(像素)
MIN_FILL_RATIO = 0.80 # 最小填充率=面积/外接圆面积，正常圆形≈0.9+
CENTER_OFFSET = 40 # 圆心允许偏移(像素)
# =============================================================
DETECT_INTERVAL = 0.2 # 检测间隔(秒)
SAVE_LOG = True
LOG_FILE = "cap_inspection.log"
FONT_CANDIDATES = [r"C:\Windows\Fonts\msyh.ttc",
r"C:\Windows\Fonts\simhei.ttf",
r"C:\Windows\Fonts\simsun.ttc"]
FONT_SIZE = 22
# ---------- 导入海康 MVS Python SDK ----------
if MVS_IMPORT_PATH and os.path.isdir(MVS_IMPORT_PATH):
sys.path.insert(0, MVS_IMPORT_PATH)
try:
from MvCameraControl_class import *
except ImportError:
print("[错误] 找不到海康 MVS 的 Python SDK（MvCameraControl_class.py）")
print("请把 MVS 安装目录下 MvImport 文件夹里的 .py 文件复制到本程序同目录，")
print("或在本文件 MVS_IMPORT_PATH 填写 MvImport 完整路径。")
sys.exit(1)
def log(msg):
line = "[{}] {}".format(time.strftime("%Y-%m-%d %H:%M:%S"), msg)
print(line, flush=True)
if SAVE_LOG:
try:
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(line + "\n")
except OSError:
pass
# 像素格式编码（GigE Vision 标准）
PF_MONO8 = 0x01080001
PF_RGB8 = 0x02180014
PF_BAYER_GR8 = 0x01080008
PF_BAYER_RG8 = 0x01080009
PF_BAYER_GB8 = 0x0108000A
PF_BAYER_BG8 = 0x0108000B
class HKCamera(object):
"""海康机器人工业相机封装（MVS SDK）"""
def __init__(self, index=0):
self.index = index
self.cam = None
self.payload_size = 0
self.data_buf = None
self.frame_info = None
def connect(self):
self.release()
try:
device_list = MV_CC_DEVICE_INFO_LIST()
ret = MvCamera.MV_CC_EnumDevices(MV_GIGE_DEVICE | MV_USB_DEVICE, device_list)
if ret != 0:
log("枚举设备失败，错误码 0x{:X}".format(ret))
return False
if device_list.nDeviceNum == 0:
log("未找到海康工业相机：请检查连接/驱动，或 MVS 客户端是否占用")
return False
if self.index >= device_list.nDeviceNum:
log("相机索引 {} 超出范围（共 {} 台）".format(self.index, device_list.nDeviceNum))
return False
st_dev = cast(device_list.pDeviceInfo[self.index],
POINTER(MV_CC_DEVICE_INFO)).contents
self.cam = MvCamera()
ret = self.cam.MV_CC_CreateHandle(st_dev)
if ret != 0:
log("创建句柄失败，错误码 0x{:X}".format(ret))
self.release()
return False
ret = self.cam.MV_CC_OpenDevice(MV_ACCESS_Exclusive, 0)
if ret != 0:
log("打开设备失败，错误码 0x{:X}。请确认 MVS 客户端已关闭".format(ret))
self.release()
return False
self.cam.MV_CC_SetEnumValue("TriggerMode", MV_TRIGGER_MODE_OFF)
st_param = MVCC_INTVALUE()
memset(byref(st_param), 0, sizeof(st_param))
ret = self.cam.MV_CC_GetIntValue("PayloadSize", st_param)
if ret != 0:
log("获取 PayloadSize 失败，错误码 0x{:X}".format(ret))
self.release()
return False
self.payload_size = st_param.nCurValue
ret = self.cam.MV_CC_StartGrabbing()
if ret != 0:
log("开始取流失败，错误码 0x{:X}".format(ret))
self.release()
return False
self.data_buf = (c_ubyte * self.payload_size)()
self.frame_info = MV_FRAME_OUT_INFO_EX()
log("海康相机打开成功：第 {} 台".format(self.index))
return True
except Exception as e:
log("连接相机异常: {}".format(e))
self.release()
return False
def read(self):
if self.cam is None:
return False, None
try:
memset(byref(self.frame_info), 0, sizeof(self.frame_info))
ret = self.cam.MV_CC_GetOneFrameTimeout(
byref(self.data_buf), self.payload_size, self.frame_info, 1000)
if ret != 0:
return False, None
frame = self._to_bgr()
return frame is not None, frame
except Exception:
return False, None
def _to_bgr(self):
w = self.frame_info.nWidth
h = self.frame_info.nHeight
if w <= 0 or h <= 0:
return None
pt = self.frame_info.enPixelType
n = w * h
try:
if pt == PF_MONO8:
img = np.frombuffer(self.data_buf, dtype=np.uint8, count=n).reshape(h, w)
return cv2.cvtColor(img, cv2.COLOR_GRAY2BGR)
elif pt == PF_RGB8:
img = np.frombuffer(self.data_buf, dtype=np.uint8, count=n * 3).reshape(h, w, 3)
return cv2.cvtColor(img, cv2.COLOR_RGB2BGR)
elif pt == PF_BAYER_RG8:
img = np.frombuffer(self.data_buf, dtype=np.uint8, count=n).reshape(h, w)
return cv2.cvtColor(img, cv2.COLOR_BayerRG2BGR)
elif pt == PF_BAYER_GB8:
img = np.frombuffer(self.data_buf, dtype=np.uint8, count=n).reshape(h, w)
return cv2.cvtColor(img, cv2.COLOR_BayerGB2BGR)
elif pt == PF_BAYER_GR8:
img = np.frombuffer(self.data_buf, dtype=np.uint8, count=n).reshape(h, w)
return cv2.cvtColor(img, cv2.COLOR_BayerGR2BGR)
elif pt == PF_BAYER_BG8:
img = np.frombuffer(self.data_buf, dtype=np.uint8, count=n).reshape(h, w)
return cv2.cvtColor(img, cv2.COLOR_BayerBG2BGR)
else:
log("不支持的像素格式 0x{:X}，请用 MVS 客户端改为 Mono8 或 RGB8".format(pt))
return None
except Exception as e:
log("图像转换失败: {}".format(e))
return None
def release(self):
if self.cam is not None:
for fn in ("MV_CC_StopGrabbing", "MV_CC_CloseDevice", "MV_CC_DestroyHandle"):
try:
getattr(self.cam, fn)()
except Exception:
pass
self.cam = None
def resize_if_needed(frame):
if RESIZE_MAX_SIDE <= 0:
return frame
h, w = frame.shape[:2]
longest = max(h, w)
if longest > RESIZE_MAX_SIDE:
scale = RESIZE_MAX_SIDE / float(longest)
frame = cv2.resize(frame, (int(w * scale), int(h * scale)), interpolation=cv2.INTER_AREA)
return frame
def detect_cap(roi):
"""根据 DETECT_MODE 分发到对应检测方法。"""
if DETECT_MODE == "color":
return detect_by_color(roi)
return detect_by_gray(roi)
def detect_by_gray(roi):
"""灰度 Otsu 自动阈值检测：瓶盖亮、背景暗即可，红/绿/白盖通用，免调颜色。"""
h, w = roi.shape[:2]
info = None
gray = cv2.cvtColor(roi, cv2.COLOR_BGR2GRAY)
blurred = cv2.GaussianBlur(gray, (9, 9), 0)
flag = cv2.THRESH_BINARY if BRIGHT_CAP else cv2.THRESH_BINARY_INV
_, binary = cv2.threshold(blurred, 0, 255, flag + cv2.THRESH_OTSU)
kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (9, 9))
binary = cv2.morphologyEx(binary, cv2.MORPH_OPEN, kernel, iterations=1)
binary = cv2.morphologyEx(binary, cv2.MORPH_CLOSE, kernel, iterations=3)
contours, _ = cv2.findContours(binary, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
if not contours:
return False, ["未检测到瓶盖(可能缺盖)"], info, binary
largest = max(contours, key=cv2.contourArea)
area = cv2.contourArea(largest)
(cx, cy), radius = cv2.minEnclosingCircle(largest)
cx, cy, radius = int(cx), int(cy), int(radius)
enclosing_area = np.pi * radius * radius
fill_ratio = area / enclosing_area if enclosing_area > 0 else 0.0
info = {"cx": cx, "cy": cy, "radius": radius,
"area": int(area), "fill_ratio": round(fill_ratio, 3)}
defects = []
if area < MIN_AREA:
defects.append("面积过小(缺盖/残缺)")
elif area > MAX_AREA:
defects.append("面积过大(异常)")
if fill_ratio < MIN_FILL_RATIO:
defects.append("破损/缺口")
if ROI is not None:
offset = np.hypot(cx - w / 2.0, cy - h / 2.0)
info["offset"] = round(offset, 1)
if offset > CENTER_OFFSET:
defects.append("位置偏移/歪斜")
return len(defects) == 0, defects, info, binary
def detect_by_color(roi):
"""HSV 颜色分割检测（备用模式，DETECT_MODE="color" 时使用）。"""
h, w = roi.shape[:2]
info = None
blurred = cv2.GaussianBlur(roi, (7, 7), 0)
hsv = cv2.cvtColor(blurred, cv2.COLOR_BGR2HSV)
mask = cv2.inRange(hsv, CAP_COLOR_LOWER, CAP_COLOR_UPPER)
kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (9, 9))
mask = cv2.morphologyEx(mask, cv2.MORPH_OPEN, kernel, iterations=1)
mask = cv2.morphologyEx(mask, cv2.MORPH_CLOSE, kernel, iterations=3)
contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
if not contours:
return False, ["未检测到瓶盖(可能缺盖)"], info, mask
largest = max(contours, key=cv2.contourArea)
area = cv2.contourArea(largest)
(cx, cy), radius = cv2.minEnclosingCircle(largest)
cx, cy, radius = int(cx), int(cy), int(radius)
enclosing_area = np.pi * radius * radius
fill_ratio = area / enclosing_area if enclosing_area > 0 else 0.0
info = {"cx": cx, "cy": cy, "radius": radius,
"area": int(area), "fill_ratio": round(fill_ratio, 3)}
defects = []
if area < MIN_AREA:
defects.append("面积过小(缺盖/残缺)")
elif area > MAX_AREA:
defects.append("面积过大(异常)")
if fill_ratio < MIN_FILL_RATIO:
defects.append("破损/缺口")
if ROI is not None:
offset = np.hypot(cx - w / 2.0, cy - h / 2.0)
info["offset"] = round(offset, 1)
if offset > CENTER_OFFSET:
defects.append("位置偏移/歪斜")
return len(defects) == 0, defects, info, mask
def draw_chinese(frame, text, pos, color, font):
pil = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
d = ImageDraw.Draw(pil)
if font is not None:
d.text(pos, text, fill=(color[2], color[1], color[0]), font=font)
return cv2.cvtColor(np.array(pil), cv2.COLOR_RGB2BGR)
def load_font():
try:
for p in FONT_CANDIDATES:
if os.path.exists(p):
return ImageFont.truetype(p, FONT_SIZE)
log("警告: 未找到中文字体，画面中文标注将不显示")
return None
except Exception as e:
log("警告: 中文字体加载失败({})".format(e))
return None
def main():
p = argparse.ArgumentParser()
p.add_argument("--camera", type=int, default=CAMERA_INDEX)
p.add_argument("--interval", type=float, default=DETECT_INTERVAL)
p.add_argument("--no-window", action="store_true")
args = p.parse_args()
show_window = not args.no_window
font = load_font()
cam = HKCamera(args.camera)
if not cam.connect():
log("相机打开失败，程序退出。请关闭 MVS 客户端后重试，或换 --camera 1")
return
log("开始瓶盖缺陷检测：间隔 {}s，模式={}".format(args.interval, DETECT_MODE))
last_detect = 0.0
last_state = None
last_defects = []
show_mask = False
fps_start, fps_frames = time.time(), 0
while True:
ok, frame = cam.read()
if not ok:
log("相机取流中断，重连中...")
if not cam.connect():
time.sleep(3)
continue
continue
frame = resize_if_needed(frame)
H, W = frame.shape[:2]
if ROI is not None:
x, y, rw, rh = ROI
x = max(0, min(x, W - 1)); y = max(0, min(y, H - 1))
rw = min(rw, W - x); rh = min(rh, H - y)
roi = frame[y:y + rh, x:x + rw]
else:
x, y, rw, rh = 0, 0, W, H
roi = frame
if show_window:
key = cv2.waitKey(1) & 0xFF
else:
key = -1
time.sleep(0.01)
if key == ord("d"):
show_mask = not show_mask
do_detect = (time.time() - last_detect >= args.interval)
if do_detect:
last_detect = time.time()
is_ok, defects, info, mask = detect_cap(roi)
last_defects = defects
state = "OK" if is_ok else "NG"
if state != last_state:
last_state = state
if is_ok:
log("结果: OK 面积={} 填充率={}".format(info["area"], info["fill_ratio"]) if info else "结果: OK")
else:
log("结果: NG 缺陷: {}".format("、".join(defects)))
else:
is_ok = (last_state == "OK")
info = None
mask = None
if show_mask and mask is not None:
display = cv2.cvtColor(mask, cv2.COLOR_GRAY2BGR)
else:
display = frame
if ROI is not None:
cv2.rectangle(display, (x, y), (x + rw, y + rh), (255, 200, 0), 2)
color = (0, 255, 0) if is_ok else (0, 0, 255)
status_text = "OK" if is_ok else "NG: " + "、".join(last_defects)
if font is not None:
display = draw_chinese(display, status_text, (10, 10), color, font)
else:
cv2.putText(display, status_text, (10, 35),
cv2.FONT_HERSHEY_SIMPLEX, 1.0, color, 2)
if not show_mask and info is not None:
cv2.circle(display, (x + info["cx"], y + info["cy"]), info["radius"], color, 3)
cv2.circle(display, (x + info["cx"], y + info["cy"]), 3, color, -1)
cv2.putText(display, "area:{} fill:{:.2f}".format(info["area"], info["fill_ratio"]),
(x + info["cx"] - info["radius"], max(10, y + info["cy"] - info["radius"] - 10)),
cv2.FONT_HERSHEY_SIMPLEX, 0.6, color, 2)
fps_frames += 1
if time.time() - fps_start >= 1.0:
fps = fps_frames / (time.time() - fps_start)
cv2.putText(display, "FPS:{:.1f}".format(fps), (W - 140, 30),
cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 255), 2)
fps_start, fps_frames = time.time(), 0
if show_window:
cv2.imshow("Cap Defect Inspection (q=退出 s=截图 d=分割图)", display)
if key == ord("q"):
break
elif key == ord("s"):
fn = "cap_{}.jpg".format(time.strftime("%Y%m%d_%H%M%S"))
cv2.imwrite(fn, display)
log("已保存截图: {}".format(fn))
cam.release()
cv2.destroyAllWindows()
log("程序结束")
if __name__ == "__main__":
main()
修改代码后的运行测试图：
4 IoTDA中转空间构建
4.0 IoTDA构造
4.1 产品及其模型构建
4.1.1 温湿度计
4.1.2 吸顶灯
4.2 设备实例化应用
4.2.1 温湿度计设备注册及模拟测试
4.2.2 吸顶灯设备注册及模拟测试
5 CodeArts智慧路灯快速构建与部署
5.1 应用项目创建
5.2 代码托管操作
5.3 构建任务设立
5.4 项目构建实现
5.5 项目部署运行
1 工效组合展示
鲲鹏昇腾开发者社区是面向全社会开放的“联接全球计算开发者，聚合华为+生态”的社区，内容涵盖鲲鹏、昇腾资源，帮助开发者快速获取所需的知识、经验、软件、工具、算力，支撑开发者易学、好用、成功，成为核心开发者。
更多推荐
3
0- 0
已为社区贡献1条内容
所有评论(0)