import threading import time as time_sel from typing import Dict import cv2 import numpy as np import detect_task from datetime import time import requests import os from datetime import datetime import torch import logging import jieba.posseg as pseg from transformers import AutoModel, AutoTokenizer, AutoModelForCausalLM, AutoConfig, AutoProcessor, \ AutoModelForVision2Seq from pymilvus import connections, Collection from logging.handlers import RotatingFileHandler 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 self.threads: Dict[str, threading.Thread] = {} self.lock = threading.Lock() # 配置日志 # 确保日志目录存在 log_dir = "logs" os.makedirs(log_dir, exist_ok=True) self.device = "cuda:0" if torch.cuda.is_available() else "cpu" # 初始化向量模型 self.qwen_tokenizer = AutoProcessor.from_pretrained("Qwen2-VL-2B", trust_remote_code=True) self.qwen_model = AutoModelForVision2Seq.from_pretrained("Qwen2-VL-2B", device_map="auto", trust_remote_code=True).eval() # 初始化Milvus集合 connections.connect("default", host=self.config.get("milvusurl"), port=self.config.get("milvusport")) # 加载集合 self.collection = Collection(name="smartobject") self.collection.load() # 配置日志 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 cross_collection_vector_compare(self,image_id,video_point_id,rule_id,video_image_time,frame_id,task_id,video_path,videotaskurl): try: # 1. 从集合A获取向量和元数据 res_a = self.collection.query( expr=f"id in {image_id}", output_fields=["id", "zh_desc_class", "text_vector", "bounding_box", "object_label", "task_id", "event_level_id", "video_point_id", "detect_num", "detect_id", "detect_time", "image_path", "video_path"], consistency_level="Strong" ) # 2. 从集合B获取向量和元数据 res_b = self.collection_rule.query( expr=f"rule_id in {rule_id}", output_fields=["id", "zh_desc_class", "text_vector","range_value"], consistency_level="Strong" ) # 3. 计算两两相似度 results = [] pos_weights = { ('n', 'n'): 1.0, # 名词匹配 ('v', 'v'): 0.8, # 动词匹配 ('n', 'v'): 0.3 # 名词-动词交叉 } # 循环结果集 for item_a in res_a: if item_a["detect_num"]>0 : # 有检测到目标时进行比对 for item_b in res_b: # 和每个规则项进行对比 similarity = 0 # 1. 向量相似度（60%权重） # 将向量转为numpy数组 vec_a = np.array(item_a["text_vector"]) vec_b = np.array(item_b["text_vector"]) assert vec_a.shape == vec_b.shape, f"维度不匹配: {vec_a.shape} vs {vec_b.shape}" vec_a = vec_a.astype(np.float64) # 提升计算精度 vec_b = vec_b.astype(np.float64) # 计算相似度 vector_sim = np.dot(vec_a, vec_b) / (np.linalg.norm(vec_a) * np.linalg.norm(vec_b)) # 2. 词性匹配度（40%权重） desc_a = ast.literal_eval(item_a['zh_desc_class']) # 获取目标语义 desc_b = ast.literal_eval(item_b['zh_desc_class']) for clas_a in desc_a: # 和每个目标语义进行对比 pos_match = 0 for (_, pos_a, _), (_, pos_b, _) in zip(clas_a, desc_b): pos_match += pos_weights.get((pos_a, pos_b), 0) pos_match /= max(len(clas_a), len(desc_b)) #按权重组装 当前相似度 并赋值最大值 stem_int = 0.6 * vector_sim + 0.4 * pos_match if stem_int > similarity: similarity = stem_int # 3.相似度如果大于规则匹配阈值,组装预警数据 if similarity > item_b["range_value"]: logging.info(f"相似度：{similarity}：{item_a['zh_desc_class']} {item_b['zh_desc_class']}") results.append({ "a_id": item_a["id"], "b_id": item_b["id"], "similarity": round(float(similarity), 4) }) # 4. 按相似度排序 if len(results) > 0: comparison_results = sorted(results, key=lambda x: x["similarity"], reverse=True) # 集合排序 # 保存录像视频 asyncio.run(self.video_task(video_point_id,video_image_time,"basic",frame_id,video_path,videotaskurl)) # 保存到milvus self.update_milvus(image_id,res_a,1,comparison_results[0]['similarity'],rule_id) else: self.update_milvus(image_id,res_a,0,0.0,rule_id) except Exception as e: self.logger.info(f"{video_point_id}线程：规则对比时出错:{image_id,rule_id}： {e}") #启动录像 async def video_task(self,video_point_id,video_image_time,task_id,frame_id,video_path,videotaskurl): try: json_data = { "cameraId": f"{video_point_id}", "timestamp": video_image_time, "preSeconds": 10, "postSeconds": 10, "taskId": task_id, "fid": frame_id, "uploadUrl": video_path, "uploadType": "local" } self.logger.info(f"{video_point_id}线程：调用录像服务:{videotaskurl, json_data}") # 定义请求的 URL # 发送 GET 请求 response = requests.post(videotaskurl,json=json_data,timeout=(0.03, 0.03)) # 检查响应状态码 if response.status_code == 200: data = response.json() print(data) except Exception as e: self.logger.info(f"{self._thread_name}线程：调用录像时出错:地址：{videotaskurl}：{e}") def update_milvus(self,image_id,res_a,is_waning,similarity,rule_id_list): try: # 保存到milvus self.collection.upsert( data=[{ "id": image_id[0], "text_vector": res_a[0]["text_vector"], "is_waning": is_waning, "waning_value": similarity, "rule_id": rule_id_list, "zh_desc_class": res_a[0]['zh_desc_class'], # text_vector "bounding_box": res_a[0]['bounding_box'], # bounding_box "object_label": res_a[0]['object_label'], # desc "task_id": res_a[0]['task_id'], # task_id "event_level_id": res_a[0]['event_level_id'], # event_level_id "video_point_id": res_a[0]['video_point_id'], # video_point_id "detect_num": res_a[0]['detect_num'], "detect_id": res_a[0]['detect_id'], # detect_id "detect_time": res_a[0]['detect_time'], # detect_time "image_path": res_a[0]['image_path'], # image_path "video_path": res_a[0]['video_path'] # video_path }] ) except Exception as e: self.logger.info(f"{self._thread_name}线程：规则对比后修改数据时出错:{image_id}：是否预警{is_waning}：预警值：{similarity}：规则id:{rule_id_list}:数据集合：{len(res_a)} :{e}") def tark_do(): try : # 1. 从集合A获取向量和元数据 res_a = collection.query( expr=f"id == {image_id}", output_fields=["id", "zh_desc_class", "text_vector", "bounding_box", "object_label", "task_id", "event_level_id", "video_point_id", "detect_num", "is_waning", "waning_value", "rule_id", "detect_id", "detect_time", "image_path", "video_path"], consistency_level="Strong" ) # 图片和视频地址 image = Image.open(image_path) # 替换为您的图片路径 image = image.thumbnail((512, 512)) text = "描述这张图片的内容" # 处理输入 inputs = qwen_tokenizer(text=text, images=image, return_tensors="pt").to("cuda") # 生成输出 with torch.no_grad(): outputs = qwen_model.generate(**inputs, max_new_tokens=50) response = qwen_tokenizer.decode(outputs[0], skip_special_tokens=True) print(response) # 保存到milvus collection.upsert( data=[{ "id": image_id, "text_vector": res_a[0]["text_vector"], "is_waning": res_a[0]["is_waning"], "waning_value": res_a[0]["waning_value"], "rule_id": res_a[0]["rule_id"], "zh_desc_class": res_a[0]['zh_desc_class'], # text_vector "bounding_box": res_a[0]['bounding_box'], # bounding_box "object_label": f"{response}", # desc "task_id": res_a[0]['task_id'], # task_id "event_level_id": res_a[0]['event_level_id'], # event_level_id "video_point_id": res_a[0]['video_point_id'], # video_point_id "detect_num": res_a[0]['detect_num'], "detect_id": res_a[0]['detect_id'], # detect_id "detect_time": res_a[0]['detect_time'], # detect_time "image_path": res_a[0]['image_path'], # image_path "video_path": res_a[0]['video_path'] # video_path }] ) print(f"执行任务{image_id,image_path}：结束") return 1 except Exception as e: print(f"执行模型解析时出错:{image_id,image_path} :{e}") return 0 # 使用示例 if __name__ == "__main__": pool = ThreadPool() while True: try: pool.tark_do() except Exception as e: logging.info(f"主线程未知错误:{e}")