工业视觉识别 + IoTDA 物联网 + CodeArts 应用开发

T康纳

328人浏览 · 2026-09-17 11:45:17

T康纳 · 2026-09-17 11:45:17 发布

1 工效组合展示

Ert

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

2.1 工具软件安装

在PC桌面左下方右键单击，单击“设备管理器”选项。

选择“设备管理器 > 其他设备”，如图所示。

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

单击“网络连接”选项，选择“高级网络设置”。

在“更多适配器选项”单击“编辑”进入“本地连接属性”界面。

“Internet 协议版本 4（TCP/IPv4）”，单击“属性”。

勾选“使用下面的 IP 地址”选项，填写IP地址（图示以192.168.0.101为例）和子网掩码，默认网关与DNS服务器地址为空，单击“确定”保存。

使用快捷键“Win+R”，在运行窗口输入cmd进入命令行窗口。输入ipconfig命令查询PC接口IP地址是否修改成功。

双击“MobaXterm_Personal_22.2.exe”启动SSH登录工具。

单击左上方的“Session”进入界面。

单击左上方的“SSH”进入SSH连接配置界面

根据硬件连接方式填写开发者套件连接PC的接口实际IP地址，勾选“Specify username”选项，填写用户名（以root为例）。

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

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

3.1 IMA-DS借力

3.2 运行环境构造

3.3 编码实现

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

"""

主程序：USB摄像头实时OCR

- 画面上用方框标出文字

- PowerShell里打印识别到的文字

- 后台线程OCR，画面流畅不卡

- 按 q 退出，按 s 保存截图

用法：

python ocr_camera.py # 默认用 config.py 里的摄像头编号

python ocr_camera.py --cam 0 # 指定摄像头编号

python ocr_camera.py --url rtsp://... # 也支持网络摄像头/RTSP

"""

import sys

import time

import argparse

from collections import defaultdict

import cv2

import numpy as np

import config

from ocr_engine import OCREngine

def parse_args():

parser = argparse.ArgumentParser(description="USB摄像头实时OCR")

parser.add_argument("--cam", type=int, default=config.CAMERA_INDEX,

help=f"USB摄像头编号，默认 {config.CAMERA_INDEX}")

parser.add_argument("--url", type=str, default="",

help="网络摄像头RTSP地址（填了就不用--cam）")

parser.add_argument("--width", type=int, default=config.CAMERA_WIDTH)

parser.add_argument("--height", type=int, default=config.CAMERA_HEIGHT)

return parser.parse_args()

def open_camera(args):

"""打开摄像头，支持USB和RTSP"""

if args.url:

print(f"正在打开网络摄像头: {args.url}")

cap = cv2.VideoCapture(args.url)

else:

print(f"正在打开USB摄像头，编号 = {args.cam} ...")

# Windows上用 DSHOW 后端，兼容性最好

cap = cv2.VideoCapture(args.cam, cv2.CAP_DSHOW)

if not cap.isOpened():

# 再试一次默认后端

cap = cv2.VideoCapture(args.cam)

if not cap.isOpened():

print("❌ 摄像头打开失败！")

print(" 请运行 python scan_cam.py 查看可用编号")

print(" 或检查USB摄像头是否插好")

sys.exit(1)

cap.set(cv2.CAP_PROP_FRAME_WIDTH, args.width)

cap.set(cv2.CAP_PROP_FRAME_HEIGHT, args.height)

print("✅ 摄像头已打开")

return cap

def draw_boxes(frame, ocr_results):

"""在画面上画方框和文字标签"""

if not ocr_results:

return frame

for item in ocr_results:

poly = item["poly"]

text = item["text"]

score = item["score"]

if poly is None:

continue

# poly 是 4个点 [[x,y],[x,y],[x,y],[x,y]]

try:

pts = np.array(poly, dtype=np.int32).reshape((-1, 1, 2))

cv2.polylines(frame, [pts], True, config.BOX_COLOR, config.BOX_THICKNESS)

# 在框上方写文字 + 置信度

top_left = pts[0][0]

label = f"{text} ({score:.2f})"

# 文字背景

(tw, th), _ = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX,

config.TEXT_FONT_SCALE, config.TEXT_THICKNESS)

cv2.rectangle(frame,

(int(top_left[0]), int(top_left[1]) - th - 8),

(int(top_left[0]) + tw, int(top_left[1])),

config.BOX_COLOR, -1)

cv2.putText(frame, label,

(int(top_left[0]), int(top_left[1]) - 4),

cv2.FONT_HERSHEY_SIMPLEX,

config.TEXT_FONT_SCALE,

(255, 255, 255),

config.TEXT_THICKNESS)

except Exception:

continue

return frame

def print_to_console(ocr_results, last_printed):

"""在PowerShell里打印识别到的文字，带去重"""

if not config.PRINT_TO_CONSOLE or not ocr_results:

return last_printed

now = time.time()

new_texts = []

for item in ocr_results:

text = item["text"]

score = item["score"]

# 去重：相同文字在 DEDUP_SECONDS 内不重复打印

if text in last_printed and (now - last_printed[text]) < config.DEDUP_SECONDS:

continue

last_printed[text] = now

new_texts.append((text, score))

if new_texts:

print("=" * 50)

for text, score in new_texts:

print(f" �� 识别文字: {text} (置信度: {score:.2f})")

print("=" * 50)

# 清理过期的去重记录

expired = [k for k, v in last_printed.items() if (now - v) > config.DEDUP_SECONDS * 5]

for k in expired:

del last_printed[k]

return last_printed

def main():

args = parse_args()

# 1. 打开摄像头

cap = open_camera(args)

# 2. 初始化OCR引擎（后台线程）

engine = OCREngine()

engine.start()

# 3. 主循环

last_printed = {}

fps_time = time.time()

fps_count = 0

fps = 0

saved_count = 0

print("\n�� 实时OCR已启动！")

print(" 画面上绿色方框 = 检测到的文字")

print(" PowerShell里会打印识别结果")

print(" 按 q 退出，按 s 保存当前画面\n")

try:

while True:

ret, frame = cap.read()

if not ret:

print("⚠️ 读取帧失败，重试...")

time.sleep(0.1)

continue

# 镜像翻转（可选，自拍模式更自然，不需要就注释掉）

# frame = cv2.flip(frame, 1)

# 把帧交给后台OCR线程（不阻塞）

engine.submit_frame(frame)

# 拿最新OCR结果画框

ocr_results = engine.get_latest_result()

frame = draw_boxes(frame, ocr_results)

# 控制台输出

last_printed = print_to_console(ocr_results, last_printed)

# 显示FPS

fps_count += 1

if time.time() - fps_time >= 1.0:

fps = fps_count

fps_count = 0

fps_time = time.time()

cv2.putText(frame, f"FPS: {fps} 文字数: {len(ocr_results)}",

(10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 255), 2)

cv2.imshow(config.WINDOW_NAME, frame)

key = cv2.waitKey(1) & 0xFF

if key == ord('q'):

print("\n�� 已退出")

break

elif key == ord('s'):

filename = f"ocr_screenshot_{saved_count}.jpg"

cv2.imwrite(filename, frame)

print(f"�� 已保存截图: {filename}")

saved_count += 1

except KeyboardInterrupt:

print("\n�� 已中断")

finally:

engine.stop()

cap.release()

cv2.destroyAllWindows()

print("✅ 资源已释放")

if __name__ == "__main__":

main()

3.4 运行测试

4 IoTDA中转空间构建

4.0 IoTDA构造

4.1 产品及其模型构建

通过以下链接访问华为云 IoTDA 管理控制台：

https://console.huaweicloud.com/iotdm/#/dm-portal/home

或者通过华为云官网 https://www.huaweicloud.com/，使用实验“DevEco Studio 安装配置与应用实验”中创

建的 IAM 用户访问 IoTDA 设备接入服务。

4.1.1 温湿度计

温湿度计产品信息如下：

产品信息：产品名称：温湿度计协议类型：MQTT数据格式：JSON所属行业：智能家居所属子行业：智能家居服务列表：服务 ID：T&H属性名称：temperature

数据类型：decimal访问权限：可读可写取值范围：-100~100属性名称：humidity数据类型：decimal访问权限：可读可写取值范围：0~100

点击“产品创建产品”。

4.1.2 吸顶灯

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

产品信息：

产品名称：智能吸顶灯

协议类型：MQTT

数据格式：JSON

所属行业：智能家居

所属子行业：智能家居

服务列表:服务 ID：Light属性名称：光照强度数据类型：decimal访问权限：可读可写取值范围：0~5000属性名称：功率数据类型：decimal访问权限：可读可写取值范围：0~10000命令名称：Light 下发参数名称：turn 数据类型：string长度：10下发参数名称：power 数据类型：decimal取值范围：0~10000服务 ID：Color属性名称：当前颜色数据类型：string长度：10命令名称：设置颜色下发参数名称：color 数据类型：string长度：10