从镜头到云端：具身智能工业视觉识别与物联监护一体化实践——基于海康工业相机与华为云 IoTDA 的端云协同方案

范 · 高℡

279人浏览 · 2026-09-17 16:42:21

范 · 高℡ · 2026-09-17 16:42:21 发布

摘要：本实训基于华为云 IoTDA 与 PaddleOCR，完成工业视觉工位搭建、USB 相机中文识别、设备接入与数据上报，并实现 CodeArts 智慧路灯项目构建部署，打通图像采集、文字识别到物联监控的端到端应用链路。

1 工效组合展示

1. 将海康工业相机、环形光源、镜头、支架与上位机组装为视觉识别工位。

2. 连接相机 USB3.0 线并接通环形光源电源。

3. 调整支架高度与镜头角度，使待识别区域位于画面中央。

4. 拍摄工位整体照片，作为工效组合展示图。

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

2.1 工具软件安装

1. 安装 Python 3.10/3.11，安装时勾选“Add Python to PATH”。

2. 安装海康 MVS 客户端，获得工业相机驱动与 SDK。

3. 安装 MQTT.fx 1.7.1，用于物联网设备模拟。

4. 命令行执行 python --version 验证 Python 安装成功。

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

1. 用 USB3.0 线连接海康工业相机与上位机。

2. 打开 MVS 客户端，确认设备列表中出现相机并能正常出图。

3. 转动镜头对焦环并调节环形光源，使目标文字清晰、亮度均匀。

4. 采集一帧图像，检查画面质量满足识别要求。

在工业视觉环境构建与应用过程中，可能遇到以下常见问题，可按表格中的方法逐一排查解决。

常见问题	可能原因	解决方法
相机无法识别	USB3.0 线未插好、驱动未安装、相机被其他程序占用、设备编号错误	1. 检查 USB3.0 线连接与 MVS 驱动；2. 关闭微信、QQ 等占用摄像头的程序；3. 在 MVS 客户端中确认设备列表是否出现相机；4. 更换 USB 接口或重启上位机后重试
画面模糊	镜头未对焦、镜头表面有污渍、工作距离过近或过远	1. 转动镜头对焦环直至文字边缘清晰；2. 用镜头纸清洁镜头表面；3. 调整支架高度，使目标位于镜头最佳工作距离内
光源不均匀	环形光源角度不当、亮度不足、环境光干扰	1. 调整环形光源角度与高度，使光线均匀覆盖目标区域；2. 适当调高光源亮度；3. 关闭或遮挡环境强光，减少反光与阴影干扰
画面黑屏或花屏	相机供电不足、USB 带宽不够、线缆过长或质量差	1. 使用独立供电的 USB3.0 线或带供电的集线器；2. 缩短线缆长度或更换高质量 USB3.0 线；3. 关闭其他占用 USB 带宽的设备
图像过曝或过暗	曝光时间设置不当、光源过强或过弱、增益过高	1. 在 MVS 客户端中调整曝光时间与增益；2. 调节环形光源亮度；3. 开启自动曝光或自动白平衡后微调

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

1. 连接昇腾工业视觉终端电源与网络。

2. 部署视觉识别应用至终端。

3. 运行应用，展示终端实时识别效果。

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

3.1 IMA-DS借力

1. 打开 IMA-DS 智能助手。

2. 输入如下提示词，说明识别目标、运行环境与编码要求。

3. 获取生成的 Python 代码，保存为 ocr_usb.py。

给出 Python 虚拟环境 OCR 文字识别的编程及其运行过程，包括汉字，Windows 环境，CPU 运行，非 Conda，使用本机 USB 连接的海康工业相机，编码用一个完整文件。

3.2 运行环境构造

1. 创建项目目录，执行 python -m venv venv 建立虚拟环境。

2. 执行 venv\Scripts\activate 激活环境。

3. 使用清华源安装依赖：pip install paddlepaddle paddleocr opencv-python pillow numpy -i https://pypi.tuna.tsinghua.edu.cn/simple。

4. 首次运行会自动下载 OCR 模型，保持联网。

3.3 编码实现

1. 编写主程序，包含参数解析、摄像头打开、后台识别线程与结果绘制模块。

2. 采用 PaddleOCR 并关闭方向分类、矫正等模块以提升速度。

3. 识别放在后台线程，主线程只负责显示画面，保证视频流畅。

4. 使用 Pillow 绘制中文标签（cv2.putText 不支持中文）。

#!/usr/bin/env python3

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

"""

USB 摄像头中文 OCR 识别（高精度版）

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

在「流畅版」基础上，把默认分辨率提高到 1280x720，文字识别更准。

同时保留低分辨率模式，想更快时加 --res low 即可切换。

主要特性：

1. OCR 识别放到后台线程，主线程只负责显示画面，视频不卡；

2. 默认 1280x720 高分辨率，识别精度更高；

3. 识别结果只在更新时打印一次，不刷屏；

4. 画面左上角显示实时 FPS，方便判断卡顿来源。

用法：

python ocr_usb.py # 高精度（1280x720）

python ocr_usb.py --res low # 快速（640x480）

python ocr_usb.py --camera 1 # 画面黑屏/不对就换编号

python ocr_usb.py --scale 0.8 # 识别前再略微缩小（更快，精度略降）

python ocr_usb.py --save # 识别到文字时自动保存截图

按键：

ESC 退出

s 手动保存当前画面

"""

import argparse

import os

import sys

import threading

import time

from datetime import datetime

import cv2

import numpy as np

from PIL import Image, ImageDraw, ImageFont

from paddleocr import PaddleOCR

# 分辨率档位：high = 更准但更慢；low = 更快但略粗

RESOLUTIONS = {

"high": (1280, 720),

"low": (640, 480),

}

CAMERA_FPS = 30 # 目标帧率，摄像头不支持会自动忽略

WINDOWS_CN_FONTS = [

"C:/Windows/Fonts/msyh.ttc", # 微软雅黑

"C:/Windows/Fonts/msyhbd.ttc", # 微软雅黑粗体

"C:/Windows/Fonts/simhei.ttf", # 黑体

"C:/Windows/Fonts/simsun.ttc", # 宋体

]

def fix_windows_console_encoding():

if sys.platform == "win32":

for stream in (sys.stdout, sys.stderr):

try:

stream.reconfigure(encoding="utf-8")

except Exception:

pass

def parse_args():

p = argparse.ArgumentParser(description="USB 摄像头中文 OCR 识别（高精度版）")

p.add_argument("--camera", "-c", type=int, default=0,

help="摄像头编号，一般 0 或 1")

p.add_argument("--res", choices=["high", "low"], default="high",

help="分辨率档位：high=1280x720 更准，low=640x480 更快")

p.add_argument("--interval", "-i", type=float, default=1.0,

help="每隔多少秒识别一次")

p.add_argument("--scale", type=float, default=1.0,

help="识别前缩小画面的比例（0.1~1.0），越小越快、精度越低")

p.add_argument("--score-thresh", type=float, default=0.5,

help="只保留置信度不低于该值的识别结果")

p.add_argument("--save", "-s", action="store_true",

help="识别到文字时自动保存截图")

p.add_argument("--output", "-o", default="output",

help="截图保存目录")

return p.parse_args()

def load_cn_font(size):

for path in WINDOWS_CN_FONTS:

if os.path.exists(path):

return ImageFont.truetype(path, size)

print("[警告] 未找到系统中文字体，画面上的中文可能显示为方块。", file=sys.stderr)

return ImageFont.load_default()

def _get_field(res, key):

try:

return res[key]

except (TypeError, KeyError):

return getattr(res, key, None)

def parse_ocr_result(result):

"""把 PaddleOCR 返回结果解析成 [{"text", "score", "box"}, ...]。"""

items = []

if not result:

return items

for res in result:

texts = _get_field(res, "rec_texts")

scores = _get_field(res, "rec_scores")

boxes = _get_field(res, "rec_boxes")

# 这几个字段可能是 numpy 数组，不能用 `or []` 判空，只能用 is None 判断

if texts is None:

texts = []

if scores is None:

scores = []

if boxes is None:

boxes = []

for text, score, box in zip(texts, scores, boxes):

x1, y1, x2, y2 = [int(float(v)) for v in box]

items.append({"text": str(text), "score": float(score),

"box": [x1, y1, x2, y2]})

return items

def draw_results(frame, items, font):

"""在画面上画框 + 用 Pillow 写中文。"""

if not items:

return frame

for it in items:

x1, y1, x2, y2 = it["box"]

cv2.rectangle(frame, (x1, y1), (x2, y2), (0, 255, 0), 2)

try:

rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)

pil = Image.fromarray(np.ascontiguousarray(rgb))

draw = ImageDraw.Draw(pil)

for it in items:

x1, y1 = it["box"][0], it["box"][1]

label = f"{it['text']} {it['score']:.2f}"

draw.text((x1, max(0, y1 - 28)), label, font=font, fill=(0, 255, 0))

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

except Exception:

return frame

class OcrWorker:

"""后台 OCR 线程：不阻塞画面显示，识别结果只在有更新时通知主线程。"""

def __init__(self, ocr, scale=1.0):

self.ocr = ocr

self.scale = scale

self._lock = threading.Lock()

self._pending_frame = None

self.results = []

self._new_results = False

self._running = True

self._thread = threading.Thread(target=self._loop, daemon=True)

self._thread.start()

def submit(self, frame):

"""主线程调用：丢一帧最新画面进去（非阻塞）。"""

with self._lock:

self._pending_frame = frame.copy()

def get_results(self):

"""返回 (识别结果列表, 是否有新结果)。"""

with self._lock:

return list(self.results), self._new_result

def mark_consumed(self):

with self._lock:

self._new_results = False

def _loop(self):

while self._running:

with self._lock:

frame = self._pending_frame

self._pending_frame = None

if frame is None:

time.sleep(0.01)

continue

# 缩小后再识别，速度更快

if self.scale < 1.0:

h, w = frame.shape[:2]

frame = cv2.resize(frame, (int(w * self.scale), int(h * self.scale)),

interpolation=cv2.INTER_AREA)

try:

items = parse_ocr_result(self.ocr.predict(frame))

except Exception:

items = []

# 把缩小后的坐标还原回原图

if self.scale < 1.0 and items:

for it in items:

it["box"] = [int(v / self.scale) for v in it["box"]]

with self._lock:

self.results = items

self._new_results = True

def stop(self):

self._running = False

def open_usb_camera(index, width, height):

"""打开 USB 摄像头。Windows 下用 DirectShow 后端更稳定。"""

if sys.platform == "win32":

cap = cv2.VideoCapture(index, cv2.CAP_DSHOW)

else:

cap = cv2.VideoCapture(index)

if not cap.isOpened():

cap.release()

raise RuntimeError(

f"无法打开摄像头 #{index}。\n"

"请检查：1) 摄像头是否插好 2) 是否被微信/QQ/其他程序占用 "

"3) 换个编号试：python ocr_usb.py --camera 1"

)

cap.set(cv2.CAP_PROP_FRAME_WIDTH, width)

cap.set(cv2.CAP_PROP_FRAME_HEIGHT, height)

cap.set(cv2.CAP_PROP_FPS, CAMERA_FPS)

return cap

def save_snapshot(frame, output_dir):

os.makedirs(output_dir, exist_ok=True)

name = datetime.now().strftime("frame_%Y%m%d_%H%M%S.jpg")

path = os.path.join(output_dir, name)

cv2.imwrite(path, frame)

return path

def main():

fix_windows_console_encoding()

args = parse_args()

print("正在初始化 PaddleOCR（首次运行会下载模型，请耐心等待）...")

ocr = PaddleOCR(

use_doc_orientation_classify=False,

use_doc_unwarping=False,

use_textline_orientation=False,

device="cpu",

)

print("PaddleOCR 初始化完成。\n")

width, height = RESOLUTIONS[args.res]

font = load_cn_font(22)

cap = open_usb_camera(args.camera, width, height)

w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))

h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))

print(f"摄像头已打开（编号 {args.camera}，分辨率 {w}x{h}，档位 {args.res}）。")

print("按 ESC 退出；按 s 保存当前画面。")

print("画面左上角的 FPS 数字：30 左右说明流畅；个位数说明摄像头/光线问题。\n")

worker = OcrWorker(ocr, scale=args.scale)

last_submit_time = 0.0

# 帧率统计（滑动平均）

fps = 0.0

last_t = time.time()

try:

while True:

ret, frame = cap.read()

if not ret:

print("[警告] 读不到画面，1 秒后重试 ...")

time.sleep(1)

continue

now = time.time()

dt = now - last_t

last_t = now

if dt > 0:

cur_fps = 1.0 / dt

fps = fps * 0.9 + cur_fps * 0.1

# 每隔 interval 秒丢一帧给后台识别（不阻塞显示）

if now - last_submit_time >= args.interval:

last_submit_time = now

worker.submit(frame)

items, is_new = worker.get_results()

items = [it for it in items if it["score"] >= args.score_thresh]

# 只在识别结果「更新」时打印一次，避免刷屏卡顿

if is_new:

for it in items:

print(f"[{it['score']:.3f}] {it['text']}")

if args.save and items:

print(f"已保存截图：{save_snapshot(frame, args.output)}")

worker.mark_consumed()

draw_results(frame, items, font)

# 显示帧率

cv2.putText(frame, f"FPS: {fps:.1f}", (10, 30),

cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 255), 2)

cv2.imshow("USB Camera OCR - ESC 退出 / s 保存", frame)

key = cv2.waitKey(1) & 0xFF

if key == 27: # ESC

break

if key == ord("s"):

print(f"已手动保存截图：{save_snapshot(frame, args.output)}")

except KeyboardInterrupt:

print("\n已手动中断。")

finally:

worker.stop()

cap.release()

cv2.destroyAllWindows()

print("程序已退出。")

if __name__ == "__main__":

main()

3.4 常见问题与排查

运行 ocr_usb.py 时可能遇到以下典型错误，可按表格中的方法逐一排查解决。

常见问题	可能原因	解决方法
相机无法打开	USB 线未插好、驱动未安装、摄像头被其他程序占用、编号错误	1. 检查 USB3.0 线连接与 MVS 驱动；2. 关闭微信、QQ 等占用摄像头的程序；3. 换编号运行：python ocr_usb.py --camera 1
PaddleOCR 模型下载失败	网络不稳定、未保持联网、下载源访问受限	1. 保持网络畅通后重试；2. 配置国内镜像源加速下载；3. 手动下载模型文件放入 ~/.paddleocr 目录
中文乱码	控制台编码不是 UTF-8、系统中文字体缺失	1. 程序已自动设置 UTF-8 输出，若仍乱码请执行 chcp 65001；2. 确认系统装有微软雅黑或黑体等中文字体
画面卡顿、FPS 过低	分辨率过高、识别间隔过短、CPU 负载过大	1. 使用 --res low 切换 640x480；2. 调大 --interval 识别间隔；3. 使用 --scale 0.5 缩小识别画面
识别结果为空或不准	光线不足、文字过小、对焦不准、置信度阈值过高	1. 调节环形光源亮度与镜头对焦；2. 将文字置于画面中央并靠近镜头；3. 调低 --score-thresh 阈值

3.4 运行测试

1. 执行 python ocr_usb.py --scale 0.5 启动识别。

2. 将文字置于镜头下，观察命令行识别结果与画面标注。

3. 按需调整 --scale、--score-thresh、--camera 等参数。

4. 按 s 保存截图，按 ESC 退出。

4 华为云 IoTDA 设备接入与数据上报

4.1 设备注册

1. 登录华为云控制台，进入“设备接入 IoTDA”服务，选择已创建的标准版实例。

2. 在左侧导航栏选择“设备”，点击“注册设备”。

3. 填写设备名称（如 iotda_temp_humidity），选择所属产品（如温湿度计），点击“确定”完成注册。

4. 注册成功后，系统生成设备 ID（device_id）与设备密钥（secret），需妥善保存，后续 MQTT 连接将使用。

5. 在设备列表中可查看设备状态，初始为“未激活”，待设备首次连接后变为“在线”。

4.2 产品模型定义

1. 在 IoTDA 控制台选择“产品”，创建产品（如“温湿度计”），协议类型选择 MQTT，数据格式选择 JSON。

2. 进入产品详情，点击“模型定义”，添加服务（Service），例如服务 ID 为 T&H。

3. 在服务下新增属性（Property）：

temperature：数据类型 decimal，访问权限可读可写，取值范围 -100~100。
humidity：数据类型 decimal，访问权限可读可写，取值范围 0~100。

4. 如需支持命令下发，在服务下新增命令（Command），例如设置命令 light，包含参数 turn（int 型）与 power（int 型）。

5. 保存并发布产品模型，使模型对设备生效。

4.3 MQTT 连接参数配置

华为云 IoTDA 使用 MQTT 协议接入，需配置以下连接参数：

参数	说明	示例值
接入地址	IoTDA 实例的 MQTT 接入地址	xxxx.iot-mqtts.cn-north-4.myhuaweicloud.com
端口	MQTT 端口，非加密为 1883，加密为 8883	1883
ClientID	格式：{device_id}_0_0_{timestamp}	65a1b2c3d4e5f6_0_0_20240101120000
Username	设备 ID	65a1b2c3d4e5f6
Password	使用 HMAC-SHA256 对密钥与时间戳签名生成	由工具生成的一串十六进制字符串

可使用华为云提供的“MQTT 连接参数生成工具”，输入设备 ID、密钥与时间戳，自动生成 ClientID、Username 与 Password 三件套。

4.4 属性上报与命令下发

4.4.1 属性上报

设备通过 MQTT 向平台上报属性数据，上报主题为：

$oc/devices/{device_id}/sys/properties/report

上报消息体为 JSON 格式，示例如下：

{
  "services": [
    {
      "service_id": "T&H",
      "properties": {
        "temperature": 25.6,
        "humidity": 60.2
      }
    }
  ]
}

4.4.2 命令下发

平台向设备下发命令时，设备需订阅命令主题：

$oc/devices/{device_id}/sys/commands/#

设备收到命令后，需向响应主题回复执行结果：

$oc/devices/{device_id}/sys/commands/response/request_id={request_id}

4.5 Python 代码示例

以下代码使用 paho-mqtt 库实现设备接入、属性上报与命令响应，可直接运行。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
华为云 IoTDA 设备接入与数据上报示例
====================================
功能：
1. 使用 MQTT 连接华为云 IoTDA 平台
2. 上报温湿度属性数据
3. 订阅并响应平台下发的命令
依赖：
pip install paho-mqtt
"""
import json
import time
import hmac
import hashlib
import random
import threading
import paho.mqtt.client as mqtt
========== 配置区：请替换为实际值 ==========
HOST = "xxxx.iot-mqtts.cn-north-4.myhuaweicloud.com"  # MQTT 接入地址
PORT = 1883                                            # MQTT 端口
DEVICE_ID = "65a1b2c3d4e5f6"                           # 设备 ID
SECRET = "your_device_secret"                          # 设备密钥
===========================================
def generate_password(secret, timestamp):
"""使用 HMAC-SHA256 生成 MQTT 连接密码"""
key = secret.encode("utf-8")
message = timestamp.encode("utf-8")
return hmac.new(key, message, hashlib.sha256).hexdigest()
def build_client_id(device_id, timestamp):
"""构造 ClientID：{device_id}0_0{timestamp}"""
return f"{device_id}0_0{timestamp}"
def on_connect(client, userdata, flags, rc):
"""连接成功回调"""
if rc == 0:
print("连接成功！")
# 订阅命令下发主题
client.subscribe(f"$oc/devices/{DEVICE_ID}/sys/commands/#")
print("已订阅命令主题")
else:
print(f"连接失败，返回码：{rc}")
def on_message(client, userdata, msg):
"""收到消息回调：处理平台下发的命令"""
print(f"收到命令：{msg.topic}")
try:
payload = json.loads(msg.payload.decode("utf-8"))
print(f"命令内容：{json.dumps(payload, ensure_ascii=False, indent=2)}")
    # 解析命令
    command_name = payload.get("command_name", "")
    paras = payload.get("paras", {})
    service_id = payload.get("service_id", "")
    request_id = payload.get("request_id", "")
# 处理命令
if command_name == "light":
    turn = paras.get("turn", 0)
    power = paras.get("power", 0)
    print(f"执行开灯命令：turn={turn}, power={power}")
回复命令响应
response_topic = (
f"$oc/devices/{DEVICE_ID}/sys/commands/response/request_id={request_id}"
)
response = {
"result_code": 0,
"response_name": command_name,
"paras": {"result": "success"}
}
client.publish(response_topic, json.dumps(response))
print("已回复命令响应")
except Exception as e:
print(f"处理命令出错：{e}")
def report_properties(client):
"""上报温湿度属性"""
topic = f"$oc/devices/{DEVICE_ID}/sys/properties/report"
payload = {
"services": [
{
"service_id": "T&H",
"properties": {
"temperature": round(random.uniform(20.0, 30.0), 1),
"humidity": round(random.uniform(40.0, 70.0), 1)
}
}
]
}
client.publish(topic, json.dumps(payload))
print(f"已上报属性：{json.dumps(payload, ensure_ascii=False)}")
def main():
生成时间戳
timestamp = str(int(time.time() * 1000))
生成连接三件套
client_id = build_client_id(DEVICE_ID, timestamp)
username = DEVICE_ID
password = generate_password(SECRET, timestamp)
print(f"ClientID：{client_id}")
print(f"Username：{username}")
print(f"Password：{password}")
创建 MQTT 客户端
client = mqtt.Client(client_id=client_id, protocol=mqtt.MQTTv311)
client.username_pw_set(username, password)
client.on_connect = on_connect
client.on_message = on_message
连接平台
client.connect(HOST, PORT, keepalive=120)
client.loop_start()
每 10 秒上报一次属性
try:
while True:
report_properties(client)
time.sleep(10)
except KeyboardInterrupt:
print("\n程序已手动中断。")
finally:
client.loop_stop()
client.disconnect()
print("已断开连接。")
if name == "main":
main()

4.6 运行结果验证

1. 安装依赖：pip install paho-mqtt。

2. 将代码中的 HOST、DEVICE_ID、SECRET 替换为实际值，执行 python iotda_report.py。

3. 观察控制台输出，确认连接成功、属性上报成功。

4. 登录 IoTDA 控制台，进入设备详情页，在“设备影子”或“消息跟踪”中查看上报的属性数据。

5. 在控制台“命令下发”页面发送命令，观察设备端是否收到并正确响应。

运行截图如下：

4 IoTDA中转空间构建

4.0 IoTDA构造

1. 登录华为云，搜索并开通“设备接入 IoTDA”服务。

2. 创建标准版实例，区域选择华北-北京四。

3. 进入实例总览，记录 MQTT 接入地址（端口 1883）。

4.1 产品及其模型构建

4.1.1 温湿度计

1. 创建产品：名称“温湿度计”，协议 MQTT，数据格式 JSON。

2. 添加服务 T&H，新增属性 temperature、humidity（decimal 型）。

3. 保存并发布产品模型。

温湿度计产品信息如下：

产品信息：产品名称：温湿度计; 协议类型：MQTT; 数据格式：JSON;

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

服务列表：服务 ID：T&H; 属性名称：temperature; 数据类型：decimal

访问权限：可读可写; 取值范围：-100~100; 属性名称：humidity

数据类型：decimal; 访问权限：可读可写; 取值范围：0~100

点击“产品创建产品”。

4.1.2 吸顶灯

1. 创建产品“智能吸顶灯”（MQTT / JSON）。

2. 添加服务 light：属性 luminance、power；命令 light（参数 turn、power）。

3. 添加服务 color：属性 crtColor；命令 color（参数 color）。

4. 保存并发布产品模型。

4.2 设备实例化应用

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

1. 注册设备，获取设备 ID 与密钥。

2. 用华为云工具生成 MQTT 连接三件套（ClientID / Username / Password）。

3. MQTT.fx 连接平台（接入地址、端口 1883、三件套，SSL 不勾选）。

4. 向 $oc/devices/{设备ID}/sys/properties/report 上报温湿度属性。

5. 在设备详情与消息跟踪中验证数据上报。

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

1. 注册吸顶灯设备，获取设备 ID 与密钥并生成三件套。

2. MQTT.fx 连接并订阅命令主题 $oc/devices/{设备ID}/sys/commands/#。

3. 上报 light / color 服务属性（luminance、power、crtColor）。

4. 平台命令下发：开灯、调功率、设置颜色。

5. 设备回复命令响应，消息跟踪确认响应成功。

4 设备接入与数据上报

本章将使用 MQTT.fx 模拟设备接入华为云 IoTDA 平台，完成从产品创建、设备注册、MQTT 连接参数配置到识别结果数据上报的完整流程，并给出关键配置截图占位符与 MQTT 报文示例。

4.1 创建产品

1. 登录华为云控制台，进入“设备接入 IoTDA”服务，选择已创建的标准版实例。

2. 在左侧导航栏选择“产品”，点击“创建产品”。

3. 填写产品信息：产品名称填写“工业视觉识别设备”，协议类型选择 MQTT，数据格式选择 JSON，所属行业选择“智慧工业”。

4. 点击“确定”完成产品创建。

5. 进入产品详情页，点击“模型定义”，添加服务（Service），服务 ID 填写 ocr_result。

6. 在服务下新增属性（Property）：

text：数据类型 string，访问权限可读可写，用于存放 OCR 识别出的文字内容。
score：数据类型 decimal，访问权限可读可写，用于存放识别置信度。
timestamp：数据类型 string，访问权限可读可写，用于存放识别时间戳。

7. 保存并发布产品模型，使模型对设备生效。

【截图占位符：产品创建页面截图】

【截图占位符：产品模型定义页面截图】

4.2 注册设备

1. 在 IoTDA 控制台左侧导航栏选择“设备”，点击“注册设备”。

2. 填写设备名称（如 ocr_device_01），选择所属产品（工业视觉识别设备），点击“确定”完成注册。

3. 注册成功后，系统生成设备 ID（device_id）与设备密钥（secret），需妥善保存，后续 MQTT 连接将使用。

4. 在设备列表中可查看设备状态，初始为“未激活”，待设备首次连接后变为“在线”。

【截图占位符：设备注册页面截图】

【截图占位符：设备注册成功后的设备 ID 与密钥截图】

4.3 配置 MQTT 连接参数

华为云 IoTDA 使用 MQTT 协议接入，需配置以下连接参数：

参数	说明	示例值
接入地址	IoTDA 实例的 MQTT 接入地址	xxxx.iot-mqtts.cn-north-4.myhuaweicloud.com
端口	MQTT 端口，非加密为 1883，加密为 8883	1883
ClientID	格式：{device_id}_0_0_{timestamp}	65a1b2c3d4e5f6a7b8c9d0e1_0_0_1690000000000
Username	设备 ID（device_id）	65a1b2c3d4e5f6a7b8c9d0e1
Password	使用 HMAC-SHA256 算法对密钥与时间戳计算生成	生成的加密字符串

可使用华为云提供的“MQTT 连接参数生成工具”，输入设备 ID、密钥与时间戳，自动生成 ClientID、Username 与 Password 三件套。

【截图占位符：MQTT 连接参数生成工具截图】

4.4 使用 MQTT.fx 连接平台

1. 打开 MQTT.fx 1.7.1，点击齿轮图标进入连接配置页面。

2. 在“General”选项卡中填写以下信息：

Profile Name：填写 iotda_ocr_device。
Broker Address：填写 IoTDA 实例的 MQTT 接入地址。
Broker Port：填写 1883。
Client ID：填写生成的 ClientID。

3. 在“User Credentials”选项卡中填写 Username 与 Password。

4. 非加密连接不勾选 SSL/TLS 选项，点击“OK”保存配置。

5. 点击“Connect”按钮连接平台，连接成功后 MQTT.fx 右上角指示灯变为绿色。

【截图占位符：MQTT.fx 连接配置页面截图】

【截图占位符：MQTT.fx 连接成功指示灯截图】

4.5 上报识别结果数据

设备通过 MQTT 向平台上报属性数据，上报主题为：

$oc/devices/{device_id}/sys/properties/report

上报消息体为 JSON 格式，示例如下：

{
  "services": [
    {
      "service_id": "ocr_result",
      "properties": {
        "text": "产品型号：HIK-12345",
        "score": 0.98,
        "timestamp": "2026-09-17 16:20:30"
      }
    }
  ]
}

在 MQTT.fx 中操作步骤如下：

1. 点击“Publish”选项卡。

2. 在 Topic 输入框中填写上报主题：$oc/devices/{device_id}/sys/properties/report。

3. 在 Payload 输入框中粘贴上述 JSON 报文。

4. 点击“Publish”按钮发送数据。

5. 登录 IoTDA 控制台，进入设备详情页，在“设备影子”或“消息跟踪”中查看上报的属性数据。

【截图占位符：MQTT.fx Publish 页面截图】

【截图占位符：IoTDA 控制台设备影子数据截图】

【截图占位符：IoTDA 控制台消息跟踪截图】

5 常见问题与排查

在实训过程中，从工业相机识别到物联网设备接入，可能遇到以下典型问题，可按表格中的方法逐一排查解决。

常见问题	现象	可能原因	解决步骤
海康相机在 MVS 中无法识别	打开 MVS 客户端后，设备列表为空，无法看到相机	USB3.0 线未插好、驱动未安装、相机被其他程序占用、USB 接口供电不足	1. 检查 USB3.0 线是否插紧，更换 USB 接口或线缆重试；2. 重新安装 MVS 客户端及相机驱动；3. 关闭微信、QQ 等占用摄像头的程序；4. 在 MVS 中点击刷新设备列表；5. 重启上位机后重新打开 MVS
PaddleOCR 首次运行下载模型失败	运行 ocr_usb.py 时提示模型下载超时或网络错误，程序无法继续	网络不稳定、默认下载源访问受限、代理设置干扰	1. 检查网络连接，确保可正常访问外网；2. 配置国内镜像源下载模型；3. 手动下载模型文件并放置到 PaddleOCR 模型目录；4. 关闭代理或配置代理白名单后重试；5. 多次重试，模型支持断点续传
中文标签乱码	画面上的中文识别结果显示为方块或乱码	cv2.putText 不支持中文、系统中文字体缺失、控制台编码不正确	1. 使用 Pillow 绘制中文标签（ImageDraw + ImageFont）；2. 确认系统存在中文字体（如微软雅黑 msyh.ttc）；3. 在代码开头调用 fix_windows_console_encoding() 设置 UTF-8 编码；4. 检查终端字体是否支持中文显示
视频卡顿或识别延迟高	画面显示不流畅，FPS 较低，识别结果更新缓慢	分辨率设置过高、识别线程阻塞主线程、CPU 性能不足、光源不足导致帧率下降	1. 使用 --res low 切换低分辨率模式；2. 使用 --scale 0.5 缩小识别画面；3. 增大 --interval 减少识别频率；4. 确认识别在后台线程运行，主线程只负责显示；5. 调节环形光源亮度，避免因曝光时间过长导致帧率下降
MQTT.fx 连接 IoTDA 失败	MQTT.fx 提示连接超时或返回错误码，无法接入平台	接入地址或端口错误、ClientID/Username/Password 三件套不正确、设备未注册、SSL 配置错误	1. 核对 IoTDA 实例的 MQTT 接入地址与端口（1883）；2. 使用华为云工具重新生成三件套，确认时间戳与设备 ID 正确；3. 确认设备已在 IoTDA 控制台注册且状态正常；4. 非加密连接不勾选 SSL；5. 检查防火墙或网络策略是否放行 1883 端口

参考资料

本实训涉及的主要官方文档与工具链接如下，供读者查阅与下载：

华为云 IoTDA 产品文档：Security Verification，用于了解设备接入、产品模型定义、MQTT 连接参数配置与消息跟踪等操作说明。
PaddleOCR GitHub 仓库与文档：https://github.com/PaddlePaddle/PaddleOCR，用于获取 OCR 识别库源码、安装说明与模型下载指引。
海康 MVS 客户端下载页：https://www.hikrobotics.com/cn/support/support-down，用于下载工业相机客户端与驱动，完成相机连接与出图配置。
MQTT.fx 官方下载页：https://mqttfx.org/，用于下载 MQTT.fx 客户端，模拟设备接入华为云 IoTDA 平台并完成属性上报与命令下发。
CodeArts 产品文档：Security Verification，用于了解项目创建、代码托管、构建与部署等操作说明。

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

5.1 应用项目创建

1. 进入华为云 CodeArts，创建“智慧路灯”项目。

2. 选择开发流程模板并完成项目初始化。

5.2 代码托管操作

1. 在项目中创建代码仓库。

2. 克隆仓库到本地，提交并推送智慧路灯代码。

5.3 构建任务设立

1. 创建构建任务，选择代码仓库与分支。

2. 配置构建环境、构建命令与产物路径。

5.4 项目构建实现

1. 执行构建任务。

2. 查看构建日志，确认构建成功并生成软件包。

5.5 项目部署运行

1. 创建部署任务并关联构建产物。

2. 部署至目标主机，运行并验证应用。

鲲鹏昇腾开发者社区是面向全社会开放的“联接全球计算开发者，聚合华为+生态”的社区，内容涵盖鲲鹏、昇腾资源，帮助开发者快速获取所需的知识、经验、软件、工具、算力，支撑开发者易学、好用、成功，成为核心开发者。

更多推荐

[人工智能]2026人工智能与大语言模型芯片家族参考V02：华为、Nvidia、AMD、Intel、Qualcomm

鲲鹏昇腾开发者社区

华为昇腾Ascend深度学习开发全解析｜从架构原理到MindSpore实战避坑指南

华为昇腾（Ascend）是华为推出的全栈自研AI NPU算力平台，区别于传统GPU通用计算，昇腾是专为AI深度学习设计的异构算力芯片，主打高算力、低功耗、国产化、全场景适配。昇腾芯片广泛应用于：AI训练集群、云端推理、边缘智能设备、自动驾驶、工业质检、大模型微调等场景，是目前国内最成熟的国产化AI算力生态之一。继承nn.Cell基类，__init__定义网络层，construct定义前向传播逻辑，