import threading import time as time_sel from typing import Dict import qwen_task import requests import os import torch import logging from transformers import AutoProcessor, AutoModelForVision2Seq from pymilvus import connections, Collection from logging.handlers import RotatingFileHandler import get_mem class ThreadPool: def __init__(self): #读取配置文件 self.config = {} with open('./conf.txt', 'r', encoding='utf-8') as file: for line in file: # 去除每行的首尾空白字符（包括换行符） line = line.strip() # 跳过空行 if not line: continue # 分割键和值 if '=' in line: key, value = line.split('=', 1) # 去除键和值的首尾空白字符 key = key.strip() value = value.strip() # 将键值对添加到字典中 self.config[key] = value # 配置日志 # 确保日志目录存在 log_dir = "logs" os.makedirs(log_dir, exist_ok=True) self.threads: Dict[str, threading.Thread] = {} self.lock = threading.Lock() self.device = "cuda" if torch.cuda.is_available() else "cpu" # 初始化qwenvl检测模型 # 指定本地路径 model_path = "./Qwen2.5-VL-3B-Instruct-AWQ" # 加载模型 self.qwen_model = AutoModelForVision2Seq.from_pretrained( model_path, device_map="auto" ).eval() # 加载处理器 self.qwen_tokenizer = AutoProcessor.from_pretrained(model_path,use_fast=True) # 初始化Milvus集合 connections.connect("default", host=self.config.get("milvusurl"), port=self.config.get("milvusport")) # 加载集合 self.collection = Collection(name="smartobject") self.collection.load() #是否更新 self._isupdate = False # 初始化共享内存 get_mem.smem_init() # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(filename)s:%(lineno)d - %(funcName)s() - %(levelname)s: %(message)s', datefmt='%Y-%m-%d %H:%M:%S', handlers=[ # 按大小轮转的日志文件（最大10MB，保留3个备份） RotatingFileHandler( filename=os.path.join(log_dir, 'start_log.log'), maxBytes=10 * 1024 * 1024, # 10MB backupCount=3, encoding='utf-8' ), # 同时输出到控制台 logging.StreamHandler() ] ) #启动线程 def safe_start(self, target_func, camera_id): """线程安全启动方法""" def wrapped(): thread_name = threading.current_thread().name try: target_func(camera_id) except Exception as e: logging.error(f"线程异常: {str(e)}", exc_info=True) with self.lock: # 确保线程安全创建 t = threading.Thread( target=wrapped, daemon=True # 设置为守护线程 ) t.start() self.threads[camera_id] = t return t # 启动线程任务 def worker(self, camera_id): # 初始化 detect = qwen_task.detect_tasks() detect._thread_name = f"{camera_id}" # 设置名称 detect.init_logging(f"{camera_id}") # 初始化检测模型 detect.device = self.device # 初始化qwen模型 detect.qwen_tokenizer = self.qwen_tokenizer detect.qwen_model = self.qwen_model # 初始化Milvus集合 detect.collection = self.collection # 设置线程启动状态 detect._running = True try: # 读取共享内存中的图片 image_id_list = [458671636370971032] for image_id in image_id_list: logging.info(f"读取图像成功: {image_id}") image_id = detect.tark_do(image_id,self.config.get("filesavepath"),self.config.get("ragurl"),self.config.get("max_tokens")) logging.info(f"处理图像成功: {image_id}") except Exception as e: logging.info(f"{detect._thread_name}线程错误:{e}") #调用是否需要更新 def isUpdate(self): try: # 定义请求的 URL url = self.config.get("isupdateurl") # 发送 GET 请求 response = requests.get(url) # 检查响应状态码 if response.status_code == 200: data = response.json().get("data") if data.get("isChange") == 1: return True else: return False except Exception as e: logging.info(f"调用是否需要更新时出错:URL:{self.config.get('isupdateurl')}:{e}") return False #修改是否更新状态 def update_status(self): try: # 更新状态 url = self.config.get("updatestatusurl") # 发送 GET 请求 response = requests.post(url) # 检查响应状态码 if response.status_code == 200: return True else: return False except Exception as e: logging.info(f"修改是否更新状态时出错:URL:{self.config.get('updatestatusurl')}:{e}") return False def shutdown_all(self) -> None: """清理所有线程""" with self.lock: for camera_id, thread in list(self.threads.items()): if thread.is_alive(): thread.join(timeout=1) del self.threads[camera_id] #获取任务 def getTaskconf(self,isupdate): try: # 定义请求的 URL url = self.config.get("gettaskconfurl") # 发送 GET 请求 response = requests.get(url) # 检查响应状态码 if response.status_code == 200: data = response.json() if isupdate: # 更新状态 self.update_status() return data.get("data") else: return [] except Exception as e: logging.info(f"调用获取任务时出错:URL:{self.config.get('gettaskconfurl')}:{e}") return [] # 使用示例 if __name__ == "__main__": pool = ThreadPool() is_init = True camera_data = pool.getTaskconf(False) while True: try: pool._isupdate = False # 是否更新数据 # 是否需要更新任务数据 if pool.isUpdate(): # 获取摄像机任务 camera_data = pool.getTaskconf(True) pool._isupdate = True # 更新数据 if is_init: if camera_data: for camera in camera_data: thread = pool.threads.get(camera.get("camera_id")) if not thread: logging.info(f"开始创建{camera.get('camera_id')}线程") pool.safe_start(pool.worker, camera.get('camera_id')) logging.info(f"{camera.get('camera_id')}线程创建完毕") if pool._isupdate: logging.info(f"更新线程开始") pool.shutdown_all() if camera_data: for camera in camera_data: thread = pool.threads.get(camera.get("camera_id")) if not thread: logging.info(f"开始创建{camera.get('camera_id')}线程") pool.safe_start(pool.worker, camera.get('camera_id')) logging.info(f"{camera.get('camera_id')}线程创建完毕") logging.info(f"更新线程结束") is_init = False time_sel.sleep(1) except Exception as e: logging.info(f"主线程未知错误:{e}")