gitadmin
/
tuoheng_algN
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

merge into: gitadmin:master
Branches Tags
gitadmin:master
gitadmin:zsl
gitadmin:dsj
gitadmin:bazong
gitadmin:jcq
gitadmin:wj
...
pull from: gitadmin:dsj
Branches Tags
gitadmin:master
gitadmin:zsl
gitadmin:dsj
gitadmin:bazong
gitadmin:jcq
gitadmin:wj

5 Commits
master ... dsj

Author SHA1 Message Date
jiangchaoqing 610670fd46 update 2025-08-11 10:30:02 +08:00
jiangchaoqing d61146f244 Fileupload 2025-08-11 09:46:44 +08:00
jiangchaoqing 8e49836d6e dsj-branch 2025-07-26 14:34:36 +08:00
jiangchaoqing 030d30e4e8 大数据AI视频独立显示 2025-07-26 14:17:01 +08:00
jiangchaoqing 855ba1bb40 dsj分支-增加大模型 2025-07-22 14:32:43 +08:00
4 changed files with 271 additions and 195 deletions
Show Stats Expand all files Collapse all files

10
README.md
View File

@ -1,8 +1,2 @@
1.2025.01.21把之前的tuoheng alg仓库代码重新开个仓库 (1)在config/service/dsp_test_service.yml里面添加参数，控制存储用的oss还是minio storage_source: 1 2.2025.02.06 (1)修改代码，把mqtt读取加入到系统中。config/service/dsp_test_service.yml，中添加mqtt_flag,决定是否启用。 (2)修改了minio情况下的，文件名命名方式。 3.2025.02.12 (1)增加了对alg算法开发的代码。可以通过配置文件config/service/dsp_test_service.yml中algSwitch: true，决定是否启用。 1.dsj
2.0811
4、2025.07.10 周树亮 - 增加人群计数，自研车牌模型，裸土覆盖3个场景
5、江朝庆 -- 0715
1）代码整理，删除冗余代码。
2）增加requirements.txt,方便部署
3) logs

296
concurrency/FileUploadThread.py
View File

@ -6,21 +6,25 @@ from traceback import format_exc
from loguru import logger from loguru import logger
import cv2 import cv2
import requests
import json
import base64
from entity.FeedBack import message_feedback from entity.FeedBack import message_feedback
from enums.ExceptionEnum import ExceptionType from enums.ExceptionEnum import ExceptionType
from enums.ModelTypeEnum import ModelType
from exception.CustomerException import ServiceException from exception.CustomerException import ServiceException
from util.AliyunSdk import AliyunOssSdk from util.AliyunSdk import AliyunOssSdk
from util.MinioSdk import MinioSdk from util.MinioSdk import MinioSdk
from util import TimeUtils from util import TimeUtils
from enums.AnalysisStatusEnum import AnalysisStatus from enums.AnalysisStatusEnum import AnalysisStatus
from util.PlotsUtils import draw_painting_joint, draw_name_ocr, draw_name_crowd from util.PlotsUtils import draw_painting_joint
from util.QueUtil import put_queue, get_no_block_queue, clear_queue from util.QueUtil import put_queue, get_no_block_queue, clear_queue
import io import io
from util.LocationUtils import locate_byMqtt from util.LocationUtils import locate_byMqtt
class FileUpload(Thread): class FileUpload(Thread):
__slots__ = ('_fb_queue', '_context', '_image_queue', '_analyse_type', '_msg','_mqtt_list') __slots__ = ('_fb_queue', '_context', '_image_queue', '_analyse_type', '_msg', '_mqtt_list')
def __init__(self, *args): def __init__(self, *args):
super().__init__() super().__init__()
@ -32,6 +36,16 @@ class FileUpload(Thread):
self._algSwitch = self._context['service']['algSwitch'] self._algSwitch = self._context['service']['algSwitch']
self.server_ip = self._context['llm']['server_ip']
self.server_port = self._context['llm']['server_port']
self.is_use_llm = self._context['llm']['is_use_llm']
self.cnt = self._context['llm']['cnt']
#0521: #0521:
default_enabled = str(self._msg.get("defaultEnabled", "True")).lower() == "true" default_enabled = str(self._msg.get("defaultEnabled", "True")).lower() == "true"
@ -44,15 +58,74 @@ class FileUpload(Thread):
# 这里放非默认逻辑的代码 # 这里放非默认逻辑的代码
self._algSwitch = False self._algSwitch = False
print("---line46 :FileUploadThread.py---",self._algSwitch) # 如果任务是在线、离线处理，则用此类
#如果任务是在线、离线处理，则用此类
class ImageFileUpload(FileUpload): class ImageFileUpload(FileUpload):
__slots__ = () __slots__ = ()
def extract_answers(self, text): # 提取大模型中的分点文本内容
lines = [line.strip() for line in text.split('.') if line.strip()]
answers = []
for i, line in enumerate(lines):
if i > 0:
if i < len(lines) - 1:
answers.append(line[:-1])
else:
answers.append(line)
return answers
def get_llm_res(self, image, prompt,SERVER_IP,API_PORT):
# SERVER_IP = "192.168.10.11"
# API_URL = f"http://{SERVER_IP}:8000/generate"
SERVER_IP = SERVER_IP
API_URL = f"http://{SERVER_IP}:{API_PORT}/generate"
# _, buffer = cv2.imencode('.jpg', image)
# is_success, buffer = cv2.imencode('.png', image)
is_success, buffer = cv2.imencode('.webp', image, [cv2.IMWRITE_WEBP_QUALITY, 100])
image_base64 = base64.b64encode(buffer).decode('utf-8')
conversation = {
"prompt": prompt,
"image_base64": image_base64,
}
try:
logger.info(f"正在向 {API_URL} 发送请求...")
response = requests.post(API_URL, json=conversation, timeout=60) # 设置超时
response.raise_for_status() # 如果状态码不是2xx，会抛出异常
resp = response.json()
result = resp.get("response")
logger.info("\n--- 请求成功 ---")
logger.info(f"原始Prompt: {prompt}")
logger.info(f"模型生成结果: {result}")
except requests.exceptions.RequestException as e:
logger.info(f"\n--- 请求失败 ---")
logger.info(f"发生错误: {e}")
except json.JSONDecodeError:
logger.info("\n--- 解析失败 ---")
logger.info(f"无法解析服务器返回的响应: {response.text}")
new_result = self.extract_answers(result)
print("###line116",new_result)
new_result = new_result[1] # 获取第二点内容
# flag = "不" in result or ("没" in result and "没错" not in result) or "否" in result or "未" in result or "无" in result
flag = "不存在" in new_result or ("没" in new_result and "没错" not in new_result) or (
"否" in new_result and "是否" not in new_result) or "未" in new_result or "无法" in new_result
return not flag, new_result
#@staticmethod #@staticmethod
def handle_image(self,frame_msg, frame_step): def handle_image(self, frame_msg, frame_step):
# (high_score_image["code"], all_frames, draw_config["font_config"]) # (high_score_image["code"], all_frames, draw_config["font_config"])
# high_score_image["code"][code][cls] = (frame, frame_index_list[i], cls_list) # high_score_image["code"][code][cls] = (frame, frame_index_list[i], cls_list)
det_xywh, frame, current_frame, all_frames, font_config = frame_msg det_xywh, frame, current_frame, all_frames, font_config = frame_msg
@ -67,7 +140,6 @@ class ImageFileUpload(FileUpload):
''' '''
print('*'*100,' mqtt_list:',len(self._mqtt_list)) print('*'*100,' mqtt_list:',len(self._mqtt_list))
model_info = [] model_info = []
# 更加模型编码解析数据 # 更加模型编码解析数据
for code, det_list in det_xywh.items(): for code, det_list in det_xywh.items():
@ -76,27 +148,15 @@ class ImageFileUpload(FileUpload):
if len(target_list) > 0: if len(target_list) > 0:
aFrame = frame.copy() aFrame = frame.copy()
for target in target_list: for target in target_list:
# 自研车牌模型判断 draw_painting_joint(target[1], aFrame, target[3], target[2], target[4], font_config, target[5])
if ModelType.CITY_CARPLATE_MODEL.value[1] == str(code): igH,igW = aFrame.shape[0:2]
box = [target[1][0][0], target[1][0][1], target[1][3][0], target[1][3][1]] if len(self._mqtt_list)>=1:
draw_name_ocr(box, aFrame, target[4], target[0]) #camParas = self._mqtt_list[0]['data']
cls = 0
elif ModelType.CITY_DENSECROWDCOUNT_MODEL.value[1] == str(code):
draw_name_crowd(target[3], aFrame, target[4], cls)
cls = 0
else:
draw_painting_joint(target[1], aFrame, target[3], target[2], target[4], font_config,
target[5])
igH, igW = aFrame.shape[0:2]
if len(self._mqtt_list) >= 1:
# camParas = self._mqtt_list[0]['data']
camParas = self._mqtt_list[0] camParas = self._mqtt_list[0]
gps = locate_byMqtt(target[1], igW, igH, camParas, outFormat='wgs84') gps = locate_byMqtt(target[1],igW,igH,camParas,outFormat='wgs84')
else: else:
gps = [None, None] gps=[None,None]
model_info.append( model_info.append({"modelCode": str(code), "detectTargetCode": str(cls), "aFrame": aFrame,'gps':gps})
{"modelCode": str(code), "detectTargetCode": str(cls), "aFrame": aFrame, 'gps': gps})
if len(model_info) > 0: if len(model_info) > 0:
image_result = { image_result = {
"or_frame": frame, "or_frame": frame,
@ -135,7 +195,7 @@ class ImageFileUpload(FileUpload):
# 获取队列中的消息 # 获取队列中的消息
image_msg = get_no_block_queue(image_queue) image_msg = get_no_block_queue(image_queue)
if image_msg is not None: if image_msg is not None:
if image_msg[0] == 2: if image_msg[0] == 2:
logger.info("图片上传线程收到命令:{}, requestId: {}",image_msg[1] ,request_id) logger.info("图片上传线程收到命令:{}, requestId: {}",image_msg[1] ,request_id)
if 'stop' == image_msg[1]: if 'stop' == image_msg[1]:
@ -143,7 +203,7 @@ class ImageFileUpload(FileUpload):
break break
if 'algStart' == image_msg[1]: self._algStatus = True; logger.info("图片上传线程,执行算法开启命令, requestId:{}", request_id) if 'algStart' == image_msg[1]: self._algStatus = True; logger.info("图片上传线程,执行算法开启命令, requestId:{}", request_id)
if 'algStop' == image_msg[1]: self._algStatus = False; logger.info("图片上传线程,执行算法关闭命令, requestId:{}", request_id) if 'algStop' == image_msg[1]: self._algStatus = False; logger.info("图片上传线程,执行算法关闭命令, requestId:{}", request_id)
if image_msg[0] == 1: if image_msg[0] == 1:
image_result = self.handle_image(image_msg[1], frame_step) image_result = self.handle_image(image_msg[1], frame_step)
if image_result is not None: if image_result is not None:
@ -153,30 +213,89 @@ class ImageFileUpload(FileUpload):
image_result["last_frame"], image_result["last_frame"],
analyse_type, analyse_type,
"OR", "0", "0", request_id) "OR", "0", "0", request_id)
if self._storage_source==1: if self._storage_source==1:
or_future = t.submit(minioSdk.put_object, or_image,or_image_name) or_future = t.submit(minioSdk.put_object, or_image, or_image_name)
else: else:
or_future = t.submit(aliyunOssSdk.put_object, or_image_name, or_image.tobytes()) or_future = t.submit(aliyunOssSdk.put_object, or_image_name, or_image.tobytes())
task.append(or_future) task.append(or_future)
model_info_list = image_result["model_info"] model_info_list = image_result["model_info"]
llm_flag_list = []
msg_list = [] msg_list = []
for model_info in model_info_list:
ai_image = cv2.imencode(".jpg", model_info["aFrame"])[1]
ai_image_name = build_image_name(image_result["current_frame"], for model_info in model_info_list:
image_result["last_frame"],
analyse_type, fire_flag = model_info.get("modelCode")
"AI", print("###line229",fire_flag)
model_info["modelCode"],
model_info["detectTargetCode"], if fire_flag == "008":
request_id)
if self._storage_source==1: aFrame = model_info["aFrame"]
ai_future = t.submit(minioSdk.put_object, ai_image, # aFrame_rgb = cv2.cvtColor(aFrame, cv2.COLOR_BGR2RGB)
ai_image_name)
else: tar_cls = model_info["detectTargetCode"]
ai_future = t.submit(aliyunOssSdk.put_object, ai_image_name, logger.info("目标类别：{}", tar_cls)
ai_image.tobytes())
# 0715 ： 这块也不对，因为需要具体到目标才可以，否则只看0 1 ，是无法做到判断正确的火情处理，还有大量其他的部分
# is_use_llm = True
if tar_cls == '0':
# tar = "明火"
tar = "2. 判断图中是否有明火。3. 如果有，识别火势的大小，比如：[无情况/小/中/大]"
elif tar_cls == '1':
# tar = "某种物质燃烧产生的烟雾"
tar = "2. 判断图中是否有烟雾。3. 如果有，识别烟雾的颜色，比如：[无烟雾/白/灰白/灰黑]"
else:
self.is_use_llm = False
llm_flag = True
if self.is_use_llm:
# prompt = f"我看到图中有{tar}，请你判断是否正确。简要回答。"
prompt = f"分点进行输出：1. 简短描述图像中的场景。{tar}"
logger.info("检测到目标，大模型提示词：{}", prompt)
# llm_flag, llm_res = self.get_llm_res(aFrame, prompt)
llm_flag, llm_res = self.get_llm_res(aFrame, prompt,self.server_ip,self.server_port)
llm_flag_list.append(llm_flag)
logger.info("检测到目标，大模型识别图中是否存在目标：{}, 大模型输出：{}", llm_flag, llm_res)
if llm_flag:
logger.info("经大模型筛查，小模型识别正确!!!")
# cv2.imwrite(f"/home/thsw2/wei/image_res/{self.cnt}_{tar_cls}_1.jpg", aFrame, [cv2.IMWRITE_JPEG_QUALITY, 80])
# self.cnt = self.cnt + 1
else:
logger.info("经大模型筛查，小模型识别错误!!!")
# cv2.imwrite(f"/home/thsw2/wei/image_res/{self.cnt}_{tar_cls}_0.jpg", aFrame, [cv2.IMWRITE_JPEG_QUALITY, 80])
# self.cnt = self.cnt + 1
else:
llm_flag_list.append(llm_flag)
ai_image = cv2.imencode(".jpg", aFrame)[1]
ai_image_name = build_image_name(image_result["current_frame"],
image_result["last_frame"],
analyse_type,
"AI",
model_info["modelCode"],
model_info["detectTargetCode"],
request_id)
if llm_flag: # 根据大模型结果判定是否上传
if self._storage_source==1:
ai_future = t.submit(minioSdk.put_object, ai_image,
ai_image_name)
else:
ai_future = t.submit(aliyunOssSdk.put_object, ai_image_name,
ai_image.tobytes())
task.append(ai_future)
# print("###line288",len(task))
task.append(ai_future)
#msg_list.append(message_feedback(request_id, #msg_list.append(message_feedback(request_id,
# AnalysisStatus.RUNNING.value, # AnalysisStatus.RUNNING.value,
# analyse_type, "", "", "", # analyse_type, "", "", "",
@ -184,22 +303,26 @@ class ImageFileUpload(FileUpload):
# ai_image_name, # ai_image_name,
# model_info['modelCode'], # model_info['modelCode'],
# model_info['detectTargetCode'])) # model_info['detectTargetCode']))
remote_image_list=[]
remote_image_list = []
for tk in task: for tk in task:
remote_image_list.append( tk.result()) remote_image_list.append( tk.result() )
for ii,model_info in enumerate(model_info_list):
msg_list.append( message_feedback(request_id,
AnalysisStatus.RUNNING.value,
analyse_type, "", "", "",
remote_image_list[0],
remote_image_list[ii+1],
model_info['modelCode'],
model_info['detectTargetCode'],
longitude=model_info['gps'][0],
latitude=model_info['gps'][1],
) )
for ii, model_info in enumerate(model_info_list):
if llm_flag_list[ii]:
msg_list.append( message_feedback(request_id,
AnalysisStatus.RUNNING.value,
analyse_type, "", "", "",
remote_image_list[0],
remote_image_list[ii+1],
model_info['modelCode'],
model_info['detectTargetCode'],
longitude=model_info['gps'][0],
latitude=model_info['gps'][1],
) )
if (not self._algSwitch) or ( self._algStatus and self._algSwitch): if (not self._algSwitch) or ( self._algStatus and self._algSwitch):
for msg in msg_list: for msg in msg_list:
put_queue(fb_queue, msg, timeout=2, is_ex=False) put_queue(fb_queue, msg, timeout=2, is_ex=False)
@ -225,7 +348,7 @@ def build_image_name(*args):
time_now = TimeUtils.now_date_to_str("%Y-%m-%d-%H-%M-%S") time_now = TimeUtils.now_date_to_str("%Y-%m-%d-%H-%M-%S")
return "%s/%s_frame-%s-%s_type_%s-%s-%s-%s_%s.jpg" % (request_id, time_now, current_frame, last_frame, return "%s/%s_frame-%s-%s_type_%s-%s-%s-%s_%s.jpg" % (request_id, time_now, current_frame, last_frame,
random_num, mode_type, modeCode, target, image_type) random_num, mode_type, modeCode, target, image_type)
#如果任务是图像处理，则用此类 #如果任务是图像处理，则用此类
class ImageTypeImageFileUpload(Thread): class ImageTypeImageFileUpload(Thread):
@ -254,20 +377,12 @@ class ImageTypeImageFileUpload(Thread):
if target_list is not None and len(target_list) > 0: if target_list is not None and len(target_list) > 0:
aiFrame = copy_frame.copy() aiFrame = copy_frame.copy()
for target in target_list: for target in target_list:
# 自研车牌模型判断 draw_painting_joint(target[1], aiFrame, target[3], target[2], target[4], font_config)
if ModelType.CITY_CARPLATE_MODEL.value[1] == str(code): model_info.append({
draw_name_ocr(target[1], aiFrame, font_config[cls], target[0]) "modelCode": str(code),
elif ModelType.CITY_DENSECROWDCOUNT_MODEL.value[1] == str(code): "detectTargetCode": str(cls),
draw_name_crowd(target[1],aiFrame,font_config[cls],target[0]) "frame": aiFrame
else: })
draw_painting_joint(target[1], aiFrame, target[3], target[2], target[4], font_config)
model_info.append({
"modelCode": str(code),
"detectTargetCode": str(cls),
"frame": aiFrame
})
if len(model_info) > 0: if len(model_info) > 0:
image_result = { image_result = {
"or_frame": copy_frame, "or_frame": copy_frame,
@ -291,7 +406,7 @@ class ImageTypeImageFileUpload(Thread):
minioSdk = MinioSdk(base_dir, env, request_id ) minioSdk = MinioSdk(base_dir, env, request_id )
else: else:
aliyunOssSdk = AliyunOssSdk(base_dir, env, request_id) aliyunOssSdk = AliyunOssSdk(base_dir, env, request_id)
start_time = time() start_time = time()
while True: while True:
try: try:
@ -312,12 +427,12 @@ class ImageTypeImageFileUpload(Thread):
if det_xywh is None: if det_xywh is None:
ai_image_name = build_image_name(0, 0, analyse_type, "AI", result.get("modelCode"), ai_image_name = build_image_name(0, 0, analyse_type, "AI", result.get("modelCode"),
result.get("type"), request_id) result.get("type"), request_id)
if self._storage_source==1: if self._storage_source==1:
ai_future = t.submit(minioSdk.put_object, copy_frame,ai_image_name) ai_future = t.submit(minioSdk.put_object, copy_frame,ai_image_name)
else: else:
ai_future = t.submit(aliyunOssSdk.put_object, ai_image_name, copy_frame) ai_future = t.submit(aliyunOssSdk.put_object, ai_image_name, copy_frame)
task.append(ai_future) task.append(ai_future)
remote_names.append(ai_image_name) remote_names.append(ai_image_name)
#msg_list.append(message_feedback(request_id, #msg_list.append(message_feedback(request_id,
@ -331,7 +446,7 @@ class ImageTypeImageFileUpload(Thread):
else: else:
image_result = self.handle_image(det_xywh, copy_frame, font_config) image_result = self.handle_image(det_xywh, copy_frame, font_config)
if image_result: if image_result:
# 图片帧数编码 # 图片帧数编码
if image_url is None: if image_url is None:
or_result, or_image = cv2.imencode(".jpg", image_result.get("or_frame")) or_result, or_image = cv2.imencode(".jpg", image_result.get("or_frame"))
@ -339,8 +454,8 @@ class ImageTypeImageFileUpload(Thread):
image_result.get("last_frame"), image_result.get("last_frame"),
analyse_type, analyse_type,
"OR", "0", "O", request_id) "OR", "0", "O", request_id)
if self._storage_source==1: if self._storage_source==1:
or_future = t.submit(minioSdk.put_object, or_image,image_url_0) or_future = t.submit(minioSdk.put_object, or_image,image_url_0)
else: else:
or_future = t.submit(aliyunOssSdk.put_object, image_url_0, or_future = t.submit(aliyunOssSdk.put_object, image_url_0,
@ -357,9 +472,9 @@ class ImageTypeImageFileUpload(Thread):
model_info.get("modelCode"), model_info.get("modelCode"),
model_info.get("detectTargetCode"), model_info.get("detectTargetCode"),
request_id) request_id)
if self._storage_source==1: if self._storage_source==1:
ai_future = t.submit(minioSdk.put_object, ai_image, ai_image_name) ai_future = t.submit(minioSdk.put_object, ai_image, ai_image_name)
else: else:
ai_future = t.submit(aliyunOssSdk.put_object, ai_image_name, ai_future = t.submit(aliyunOssSdk.put_object, ai_image_name,
ai_image.tobytes()) ai_image.tobytes())
task.append(ai_future) task.append(ai_future)
@ -375,8 +490,9 @@ class ImageTypeImageFileUpload(Thread):
remote_url_list = [] remote_url_list = []
for thread_result in task: for thread_result in task:
remote_url_list.append(thread_result.result()) remote_url_list.append(thread_result.result())
# 以下代码是为了获取图像上传后，返回的全路径地址
#以下代码是为了获取图像上传后，返回的全路径地址
if det_xywh is None: if det_xywh is None:
msg_list.append(message_feedback(request_id, msg_list.append(message_feedback(request_id,
AnalysisStatus.RUNNING.value, AnalysisStatus.RUNNING.value,
@ -408,7 +524,11 @@ class ImageTypeImageFileUpload(Thread):
model_info_list[ii].get('modelCode'), model_info_list[ii].get('modelCode'),
model_info_list[ii].get('detectTargetCode'), model_info_list[ii].get('detectTargetCode'),
analyse_results=result)) analyse_results=result))
for msg in msg_list: for msg in msg_list:
put_queue(fb_queue, msg, timeout=2, is_ex=False) put_queue(fb_queue, msg, timeout=2, is_ex=False)
else: else:

153
concurrency/PushVideoStreamProcess.py
View File

@ -23,7 +23,7 @@ from util.Cv2Utils import video_conjuncing, write_or_video, write_ai_video, push
from util.ImageUtils import url2Array, add_water_pic from util.ImageUtils import url2Array, add_water_pic
from util.LogUtils import init_log from util.LogUtils import init_log
from util.PlotsUtils import draw_painting_joint, filterBox, xywh2xyxy2, xy2xyxy, draw_name_joint, plot_one_box_auto, draw_name_ocr,draw_name_crowd from util.PlotsUtils import draw_painting_joint, filterBox, xywh2xyxy2, draw_name_joint
from util.QueUtil import get_no_block_queue, put_queue, clear_queue from util.QueUtil import get_no_block_queue, put_queue, clear_queue
@ -37,7 +37,7 @@ class PushStreamProcess(Process):
self._msg, self._push_queue, self._image_queue, self._push_ex_queue, self._hb_queue, self._context = args self._msg, self._push_queue, self._image_queue, self._push_ex_queue, self._hb_queue, self._context = args
self._algStatus = False # 默认关闭 self._algStatus = False # 默认关闭
self._algSwitch = self._context['service']['algSwitch'] self._algSwitch = self._context['service']['algSwitch']
#0521: #0521:
default_enabled = str(self._msg.get("defaultEnabled", "True")).lower() == "true" default_enabled = str(self._msg.get("defaultEnabled", "True")).lower() == "true"
@ -49,9 +49,16 @@ class PushStreamProcess(Process):
print("执行替代程序（defaultEnabled=False）") print("执行替代程序（defaultEnabled=False）")
# 这里放非默认逻辑的代码 # 这里放非默认逻辑的代码
self._algSwitch = False self._algSwitch = False
print("---line53 :PushVideoStreamProcess.py---",self._algSwitch) print("---line53 :PushVideoStreamProcess.py---",self._algSwitch)
def build_logo_url(self): def build_logo_url(self):
logo = None logo = None
if self._context["video"]["video_add_water"]: if self._context["video"]["video_add_water"]:
@ -148,39 +155,17 @@ class OnPushStreamProcess(PushStreamProcess):
font_config, allowedList = draw_config["font_config"], draw_config[code]["allowedList"] font_config, allowedList = draw_config["font_config"], draw_config[code]["allowedList"]
rainbows, label_arrays = draw_config[code]["rainbows"], draw_config[code]["label_arrays"] rainbows, label_arrays = draw_config[code]["rainbows"], draw_config[code]["label_arrays"]
for qs in det_result: for qs in det_result:
# 自研车牌模型处理 try: # 应对NaN情况
if ModelType.CITY_CARPLATE_MODEL.value[1] == str(code): box, score, cls = xywh2xyxy2(qs)
cls = 0 except:
ocrlabel, xybox = qs continue
box = xy2xyxy(xybox) if cls not in allowedList or score < frame_score:
score = None continue
color = rainbows[cls] label_array, color = label_arrays[cls], rainbows[cls]
label_array = None if ModelType.CHANNEL2_MODEL.value[1] == str(code) and cls == 2:
rr = t.submit(draw_name_ocr, xybox, copy_frame, color, ocrlabel) rr = t.submit(draw_name_joint, box, copy_frame, draw_config[code]["label_dict"], score, color, font_config, qs[6])
elif ModelType.CITY_DENSECROWDCOUNT_MODEL.value[1] == str(code):
cls = 0
crowdlabel, points = qs
box = [(0, 0), (0, 0), (0, 0), (0, 0)]
score = None
color = rainbows[cls]
label_array = None
rr = t.submit(draw_name_crowd, points, copy_frame, color, crowdlabel)
else: else:
try: # 应对NaN情况 rr = t.submit(draw_painting_joint, box, copy_frame, label_array, score, color, font_config)
box, score, cls = xywh2xyxy2(qs)
except:
continue
if cls not in allowedList or score < frame_score:
continue
label_array, color = label_arrays[cls], rainbows[cls]
if ModelType.CHANNEL2_MODEL.value[1] == str(code) and cls == 2:
rr = t.submit(draw_name_joint, box, copy_frame,
draw_config[code]["label_dict"], score, color,
font_config, qs[6])
else:
rr = t.submit(draw_painting_joint, box, copy_frame, label_array,
score, color, font_config)
thread_p.append(rr) thread_p.append(rr)
if det_xywh.get(code) is None: if det_xywh.get(code) is None:
det_xywh[code] = {} det_xywh[code] = {}
@ -199,18 +184,22 @@ class OnPushStreamProcess(PushStreamProcess):
box[2][0], box[2][1], box[3][0], box[3][1], box[2][0], box[2][1], box[3][0], box[3][1],
score, cls, code],dtype=np.float32) score, cls, code],dtype=np.float32)
qs_np = np.row_stack((qs_np, result_li)) qs_np = np.row_stack((qs_np, result_li))
if logo: if logo:
frame = add_water_pic(frame, logo, request_id) frame = add_water_pic(frame, logo, request_id)
copy_frame = add_water_pic(copy_frame, logo, request_id) copy_frame = add_water_pic(copy_frame, logo, request_id)
if len(thread_p) > 0: if len(thread_p) > 0:
for r in thread_p: for r in thread_p:
r.result() r.result()
#print('----line173:',self._algSwitch,self._algStatus) #print('----line173:',self._algSwitch,self._algStatus)
if self._algSwitch and (not self._algStatus): if self._algSwitch and (not self._algStatus):
frame_merge = video_conjuncing(frame, frame.copy()) # frame_merge = video_conjuncing(frame, frame.copy())
else:
frame_merge = video_conjuncing(frame, copy_frame) frame_merge = frame.copy()
else:
# frame_merge = video_conjuncing(frame, copy_frame)
frame_merge = copy_frame
# 写原视频到本地 # 写原视频到本地
write_or_video_result = t.submit(write_or_video, frame, orFilePath, or_video_file, write_or_video_result = t.submit(write_or_video, frame, orFilePath, or_video_file,
or_write_status, request_id) or_write_status, request_id)
@ -222,7 +211,7 @@ class OnPushStreamProcess(PushStreamProcess):
# 如果有问题, 走下面的逻辑 # 如果有问题, 走下面的逻辑
if qs_np is not None: if qs_np is not None:
if len(qs_np.shape) == 1: if len(qs_np.shape) == 1:
qs_np = qs_np[np.newaxis,...] qs_np = qs_np[np.newaxis,...]
qs_np_id = qs_np.copy() qs_np_id = qs_np.copy()
b = np.ones(qs_np_id.shape[0]) b = np.ones(qs_np_id.shape[0])
qs_np_id = np.column_stack((qs_np_id,b)) qs_np_id = np.column_stack((qs_np_id,b))
@ -248,7 +237,7 @@ class OnPushStreamProcess(PushStreamProcess):
if q[11] >= 1: if q[11] >= 1:
cls = int(q[9]) cls = int(q[9])
if not (cls in new_lab): if not (cls in new_lab):
continue # 为了防止其他类别被带出 continue # 为了防止其他类别被带出
code = str(int(q[10])).zfill(3) code = str(int(q[10])).zfill(3)
if det_xywh2.get(code) is None: if det_xywh2.get(code) is None:
det_xywh2[code] = {} det_xywh2[code] = {}
@ -261,11 +250,6 @@ class OnPushStreamProcess(PushStreamProcess):
is_new = False is_new = False
if q[11] == 1: if q[11] == 1:
is_new = True is_new = True
if ModelType.CITY_CARPLATE_MODEL.value[1] == str(code):
cls = ocrlabel
elif ModelType.CITY_DENSECROWDCOUNT_MODEL.value[1] == str(code):
cls = crowdlabel
label_array = points
if cd is None: if cd is None:
det_xywh2[code][cls] = [[cls, box, score, label_array, color, is_new]] det_xywh2[code][cls] = [[cls, box, score, label_array, color, is_new]]
else: else:
@ -288,7 +272,7 @@ class OnPushStreamProcess(PushStreamProcess):
ex_status = False ex_status = False
logger.info("停止推流进程, requestId: {}", request_id) logger.info("停止推流进程, requestId: {}", request_id)
break break
del push_r del push_r
else: else:
sleep(1) sleep(1)
@ -383,45 +367,24 @@ class OffPushStreamProcess(PushStreamProcess):
qs_reurn = [] qs_reurn = []
for det in push_objs[i]: for det in push_objs[i]:
code, det_result = det code, det_result = det
# 每个单独模型处理 # 每个单独模型处理
# 模型编号、100帧的所有问题, 检测目标、颜色、文字图片 # 模型编号、100帧的所有问题, 检测目标、颜色、文字图片
if len(det_result) > 0: if len(det_result) > 0:
font_config, allowedList = draw_config["font_config"], draw_config[code]["allowedList"] font_config, allowedList = draw_config["font_config"], draw_config[code]["allowedList"]
rainbows, label_arrays = draw_config[code]["rainbows"], draw_config[code]["label_arrays"] rainbows, label_arrays = draw_config[code]["rainbows"], draw_config[code]["label_arrays"]
for qs in det_result: for qs in det_result:
# 自研车牌模型处理 box, score, cls = xywh2xyxy2(qs)
if ModelType.CITY_CARPLATE_MODEL.value[1] == str(code):
cls = 0 if cls not in allowedList or score < frame_score:
ocrlabel, xybox = qs continue
box = xy2xyxy(xybox) label_array, color = label_arrays[cls], rainbows[cls]
score = None if ModelType.CHANNEL2_MODEL.value[1] == str(code) and cls == 2:
color = rainbows[cls] rr = t.submit(draw_name_joint, box, copy_frame, draw_config[code]["label_dict"], score, color, font_config, qs[6])
label_array = None
label_arrays = [None]
rr = t.submit(draw_name_ocr,xybox,copy_frame,color,ocrlabel)
elif ModelType.CITY_DENSECROWDCOUNT_MODEL.value[1] == str(code):
cls = 0
crowdlabel, points = qs
box = [(0,0),(0,0),(0,0),(0,0)]
score = None
color = rainbows[cls]
label_array = None
rr = t.submit(draw_name_crowd, points, copy_frame, color, crowdlabel)
else: else:
box, score, cls = xywh2xyxy2(qs) rr = t.submit(draw_painting_joint, box, copy_frame, label_array, score, color, font_config)
if cls not in allowedList or score < frame_score:
continue
label_array, color = label_arrays[cls], rainbows[cls]
if ModelType.CHANNEL2_MODEL.value[1] == str(code) and cls == 2:
rr = t.submit(draw_name_joint, box, copy_frame, draw_config[code]["label_dict"], score, color, font_config, qs[6])
else:
rr = t.submit(draw_painting_joint, box, copy_frame, label_array, score, color, font_config)
thread_p.append(rr) thread_p.append(rr)
if det_xywh.get(code) is None: if det_xywh.get(code) is None:
det_xywh[code] = {} det_xywh[code] = {}
cd = det_xywh[code].get(cls) cd = det_xywh[code].get(cls)
@ -429,17 +392,17 @@ class OffPushStreamProcess(PushStreamProcess):
if cd is None: if cd is None:
det_xywh[code][cls] = [[cls, box, score, label_array, color]] det_xywh[code][cls] = [[cls, box, score, label_array, color]]
else: else:
det_xywh[code][cls].append([cls, box, score, label_array, color]) det_xywh[code][cls].append([cls, box, score, label_array, color])
if qs_np is None: if qs_np is None:
qs_np = np.array([box[0][0], box[0][1], box[1][0], box[1][1], qs_np = np.array([box[0][0], box[0][1], box[1][0], box[1][1],
box[2][0], box[2][1], box[3][0], box[3][1], box[2][0], box[2][1], box[3][0], box[3][1],
score, cls, code],dtype=np.float32) score, cls, code],dtype=np.float32)
else: else:
result_li = np.array([box[0][0], box[0][1], box[1][0], box[1][1], result_li = np.array([box[0][0], box[0][1], box[1][0], box[1][1],
box[2][0], box[2][1], box[3][0], box[3][1], box[2][0], box[2][1], box[3][0], box[3][1],
score, cls, code],dtype=np.float32) score, cls, code],dtype=np.float32)
qs_np = np.row_stack((qs_np, result_li)) qs_np = np.row_stack((qs_np, result_li))
if logo: if logo:
frame = add_water_pic(frame, logo, request_id) frame = add_water_pic(frame, logo, request_id)
copy_frame = add_water_pic(copy_frame, logo, request_id) copy_frame = add_water_pic(copy_frame, logo, request_id)
@ -447,9 +410,11 @@ class OffPushStreamProcess(PushStreamProcess):
for r in thread_p: for r in thread_p:
r.result() r.result()
if self._algSwitch and (not self._algStatus): if self._algSwitch and (not self._algStatus):
frame_merge = video_conjuncing(frame, frame.copy()) # frame_merge = video_conjuncing(frame, frame.copy())
else: frame_merge = frame.copy()
frame_merge = video_conjuncing(frame, copy_frame) else:
# frame_merge = video_conjuncing(frame, copy_frame)
frame_merge = copy_frame
# 写识别视频到本地 # 写识别视频到本地
write_ai_video_result = t.submit(write_ai_video, frame_merge, aiFilePath, write_ai_video_result = t.submit(write_ai_video, frame_merge, aiFilePath,
ai_video_file, ai_video_file,
@ -457,9 +422,10 @@ class OffPushStreamProcess(PushStreamProcess):
push_stream_result = t.submit(push_video_stream, frame_merge, push_p, push_url, push_stream_result = t.submit(push_video_stream, frame_merge, push_p, push_url,
p_push_status, request_id) p_push_status, request_id)
if qs_np is not None: if qs_np is not None:
if len(qs_np.shape) == 1: if len(qs_np.shape) == 1:
qs_np = qs_np[np.newaxis,...] qs_np = qs_np[np.newaxis,...]
qs_np_id = qs_np.copy() qs_np_id = qs_np.copy()
b = np.ones(qs_np_id.shape[0]) b = np.ones(qs_np_id.shape[0])
qs_np_id = np.column_stack((qs_np_id,b)) qs_np_id = np.column_stack((qs_np_id,b))
@ -486,7 +452,7 @@ class OffPushStreamProcess(PushStreamProcess):
if q[11] >= 1: if q[11] >= 1:
cls = int(q[9]) cls = int(q[9])
if not (cls in new_lab): if not (cls in new_lab):
continue # 为了防止其他类别被带出 continue # 为了防止其他类别被带出
code = str(int(q[10])).zfill(3) code = str(int(q[10])).zfill(3)
if det_xywh2.get(code) is None: if det_xywh2.get(code) is None:
det_xywh2[code] = {} det_xywh2[code] = {}
@ -499,13 +465,6 @@ class OffPushStreamProcess(PushStreamProcess):
is_new = False is_new = False
if q[11] == 1: if q[11] == 1:
is_new = True is_new = True
if ModelType.CITY_CARPLATE_MODEL.value[1] == str(code):
cls = ocrlabel
elif ModelType.CITY_DENSECROWDCOUNT_MODEL.value[1] == str(code):
cls = crowdlabel
label_array = points
if cd is None: if cd is None:
det_xywh2[code][cls] = [[cls, box, score, label_array, color, is_new]] det_xywh2[code][cls] = [[cls, box, score, label_array, color, is_new]]
else: else:

7
config/service/dsp_test_service.yml
View File

@ -25,6 +25,10 @@ service:
# 2 模型追踪 # 2 模型追踪
model_type: 1 model_type: 1
limit: 3 limit: 3
# 新增：大模型识别配置
llm:
server_ip: "192.168.10.11"
api_url: "http://192.168.10.11:8000/generate"
task: task:
# 任务限制5个 # 任务限制5个
limit: 5 limit: 5
@ -35,5 +39,4 @@ service:
#是否启用mqtt，0--不用，1--启用 #是否启用mqtt，0--不用，1--启用
mqtt_flag: 0 mqtt_flag: 0
#是否启用alg控制功能 #是否启用alg控制功能
algSwitch: False algSwitch: False