import os import logging from logging.handlers import RotatingFileHandler import cv2 import torch from PIL import Image from pymilvus import connections, Collection from transformers import AutoTokenizer, AutoModelForVision2Seq, AutoProcessor, BitsAndBytesConfig from flask import Flask, request, jsonify app = Flask(__name__) #规则比对 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(image_id,image_path,milvusurl,milvusport): try : # 初始化qwenvl检测模型 qwen_tokenizer = AutoProcessor.from_pretrained("Qwen2-VL-2B", trust_remote_code=True) qwen_model = AutoModelForVision2Seq.from_pretrained("Qwen2-VL-2B", device_map="auto", trust_remote_code=True).eval() # 初始化Milvus集合 connections.connect("default", host=milvusurl, port=milvusport) # 加载集合 collection = Collection(name="smartobject") collection.load() # 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 # 示例：文本处理接口 @app.route('/process', methods=['POST']) def process_text(): data = request.json # 调用您的处理函数（例如 Qwen2-VL 模型） try: return tark_do(data.get("image_id"),data.get("image_path"),data.get("milvusurl"),data.get("milvusport")) # 替换为实际函数 except Exception as e: # 错误处理 return 0 if __name__ == '__main__': app.run(host='0.0.0.0', port=5000, debug=True)