detect-gui/components/image_framework.py

import os
import threading
from ctypes import *
import numpy as np
from PyQt5.QtCore import QObject, pyqtSignal
from dynaconf.base import Settings
from numpy.ctypeslib import as_array
import platform
import time
import cv2

from core.edge_component import action, EdgeComponent, service
from util import get_system_and_library_suffix, imgProcess


# 给回调函数使用
class _SubRoiData_(Structure):
    _fields_ = [
        ('mPparentId', c_long),
        ('mId', c_long),
        ('mInfo', c_char * 256),
        ('isGood', c_bool),
        ('vpOffset', c_float * 2),
        ('mRoiPoints', c_float * 8),  # 4 * 2
        ('mInnerConners', c_float * 8),  # 4 * 2
        ('mInnerConners3D', c_float * 12),  # 4 * 3
        ('edges', c_float * 4),
        ('center', c_float * 2),
        ('cloud_size', c_int),
        ('mVc2D', POINTER(c_float)),
        ('mVc3D', POINTER(c_float)),
        ('H', c_float * 9),
    ]

SubRoiData = _SubRoiData_

class _CallbackInfo_(Structure):
    _fields_ = [
        ('code', c_int),
        ('errorInfo', c_char * 256),
        ('bGetData', c_bool),
        ('mId', c_long),
        ('mInfo', c_char * 256),
        ('mImPath', c_char * 256),
        ('mIm2D3DPtr', c_char * 256),
        ('roi_size', c_int),
        ('subRois', SubRoiData * 16),
    ]

CallbackInfo = _CallbackInfo_
# 回调函数类型定义
CallbackType = CFUNCTYPE(None, POINTER(CallbackInfo))

# emit的结构体
MSG_ONLY_RECORD         = -1
MSG_LOCATION_RECORD     = 0
MSG_DETECTION_RECORD    = 1

class Msg:
    def __init__(self, msg_type, record, im2D):
        self.msg_type = msg_type  # 0：刷新广角图片， 1：刷新高清， -1：不刷新图片，纯消息
        self.record = record
        self.im2D = im2D


TO_CLIENT_NOTIFY_BASE =  1000
SYS_BASE = 1000
CAL_2D3D_BASE = 2000
CAMERA_BASE = 4000
LIDAR_BASE = 5000
SYS_OK = 0
CAL_OK = 0
LIDAR_OK = 0
MSG_LIST = [
    { 1 + TO_CLIENT_NOTIFY_BASE : "开始相机拍摄" },
    { 2 + TO_CLIENT_NOTIFY_BASE : "相机拍摄已完成" },
    { 3 + TO_CLIENT_NOTIFY_BASE : "开始雷达采集数据" },
    { 4 + TO_CLIENT_NOTIFY_BASE : "雷达采集数据已完成" },
    { 5 + TO_CLIENT_NOTIFY_BASE : "开始定位检测" }, # 这个消息是相机实时定位，不用打印
    { 6 + TO_CLIENT_NOTIFY_BASE : "定位检测已完成" }, 
    { 7 + TO_CLIENT_NOTIFY_BASE : "开始角点2D检测" },
    { 8 + TO_CLIENT_NOTIFY_BASE : "角点2D检测已完成" },
    { 9 + TO_CLIENT_NOTIFY_BASE : "开始角点3D检测" },
    { 10 + TO_CLIENT_NOTIFY_BASE : "角点3D检测已完成" },
    { 11 + TO_CLIENT_NOTIFY_BASE : "开始计算预埋件尺寸" },
    { 12 + TO_CLIENT_NOTIFY_BASE : "预埋件尺寸计算已完成" },

    { 1 - SYS_BASE : "驱动初始化错误" },
    { 2 - SYS_BASE : "NPU计算单元初始化错误" },
    { 3 - SYS_BASE : "系统错误" },
    { 4 - SYS_BASE : "相机初始化错误" },
    { 5 - SYS_BASE : "当前没有可以读取的高清相片"},

    { 1 - CAL_2D3D_BASE : "读取图片文件错误" }, 
    { 2 - CAL_2D3D_BASE : "写图片文件错误" },
    { 3 - CAL_2D3D_BASE : "空指针" },
    { 4 - CAL_2D3D_BASE : "空参数" },
    { 5 - CAL_2D3D_BASE : "读取图像错误" },
    { 6 - CAL_2D3D_BASE : "NPU加载数据错误" },
    { 7 - CAL_2D3D_BASE : "NPU模型执行错误" },
    { 8 - CAL_2D3D_BASE : "NPU获取结果错误" },
    { 9 - CAL_2D3D_BASE : "NPU卸载数据错误" },
    { 10 - CAL_2D3D_BASE : "YOLO推理无可信ROI区域" },
    { 11 - CAL_2D3D_BASE : "ROI处理线程错误" },
    { 12 - CAL_2D3D_BASE : "H计算错误" },
    { 13 - CAL_2D3D_BASE : "3D角点检测错误" },
    { 14 - CAL_2D3D_BASE : "2D角点定位错误" },
    { 15 - CAL_2D3D_BASE : "3D投影计算错误" },
    { 16 - CAL_2D3D_BASE : "3D点云为空" },
    { 17 - CAL_2D3D_BASE : "未检测到2D角点" },
    { 18 - CAL_2D3D_BASE : "未检测到预埋件" },
    { 19 - CAL_2D3D_BASE : "相机捕获图像失败" },

    { 1 - CAMERA_BASE : "相机DLL初始化失败" },

    { 1 - LIDAR_BASE : "雷达初始化SDK失败" },
    { 2 - LIDAR_BASE : "雷达SO初始化失败" },
    { 3 - LIDAR_BASE : "雷达广播失败" },
    { 5 - LIDAR_BASE : "雷达开始采样失败" },
    { 6 - LIDAR_BASE : "雷达结束采样失败" },

    { -10001 : "已经在检测模式中"},
    { -10002 : "定位模式不支持执行该操作"}
]

#################################################################################### 
#                                   全局变量定义
#################################################################################### 

# 全局 msg 缓存
g_msg_cache = []

# 全局 frame 缓存
g_frame_cache = []

# 运行模式
RUNNING_MODE_NONE = -1 # 停止状态
RUNNING_MODE_LOCATION = 0 # 广角相机运行模式
RUNNING_MODE_DETECTION = 1 # 高清检测运行模式
g_running_mode = RUNNING_MODE_LOCATION # 默认是广角相机运行模式

#################################################################################### 
#                                   内部接口定义
#################################################################################### 
# 获取消息
def _get_msg_info(code) -> str: 
    global MSG_LIST
    for msg in MSG_LIST:
        if code == msg.keys()[0]:
            return msg.values
    return "未知错误"

# 加工检测图像
def _update_location_image(frame):
    global needHandle, recordCache, ckinfoCache
    def switch(choice):
        if ckinfoCache.code == 1006: #TO_CLIENT_NOTIFY_DETECTION_END: 1006
            for i in range(recordCache.roi_size):
                subRois = recordCache.subRois[i]
                s = ""
                for j in range(8):
                    s += str(subRois.mRoiPoints[j])+ " "
        else:
            print("default Case")

# 在回调函数中将 record 数据拷贝出来
def _copy_record(record):
    new_record = CallbackInfo()
    new_record.code = record.code
    new_record.errorInfo = record.errorInfo
    new_record.bGetData = record.bGetData
    for i in range(record.roi_size):
        subRois = record.subRois[i]
        newSubRois = new_record.subRois[i]
        for j in range(8):
            newSubRois.mRoiPoints[j] = subRois.mRoiPoints[j]
            newSubRois.mInnerConners[j] = subRois.mInnerConners[j]
        for j in range(12):
            newSubRois.mInnerConners3D[j] = subRois.mInnerConners3D[j]

    return new_record


class ImageFrameworkSignals(QObject):
    # QT 信号
    on_image_detect_result = pyqtSignal(Msg)


@action("image-framework", auto_start=True)
class ImageFramework(EdgeComponent):
    signals = ImageFrameworkSignals()

    def __init__(self, context):
        super().__init__(context)
        self.image_framework_sdk = None
        self.data_callback = None
        self.location_thread = None
        self.process_thread = None

    def configure(self, setting: Settings) -> None:
        self.init_image_framework_sdk()
        self.location_thread = threading.Thread(target=self._location_processing_thread)
        self.location_thread.daemon = True
        self.process_thread = threading.Thread(target=self._main_processing_thread)
        self.process_thread.daemon = True

        self.logger.info(f"Image framework configure done.")

    def init_image_framework_sdk(self):
        try:
            # 加载C++库
            current_file_dir = os.path.dirname(os.path.abspath(__file__))
            (arch, suffix) = get_system_and_library_suffix()
            self.image_framework_sdk = CDLL(os.path.join(current_file_dir, f'../vendors/image-framework/{arch}/image_framework.{suffix}'))
            # 设置回调函数
            self.init_callback()
        except Exception as e:
            self.logger.error(f"Load Image framework sdk failed: {str(e)}")

    def init_callback(self):
        def _callback(data):
            global g_msg_cache, g_frame_cache, g_running_mode

            record = CallbackInfo(code=SYS_OK, errorInfo=b"", bGetData=False)
            frame = None
            ret = SYS_OK
            record.code = data.contents.code
            record.errorInfo = data.contents.errorInfo
            # 当定位模式下，接收到【定位完成】消息
            if g_running_mode == RUNNING_MODE_LOCATION and record.code == (6 + TO_CLIENT_NOTIFY_BASE):
                if len(g_frame_cache) == 0:
                    self.logger.error("当前没有广角数据")
                    return

                frame = g_frame_cache.pop()  # 从图像缓存中去除图像
                g_frame_cache = []  # 清空缓存，避免无限增长
                msg = Msg(MSG_LOCATION_RECORD, _copy_record(record), frame)
            # 当检测模式下，接收到【拍照完成】消息
            elif g_running_mode == RUNNING_MODE_DETECTION and record.code == (2 + TO_CLIENT_NOTIFY_BASE):
                frame.zeros(9344, 7000, cv2.CV_8UC3)
                src_frame_ptr = c_char_p(frame.data.tobytes())
                ret = self.image_framework_sdk.LibapiGetHQImage(
                    src_frame_ptr, frame.shape[1], frame.shape[0], 3, c_char_p(b"PythonProcessing"))
                if ret != SYS_OK:
                    record = CallbackInfo(code=ret, errorInfo=_get_msg_info(ret), bGetData=False)
                    msg = Msg(MSG_ONLY_RECORD, record, None)
                    # emit msg
                    return
                msg = Msg(MSG_DETECTION_RECORD, _copy_record(record), frame)
            elif g_running_mode == RUNNING_MODE_LOCATION or g_running_mode == RUNNING_MODE_DETECTION:
                msg = Msg(MSG_ONLY_RECORD, _copy_record(record), None)
            else:
                print("未知回调")
                return

            g_msg_cache.append(msg)
            # self.logger.info("image framework callback done.")

        self.data_callback = CallbackType(_callback)

    @service()
    def start(self):
        super().start()
        if self.image_framework_sdk is None:
            raise RuntimeError("Image framework SDK未正确加载!")

        ret = self.image_framework_sdk.LibapiInit(1, "", self.data_callback)
        if ret != 0:
            raise RuntimeError("Image framework 初始化失败!")

        self.logger.info("Image framework started.")

    @service()
    def stop(self):
        if self.location_thread is not None:
            self.location_thread.join()
        if self.process_thread is not None:
            self.process_thread.join()

        self.image_framework_sdk.stop_sdk()
        self.logger.info("Image framework stopped.")

    @service()
    def start_location(self):
        self.location_thread.start()
        self.process_thread.start()

    @service()
    def start_detect(self):
        global g_running_mode
        if g_running_mode != RUNNING_MODE_DETECTION:
            g_running_mode = RUNNING_MODE_DETECTION  # 切换为检测模式

        ret = self.image_framework_sdk.LibapiStartDetection()
        if ret != 0:
            raise RuntimeError("启动检测失败!")

    @service()
    def continue_detect(self):
        ret = self.image_framework_sdk.LibapiContinuetDetection()
        if ret != 0:
            raise RuntimeError("继续检测失败!")

    @service()
    def stop_detect(self):
        global g_running_mode
        if g_running_mode == RUNNING_MODE_LOCATION:  # 如果在定位任务模式下，该接口不执行
            raise RuntimeError("正在检测，请稍后停止!")

        g_running_mode = RUNNING_MODE_NONE  # 切换为无模式
        ret = self.image_framework_sdk.LibapStopDetection()
        if ret != 0:
            raise RuntimeError("停止检测失败!")

    # 广角相机运行定位处理线程
    def _location_processing_thread(self):
        global g_msg_cache, g_frame_cache, g_running_mode
        # 打开广角摄像头
        capture = cv2.VideoCapture(0)
        while True:
            # 如果当前不是定位模式，不处理任何事务
            if g_running_mode != RUNNING_MODE_LOCATION:
                time.sleep(0.5)
                continue
            # 获取一帧图像
            _, frame = capture.read()
            # 如果图像为空，等待一秒，上报消息
            if frame is None:
                time.sleep(0.1)
                self.logger.error("[ERROR] 广角相机未捕获到图像")
                continue

            src_frame_ptr = c_char_p(frame.data.tobytes())
            ret = self.image_framework_sdk.LibapiCameraSendMsgWithImage(
                src_frame_ptr, frame.shape[1], frame.shape[0], 3,
                c_char_p(b"PythonProcessing"), c_char_p(b"_t_PluginCameraPro"))
            if ret != SYS_OK:
                self.logger.error(_get_msg_info(ret))
                # emit msg
                continue
            # 向 g_frame_cache 中加入消息，对图像进行二次加工
            g_frame_cache.append(frame)

    # 高清相机运行检测处理线程
    def _main_processing_thread(self):
        global g_msg_cache, g_frame_cache, g_running_mode

        while True:
            if len(g_msg_cache) <= 0:
                time.sleep(0.01)
                continue

            msg = g_msg_cache.pop()
            if msg.msg_type == MSG_LOCATION_RECORD:
                # cv2.imshow("LOCATION", msg.im2D)
                # cv2.waitKey(1)
                msg = imgProcess.process(msg)
                self.signals.on_image_detect_result.emit(msg)