1 工效组合展示

本环节以“端—边—云”协同为主线，集中展示了具身智能在工业视觉分析与物联监护场景下的综合应用成效。首先，通过 IoTDA 物联网平台完成智能吸顶灯等终端设备的接入与配置，实现设备属性定义、数据格式（JSON）与 MQTT 协议的规范化管理，为后续数据上报与指令下发打通链路；其次，依托边缘侧视觉分析能力，对现场画面进行实时智能识别，可准确检出人员及常见物品并输出置信度标注，识别结果同步上传至云端平台，形成“视觉感知—智能分析—物联上报”的闭环；最后，在云端完成服务的在线部署与联调，验证了从设备接入、模型推理到数据可视化的全流程贯通。该组合方案充分体现了具身智能技术在工业现场监控、安全监护与设备联动中的应用价值，为后续视觉获取与分析应用的深化演练奠定了基础。

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

2.1 工具软件安装

工业视觉开发所需的工具软件依次完成安装与配置：通过 Ascend AI Devkit Imager 将昇腾系统镜像烧录至开发板，搭建基础运行环境；安装海康机器视觉 MVS 客户端，用于工业相机的发现、连接、参数配置与图像采集调试；安装 MobaXterm 远程操控终端，实现对开发板的 SSH 登录与命令行操作。安装完成后，通过设备管理器确认相关驱动识别正常、各工具可正常启动，为后续环境构建与终端应用开发提供支撑。

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

在开发板上搭建工业视觉运行环境，配置昇腾 AI 计算相关依赖与视觉开发组件，并通过海康 MVS 客户端连接工业相机，完成设备发现、参数配置与图像采集验证，实现现场画面的实时显示，为后续视觉识别应用的开发与运行奠定基础。

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

昇腾开发板上运行的工业视觉终端应用，完整展示了边缘设备的图像采集、模型推理与结果呈现能力，验证端侧算力在工业场景下的实际运行效果，体现具身智能终端侧的自主感知与处理能力。

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

3.1 IMA-DS借力

IMA-DS 为工业视觉识别开发提供高效的工具支撑，借助其预置能力快速完成工程框架搭建与资源调用，可显著降低开发门槛、提升编码效率。

3.2 运行环境构造

编码前先搭建了运行环境，主要步骤： 用 python -m venv 创建虚拟环境（不依赖 Conda）； 安装 opencv-python、pytesseract、numpy 等依赖库； 安装 Tesseract-OCR 引擎及中文语言包（chi_sim），配置 tesseract.exe 路径； 准备网络摄像机（海康/大华等）的 RTSP 地址与账号信息。 环境就绪后做了连通性测试，确认摄像机可正常取流。

3.3 编码实现

# -*- coding: utf-8 -*-

"""

========================================================================

 Windows + CPU 网络摄像头中文 OCR（单文件，venv 非 Conda）

------------------------------------------------------------------------

 摄像头：OpenCV 拉取网络摄像机流（RTSP / HTTP-MJPEG / 本地摄像头）

 OCR   ：RapidOCR（ONNX Runtime 推理，CPU 即可，中文+英文识别）

 功能   ：实时窗口显示；画面叠加中文识别框；控制台实时打印文字；

         按键 S 截图 / C 保存文字 / Q 退出；支持 --image 离线识别单图

========================================================================

运行示例：

  # 1) 先离线验证 OCR（不依赖摄像头）：

  python ocr_camera.py --image 测试图.png

  # 2) 网络摄像头（RTSP，按需替换 IP/账号/密码/通道）：

  python ocr_camera.py --source "rtsp://admin:123456@192.168.1.64:554/stream1"

  # 3) 本地 USB 摄像头：

  python ocr_camera.py --source 0

"""

import os

import sys

import time

import argparse

import logging

import cv2

import numpy as np

from PIL import Image, ImageDraw, ImageFont

try:

    from rapidocr_onnxruntime import RapidOCR

except ImportError:  # 兼容部分版本包名/导入名差异

    from rapidocr import RapidOCR

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

log = logging.getLogger("ocr_camera")

# ======================== 配置区 ========================

# 网络摄像头地址：RTSP 或 HTTP-MJPEG；本地摄像头填 0/1

DEFAULT_SOURCE = os.environ.get("CAM_SOURCE", "0")

# 每 N 帧做一次 OCR（CPU 推理较慢，不必逐帧识别；帧率高的摄像头建议加大）

OCR_INTERVAL = int(os.environ.get("OCR_INTERVAL", "3"))

# OCR 前对帧的缩放比例（0.5 表示缩到一半，速度更快；1.0 原始尺寸，更准）

OCR_SCALE = float(os.environ.get("OCR_SCALE", "0.8"))

# 截图/文字保存目录

SAVE_DIR = os.environ.get("SAVE_DIR", "./ocr_output")

# 置信度阈值，低于该值的识别结果丢弃

SCORE_THRESHOLD = float(os.environ.get("SCORE_THRESHOLD", "0.5"))

def find_chinese_font():

    """在 Windows 上自动查找可用的中文字体，用于画面叠加中文。"""

    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",  # 宋体

        r"C:\Windows\Fonts\simkai.ttf",  # 楷体

        # Linux 沙箱/CI 兜底

        "/usr/share/fonts/opentype/noto/NotoSerifCJK-Regular.ttc",

        "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc",

    ]

    for p in candidates:

        if os.path.exists(p):

            return p

    log.warning("未找到中文字体，画面内中文将显示为方框（不影响控制台文字输出）。")

    return None

def parse_result(result):

    """把 RapidOCR 返回结果解析成 (boxes, texts)。"""

    boxes, texts = [], []

    if not result:

        return boxes, texts

    for item in result:

        box, text, score = item[0], item[1], float(item[2])

        if score < SCORE_THRESHOLD:

            continue

        boxes.append(np.array(box, dtype=np.int32).reshape(-1, 2))

        texts.append(f"{text} ({score:.2f})")

    return boxes, texts

def draw_chinese(frame, boxes, texts, font_path):

    """在帧上绘制检测框与中文文本（OpenCV 不支持中文，用 PIL 中转）。"""

    if frame is None:

        return frame

    img = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))

    draw = ImageDraw.Draw(img)

    try:

        font = ImageFont.truetype(font_path, 22) if font_path else ImageFont.load_default()

    except Exception:

        font = ImageFont.load_default()

    for box, txt in zip(boxes, texts):

        pts = box.reshape(-1, 2)

        draw.polygon([tuple(p) for p in pts], outline=(0, 255, 0))

        x, y = int(pts[:, 0].min()), int(pts[:, 1].min())

        draw.text((x, max(0, y - 24)), txt, fill=(0, 255, 0), font=font)

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

def ocr_image(engine, img_bgr):

    """对单张 BGR 图像做 OCR，返回 (annotated_img, boxes, texts, elapse)。"""

    result, elapse = engine(img_bgr)

    boxes, texts = parse_result(result)

    return boxes, texts, elapse

def run_single_image(engine, image_path, font_path):

    if not os.path.isfile(image_path):

        log.error("图片不存在：%s", image_path)

        sys.exit(1)

    frame = cv2.imread(image_path)

    if frame is None:

        log.error("无法读取图片（路径含中文请用绝对路径）：%s", image_path)

        sys.exit(1)

    boxes, texts, elapse = ocr_image(engine, frame)

    log.info("识别耗时约 %.1f ms，共识别 %d 段文字", sum(elapse) if elapse else 0.0, len(texts))

    for t in texts:

        print("  [识别]", t)

    annotated = draw_chinese(frame, boxes, texts, font_path)

    os.makedirs(SAVE_DIR, exist_ok=True)

    out = os.path.join(SAVE_DIR, "ocr_result.png")

    cv2.imwrite(out, annotated)

    log.info("标注结果已保存：%s", out)

def open_camera(source):

    """打开网络摄像头，失败自动重试。"""

    if str(source).isdigit():

        source = int(source)

    cap = cv2.VideoCapture(source)

    if isinstance(source, str) and source.lower().startswith("rtsp"):

        # 降低 RTSP 缓冲，减小延迟

        cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)

    return cap

def run_camera(source, engine, font_path):

    cap = open_camera(source)

    if not cap.isOpened():

        log.error("无法打开摄像头：%s", source)

        log.error("请检查：1) 摄像头在线；2) RTSP 地址/账号/密码；3) 用 VLC 能否播放该地址。")

        sys.exit(1)

    log.info("摄像头已打开：%s（每 %d 帧 OCR 一次，Q 退出，S 截图，C 保存文字）", source, OCR_INTERVAL)

    os.makedirs(SAVE_DIR, exist_ok=True)

    frame_idx = 0

    last_boxes, last_texts = [], []

    while True:

        ret, frame = cap.read()

        if not ret or frame is None:

            log.warning("读帧失败，尝试重连 ...")

            cap.release()

            time.sleep(2)

            cap = open_camera(source)

            if not cap.isOpened():

                log.error("重连失败，退出。")

                break

            continue

        frame_idx += 1

        annotated = frame

        if frame_idx % OCR_INTERVAL == 0:

            # 可选缩放，加速 CPU 推理

            if OCR_SCALE < 1.0:

                small = cv2.resize(frame, None, fx=OCR_SCALE, fy=OCR_SCALE,

                                   interpolation=cv2.INTER_AREA)

            else:

                small = frame

            boxes_small, texts, elapse = ocr_image(engine, small)

            # 把检测框坐标映射回原图

            if OCR_SCALE < 1.0:

                inv = 1.0 / OCR_SCALE

                boxes = [(b * inv).astype(np.int32) for b in boxes_small]

            else:

                boxes = boxes_small

            last_boxes, last_texts = boxes, texts

            for t in texts:

                print("  [识别]", t)

            annotated = draw_chinese(frame, boxes, texts, font_path)

        cv2.imshow("OCR Camera - Q退出/S截图/C保存文字", annotated)

        key = cv2.waitKey(1) & 0xFF

        if key == ord("q") or key == ord("Q") or key == 27:  # ESC

            break

        elif key == ord("s") or key == ord("S"):

            ts = time.strftime("%Y%m%d_%H%M%S")

            f = os.path.join(SAVE_DIR, f"shot_{ts}.png")

            cv2.imwrite(f, annotated)

            log.info("截图已保存：%s", f)

        elif key == ord("c") or key == ord("C"):

            ts = time.strftime("%Y%m%d_%H%M%S")

            f = os.path.join(SAVE_DIR, f"text_{ts}.txt")

            with open(f, "w", encoding="utf-8") as fp:

                fp.write("\n".join(last_texts))

            log.info("文字已保存：%s", f)

    cap.release()

    cv2.destroyAllWindows()

def main():

    global OCR_INTERVAL, OCR_SCALE, SAVE_DIR

    ap = argparse.ArgumentParser(description="Windows CPU 网络摄像头中文 OCR（RapidOCR）")

    ap.add_argument("--source", default=DEFAULT_SOURCE,

                    help="摄像头地址：RTSP/HTTP-MJPEG 或本地摄像头索引 0/1")

    ap.add_argument("--image", default=None, help="离线识别单张图片（先验证 OCR 用）")

    ap.add_argument("--interval", type=int, default=OCR_INTERVAL, help="每 N 帧 OCR 一次")

    ap.add_argument("--scale", type=float, default=OCR_SCALE, help="OCR 前缩放比例(0.1~1.0)")

    ap.add_argument("--save-dir", default=SAVE_DIR, help="截图/文字保存目录")

    args = ap.parse_args()

    OCR_INTERVAL, OCR_SCALE, SAVE_DIR = args.interval, args.scale, args.save_dir

    log.info("初始化 RapidOCR（首次运行会自动下载内置模型，请稍候）...")

    try:

        engine = RapidOCR()

    except Exception as e:

        log.error("RapidOCR 初始化失败：%s", e)

        log.error("提示：首次运行需要联网下载模型；若网络受限，可离线配置模型目录。")

        sys.exit(1)

    font_path = find_chinese_font()

    if args.image:

        run_single_image(engine, args.image, font_path)

    else:

        run_camera(args.source, engine, font_path)

if __name__ == "__main__":

    main()

3.4 运行测试

通过输入测试图像或实时采集画面对识别应用进行运行测试，观察识别结果与置信度输出，验证模型推理的正确性与运行稳定性，完成端侧识别能力的功能验证。

4 IoTDA中转空间构建

基于华为云 IoTDA 物联网平台搭建中转空间，依次开展平台资源创建、产品及其模型构建与设备实例化应用，为设备接入、数据上报与指令下发建立统一的物联通道。

4.0 IoTDA构造

4.1 产品及其模型构建

在 IoTDA 平台上定义产品及其模型，设置产品名称、协议类型、数据格式与所属行业等信息，并构建服务与属性模型，为设备数据的规范化上报与解析提供依据。

4.1.1 温湿度计

温湿度计产品信息如下：

产品信息：

- 产品名称：温湿度计

- 协议类型：MQTT

- 数据格式：JSON

- 所属行业：智能家居

- 所属子行业：智能家居

服务列表：

- 服务 ID：T&H

- 属性名称：temperature

- 数据类型：decimal

- 访问权限：可读可写

- 取值范围：-100~100

- 属性名称：humidity

- 数据类型：decimal

- 访问权限：可读可写

- 取值范围：0~100

点击“产品 》 创建产品”。

4.1.2 吸顶灯

智能吸顶灯产品信息如下：

产品信息：

-- 产品名称：智能吸顶灯-- 协议类型：MQTT

-- 数据格式：JSON

-- 所属行业：智能家居

-- 所属子行业：智能家居

服务列表：

4.2 设备实例化应用

运行 MQTT 设备模拟器。

填写对接参数后，单击“Connect”按钮：

- SSL Connection：勾选则使用 8883 端口，不勾选则使用 1883 端口；

- Server Address：设备接入服务实例的设备侧 MQTT(S)接入地址；

- Device ID：填写注册设备成功后生成的设备 ID；

- Device Secret：填写注册设备时使用的设备密钥。

4.2.1 温湿度计设备注册及模拟测试

进入 IoTDA 平台的设备详情页面，可以查看到 MQTT 模拟器上报的数据。

4.2.2 吸顶灯设备注册及模拟测试

吸顶灯设备的注册与模拟测试通过 MQTT 设备模拟器完成：配置对接参数并连接平台，模拟设备数据上报与指令接收，验证吸顶灯产品模型与设备接入链路的正确性。

5 CodeArts智慧路灯快速构建与部署

基于 CodeArts 完成智慧路灯应用项目的快速构建与部署，依次开展项目创建、代码托管、构建任务设立、项目构建实现与部署运行，验证云端一体化开发部署流程。

5.1 应用项目创建

进入“代码 > 代码托管”页面，选择“按模版新建”。

搜索“IoT_Streetlight_Demo”，点击选中，然后点击“下一步”。

5.2 代码托管操作

5.3 构建任务设立

5.4 项目构建实现

在“新建编译构建任务 > 选择构建模板”中，选择“Maven”，单击“确定”。

5.5 项目部署运行