60 lines
2.7 KiB
Python
60 lines
2.7 KiB
Python
|
||
import importlib
|
||
from django.conf import settings
|
||
import logging
|
||
myLogger = logging.getLogger('log')
|
||
|
||
|
||
algo_dict = {
|
||
"helmet": "apps.ai.client.helmet",
|
||
"fire1": "apps.ai.client.fire1",
|
||
"fangtang": "apps.ai.client.fangtang",
|
||
"jingjiedai": "apps.ai.jingjiedai",
|
||
"qiping": "apps.ai.qiping"
|
||
}
|
||
|
||
|
||
def ai_analyse(codes: list, global_img: str, face_img: str = None):
|
||
"""算法分析图片
|
||
|
||
Args:
|
||
codes: 算法列表
|
||
global_img (str): 全景图片地址
|
||
face_img (str): 人脸图片地址
|
||
"""
|
||
results = {} # dict key: 触发的事件标识字符 value: 多个矩形框坐标列表
|
||
for i in codes:
|
||
if i in algo_dict and i not in results: # 如果算法支持且没有识别过
|
||
module, func = algo_dict[i].rsplit(".", 1)
|
||
m = importlib.import_module(module)
|
||
f = getattr(m, func)
|
||
try:
|
||
is_happend, res, rectangle_dict = False, None, {}
|
||
if i == 'helmet': # 如果是安全帽抠图识别
|
||
if face_img:
|
||
is_happend, res = f(ip=settings.AI_IP, pic_url=face_img)
|
||
else:
|
||
is_happend, res = getattr(m, 'helmet2')(ip=settings.AI_IP, pic_url=global_img)
|
||
else:
|
||
is_happend, res = f(ip=settings.AI_IP, pic_url=global_img)
|
||
if i in ['fire1', 'jingjiedai', 'qiping']: # 如果是这3类算法就无需再识别，都在一个算法里处理了
|
||
for x in res.FireinfoList:
|
||
has_fire = False # 默认没有灭火器
|
||
has_jingjiedai = False # 默认没有警戒带
|
||
qiping_qd = False # 气瓶倾倒未发生
|
||
rectangle_dict .update({'qiping':[]}) # 气瓶倾倒的坐标字典
|
||
if x.fire == 3 and 'qiping' in codes: # 气瓶倾倒
|
||
qiping_qd = True
|
||
rectangle_dict['qiping'].append([x.coord.x, x.coord.y]) # 加入矩形框
|
||
if x.fire == 0:
|
||
has_fire = True
|
||
if x.fire == 2:
|
||
has_jingjiedai = True
|
||
if (i == 'fire1' and not has_fire) or (i == 'jingjiedai' and not has_jingjiedai) or (i == 'qiping' and qiping_qd):
|
||
results.update({i: {'rectangles':rectangle_dict.get(i, [])}})
|
||
if is_happend and (i not in results):
|
||
results.update({i: {'rectangles':rectangle_dict.get(i, [])}})
|
||
except Exception:
|
||
myLogger.error('算法处理错误', exc_info=True)
|
||
return results
|