import sys
|
import time
|
import cv2
|
import numpy as np
|
import ctypes
|
import ctypes.wintypes
|
import yaml
|
import win32api
|
import win32con
|
import win32gui
|
import multiprocessing
|
from safety_detect import SAFETY_DETECT
|
from cam_util import CAM_UTIL
|
from identifier import IDENTIFIER
|
import os
|
from logger_config import logger
|
import threading
|
|
|
multiprocessing.freeze_support()
|
|
|
def variance_of_laplacian(image):
|
# 计算输入图像的拉普拉斯响应的方差
|
return cv2.Laplacian(image, cv2.CV_64F).var()
|
# 调用另一个长焦镜头，拍摄清晰的局部药材图片
|
def get_image():
|
logger.info("识别线程启动")
|
global is_loaded, class_count, class_count_max, class_sum
|
camera2_index = webcams.get(cam2)
|
print("第二个摄像头索引：" + str(camera2_index))
|
# 打开摄像头
|
capture = cv2.VideoCapture(camera2_index, cv2.CAP_DSHOW)
|
# 设置分辨率y
|
capture.set(cv2.CAP_PROP_FRAME_WIDTH, 2048) # 宽度
|
capture.set(cv2.CAP_PROP_FRAME_HEIGHT, 1540) # 高度
|
# 检查摄像头是否成功打开
|
if not capture.isOpened():
|
print("无法打开摄像头2")
|
logger.error("无法打开摄像头2")
|
exit()
|
width2 = capture.get(cv2.CAP_PROP_FRAME_WIDTH)
|
height2 = capture.get(cv2.CAP_PROP_FRAME_HEIGHT)
|
print("摄像头2分辨率:", width2, "x", height2)
|
logger.info(f"摄像头2分辨率:, {width2}, x, {height2}")
|
# 循环读取摄像头画面
|
# Shadows name 'width' from outer scope
|
count = 0
|
while True:
|
ret2, frame2 = capture.read()
|
if not ret2:
|
print("无法读取摄像头画面")
|
logger.error("无法读取摄像头画面")
|
break
|
count += 1
|
|
if count == 2:
|
herb_probabilities = herb_identifier(frame2)
|
top_five_classes = np.argsort(herb_probabilities, axis=1)[0][-5:][::-1]
|
name = ""
|
for i, class_id in enumerate(top_five_classes):
|
# 保留两位小数
|
probability = round(herb_probabilities[0][class_id], 2)
|
herb_class = herb_identifier.class_names[class_id]
|
name = name + herb_class + "=" + str(probability)
|
# 显示画面
|
# cv2.imshow('Output2', resized_frame2)
|
# 计算拉普拉斯响应的方差
|
laplacian = variance_of_laplacian(frame2)
|
# 生成保存文件名，以当前时间命名
|
save_name2 = name +"_["+ str(round(laplacian, 2)) +"]_"+ time.strftime("%Y%m%d%H%M%S", time.localtime()) + ".jpg"
|
logger.info(f"识别结果转换为保存图片名称:, {save_name2}")
|
# 判断图像的清晰度
|
# 保存调整尺寸后的图片
|
if laplacian > 200:
|
c_ = save_path + "2/c/"
|
# 判断文件是否存在，不存在则创建
|
if not os.path.exists(c_):
|
os.makedirs(c_)
|
cv2.imwrite(c_ + save_name2, frame2)
|
# 清晰条件下累计识别结果中药材名称出现的次数
|
# 累计每种药材不论名次出现的次数,累计每种药材置信度最高的次数,累计每种药材的置信度总和
|
# class_count = {}
|
# class_count_max = {}
|
# class_sum = {}
|
for i in range(len(top_five_classes)):
|
class_id = top_five_classes[i]
|
herb_class = herb_identifier.class_names[class_id]
|
# 累计每种药材不论名次出现的次数
|
if herb_class in class_count:
|
class_count[herb_class] += 1
|
else:
|
class_count[herb_class] = 1
|
# 累计每种药材置信度最高的次数
|
if i == 0 and herb_class in class_count_max:
|
class_count_max[herb_class] += 1
|
elif i == 0:
|
class_count_max[herb_class] = 1
|
# 累计每种药材的置信度总和
|
if herb_class in class_sum:
|
class_sum[herb_class] += herb_probabilities[0][class_id]
|
else:
|
class_sum[herb_class] = herb_probabilities[0][class_id]
|
is_loaded = True
|
else:
|
n_ = save_path + "2/n/"
|
# 判断文件是否存在，不存在则创建
|
if not os.path.exists(n_):
|
os.makedirs(n_)
|
cv2.imwrite(n_ + save_name2, frame2)
|
# cv2.imshow("Camera", resized_frame2)
|
print("保存图片:", save_name2)
|
break
|
# 结束线程
|
capture.release()
|
def send_result(class_count, class_count_max, class_sum):
|
global is_loaded
|
# 将三种统计结果输出到日志中
|
logger.info("class_count:"+str(class_count))
|
logger.info("class_count_max:"+str(class_count_max))
|
logger.info("class_sum:"+str(class_sum))
|
is_loaded = False
|
l.send_msg("airecognize," + f"{class_count}")
|
pass
|
|
|
def load_identify():
|
global is_loaded
|
# 摄像头索引号，通常为0表示第一个摄像头
|
camera_index = webcams.get(cam1)
|
print("第一个摄像头索引：" + str(camera_index))
|
# 打开摄像头
|
cap = cv2.VideoCapture(camera_index, cv2.CAP_DSHOW)
|
# 设置分辨率
|
# cap.set(cv2.CAP_PROP_FRAME_WIDTH, 3840) # 宽度
|
# cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 2160) # 高度
|
# 检查摄像头是否成功打开
|
if not cap.isOpened():
|
print("无法打开摄像头")
|
logger.error("无法打开摄像头")
|
exit()
|
width = cap.get(cv2.CAP_PROP_FRAME_WIDTH)
|
height = cap.get(cv2.CAP_PROP_FRAME_HEIGHT)
|
print("摄像头分辨率:", width, "x", height)
|
logger.info(f"摄像头分辨率:, {width}, x, {height}")
|
# 目标图像尺寸
|
# 计时器
|
frame_count = 0
|
start_time = time.time()
|
stime = time.time()
|
if not os.path.exists(save_path):
|
os.makedirs(save_path)
|
# 上次识别结果
|
class_old = "1"
|
# 累计次数
|
count = 0
|
# 上料状态
|
status = "没有上料"
|
|
# 累计每种药材不论名次出现的次数
|
class_count = {}
|
# 累计每种药材置信度最高的次数
|
class_count_max = {}
|
# 累计每种药材的置信度总和
|
class_sum = {}
|
# 循环读取摄像头画面
|
while True:
|
logger.info("循环读取摄像头画面")
|
# 睡眠100毫秒
|
time.sleep(0.1)
|
ret, frame = cap.read()
|
if not ret:
|
print("无法读取摄像头画面")
|
logger.error("无法读取摄像头画面")
|
break
|
# 获取当前时间
|
current_time = time.time()
|
# 安全检测
|
boxes, scores, class_ids = safety_detect(frame)
|
draw_img = safety_detect.draw_detections(frame, boxes, scores, class_ids)
|
print(boxes, scores, class_ids)
|
det_res = {}
|
if class_ids is not None:
|
# 遍历class_ids 转换成类别名称
|
for i in range(len(class_ids)):
|
class_id = class_ids[i]
|
class_name = safety_detect.class_names[class_id]
|
# 存入到det_res中
|
if class_name in det_res:
|
det_res[class_name] = det_res[class_name] if det_res[class_name] > scores[i] else scores[i]
|
else:
|
det_res[class_name] = scores[i]
|
|
logger.info(f"安全检测识别结果, {det_res}")
|
# 如果cass_ids中包含0，则表示有安全检测到人体
|
if 0 in class_ids:
|
l.send_msg("aidetect," + f"{det_res}")
|
# 上料识别
|
probabilities = load_identifier(frame)
|
# 找到最大概率的类别
|
predicted_class = np.argmax(probabilities, axis=1)[0]
|
max_probability = np.max(probabilities, axis=1)[0]
|
class_ = load_identifier.class_names[predicted_class]
|
# 计算类型重复的次数，类别更换之后重新计数
|
if class_ != class_old:
|
count = 0
|
else:
|
count += 1
|
class_old = class_
|
print(f"{class_}:{count}: {max_probability}")
|
logger.info(f"{class_}:{count}: {max_probability}")
|
# 判断是否上料并且上料次数大于10次
|
if class_ == "shangliao" and count > 10:
|
status = "正在上料"
|
# 每隔3秒取一帧图像
|
# 如果距离上一次保存已经过去1秒，则保存当前画面
|
if current_time - stime >= 10.0:
|
save_name = time.strftime("%Y%m%d%H%M%S", time.localtime()) + ".jpg"
|
# 保存调整尺寸后的图片
|
path_ = save_path + "1/"
|
if not os.path.exists(path_):
|
os.makedirs(path_)
|
cv2.imwrite(path_ + save_name, frame)
|
# 重置计时器
|
stime = time.time()
|
|
thread = threading.Thread(target=get_image)
|
thread.start()
|
|
else:
|
status = "没有上料"
|
if class_ == "meishangliao" and count == 3 and is_loaded:
|
logger.info("停止上料后发送识别结果")
|
send_result(class_count, class_count_max, class_sum)
|
if class_ == "meishangliao" and count == 1000:
|
is_loaded = False
|
logger.info("长时间未上料，重置正在上料状态")
|
# print(status)
|
|
# 计算帧速率
|
frame_count += 1
|
end_time = time.time()
|
elapsed_time = end_time - start_time
|
fps = frame_count / elapsed_time
|
# print(f"FPS: {fps:.2f}")
|
# 将FPS绘制在图像上
|
cv2.putText(frame, f"FPS: {fps:.2f}", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 2,
|
cv2.LINE_AA)
|
# 显示画面
|
cv2.imshow("Camera", draw_img)
|
# 检测按键，如果按下q键则退出循环
|
if cv2.waitKey(1) & 0xFF == ord('q'):
|
break
|
# 关闭摄像头
|
cap.release()
|
# 关闭所有窗口
|
cv2.destroyAllWindows()
|
|
|
|
# 读取配置文件
|
def read_config(file_path='./config/herb_ai.yaml'):
|
with open(file_path, 'r') as file:
|
config = yaml.safe_load(file)
|
return config
|
|
|
|
|
|
# 消息结构
|
class COPYDATASTRUCT(ctypes.Structure):
|
_fields_ = [
|
('dwData', ctypes.wintypes.LPARAM),
|
('cbData', ctypes.wintypes.DWORD),
|
('lpData', ctypes.c_char_p)
|
]
|
|
|
|
# logging.info("准备加载安全检测模型..")
|
# print("准备加载安全检测模型..")
|
# model_safe = SAFETY_DETECT(config['model']['safe'])
|
#
|
# logging.info("安全检测模型加载成功。")
|
# print("安全检测模型加载成功。")
|
# logging.info("准备加载药材识别模型..")
|
# print("准备加载药材识别模型..")
|
# model_cls = HERB_IDENTIFY(config['model']['cls'])
|
# logging.info("药材识别模型加载成功。")
|
# print("药材识别模型加载成功。")
|
|
|
class Listener:
|
def __init__(self):
|
WindowName = config['win']['windowName']
|
ClassName = config['win']['className']
|
message_map = {
|
win32con.WM_COPYDATA: self.OnCopyData,
|
win32con.WM_CLOSE: self.HandleClose,
|
}
|
wc = win32gui.WNDCLASS()
|
wc.lpfnWndProc = message_map
|
wc.lpszClassName = ClassName
|
hinst = wc.hInstance = win32api.GetModuleHandle(None)
|
classAtom = win32gui.RegisterClass(wc)
|
self.hwnd = win32gui.CreateWindow(
|
classAtom,
|
WindowName,
|
win32con.WS_OVERLAPPEDWINDOW, # 窗口样式
|
0,
|
0,
|
win32con.CW_USEDEFAULT,
|
win32con.CW_USEDEFAULT,
|
0,
|
0,
|
hinst,
|
None
|
)
|
logger.info(f"启动成功.当前句柄{self.hwnd}")
|
# # 隐藏窗口
|
# win32gui.ShowWindow(self.hwnd, win32con.SW_HIDE)
|
print("启动成功.当前句柄", self.hwnd)
|
|
|
def OnCopyData(self, hwnd, msg, wparam, lparam):
|
try:
|
# 记录开始时间
|
start_time = time.time()
|
pCDS = ctypes.cast(lparam, PCOPYDATASTRUCT)
|
s = ctypes.string_at(pCDS.contents.lpData).decode()
|
strArr = s.split(",")
|
logger.info(f"收到来自句柄{hwnd}的消息:{s}")
|
print(f"收到来自句柄{hwnd}的消息:{s}")
|
res = {}
|
msg = ""
|
# 发送指令：
|
# AI药材识别：100
|
# AI上料区安全检测：101
|
# AI下料满料识别：102
|
# AI下料区安全检测：103
|
|
# 接收指令： 药材识别接收指令：airecognize, {json}
|
# AI上料区安全检测：aidetect, {json}
|
if strArr[0] == '100':
|
pass
|
# names, confs = cls_process(strArr[1])
|
# for idx, name in enumerate(names):
|
# if name in res:
|
# res[name] = res[name] if res[name] > confs[idx] else confs[idx]
|
# else:
|
# res[name] = confs[idx]
|
# msg = "airecognize," + f"{res}"
|
elif strArr[0] == '101':
|
pass
|
# names, confs = safe_process(strArr[1])
|
# for idx, name in enumerate(names):
|
# if name in res:
|
# res[name] = res[name] if res[name] > confs[idx] else confs[idx]
|
# else:
|
# res[name] = confs[idx]
|
# msg = "aidetect," + f"{res}"
|
|
|
|
logger.info(f"识别结果：{res}")
|
self.send_msg(msg)
|
# 记录结束时间
|
end_time = time.time()
|
# 计算执行时间
|
execution_time = end_time - start_time
|
# 打印执行时间
|
print(f"程序执行时间为:{execution_time}秒")
|
logger.info(f"程序执行时间为:{execution_time,}秒")
|
logger.info("-" * 20)
|
|
except Exception as e:
|
print(e)
|
pass
|
return 1
|
|
def send_msg(self, msg):
|
# 通过寻找窗口名获取句柄
|
hwnd = FindWindow(config['tag_win']['className'], config['tag_win']['windowName'])
|
print(f"返回结果，发现窗口：{hwnd}")
|
logger.info(f"返回结果，发现窗口：{hwnd}")
|
# 或者直接根据句柄来发送
|
# hwnd = 1707560
|
cds = COPYDATASTRUCT()
|
cds.dwData = 0
|
msg_bytes = msg.encode('utf-8')
|
cds.cbData = ctypes.sizeof(ctypes.create_string_buffer(msg_bytes))
|
cds.lpData = ctypes.c_char_p(msg_bytes)
|
SendMessage(hwnd, win32con.WM_COPYDATA, 0, ctypes.byref(cds))
|
|
def HandleClose(self, hwnd, msg, wparam, lparam):
|
logger.info("关闭窗口")
|
print("handle close")
|
# 关闭当前窗口
|
win32gui.DestroyWindow(hwnd)
|
sys.exit()
|
return 0
|
|
if __name__ == '__main__':
|
cam1 = "USB Camera"
|
cam2 = "PC Camera"
|
camUtil = CAM_UTIL(cam1, cam2)
|
webcams = camUtil.list_webcams()
|
print("摄像头", webcams)
|
save_path = "data/images/"
|
# 是否上过料
|
is_loaded = False
|
# 加载ONNX模型
|
herb_identifier = IDENTIFIER("model/herb_identify.onnx")
|
load_identifier = IDENTIFIER("model/loading.onnx")
|
safety_detect = SAFETY_DETECT("model/safety_det.onnx")
|
config = read_config()
|
PCOPYDATASTRUCT = ctypes.POINTER(COPYDATASTRUCT)
|
|
# 查询窗口方法
|
FindWindow = ctypes.windll.user32.FindWindowW
|
# 发送消息方法
|
SendMessage = ctypes.windll.user32.SendMessageW
|
l = Listener()
|
load_identify()
|
win32gui.PumpMessages()
|
