《基于昇腾与海康视觉的智能识别及 IoTDA 物联监护综合演练》

ootdn

63人浏览 · 2026-09-17 11:54:17

ootdn · 2026-09-17 11:54:17 发布

1 工效组合展示

2 具身智能工业视觉获取与分析应用

2.1 工具软件安装

2.2 工业视觉环境构建与应用

2.3 昇腾工业视觉终端应用展现

3 具身智能工业视觉识别快速编码实现

3.1 IMA-DS借力

3.2 运行环境构造

3.3 编码实现

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
海康工业相机（USB）+ PaddleOCR 文字识别（含中文）
运行环境：Windows 10/11 x64，Python 3.8~3.12，CPU 推理
"""

import os
import sys
import time
import argparse
import ctypes

import numpy as np
import cv2

if sys.stdout and hasattr(sys.stdout, "reconfigure"):
    try:
        sys.stdout.reconfigure(encoding="utf-8")
    except Exception:
        pass
if sys.stderr and hasattr(sys.stderr, "reconfigure"):
    try:
        sys.stderr.reconfigure(encoding="utf-8")
    except Exception:
        pass

HIK_AVAILABLE = False
HIK_IMPORT_ERROR = None

_MVS_IMPORT_DIRS = [
    os.path.join(os.path.dirname(os.path.abspath(__file__)), "MvImport"),
    r"C:\Program Files (x86)\MVS\Development\Samples\Python\MvImport",
    r"C:\Program Files\MVS\Development\Samples\Python\MvImport",
    r"D:\MVS\Development\Samples\Python\MvImport",
]

_MVS_DLL_DIRS = [
    r"C:\Program Files (x86)\Common Files\MVS\Runtime\Win64_x64",
    r"C:\Program Files\Common Files\MVS\Runtime\Win64_x64",
]

if sys.platform == "win32":
    for _d in _MVS_DLL_DIRS:
        if os.path.isdir(_d):
            try:
                os.add_dll_directory(_d)
            except Exception:
                pass
    for _d in _MVS_IMPORT_DIRS:
        if os.path.isdir(_d) and _d not in sys.path:
            sys.path.insert(0, _d)

try:
    from MvCameraControl_class import *  # noqa
    HIK_AVAILABLE = True
except Exception as _e:
    HIK_IMPORT_ERROR = str(_e)


class HikCamera:
    """枚举 -> 建句柄 -> 打开 -> 取流 -> 取图 -> 转 BGR"""

    def __init__(self, device_index=0, device_type="usb"):
        if not HIK_AVAILABLE:
            raise RuntimeError(
                "海康 MVS SDK 未加载成功。\n"
                f"错误信息：{HIK_IMPORT_ERROR}\n"
                "请确认已安装海康 MVS，并把 MvImport 文件夹放到本脚本同级目录。"
            )

        self.device_index = device_index
        self.cam = None
        self.is_open = False
        self.is_grabbing = False

        _type_map = {
            "usb": MV_USB_DEVICE,
            "gige": MV_GIGE_DEVICE,
            "all": MV_GIGE_DEVICE | MV_USB_DEVICE,
        }
        self._type_flag = _type_map.get(device_type, MV_USB_DEVICE)

    def enumerate_devices(self):
        device_list = MV_CC_DEVICE_INFO_LIST()
        ctypes.memset(ctypes.byref(device_list), 0, ctypes.sizeof(device_list))
        ret = MvCamera.MV_CC_EnumDevices(self._type_flag, device_list)
        if ret != 0:
            raise RuntimeError(f"枚举设备失败，错误码 0x{ret & 0xFFFFFFFF:08X}")
        return device_list

    def open(self):
        device_list = self.enumerate_devices()
        n = device_list.nDeviceNum
        if n == 0:
            raise RuntimeError("未检测到海康工业相机。请检查 USB 连接、驱动，并关闭 MVS Studio。")
        if self.device_index >= n:
            raise RuntimeError(f"相机索引 {self.device_index} 超出范围（共检测到 {n} 台）。")

        st_device_info = ctypes.cast(
            device_list.pDeviceInfo[self.device_index],
            ctypes.POINTER(MV_CC_DEVICE_INFO),
        ).contents

        self.cam = MvCamera()
        ret = self.cam.MV_CC_CreateHandle(st_device_info)
        if ret != 0:
            raise RuntimeError(f"创建相机句柄失败，错误码 0x{ret & 0xFFFFFFFF:08X}")

        ret = self.cam.MV_CC_OpenDevice(MV_ACCESS_Exclusive, 0)
        if ret != 0:
            self.cam.MV_CC_DestroyHandle()
            self.cam = None
            raise RuntimeError(
                f"打开相机失败，错误码 0x{ret & 0xFFFFFFFF:08X}。"
                "常见原因：相机被其他程序（如 MVS Studio）占用。"
            )
        self.is_open = True

        try:
            self.cam.MV_CC_SetEnumValue("TriggerMode", MV_TRIGGER_MODE_OFF)
        except Exception:
            pass

        return self

    def start(self):
        if not self.is_open:
            raise RuntimeError("相机尚未打开。")
        ret = self.cam.MV_CC_StartGrabbing()
        if ret != 0:
            raise RuntimeError(f"开始取流失败，错误码 0x{ret & 0xFFFFFFFF:08X}")
        self.is_grabbing = True

    def get_frame(self, timeout_ms=1000):
        """返回 BGR 图像；超时返回 None。"""
        if not self.is_grabbing:
            return None

        st_frame = MV_FRAME_OUT()
        ctypes.memset(ctypes.byref(st_frame), 0, ctypes.sizeof(st_frame))
        ret = self.cam.MV_CC_GetImageBuffer(st_frame, timeout_ms)
        if ret != 0:
            return None

        try:
            st_info = st_frame.stFrameInfo
            n_len = st_info.nFrameLen
            if n_len <= 0 or not st_frame.pBufAddr:
                return None

            buf = (ctypes.c_ubyte * n_len)()
            ctypes.memmove(buf, st_frame.pBufAddr, n_len)

            return self._to_bgr(
                buf,
                n_len,
                st_info.nWidth,
                st_info.nHeight,
                st_info.enPixelType,
            )
        finally:
            self.cam.MV_CC_FreeImageBuffer(st_frame)

    def _to_bgr(self, buf, n_len, width, height, pixel_type):
        if pixel_type == PixelType_Gvsp_Mono8:
            gray = np.frombuffer(buf, dtype=np.uint8, count=width * height).reshape(height, width)
            return cv2.cvtColor(gray, cv2.COLOR_GRAY2BGR)

        if pixel_type == PixelType_Gvsp_BGR8_Packed:
            arr = np.frombuffer(buf, dtype=np.uint8, count=width * height * 3)
            return arr.reshape(height, width, 3).copy()

        if pixel_type == PixelType_Gvsp_RGB8_Packed:
            arr = np.frombuffer(buf, dtype=np.uint8, count=width * height * 3).reshape(height, width, 3)
            return cv2.cvtColor(arr, cv2.COLOR_RGB2BGR)

        return self._convert_to_bgr(buf, n_len, width, height, pixel_type)

    def _convert_to_bgr(self, src_buf, src_len, width, height, src_pixel_type):
        dst_len = width * height * 3
        dst_buf = (ctypes.c_ubyte * dst_len)()

        st_param = MV_CC_PIXEL_CONVERT_PARAM()
        ctypes.memset(ctypes.byref(st_param), 0, ctypes.sizeof(st_param))
        st_param.nWidth = width
        st_param.nHeight = height
        st_param.pSrcData = ctypes.cast(src_buf, ctypes.POINTER(ctypes.c_ubyte))
        st_param.nSrcDataLen = src_len
        st_param.enSrcPixelType = src_pixel_type
        st_param.enDstPixelType = PixelType_Gvsp_RGB8_Packed
        st_param.pDstBuffer = ctypes.cast(dst_buf, ctypes.POINTER(ctypes.c_ubyte))
        st_param.nDstBufferSize = dst_len

        ret = self.cam.MV_CC_ConvertPixelType(st_param)
        if ret != 0:
            raise RuntimeError(f"像素格式转换失败，错误码 0x{ret & 0xFFFFFFFF:08X}")

        arr = np.frombuffer(dst_buf, dtype=np.uint8, count=dst_len).reshape(height, width, 3)
        return cv2.cvtColor(arr, cv2.COLOR_RGB2BGR)

    def close(self):
        if self.cam is None:
            return
        for fn, guard in (
            (self.cam.MV_CC_StopGrabbing, self.is_grabbing),
            (self.cam.MV_CC_CloseDevice, self.is_open),
            (self.cam.MV_CC_DestroyHandle, True),
        ):
            if guard:
                try:
                    fn()
                except Exception:
                    pass
        self.is_grabbing = False
        self.is_open = False
        self.cam = None


class OCREngine:
    """PaddleOCR 中文识别，强制 CPU 推理。"""

    def __init__(self, lang="ch"):
        self.ocr = None
        self._init(lang)

    def _init(self, lang):
        try:
            import paddle
            try:
                paddle.set_device("cpu")
            except Exception:
                pass
        except Exception:
            pass

        from paddleocr import PaddleOCR

        last_err = None

        try:
            self.ocr = PaddleOCR(lang=lang, device="cpu", enable_mkldnn=False)
            return
        except Exception as e:
            last_err = e

        attempts = [
            dict(lang=lang, use_angle_cls=True, use_gpu=False, show_log=False),
            dict(lang=lang, use_angle_cls=True, use_gpu=False),
            dict(lang=lang, use_angle_cls=True),
            dict(lang=lang),
        ]
        for kw in attempts:
            try:
                self.ocr = PaddleOCR(**kw)
                return
            except Exception as e:
                last_err = e

        raise RuntimeError(f"PaddleOCR 初始化失败：{last_err}")

    def predict(self, img_bgr):
        if hasattr(self.ocr, "predict"):
            return self.ocr.predict(img_bgr)
        return self.ocr.ocr(img_bgr, cls=True)


def parse_ocr_results(raw):
    items = []
    if not raw:
        return items

    if isinstance(raw[0], dict):
        for r in raw:
            texts = r.get("rec_texts") or []
            scores = r.get("rec_scores") or []
            polys = r.get("rec_polys") or []
            for i, text in enumerate(texts):
                text = str(text).strip()
                if not text:
                    continue
                score = float(scores[i]) if i < len(scores) else 0.0
                poly = polys[i] if i < len(polys) else None
                items.append((poly, text, score))
        return items

    for line in raw:
        if not line:
            continue
        for box, info in line:
            text = str(info[0]).strip()
            score = float(info[1])
            if not text:
                continue
            items.append((box, text, score))
    return items


def find_chinese_font():
    candidates = [
        r"C:\Windows\Fonts\msyh.ttc",
        r"C:\Windows\Fonts\msyhbd.ttc",
        r"C:\Windows\Fonts\simhei.ttf",
        r"C:\Windows\Fonts\simsun.ttc",
    ]
    for p in candidates:
        if os.path.exists(p):
            return p
    return None


def draw_results(img_bgr, items, font_path=None):
    from PIL import Image, ImageDraw, ImageFont

    rgb = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB)
    pil = Image.fromarray(rgb)
    draw = ImageDraw.Draw(pil)

    fp = font_path or find_chinese_font()
    if fp and os.path.exists(fp):
        font = ImageFont.truetype(fp, 26)
    else:
        font = ImageFont.load_default()

    for poly, text, score in items:
        if poly is None:
            continue
        pts = [(int(float(p[0])), int(float(p[1]))) for p in poly]
        if len(pts) >= 3:
            draw.polygon(pts, outline=(0, 255, 0))
        x0, y0 = pts[0]
        label = f"{text}  {score:.2f}"
        draw.rectangle([x0, y0 - 28, x0 + len(label) * 26, y0], fill=(0, 0, 0))
        draw.text((x0 + 2, y0 - 26), label, font=font, fill=(0, 255, 0))

    return cv2.cvtColor(np.array(pil), cv2.COLOR_RGB2BGR)


def save_result(annotated, items, directory, idx):
    os.makedirs(directory, exist_ok=True)
    ts = time.strftime("%Y%m%d_%H%M%S")
    img_path = os.path.join(directory, f"ocr_{ts}_{idx}.jpg")
    txt_path = os.path.join(directory, f"ocr_{ts}_{idx}.txt")

    cv2.imwrite(img_path, annotated)
    lines = [f"{score:.4f}\t{text}" for _, text, score in items]
    with open(txt_path, "w", encoding="utf-8") as f:
        f.write("\n".join(lines))
    return img_path, txt_path


def main():
    parser = argparse.ArgumentParser(
        description="海康工业相机（USB）+ PaddleOCR 中文文字识别（CPU）"
    )
    parser.add_argument("--index", type=int, default=0, help="相机索引（默认 0）")
    parser.add_argument("--type", default="usb", choices=["usb", "gige", "all"],
                        help="相机接口类型（默认 usb）")
    parser.add_argument("--save-dir", default="", help="按 s 键保存结果的目录")
    parser.add_argument("--font", default="", help="中文字体路径")
    parser.add_argument("--no-show", action="store_true", help="不显示窗口")
    args = parser.parse_args()

    if not HIK_AVAILABLE:
        print("=" * 70)
        print("[错误] 海康 MVS SDK 加载失败：")
        print(f"  {HIK_IMPORT_ERROR}")
        print("=" * 70)
        print("请先安装海康 MVS，并把 MvImport 文件夹放到本脚本同级目录。")
        sys.exit(1)

    print("[1/3] 初始化 OCR（首次运行会自动下载模型，请联网）...")
    ocr_engine = OCREngine(lang="ch")
    print("[1/3] OCR 就绪（CPU 模式）。")

    print("[2/3] 打开海康相机 ...")
    cam = HikCamera(device_index=args.index, device_type=args.type)
    cam.open()
    cam.start()
    print("[2/3] 相机已启动，开始实时识别。")

    print("[3/3] 运行中：按 q / ESC 退出，按 s 保存当前帧。")

    frame_idx = 0
    save_dir = args.save_dir or "."
    try:
        while True:
            frame = cam.get_frame(timeout_ms=1000)
            if frame is None:
                print("取帧超时（无新图像）...")
                continue

            raw = ocr_engine.predict(frame)
            items = parse_ocr_results(raw)

            if items:
                print("-" * 60)
                for _, text, score in items:
                    print(f"  [{score:.3f}] {text}")

            annotated = draw_results(frame, items, args.font) if items else frame

            key = -1
            if not args.no_show:
                cv2.imshow("HikCamera OCR", annotated)
                key = cv2.waitKey(1) & 0xFF

            if key in (ord("q"), 27):
                break
            if key == ord("s"):
                img_path, txt_path = save_result(annotated, items, save_dir, frame_idx)
                print(f"  已保存图片: {img_path}")
                print(f"  已保存文本: {txt_path}")

            frame_idx += 1
    except KeyboardInterrupt:
        print("\n用户中断。")
    finally:
        cam.close()
        cv2.destroyAllWindows()
        print("已释放相机，退出。")


if __name__ == "__main__":
    main()