# 导入代码依赖 import cv2 from core.DBManager import mDBM from myutils.MyLogger_logger import LogHandler from myutils.ConfigManager import myCongif from core.ChannelManager import ChannelManager from core.ACLModelManager import ACLModeManger from core.WarnManager import WarnManager class ModelManager: def __init__(self): self.verify_list = ChannelManager() #模型的主要数据 -- 2024-7-5修改为类管理通道数据 self.logger = LogHandler().get_logger("ModelManager") # 本地YOLOv5仓库路径 self.yolov5_path = myCongif.get_data("yolov5_path") #self.buflen = myCongif.get_data("buffer_len") # 报警处理线程-全进程独立一个线程处理 self.warnM = None # acl初始化 -- 一个进程一个 self.model_platform = myCongif.get_data("model_platform") if self.model_platform == "acl": self.device_id = myCongif.get_data("device_id") ACLModeManger.init_acl(self.device_id) #acl -- 全程序初始化 #self.model_dic = {} # model_id model def __del__(self): self.logger.debug("释放资源") self.stop_work(0) #停止所有工作 del self.verify_list #应该需要深入的删除--待完善 if self.model_platform == "acl": #去初始化 ACLModeManger.del_acl(self.device_id) #acl -- 全程序反初始化 需要确保在执行析构前，其它资源已释放 def send_warn(self): '''发送报警信息''' pass def save_frame_to_video(self): '''把缓冲区中的画面保存为录像''' pass def start_work(self,channel_id=0): '''算法模型是在后台根据画面实时分析的 1.布防开关需要触发通道关闭和开启 2.布防策略的调整也需要关闭和重启工作 ''' # 启动告警线程 if self.warnM is None: self.warnM = WarnManager() self.warnM.start_warnmanager_th() #工作线程：cap+model if channel_id ==0: strsql = "select id,ulr,type from channel where is_work = 1;" #执行所有通道 else: strsql = f"select id,ulr,type from channel where is_work = 1 and id = {channel_id};" #单通道启动检测线程 datas = mDBM.do_select(strsql) deque_length = myCongif.get_data("buffer_len") icount_max = myCongif.get_data("RESET_INTERVAL") for data in datas: channel_id = data[0] #1.创建channel对象 #channel_id,deque_length,icount_max,warnM ret = self.verify_list.add_channel(channel_id,deque_length,icount_max,self.warnM) if ret: # 2.启动工作线程 strsql = ( f"select t1.model_id,t1.check_area,t1.polygon ,t2.duration_time,t2.proportion,t2.model_path,t1.ID," f"t2.model_name,t1.conf_thres,t1.iou_thres " f"from channel2model t1 left join model t2 on t1.model_id = t2.ID where t1.channel_id ={channel_id};") model_data = mDBM.do_select(strsql, 1) # 2024-7-12调整规则，一个通道只关联一个模型,表结构暂时不动 # cap_data[source,type] cap_data = [data[1], data[2]] if model_data and model_data[0]: # 如果该通道关联了模型 schedule = mDBM.getschedule(model_data[6]) # 获取布防计划 #启动 self.verify_list.start_channel(channel_id,cap_data,model_data,schedule,0) #cap + model else: self.logger.debug(f"{channel_id}通道没有关联模型，只运行cap采集线程") # 启动 self.verify_list.start_channel(channel_id, cap_data, None,None, 1) # cap else: print("*****有错误********") def stop_work(self,channel_id=0): #要对应start_work 1.停止model线程，2.停止cap线程。3.删除c_data '''停止工作线程(包括采集线程,并删除通道数据对象)，0-停止所有，非0停止对应通道ID的线程''' try: self.verify_list.delete_channel(channel_id) # 1，2，3 except Exception as e: print(e) if channel_id == 0: #停止告警线程 self.warnM.brun = False del self.warnM self.warnM = None def startModelWork(self,channel_id): strsql = ( f"select t1.model_id,t1.check_area,t1.polygon ,t2.duration_time,t2.proportion,t2.model_path,t1.ID," f"t2.model_name,t1.conf_thres,t1.iou_thres " f"from channel2model t1 left join model t2 on t1.model_id = t2.ID where t1.channel_id ={channel_id};") model_data = mDBM.do_select(strsql, 1) # 2024-7-12调整规则，一个通道只关联一个模型,表结构暂时不动 if model_data and model_data[0]: # 如果该通道关联了模型 schedule = mDBM.getschedule(model_data[6]) # 获取布防计划 # cap_data[source,type] cap_data = None # 启动 self.logger.debug(f"通道{channel_id}重新启动model线程") self.verify_list.start_channel(channel_id, cap_data, model_data, schedule, 2) # cap + model else: self.logger.debug(f"通道{channel_id}没有关联模型，不需要启动model线程") def restartC2M(self,channel_id): ''' 修改通道管理的算法模型后需要对该通道算法执行部分重新加载执行 -- 只重启model线程 :param channel_id: :return: ''' #停止该通道的工作线程 -- 不停cap self.verify_list.stop_channel(channel_id,2) #重新读取工作线程数据，重启该通道的工作线程 self.startModelWork(channel_id) def test1(self): # from model.plugins.RYRQ_Model_ACL.RYRQ_Model_ACL import Model # mymodel = Model("") # mymodel.testRun() import importlib # 需要导入的模块路径 module_path = "model.plugins.RYRQ_Model_ACL.RYRQ_Model_ACL" # 动态导入模块 module_name = importlib.import_module(module_path) # 从模块中加载指定的类 Model = getattr(module_name, "Model") # 使用 Model 类 model_instance = Model("") model_instance.testRun() #print(f"Current working directory (ModelManager.py): {os.getcwd()}") mMM = ModelManager() if __name__ == "__main__": mMM.start_work() print("111") # name = acl.get_soc_name() # count, ret = acl.rt.get_device_count() # print(name,count)