1 工效组合展示

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

2.1 工具软件安装

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

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

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

3.1 IMA-DS借力

给出python虚拟环境OCR文字识别的编码及其运行过程，包括汉字，windows环境，CPU,非COnDA，使用本机连接的网络摄像机

3.2 运行环境构造

3.3 编码实现

import argparse

import os

import sys

import time

import logging

from dataclasses import dataclass

from typing import List, Tuple, Optional

import cv2

import numpy as np

# 跳过 HTTPS 证书校验，解决 EasyOCR 首次下载模型时的 CERTIFICATE_VERIFY_FAILED

import ssl

if hasattr(ssl, "_create_unverified_context"):

    ssl._create_default_https_context = ssl._create_unverified_context

logging.basicConfig(

    level=logging.INFO,

    format="%(asctime)s [%(levelname)s] %(message)s",

    datefmt="%H:%M:%S",

)

log = logging.getLogger("ocr_camera")

# ----------------------------------------------------------------------------

# 1. OCR 识别器

# ----------------------------------------------------------------------------

@dataclass

class OCRResult:

    text: str

    confidence: float

    box: np.ndarray  # 4 个角点 (4,2)

class OCRRecognizer:

    """封装 EasyOCR，支持中文 ch_sim + 英文 en，纯 CPU。"""

    def __init__(self, languages: Tuple[str, ...] = ("ch_sim", "en"),

                 gpu: bool = False, download_enabled: bool = True,

                 model_dir: Optional[str] = None):

        import easyocr

        resolved_dir = model_dir or os.path.expanduser("~/.EasyOCR/model/")

        log.info("OCR 模型目录：%s", resolved_dir)

        log.info("正在加载 OCR 模型（语言=%s）...", languages)

        kwargs = dict(

            lang_list=list(languages),

            gpu=gpu,

            download_enabled=download_enabled,

            verbose=False,

        )

        if model_dir:

            kwargs["model_storage_directory"] = model_dir

        self.reader = easyocr.Reader(**kwargs)

        self.languages = languages

    def recognize(self, image_bgr: np.ndarray) -> List[OCRResult]:

        raw = self.reader.readtext(image_bgr, detail=1, paragraph=False)

        results: List[OCRResult] = []

        for item in raw:

            box, text, conf = item

            box = np.array(box, dtype=np.int32)

            results.append(OCRResult(text=text, confidence=float(conf), box=box))

        return results

# ----------------------------------------------------------------------------

# 2. 摄像机取流（带断线自动重连）

# ----------------------------------------------------------------------------

def resolve_backend(backend_name: str, source: str) -> Optional[int]:

    mapping = {

        "auto": None,

        "dshow": getattr(cv2, "CAP_DSHOW", None),

        "msmf": getattr(cv2, "CAP_MSMF", None),

        "ffmpeg": getattr(cv2, "CAP_FFMPEG", None),

        "gstreamer": getattr(cv2, "CAP_GSTREAMER", None),

    }

    b = mapping.get(backend_name)

    # Windows 下 USB 摄像头默认用 DirectShow 后端

    if b is None and source.isdigit() and os.name == "nt":

        b = getattr(cv2, "CAP_DSHOW", None)

    return b

class CameraStream:

    """打开摄像机流，读取失败时自动重连。"""

    def __init__(self, source: str, max_retry: int = 5,

                 reopen_delay: float = 2.0, backend: Optional[int] = None):

        self.source = source

        self.max_retry = max_retry

        self.reopen_delay = reopen_delay

        self.backend = backend

        self.cap: Optional[cv2.VideoCapture] = None

        self._open()

    def _open(self):

        # USB 编号传 int，网络流/文件路径传 str（OpenCV 5.x 不认字符串编号）

        src = int(self.source) if str(self.source).strip().isdigit() else self.source

        if self.backend is not None:

            self.cap = cv2.VideoCapture(src, self.backend)

        else:

            self.cap = cv2.VideoCapture(src)

        if not self.cap or not self.cap.isOpened():

            log.warning("无法打开视频源：%s", self.source)

    def read(self) -> Tuple[bool, Optional[np.ndarray]]:

        if self.cap is None:

            self._open()

        for attempt in range(self.max_retry):

            if self.cap is None:

                return False, None

            ok, frame = self.cap.read()

            if ok and frame is not None:

                return True, frame

            log.warning("读帧失败，尝试重连 (%d/%d)...", attempt + 1, self.max_retry)

            time.sleep(self.reopen_delay)

            self.release()

            self._open()

        return False, None

    def release(self):

        if self.cap is not None:

            self.cap.release()

            self.cap = None

    def __del__(self):

        self.release()

# ----------------------------------------------------------------------------

# 3. 中文标注工具

# ----------------------------------------------------------------------------

class Annotator:

    """用 PIL + 系统中文字体绘制中文文字与矩形框。"""

    def __init__(self, font_path: Optional[str] = None, font_size: int = 24):

        self.font = self._load_font(font_path, font_size)

    @staticmethod

    def _load_font(font_path: Optional[str], font_size: int):

        from PIL import ImageFont

        candidates = [

            font_path,

            "C:/Windows/Fonts/simhei.ttf",

            "C:/Windows/Fonts/msyh.ttc",

            "C:/Windows/Fonts/simsun.ttc",

        ]

        for p in candidates:

            if p and os.path.exists(p):

                try:

                    return ImageFont.truetype(p, font_size)

                except Exception:

                    continue

        log.warning("未找到中文字体，中文可能显示为方框；请指定 --font")

        return ImageFont.load_default()

    def draw(self, image_bgr: np.ndarray, results: List[OCRResult]) -> np.ndarray:

        from PIL import Image, ImageDraw

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

        pil = Image.fromarray(rgb)

        draw = ImageDraw.Draw(pil)

        for r in results:

            pts = r.box.reshape(-1, 2).tolist()

            xs = [p[0] for p in pts]

            ys = [p[1] for p in pts]

            x1, y1, x2, y2 = min(xs), min(ys), max(xs), max(ys)

            draw.rectangle([x1, y1, x2, y2], outline=(0, 255, 0), width=2)

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

            bbox = draw.textbbox((x1, y1), label, font=self.font)

            draw.rectangle([bbox[0] - 2, bbox[1] - 2, bbox[2] + 2, bbox[3] + 2],

                           fill=(0, 0, 0))

            draw.text((x1, y1), label, font=self.font, fill=(0, 255, 0))

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

# ----------------------------------------------------------------------------

# 4. 主程序

# ----------------------------------------------------------------------------

def parse_args():

    ap = argparse.ArgumentParser(description="网络摄像机 OCR 文字识别（支持中文）")

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

                    help="摄像机流地址：RTSP/HTTP，或 0 表示本机摄像头")

    ap.add_argument("--languages", default="ch_sim,en",

                    help="OCR 语言，逗号分隔，如 ch_sim,en")

    ap.add_argument("--gpu", action="store_true", help="使用 GPU 加速")

    ap.add_argument("--font", default=None, help="中文字体路径（.ttf/.ttc）")

    ap.add_argument("--output-dir", default="snapshots", help="截图保存目录")

    ap.add_argument("--test-image", default=None,

                    help="提供图片路径时，只对该图做 OCR 测试，不打开摄像头")

    ap.add_argument("--low-res", type=int, default=0,

                    help=">0 时把帧缩放到此宽度后再识别，提速但降精度")

    ap.add_argument("--model-dir", default=None,

                    help="OCR 模型目录（默认 ~/.EasyOCR/model/）")

    ap.add_argument("--backend", default="auto",

                    choices=["auto", "dshow", "msmf", "ffmpeg", "gstreamer"],

                    help="视频后端；Windows USB 摄像头打不开时改用 dshow")

    return ap.parse_args()

def collect_text(results: List[OCRResult]) -> str:

    return "\n".join(f"[{r.confidence:.2f}] {r.text}" for r in results)

def run_camera(args):

    os.makedirs(args.output_dir, exist_ok=True)

    languages = tuple(x.strip() for x in args.languages.split(",") if x.strip())

    recognizer = OCRRecognizer(languages=languages, gpu=args.gpu,

                               model_dir=args.model_dir)

    annotator = Annotator(font_path=args.font)

    if args.test_image:

        img = cv2.imread(args.test_image)

        if img is None:

            log.error("无法读取图片：%s", args.test_image)

            return

        results = recognizer.recognize(img)

        out = annotator.draw(img, results)

        dest = os.path.join(args.output_dir, "test_result.jpg")

        cv2.imwrite(dest, out)

        log.info("识别结果：\n%s", collect_text(results))

        log.info("已保存标注图：%s", dest)

        return

    backend = resolve_backend(args.backend, str(args.source))

    cam = CameraStream(str(args.source), backend=backend)

    if cam.cap is None or not cam.cap.isOpened():

        log.error("摄像机未就绪，请检查 --source 地址、账号密码与网络。")

        return

    log.info("开始实时识别，按 'q' 退出，按 's' 截图保存当前帧识别结果。")

    frame_idx = 0

    while True:

        ok, frame = cam.read()

        if not ok or frame is None:

            log.error("持续读取失败，退出。")

            break

        frame_idx += 1

        detect_frame = frame

        if args.low_res and args.low_res > 0:

            h, w = frame.shape[:2]

            scale = args.low_res / max(w, 1)

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

        t0 = time.time()

        results = recognizer.recognize(detect_frame)

        if args.low_res and args.low_res > 0:

            scale = frame.shape[1] / max(detect_frame.shape[1], 1)

            for r in results:

                r.box = (r.box * scale).astype(np.int32)

        annotated = annotator.draw(frame, results)

        fps = 1.0 / max(time.time() - t0, 1e-6)

        cv2.putText(annotated, f"FPS(OCR):{fps:.1f} | texts:{len(results)}",

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

        if results:

            log.info("[帧 %d] 识别到 %d 条：%s", frame_idx, len(results),

                     " | ".join(r.text for r in results))

        cv2.imshow("OCR Camera", annotated)

        key = cv2.waitKey(1) & 0xFF

        if key == ord("q"):

            break

        elif key == ord("s"):

            dest = os.path.join(args.output_dir, f"shot_{frame_idx}.jpg")

            cv2.imwrite(dest, annotated)

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

    cam.release()

    cv2.destroyAllWindows()

def main():

    args = parse_args()

    try:

        run_camera(args)

    except KeyboardInterrupt:

        log.info("用户中断，退出。")

    except Exception as e:  # noqa

        log.exception("运行出错：%s", e)

        sys.exit(1)

if __name__ == "__main__":

    main()

3.4 运行测试

4 IoTDA中转空间构建

4.0 IoTDA构造

4.1 产品及其模型构建

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

----属性名称：crtColor ，数据类型：string，长度：10

----命令名称：setColor，数据类型：string，长度：10

​​​​​​​​​​​​​​4.2  设备实例化应用

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

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

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

5.1 应用项目创建

5.2 代码托管操作

5.3 构建任务设立

5.4 项目构建实现

5.5 项目部署运行