1 工效组合展示

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

2.1 工具软件安装

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

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

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

3.1 IMA-DS借力

给出python虚拟环境OCR文字识别的编码及其运行过程，包括汉字，windows环境，非COnDA，使用本机连接的网络摄像机，windows只能使用CPU，整个代码需要用一个文件表示，我的python版本是3.9.10，摄像头使用的是usb连接，我现在需要调用海康机器人的工业相机，请帮我在程序中再加上可以调用MVS数据库的程序，请你也适配一下设备

3.2 运行环境构造

3.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.4 运行测试

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 项目部署运行