############################################################################# seguimineto import cv2 import numpy as np import time import threading from scipy.optimize import linear_sum_assignment from scipy.spatial.distance import cosine from ultralytics import YOLO import onnxruntime as ort import os from datetime import datetime import json import csv # ────────────────────────────────────────────────────────────────────────────── # CONFIGURACIÓN DEL SISTEMA # ────────────────────────────────────────────────────────────────────────────── USUARIO, PASSWORD, IP_DVR = "admin", "TCA200503", "192.168.1.244" SECUENCIA = [1, 7, 5, 8, 3, 6] # RED ESTABILIZADA (Timeout de 3s para evitar congelamientos de FFmpeg) os.environ["OPENCV_FFMPEG_CAPTURE_OPTIONS"] = "rtsp_transport;tcp|stimeout;3000000" URLS = [f"rtsp://{USUARIO}:{PASSWORD}@{IP_DVR}:554/Streaming/Channels/{i}02" for i in SECUENCIA] ONNX_MODEL_PATH = "osnet_x0_25_msmt17.onnx" VECINOS = { "1": ["7"], "7": ["1", "5"], "5": ["7", "8"], "8": ["5", "3"], "3": ["8", "6"], "6": ["3"] } ASPECT_RATIO_MIN = 0.6 ASPECT_RATIO_MAX = 4.0 AREA_MIN_CALIDAD = 1200 FRAMES_CALIDAD = 3 TIEMPO_MAX_AUSENCIA = 800.0 MAX_FIRMAS_MEMORIA = 10 C_CANDIDATO = (150, 150, 150) C_LOCAL = (0, 255, 0) C_GLOBAL = (0, 165, 255) C_GRUPO = (0, 0, 255) C_APRENDIZAJE = (255, 255, 0) FUENTE = cv2.FONT_HERSHEY_SIMPLEX # ────────────────────────────────────────────────────────────────────────────── # INICIALIZACIÓN OSNET # ────────────────────────────────────────────────────────────────────────────── print("Cargando cerebro de Re-Identificación (OSNet)...") try: ort_session = ort.InferenceSession(ONNX_MODEL_PATH, providers=['CPUExecutionProvider']) input_name = ort_session.get_inputs()[0].name print("Modelo OSNet cargado exitosamente.") except Exception as e: print(f"ERROR FATAL: No se pudo cargar {ONNX_MODEL_PATH}.") exit() MEAN = np.array([0.485, 0.456, 0.406], dtype=np.float32).reshape(1, 3, 1, 1) STD = np.array([0.229, 0.224, 0.225], dtype=np.float32).reshape(1, 3, 1, 1) # ────────────────────────────────────────────────────────────────────────────── # 1. EXTRACCIÓN DE FIRMAS (Deep + Color + Textura) # ────────────────────────────────────────────────────────────────────────────── def analizar_calidad(box): x1, y1, x2, y2 = box w, h = x2 - x1, y2 - y1 if w <= 0 or h <= 0: return False return (ASPECT_RATIO_MIN < (h / w) < ASPECT_RATIO_MAX) and ((w * h) > AREA_MIN_CALIDAD) def preprocess_onnx(roi): # ⚡ ESCUDO ANTI-SOMBRAS (CLAHE) # Convertimos a LAB para ecualizar solo la luz (L) sin deformar los colores reales (A y B) lab = cv2.cvtColor(roi, cv2.COLOR_BGR2LAB) l, a, b = cv2.split(lab) clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8,8)) l_eq = clahe.apply(l) lab_eq = cv2.merge((l_eq, a, b)) roi_eq = cv2.cvtColor(lab_eq, cv2.COLOR_LAB2BGR) # Preprocesamiento original para OSNet img = cv2.resize(roi_eq, (128, 256)) img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) img = img.transpose(2,0,1).astype(np.float32) / 255.0 img = (img - MEAN) / STD # Devolvemos el tensor correcto de 1 sola imagen return np.expand_dims(img, axis=0) def extraer_color_zonas(img): h_roi = img.shape[0] t1, t2 = int(h_roi * 0.15), int(h_roi * 0.55) zonas = [img[:t1, :], img[t1:t2, :], img[t2:, :]] def hist_zona(z): if z.size == 0: return np.zeros(16 * 8) hsv = cv2.cvtColor(z, cv2.COLOR_BGR2HSV) hist = cv2.calcHist([hsv], [0, 1], None, [16, 8], [0, 180, 0, 256]) cv2.normalize(hist, hist) return hist.flatten() return np.concatenate([hist_zona(z) for z in zonas]) def extraer_textura_rapida(roi): if roi.size == 0: return np.zeros(16) gray = cv2.cvtColor(roi, cv2.COLOR_BGR2GRAY) gray_eq = cv2.equalizeHist(gray) gx = cv2.Sobel(gray_eq, cv2.CV_32F, 1, 0, ksize=3) gy = cv2.Sobel(gray_eq, cv2.CV_32F, 0, 1, ksize=3) mag, _ = cv2.cartToPolar(gx, gy) hist = cv2.calcHist([mag], [0], None, [16], [0, 256]) cv2.normalize(hist, hist) return hist.flatten() def extraer_firma_hibrida(frame_hd, box_480): try: h_hd, w_hd = frame_hd.shape[:2] escala_x = w_hd / 480.0 escala_y = h_hd / 270.0 x1, y1, x2, y2 = box_480 x1_hd, y1_hd = int(x1 * escala_x), int(y1 * escala_y) x2_hd, y2_hd = int(x2 * escala_x), int(y2 * escala_y) x1_c, y1_c = max(0, x1_hd), max(0, y1_hd) x2_c, y2_c = min(w_hd, x2_hd), min(h_hd, y2_hd) roi = frame_hd[y1_c:y2_c, x1_c:x2_c] if roi.size == 0 or roi.shape[0] < 40 or roi.shape[1] < 20: return None calidad_area = (x2_c - x1_c) * (y2_c - y1_c) # ⚡ VOLVEMOS AL BATCH DE 16 PARA EVITAR EL CRASH FATAL blob = preprocess_onnx(roi) # Esto devuelve (1, 3, 256, 128) # Creamos el contenedor de 16 espacios que el modelo ONNX exige blob_16 = np.zeros((16, 3, 256, 128), dtype=np.float32) blob_16[0] = blob[0] # Metemos nuestra imagen en el primer espacio # Ejecutamos la inferencia outputs = ort_session.run(None, {input_name: blob_16}) deep_feat = outputs[0][0].flatten() # Extraemos solo el primer resultado norma = np.linalg.norm(deep_feat) if norma > 0: deep_feat = deep_feat / norma color_feat = extraer_color_zonas(roi) textura_feat = extraer_textura_rapida(roi) return {'deep': deep_feat, 'color': color_feat, 'textura': textura_feat, 'calidad': calidad_area} except Exception as e: print(f"Error en extracción: {e}") return None def similitud_hibrida(f1, f2, cross_cam=False): if f1 is None or f2 is None: return 0.0 sim_deep = max(0.0, 1.0 - cosine(f1['deep'], f2['deep'])) if f1['color'].shape == f2['color'].shape and f1['color'].size > 1: L = len(f1['color']) // 3 sim_head = max(0.0, float(cv2.compareHist(f1['color'][:L].astype(np.float32), f2['color'][:L].astype(np.float32), cv2.HISTCMP_CORREL))) sim_torso = max(0.0, float(cv2.compareHist(f1['color'][L:2*L].astype(np.float32), f2['color'][L:2*L].astype(np.float32), cv2.HISTCMP_CORREL))) sim_legs = max(0.0, float(cv2.compareHist(f1['color'][2*L:].astype(np.float32), f2['color'][2*L:].astype(np.float32), cv2.HISTCMP_CORREL))) else: sim_head, sim_torso, sim_legs = 0.0, 0.0, 0.0 if cross_cam: if sim_legs < 0.25: sim_deep -= 0.15 return max(0.0, sim_deep) sim_color = (0.10 * sim_head) + (0.60 * sim_torso) + (0.30 * sim_legs) if 'textura' in f1 and 'textura' in f2 and f1['textura'].size > 1: sim_textura = max(0.0, float(cv2.compareHist(f1['textura'].astype(np.float32), f2['textura'].astype(np.float32), cv2.HISTCMP_CORREL))) else: sim_textura = 0.0 return (sim_deep * 0.80) + (sim_color * 0.10) + (sim_textura * 0.10) # ────────────────────────────────────────────────────────────────────────────── # 2. KALMAN TRACKER # ────────────────────────────────────────────────────────────────────────────── class KalmanTrack: _count = 0 def __init__(self, box, now): self.kf = cv2.KalmanFilter(7, 4) self.kf.measurementMatrix = np.array([[1,0,0,0,0,0,0], [0,1,0,0,0,0,0], [0,0,1,0,0,0,0], [0,0,0,1,0,0,0]], np.float32) self.kf.transitionMatrix = np.eye(7, dtype=np.float32) self.kf.transitionMatrix[0,4] = 1; self.kf.transitionMatrix[1,5] = 1; self.kf.transitionMatrix[2,6] = 1 self.kf.processNoiseCov *= 0.05 self.kf.statePost = np.zeros((7, 1), np.float32) self.kf.statePost[:4] = self._convert_bbox_to_z(box) self.local_id = KalmanTrack._count KalmanTrack._count += 1 self.gid = None self.origen_global = False self.aprendiendo = False self.box = list(box) self.ts_creacion = now self.ts_ultima_deteccion = now self.time_since_update = 0 self.en_grupo = False self.frames_buena_calidad = 0 self.listo_para_id = False self.area_referencia = 0.0 self.firma_pre_grupo = None self.ts_salio_grupo = 0 self.validado_post_grupo = True def _convert_bbox_to_z(self, bbox): w = bbox[2] - bbox[0]; h = bbox[3] - bbox[1]; x = bbox[0] + w/2.; y = bbox[1] + h/2. return np.array([[x],[y],[w*h],[w/float(h+1e-6)]]).astype(np.float32) def _convert_x_to_bbox(self, x): cx, cy, s, r = float(x[0].item()), float(x[1].item()), float(x[2].item()), float(x[3].item()) w = np.sqrt(s * r); h = s / (w + 1e-6) return [cx-w/2., cy-h/2., cx+w/2., cy+h/2.] def predict(self, turno_activo=True): # ⚡ EL SUICIDIO DE CAJAS FANTASMAS # Si lleva más de 5 frames de ceguera total (YOLO no lo ve), matamos la predicción if self.time_since_update > 5: return None if (self.kf.statePost[6] + self.kf.statePost[2]) <= 0: self.kf.statePost[6] *= 0.0 self.kf.predict() if turno_activo: self.time_since_update += 1 self.aprendiendo = False self.box = self._convert_x_to_bbox(self.kf.statePre) return self.box """def predict(self, turno_activo=True): if (self.kf.statePost[6] + self.kf.statePost[2]) <= 0: self.kf.statePost[6] *= 0.0 self.kf.predict() if turno_activo: self.time_since_update += 1 self.aprendiendo = False self.box = self._convert_x_to_bbox(self.kf.statePre) return self.box""" def update(self, box, en_grupo, now): self.ts_ultima_deteccion = now self.time_since_update = 0 self.box = list(box) self.en_grupo = en_grupo self.kf.correct(self._convert_bbox_to_z(box)) if analizar_calidad(box) and not en_grupo: self.frames_buena_calidad += 1 if self.frames_buena_calidad >= FRAMES_CALIDAD: self.listo_para_id = True elif self.gid is None: self.frames_buena_calidad = max(0, self.frames_buena_calidad - 1) # ────────────────────────────────────────────────────────────────────────────── # 3. MEMORIA GLOBAL (Anti-Robos y Físicas de Tiempo) # ────────────────────────────────────────────────────────────────────────────── class GlobalMemory: def __init__(self): self.db = {} self.next_gid = 100 self.lock = threading.RLock() self.ruta_saludos = os.path.join("cache_nombres", "registro_saludos.json") self.ultimos_saludos = self._cargar_y_limpiar_saludos() def registrar_movimiento(self, nombre, cam_origen, cam_destino, duracion_seg): """Guarda el movimiento y la duración en la cámara anterior en un CSV.""" directorio = "cache_nombres" if not os.path.exists(directorio): os.makedirs(directorio) ruta_csv = os.path.join(directorio, "registro_movimientos.csv") ahora = datetime.now().strftime("%Y-%m-%d %H:%M:%S") archivo_nuevo = not os.path.exists(ruta_csv) try: with open(ruta_csv, 'a', newline='', encoding='utf-8') as f: writer = csv.writer(f) if archivo_nuevo: writer.writerow(["Fecha_Hora", "Nombre", "Origen", "Destino", "Segundos_en_Origen"]) # Redondeamos la duración a 2 decimales writer.writerow([ahora, nombre, cam_origen, cam_destino, round(duracion_seg, 2)]) f.flush() # Forzamos la escritura física en el disco print(f" [LOG] {nombre} estuvo {duracion_seg:.1f}s en Cam {cam_origen} antes de pasar a {cam_destino}") except Exception as e: print(f"Error al guardar CSV: {e}") try: with open(ruta_csv, 'a', newline='', encoding='utf-8') as f: writer = csv.writer(f) if archivo_nuevo: writer.writerow(["Fecha_Hora", "Nombre", "Camara_Origen", "Camara_Destino"]) writer.writerow([ahora, nombre, cam_origen, cam_destino]) print(f" [LOG MOVIMIENTO] {nombre} pasó de Cam {cam_origen} a Cam {cam_destino}") except Exception as e: print(f"Error al guardar movimiento: {e}") def _cargar_y_limpiar_saludos(self): """Carga el historial y elimina registros que no sean de hoy.""" hoy = datetime.now().strftime("%Y-%m-%d") if os.path.exists(self.ruta_saludos): try: with open(self.ruta_saludos, 'r') as f: datos = json.load(f) # Solo mantenemos los saludos cuya fecha sea 'hoy' return {nombre: info for nombre, info in datos.items() if info.get('fecha') == hoy} except Exception as e: print(f"Error cargando saludos: {e}") return {} def guardar_saludo(self, nombre): """Registra el saludo con la fecha actual en disco.""" hoy = datetime.now().strftime("%Y-%m-%d") self.ultimos_saludos[nombre] = { 'fecha': hoy, 'timestamp': time.time() } with open(self.ruta_saludos, 'w') as f: json.dump(self.ultimos_saludos, f) def _es_transito_posible(self, data, cam_id, now): cam_origen = str(data['last_cam']) cam_destino = str(cam_id) dt = now - data['ts'] # 1. Si es la misma cámara, aplicamos la regla normal de tiempo máximo de ausencia if cam_origen == cam_destino: return dt < TIEMPO_MAX_AUSENCIA # 2. ⚡ DEDUCCIÓN DE TOPOLOGÍA (El sistema aprende el mapa solo) # Si salta a otra cámara en menos de 1.5 segundos, es físicamente imposible # a menos que ambas cámaras apunten al mismo lugar (están solapadas). if dt < 1.5: # Le damos vía libre inmediata porque sabemos que está en una intersección return True # 3. Si tardó un tiempo normal (ej. > 1.5s), es un tránsito de pasillo válido return dt < TIEMPO_MAX_AUSENCIA def _sim_robusta(self, firma_nueva, firmas_guardadas, cross_cam=False): if not firmas_guardadas: return 0.0 sims = sorted([similitud_hibrida(firma_nueva, f, cross_cam) for f in firmas_guardadas], reverse=True) if len(sims) == 1: return sims[0] elif len(sims) <= 4: return (sims[0] * 0.6) + (sims[1] * 0.4) else: return (sims[0] * 0.50) + (sims[1] * 0.30) + (sims[2] * 0.20) # ⚡ SE AGREGÓ LÓGICA ESPACIO-TEMPORAL DINÁMICA def identificar_candidato(self, firma_hibrida, cam_id, now, active_gids, en_borde=True): self.limpiar_fantasmas() with self.lock: candidatos = [] vecinos = VECINOS.get(str(cam_id), []) for gid, data in self.db.items(): if gid in active_gids: continue dt = now - data['ts'] # Bloqueo inmediato si es físicamente imposible if dt > TIEMPO_MAX_AUSENCIA or not self._es_transito_posible(data, cam_id, now): continue if not data['firmas']: continue misma_cam = (str(data['last_cam']) == str(cam_id)) es_cross_cam = not misma_cam es_vecino = str(data['last_cam']) in vecinos sim = self._sim_robusta(firma_hibrida, data['firmas'], cross_cam=es_cross_cam) # UMBRALES DINÁMICOS INTELIGENTES if misma_cam: if dt < 2.0: umbral = 0.60 # Para parpadeos rápidos de YOLO else: umbral = 0.63 # Si desapareció un rato en la misma cámara elif es_vecino: # ⚡ EL PUNTO DULCE: 0.62. # Está por encima del ruido máximo (0.57) y por debajo de las # caídas de Omar en el cluster 7-5-8 (0.64). umbral = 0.62 else: # ⚡ LEJANOS: 0.66. # Si salta de la Cam 1 a la Cam 8, exigimos seguridad alta # para no robarle el ID a alguien que acaba de entrar por la otra puerta. umbral = 0.70 """ if misma_cam: if dt < 2.0: umbral = 0.55 elif dt < 10.0: umbral = 0.60 else: umbral = 0.60 elif es_vecino: # ⚡ Subimos a 0.66. Si un extraño saca 0.62, será rechazado y nacerá un ID nuevo. umbral = 0.66 else: # ⚡ Cámaras lejanas: Exigencia casi perfecta. umbral = 0.70""" # PROTECCIÓN VIP (Le exigimos un poquito más si ya tiene nombre para no mancharlo) if data.get('nombre') is not None: umbral += 0.05 if misma_cam else 0.02 # EL CHIVATO DE OSNET (Debug crucial) # Esto imprimirá en tu consola qué similitud real de ropa detectó al cruzar de cámara if sim > 0.35 and not misma_cam: print(f" [OSNet] Cam {cam_id} evaluando ID {gid} (Viene de Cam {data['last_cam']}) -> Similitud Ropa: {sim:.2f} (Umbral exigido: {umbral:.2f})") if sim > umbral: candidatos.append((sim, gid, misma_cam, es_vecino)) if not candidatos: nid = self.next_gid; self.next_gid += 1 self._actualizar_sin_lock(nid, firma_hibrida, cam_id, now) return nid, False candidatos.sort(reverse=True) best_sim, best_gid, best_misma, best_vecino = candidatos[0] if len(candidatos) >= 2: segunda_sim, segundo_gid, seg_misma, seg_vecino = candidatos[1] margen = best_sim - segunda_sim if margen <= 0.06 and best_sim < 0.75: peso_1 = best_sim + (0.10 if (best_misma or best_vecino) else 0.0) peso_2 = segunda_sim + (0.10 if (seg_misma or seg_vecino) else 0.0) if peso_1 > peso_2: best_gid = best_gid elif peso_2 > peso_1: best_gid = segundo_gid else: print(f"\n[ ALERTA] Empate extremo entre ID {best_gid} ({best_sim:.2f}) y ID {segundo_gid} ({segunda_sim:.2f}). Se asigna temporal.") nid = self.next_gid; self.next_gid += 1 self._actualizar_sin_lock(nid, firma_hibrida, cam_id, now) return nid, False self._actualizar_sin_lock(best_gid, firma_hibrida, cam_id, now) return best_gid, True def _actualizar_sin_lock(self, gid, firma_dict, cam_id, now): if gid not in self.db: self.db[gid] = {'firmas': [], 'last_cam': cam_id, 'ts': now} if firma_dict is not None: firmas_list = self.db[gid]['firmas'] if not firmas_list: firmas_list.append(firma_dict) else: if firma_dict['calidad'] > (firmas_list[0]['calidad'] * 1.50): vieja_ancla = firmas_list[0]; firmas_list[0] = firma_dict; firma_dict = vieja_ancla if len(firmas_list) >= MAX_FIRMAS_MEMORIA: max_sim_interna = -1.0; idx_redundante = 1 for i in range(1, len(firmas_list)): sims_con_otras = [similitud_hibrida(firmas_list[i], firmas_list[j]) for j in range(1, len(firmas_list)) if j != i] sim_promedio = np.mean(sims_con_otras) if sims_con_otras else 0.0 if sim_promedio > max_sim_interna: max_sim_interna = sim_promedio; idx_redundante = i firmas_list[idx_redundante] = firma_dict else: firmas_list.append(firma_dict) self.db[gid]['last_cam'] = cam_id self.db[gid]['ts'] = now def actualizar(self, gid, firma, cam_id, now): with self.lock: self._actualizar_sin_lock(gid, firma, cam_id, now) def limpiar_fantasmas(self): with self.lock: ahora = time.time() ids_a_borrar = [] for gid, data in self.db.items(): tiempo_inactivo = ahora - data.get('ts', ahora) if data.get('nombre') is None and tiempo_inactivo > 600.0: ids_a_borrar.append(gid) elif data.get('nombre') is not None and tiempo_inactivo > 900.0: ids_a_borrar.append(gid) for gid in ids_a_borrar: del self.db[gid] def confirmar_firma_vip(self, gid, ts): with self.lock: if gid in self.db and self.db[gid]['firmas']: firmas_actuales = self.db[gid]['firmas'] self.db[gid]['firmas'] = firmas_actuales[-3:] self.db[gid]['ts'] = ts def fusionar_ids(self, id_mantiene, id_elimina): with self.lock: if id_mantiene in self.db and id_elimina in self.db: # 1. Le pasamos toda la memoria de ropa al ID ganador self.db[id_mantiene]['firmas'].extend(self.db[id_elimina]['firmas']) # Topamos a las 15 mejores firmas para no saturar la RAM if len(self.db[id_mantiene]['firmas']) > 15: self.db[id_mantiene]['firmas'] = self.db[id_mantiene]['firmas'][-15:] self.db[id_mantiene]['ts'] = max(self.db[id_mantiene]['ts'], self.db[id_elimina]['ts']) # 2. Vaciamos al perdedor y le ponemos un "Redireccionamiento" self.db[id_elimina]['firmas'] = [] self.db[id_elimina]['fusionado_con'] = id_mantiene print(f"[FUSIÓN MÁGICA] Las firmas del ID {id_elimina} fueron absorbidas por el ID {id_mantiene}.") return True return False # ────────────────────────────────────────────────────────────────────────────── # 4. GESTOR LOCAL (Kalman Elasticity & Ghost Killer) # ────────────────────────────────────────────────────────────────────────────── def iou_overlap(boxA, boxB): xA, yA, xB, yB = max(boxA[0], boxB[0]), max(boxA[1], boxB[1]), min(boxA[2], boxB[2]), min(boxA[3], boxB[3]) inter = max(0, xB-xA) * max(0, yB-yA) areaA = (boxA[2]-boxA[0]) * (boxA[3]-boxA[1]); areaB = (boxB[2]-boxB[0]) * (boxB[3]-boxB[1]) return inter / (areaA + areaB - inter + 1e-6) class CamManager: def __init__(self, cam_id, global_mem): self.cam_id, self.global_mem, self.trackers = cam_id, global_mem, [] def _detectar_grupo(self, trk, box, todos_los_tracks): x1, y1, x2, y2 = box w_box = x2 - x1 h_box = y2 - y1 cx, cy = (x1 + x2) / 2, (y1 + y2) / 2 estado_actual = getattr(trk, 'en_grupo', False) # Si están a menos de medio cuerpo de distancia lateral, es un grupo. factor_x = 0.9 if estado_actual else 0.7 factor_y = 0.35 if estado_actual else 0.2 for other in todos_los_tracks: if other is trk: continue if not hasattr(other, 'box') or other.box is None: continue ox1, oy1, ox2, oy2 = other.box ocx, ocy = (ox1 + ox2) / 2, (oy1 + oy2) / 2 dist_x = abs(cx - ocx) dist_y = abs(cy - ocy) if dist_x < (w_box * factor_x) and dist_y < (h_box * factor_y): return True return False def _gestionar_aprendizaje_post_grupo(self, now, frame_hd): CUARENTENA = 1.2 for trk in self.trackers: if trk.gid is None or getattr(trk, 'en_grupo', False) or getattr(trk, 'firma_pre_grupo', None) is None: continue tiempo_fuera = now - getattr(trk, 'ts_salio_grupo', 0) if tiempo_fuera >= CUARENTENA: firma_actual = extraer_firma_hibrida(frame_hd, trk.box) if firma_actual is not None: sim = similitud_hibrida(firma_actual, trk.firma_pre_grupo) # ⚡ Bajamos a 0.50 como umbral de supervivencia base if sim >= 0.50: print(f"[GRUPO] ID {trk.gid} validado post-grupo ({sim:.2f}).") trk.fallos_post_grupo = 0 # Reseteamos los strikes si tenía with self.global_mem.lock: if trk.gid in self.global_mem.db: self.global_mem.db[trk.gid]['firmas'] = [trk.firma_pre_grupo, firma_actual] trk.firma_pre_grupo = None # Aprobado, limpiamos memoria else: # ⚡ SISTEMA DE 3 STRIKES trk.fallos_post_grupo = getattr(trk, 'fallos_post_grupo', 0) + 1 if trk.fallos_post_grupo >= 3: print(f" [ALERTA GRUPO] ID {trk.gid} falló 3 veces validación ({sim:.2f}). Reseteando ID.") trk.gid = None trk.firma_pre_grupo = None # Eliminado, limpiamos memoria trk.fallos_post_grupo = 0 # Reiniciamos contador else: print(f" [STRIKE {trk.fallos_post_grupo}/3] ID {trk.gid} sacó {sim:.2f}. Dando otra oportunidad...") # OJO: NO ponemos firma_pre_grupo en None aquí, # para que lo vuelva a intentar en el siguiente frame. def update(self, boxes, frame_show, frame_hd, now, turno_activo): for trk in self.trackers: if trk.gid is not None: with self.global_mem.lock: if trk.gid in self.global_mem.db and 'fusionado_con' in self.global_mem.db[trk.gid]: print(f"🔗 [HILO CAM {self.cam_id}] Tracker mutando de ID {trk.gid} a {self.global_mem.db[trk.gid]['fusionado_con']}") trk.gid = self.global_mem.db[trk.gid]['fusionado_con'] # ────────────────────────────────────────────────────────── # ⚡ FILTRO ANTI-FANTASMAS vivos_predict = [] for trk in self.trackers: caja_predicha = trk.predict(turno_activo=turno_activo) # Si el tracker lleva más de 5 frames ciego, devuelve None. Lo ignoramos. if caja_predicha is not None: vivos_predict.append(trk) self.trackers = vivos_predict # Actualizamos la lista solo con los vivos if not turno_activo: return self.trackers # ────────────────────────────────────────────────────────── # Aquí sigue tu código normal: matched, unmatched_dets, unmatched_trks = self._asignar(boxes, now) active_gids = {t.gid for t in self.trackers if t.gid is not None} for t_idx, d_idx in matched: trk = self.trackers[t_idx] box = boxes[d_idx] es_grupo_ahora = self._detectar_grupo(trk, box, self.trackers) if not getattr(trk, 'en_grupo', False) and es_grupo_ahora: if trk.gid is not None: with self.global_mem.lock: firmas = self.global_mem.db.get(trk.gid, {}).get('firmas', []) if firmas: trk.firma_pre_grupo = firmas[-1] trk.validado_post_grupo = False print(f" [GRUPO] ID {trk.gid} entró a grupo. Protegiendo firma.") trk.ts_salio_grupo = 0.0 elif getattr(trk, 'en_grupo', False) and not es_grupo_ahora: trk.ts_salio_grupo = now print(f" [GRUPO] ID {trk.gid} salió de grupo. Cuarentena de 3s.") trk.en_grupo = es_grupo_ahora trk.update(box, es_grupo_ahora, now) area_actual = (box[2] - box[0]) * (box[3] - box[1]) tiempo_desde_separacion = now - getattr(trk, 'ts_salio_grupo', 0) en_cuarentena = (tiempo_desde_separacion < 3.0) and (getattr(trk, 'ts_salio_grupo', 0) > 0) extracciones_hoy = 0 # ⚡ EL SALVAVIDAS: Contador para el frame actual if not trk.en_grupo and not en_cuarentena: # A) Bautizo de IDs Nuevos (Con Aduana Temporal) if trk.gid is None and trk.listo_para_id: firma = extraer_firma_hibrida(frame_hd, box) if firma is not None: fh, fw = frame_hd.shape[:2] bx1, by1, bx2, by2 = map(int, box) nace_en_borde = (bx1 < 25 or by1 < 25 or bx2 > fw - 25 or by2 > fh - 25) candidato_gid, es_reid = self.global_mem.identificar_candidato(firma, self.cam_id, now, active_gids, en_borde=nace_en_borde) if candidato_gid is not None: # ⚡ BLOQUEO INMEDIATO: Reservamos el ID en este milisegundo # para que ninguna otra persona en esta cámara pueda evaluarlo. active_gids.add(candidato_gid) # Si la memoria dice "Es un desconocido nuevo", lo bautizamos al instante if not es_reid: trk.gid, trk.origen_global, trk.area_referencia = candidato_gid, False, area_actual # ⚡ ADUANA TEMPORAL else: if getattr(trk, 'candidato_temporal', None) == candidato_gid: trk.votos_reid = getattr(trk, 'votos_reid', 0) + 1 else: # Si cambia de opinión, reiniciamos sin restar trk.candidato_temporal = candidato_gid trk.votos_reid = 1 if trk.votos_reid >= 2: trk.gid, trk.origen_global = candidato_gid, True with self.global_mem.lock: datos = self.global_mem.db.get(candidato_gid, {}) nombre_oficial = datos.get('nombre') cam_anterior = datos.get('last_cam') entrada_ts = datos.get('last_cam_entry_ts', now) # ⚡ REGISTRO SOLO SI TIENE NOMBRE Y VIENE DE OTRA CÁMARA if nombre_oficial and cam_anterior is not None and str(cam_anterior) != str(self.cam_id): duracion = now - entrada_ts self.global_mem.registrar_movimiento(nombre_oficial, cam_anterior, self.cam_id, duracion) # Actualizamos los datos para el siguiente salto datos['last_cam'] = self.cam_id datos['last_cam_entry_ts'] = now # B) Aprendizaje Continuo (Captura de Ángulos) elif trk.gid is not None: tiempo_ultima_firma = getattr(trk, 'ultimo_aprendizaje', 0) # ⚡ APRENDIZAJE ESCALONADO: # Mantenemos tus 0.5s perfectos, pero solo 1 persona por frame puede saturar la CPU. if (now - tiempo_ultima_firma) > 0.5 and analizar_calidad(box) and extracciones_hoy < 1: fh, fw = frame_hd.shape[:2] x1, y1, x2, y2 = map(int, box) en_borde = (x1 < 15 or y1 < 15 or x2 > fw - 15 or y2 > fh - 15) if not en_borde: firma_nueva = extraer_firma_hibrida(frame_hd, box) if firma_nueva is not None: extracciones_hoy += 1 # ⚡ Cerramos la compuerta para los demás en este frame with self.global_mem.lock: if trk.gid in self.global_mem.db and self.global_mem.db[trk.gid]['firmas']: firma_reciente = self.global_mem.db[trk.gid]['firmas'][-1] firma_original = self.global_mem.db[trk.gid]['firmas'][0] sim_coherencia = similitud_hibrida(firma_nueva, firma_reciente) sim_raiz = similitud_hibrida(firma_nueva, firma_original) # ⚡ EL BOTÓN DE PÁNICO (Anti ID-Switch) # Si la ropa de la caja actual no se parece en NADA a la original (< 0.35), # significa que Kalman le pegó el ID a la persona equivocada en un cruce. if sim_raiz < 0.35: print(f"[ID SWITCH] Ropa de ID {trk.gid} cambió drásticamente. Revocando ID.") trk.gid = None trk.listo_para_id = False trk.frames_buena_calidad = 0 continue # Rompemos el ciclo para que nazca como alguien nuevo ya_bautizado = self.global_mem.db[trk.gid].get('nombre') is not None umbral_raiz = 0.52 if ya_bautizado else 0.62 if sim_coherencia > 0.60 and sim_raiz > umbral_raiz: es_coherente = True for otro_gid, otro_data in self.global_mem.db.items(): if otro_gid == trk.gid or not otro_data['firmas']: continue sim_intruso = similitud_hibrida(firma_nueva, otro_data['firmas'][0]) if sim_intruso > sim_raiz: es_coherente = False break if es_coherente: self.global_mem._actualizar_sin_lock(trk.gid, firma_nueva, self.cam_id, now) trk.ultimo_aprendizaje = now trk.aprendiendo = True for d_idx in unmatched_dets: self.trackers.append(KalmanTrack(boxes[d_idx], now)) vivos = [] fh, fw = frame_show.shape[:2] for t in self.trackers: x1, y1, x2, y2 = t.box toca_borde = (x1 < 20 or y1 < 20 or x2 > fw - 20 or y2 > fh - 20) tiempo_oculto = now - t.ts_ultima_deteccion # ⚡ LIMPIEZA AGRESIVA PARA PROTEGER LA CPU DE MEMORY LEAKS if t.gid is None: # Si es un ID gris (no bautizado), lo matamos rápido si YOLO lo pierde limite_vida = 1.0 if toca_borde else 2.5 else: # Si es VIP, le damos paciencia para que OSNet no se fragmente limite_vida = 3.0 if toca_borde else 8.0 if tiempo_oculto < limite_vida: vivos.append(t) self.trackers = vivos self._gestionar_aprendizaje_post_grupo(now, frame_hd) return self.trackers def _asignar(self, boxes, now): n_trk = len(self.trackers); n_det = len(boxes) if n_trk == 0: return [], list(range(n_det)), [] if n_det == 0: return [], [], list(range(n_trk)) cost_mat = np.zeros((n_trk, n_det), dtype=np.float32) for t, trk in enumerate(self.trackers): tiempo_oculto = now - trk.ts_ultima_deteccion # ⚡ 2A. Aumentamos el radio dinámico mínimo para no perder gente rápida radio_dinamico = min(350.0, max(150.0, 300.0 * tiempo_oculto)) # La incertidumbre crece con el tiempo, pero la topamos en 0.5 incertidumbre = min(0.5, tiempo_oculto * 0.2) es_fantasma = getattr(trk, 'time_since_update', 0) > 1 for d, det in enumerate(boxes): iou = iou_overlap(trk.box, det) cx_t, cy_t = (trk.box[0]+trk.box[2])/2, (trk.box[1]+trk.box[3])/2 cx_d, cy_d = (det[0]+det[2])/2, (det[1]+det[3])/2 dist_pixel = np.sqrt((cx_t-cx_d)**2 + (cy_t-cy_d)**2) area_trk = (trk.box[2] - trk.box[0]) * (trk.box[3] - trk.box[1]) area_det = (det[2] - det[0]) * (det[3] - det[1]) if dist_pixel > radio_dinamico: cost_mat[t, d] = 100.0 continue # ⚡ 2B. REDUCIMOS CASTIGOS INJUSTOS # Eliminamos el "castigo_secuestro" que penalizaba excesivamente a los trackers sin IOU. # Reducimos la penalización por cambio de tamaño de 0.4 a 0.2. dist_norm = dist_pixel / radio_dinamico ratio_area = max(area_trk, area_det) / (min(area_trk, area_det) + 1e-6) # Si la caja de YOLO es mucho MÁS PEQUEÑA que la predicción de Kalman, # penalizamos fuerte (evita que la caja encoja mágicamente y pierda a la persona). if area_det < area_trk: castigo_tam = (ratio_area - 1.0) * 0.5 else: # Si la caja de YOLO es MÁS GRANDE (ej. persona acercándose/subiendo escaleras), # somos muy permisivos para que Kalman acepte el nuevo tamaño sin soltarse. castigo_tam = (ratio_area - 1.0) * 0.15 # Fórmula equilibrada cost_mat[t, d] = (1.0 - iou) + (0.5 * dist_norm) + castigo_tam + incertidumbre from scipy.optimize import linear_sum_assignment row_ind, col_ind = linear_sum_assignment(cost_mat) matched, unmatched_dets, unmatched_trks = [], [], [] for r, c in zip(row_ind, col_ind): # ⚡ UMBRAL RELAJADO: De 2.5 a 3.5. # Esto evita que rechace la caja por la latencia del procesador. if cost_mat[r, c] > 3.5: unmatched_trks.append(r); unmatched_dets.append(c) else: matched.append((r, c)) for t in range(n_trk): if t not in [m[0] for m in matched]: unmatched_trks.append(t) for d in range(n_det): if d not in [m[1] for m in matched]: unmatched_dets.append(d) return matched, unmatched_dets, unmatched_trks # ────────────────────────────────────────────────────────────────────────────── # 5. STREAM Y MAIN LOOP (Standalone) # ────────────────────────────────────────────────────────────────────────────── class CamStream: def __init__(self, url): self.url, self.cap = url, cv2.VideoCapture(url) self.cap.set(cv2.CAP_PROP_BUFFERSIZE, 1); self.frame = None threading.Thread(target=self._run, daemon=True).start() def _run(self): while True: ret, f = self.cap.read() if ret: self.frame = f; time.sleep(0.01) else: time.sleep(2); self.cap.open(self.url) def dibujar_track(frame_show, trk): try: x1, y1, x2, y2 = map(int, trk.box) except Exception: return if trk.gid is None: color, label = C_CANDIDATO, f"?{trk.local_id}" elif trk.en_grupo: color, label = C_GRUPO, f"ID:{trk.gid} [grp]" elif trk.aprendiendo: color, label = C_APRENDIZAJE, f"ID:{trk.gid} [++]" elif trk.origen_global: color, label = C_GLOBAL, f"ID:{trk.gid} [re-id]" else: color, label = C_LOCAL, f"ID:{trk.gid}" cv2.rectangle(frame_show, (x1, y1), (x2, y2), color, 2) (tw, th), _ = cv2.getTextSize(label, FUENTE, 0.55, 1) cv2.rectangle(frame_show, (x1, y1-th-6), (x1+tw+2, y1), color, -1) cv2.putText(frame_show, label, (x1+1, y1-4), FUENTE, 0.55, (0,0,0), 1) def main(): print("Iniciando Sistema V-PRO — Tracker Resiliente (Código Unificado Maestro)") model = YOLO("yolov8n.pt") global_mem = GlobalMemory() managers = {str(c): CamManager(c, global_mem) for c in SECUENCIA} cams = [CamStream(u) for u in URLS] cv2.namedWindow("SmartSoft", cv2.WINDOW_AUTOSIZE) idx = 0 while True: now = time.time() tiles = [] cam_ia = idx % len(cams) for i, cam_obj in enumerate(cams): frame = cam_obj.frame; cid = str(SECUENCIA[i]) if frame is None: tiles.append(np.zeros((270, 480, 3), np.uint8)); continue frame_show = cv2.resize(frame.copy(), (480, 270)); boxes = []; turno_activo = (i == cam_ia) if turno_activo: res = model.predict(frame_show, conf=0.50, iou=0.40, classes=[0], verbose=False, imgsz=480, device='cpu') if res[0].boxes: boxes = res[0].boxes.xyxy.cpu().numpy().tolist() tracks = managers[cid].update(boxes, frame_show, frame, now, turno_activo) for trk in tracks: if trk.time_since_update <= 1: dibujar_track(frame_show, trk) if turno_activo: cv2.circle(frame_show, (460, 20), 6, (0, 0, 255), -1) con_id = sum(1 for t in tracks if t.gid and t.time_since_update==0) cv2.putText(frame_show, f"CAM {cid} [{con_id} ID]", (10, 28), FUENTE, 0.7, (255, 255, 255), 2) tiles.append(frame_show) if len(tiles) == 6: cv2.imshow("SmartSoft", np.vstack([np.hstack(tiles[0:3]), np.hstack(tiles[3:6])])) idx += 1 if cv2.waitKey(1) == ord('q'): break cv2.destroyAllWindows() if __name__ == "__main__": main() ########################################## reconocimineto import os os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3' os.environ['CUDA_VISIBLE_DEVICES'] = '-1' # ⚡ NUEVOS CANDADOS: Silenciamos la burocracia de OpenCV y Qt os.environ['QT_LOGGING_RULES'] = '*=false' os.environ['OPENCV_LOG_LEVEL'] = 'ERROR' import cv2 import numpy as np import pickle import time import threading import asyncio import edge_tts import subprocess from datetime import datetime import warnings import json # ⚡ IMPORTACIÓN DE INSIGHTFACE from insightface.app import FaceAnalysis warnings.filterwarnings("ignore") # ────────────────────────────────────────────────────────────────────────────── # CONFIGURACIÓN # ────────────────────────────────────────────────────────────────────────────── DB_PATH = "db_institucion" CACHE_PATH = "cache_nombres" VECTORS_FILE = "base_datos_rostros.pkl" TIMESTAMPS_FILE = "representaciones_timestamps.pkl" UMBRAL_SIM = 0.45 # ⚡ Ajustado a la exigencia de producción COOLDOWN_TIME = 15 # Segundos entre saludos USUARIO, PASSWORD, IP_DVR = "admin", "TCA200503", "192.168.1.200" RTSP_URL = f"rtsp://{USUARIO}:{PASSWORD}@{IP_DVR}:554/Streaming/Channels/702" for path in [DB_PATH, CACHE_PATH]: os.makedirs(path, exist_ok=True) # ────────────────────────────────────────────────────────────────────────────── # EL NUEVO CEREBRO: INSIGHTFACE (buffalo_l) # ────────────────────────────────────────────────────────────────────────────── print("Cargando motor InsightFace (buffalo_l)...") # Carga detección (SCRFD), landmarks, reconocimiento (ArcFace) y atributos (género/edad) app = FaceAnalysis(name='buffalo_l', providers=['CPUExecutionProvider']) # det_size a 320x320 es un balance perfecto entre velocidad y capacidad de detectar rostros lejanos app.prepare(ctx_id=0, det_size=(320, 320)) print("Motor cargado exitosamente.") # ────────────────────────────────────────────────────────────────────────────── # SISTEMA DE AUDIO # ────────────────────────────────────────────────────────────────────────────── def obtener_audios_humanos(genero): hora = datetime.now().hour es_mujer = genero.lower() == 'woman' suffix = "_m.mp3" if es_mujer else "_h.mp3" if 5 <= hora < 12: intro = "dias.mp3" elif 12 <= hora < 19: intro = "tarde.mp3" else: intro = "noches.mp3" cierre = ("fin_noche" if (hora >= 19 or hora < 5) else "fin_dia") + suffix return intro, cierre async def sintetizar_nombre(nombre, ruta): nombre_limpio = nombre.replace('_', ' ') try: comunicador = edge_tts.Communicate(nombre_limpio, "es-MX-DaliaNeural", rate="+10%") await comunicador.save(ruta) except Exception: pass def reproducir(archivo): if os.path.exists(archivo): subprocess.Popen( ["mpv", "--no-video", "--volume=100", archivo], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL ) def hilo_bienvenida(nombre, genero): archivo_nombre = os.path.join(CACHE_PATH, f"nombre_{nombre}.mp3") if not os.path.exists(archivo_nombre): try: asyncio.run(sintetizar_nombre(nombre, archivo_nombre)) except Exception: pass intro, cierre = obtener_audios_humanos(genero) archivos = [f for f in [intro, archivo_nombre, cierre] if os.path.exists(f)] if archivos: subprocess.Popen( ["mpv", "--no-video", "--volume=100"] + archivos, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL ) # ────────────────────────────────────────────────────────────────────────────── # GESTIÓN DE BASE DE DATOS (AHORA CON INSIGHTFACE) # ────────────────────────────────────────────────────────────────────────────── def gestionar_vectores(actualizar=False): vectores_actuales = {} if os.path.exists(VECTORS_FILE): try: with open(VECTORS_FILE, 'rb') as f: vectores_actuales = pickle.load(f) except Exception: vectores_actuales = {} if not actualizar: return vectores_actuales timestamps = {} if os.path.exists(TIMESTAMPS_FILE): try: with open(TIMESTAMPS_FILE, 'rb') as f: timestamps = pickle.load(f) except Exception: timestamps = {} ruta_generos = os.path.join(CACHE_PATH, "generos.json") dic_generos = {} if os.path.exists(ruta_generos): try: with open(ruta_generos, 'r') as f: dic_generos = json.load(f) except Exception: pass print("\nACTUALIZANDO BASE DE DATOS (modelo y Caché de Géneros)...") imagenes = [f for f in os.listdir(DB_PATH) if f.lower().endswith(('.jpg', '.png'))] nombres_en_disco = set() hubo_cambios = False cambio_generos = False for archivo in imagenes: nombre_archivo = os.path.splitext(archivo)[0] ruta_img = os.path.join(DB_PATH, archivo) nombres_en_disco.add(nombre_archivo) ts_actual = os.path.getmtime(ruta_img) ts_guardado = timestamps.get(nombre_archivo, 0) falta_genero = nombre_archivo not in dic_generos if nombre_archivo in vectores_actuales and ts_actual == ts_guardado and not falta_genero: continue try: img_db = cv2.imread(ruta_img) # ⚡ INSIGHTFACE: Extracción primaria faces = app.get(img_db) # HACK DEL LIENZO NEGRO: Si la foto está muy recortada y no ve la cara, # le agregamos un borde negro gigante para engañar al detector if len(faces) == 0: h_img, w_img = img_db.shape[:2] img_pad = cv2.copyMakeBorder(img_db, h_img, h_img, w_img, w_img, cv2.BORDER_CONSTANT, value=(0,0,0)) faces = app.get(img_pad) if len(faces) == 0: print(f" [!] Rostro irrecuperable en '{archivo}'.") continue face = max(faces, key=lambda f: (f.bbox[2]-f.bbox[0]) * (f.bbox[3]-f.bbox[1])) emb = np.array(face.normed_embedding, dtype=np.float32) # ⚡ CORRECCIÓN DE GÉNERO: InsightFace devuelve "M" o "F" if falta_genero: dic_generos[nombre_archivo] = "Man" if face.sex == "M" else "Woman" cambio_generos = True vectores_actuales[nombre_archivo] = emb timestamps[nombre_archivo] = ts_actual hubo_cambios = True print(f" Procesado: {nombre_archivo} | Género: {dic_generos.get(nombre_archivo)}") except Exception as e: print(f" Error procesando '{archivo}': {e}") for nombre in list(vectores_actuales.keys()): if nombre not in nombres_en_disco: del vectores_actuales[nombre] timestamps.pop(nombre, None) if nombre in dic_generos: del dic_generos[nombre] cambio_generos = True hubo_cambios = True print(f" Eliminado (sin foto): {nombre}") if hubo_cambios: with open(VECTORS_FILE, 'wb') as f: pickle.dump(vectores_actuales, f) with open(TIMESTAMPS_FILE, 'wb') as f: pickle.dump(timestamps, f) if cambio_generos: with open(ruta_generos, 'w') as f: json.dump(dic_generos, f) if hubo_cambios or cambio_generos: print(" Sincronización terminada.\n") else: print(" Sin cambios. Base de datos al día.\n") return vectores_actuales # ────────────────────────────────────────────────────────────────────────────── # BÚSQUEDA BLINDADA # ────────────────────────────────────────────────────────────────────────────── def buscar_mejor_match(emb_consulta, base_datos): # InsightFace devuelve normed_embedding, por lo que la magnitud ya es 1 mejor_match, max_sim = None, -1.0 for nombre, vec in base_datos.items(): sim = float(np.dot(emb_consulta, vec)) if sim > max_sim: max_sim = sim mejor_match = nombre return mejor_match, max_sim # ────────────────────────────────────────────────────────────────────────────── # LOOP DE PRUEBA Y REGISTRO # ────────────────────────────────────────────────────────────────────────────── def sistema_interactivo(): base_datos = gestionar_vectores(actualizar=False) cap = cv2.VideoCapture(RTSP_URL) ultimo_saludo = 0 persona_actual = None confirmaciones = 0 print("\n" + "=" * 50) print(" MÓDULO DE REGISTRO - INSIGHTFACE (buffalo_l)") print(" [R] Registrar nuevo rostro | [Q] Salir") print("=" * 50 + "\n") faces_ultimo_frame = [] while True: ret, frame = cap.read() if not ret: time.sleep(2) cap.open(RTSP_URL) continue h, w = frame.shape[:2] display_frame = frame.copy() tiempo_actual = time.time() # ⚡ INSIGHTFACE: Hace TODO aquí (Detección, Landmarks, Género, ArcFace) faces_raw = app.get(frame) faces_ultimo_frame = faces_raw for face in faces_raw: # Las cajas de InsightFace vienen en formato [x1, y1, x2, y2] box = face.bbox.astype(int) fx, fy, x2, y2 = box[0], box[1], box[2], box[3] fw, fh = x2 - fx, y2 - fy # Limitamos a los bordes de la pantalla fx, fy = max(0, fx), max(0, fy) fw, fh = min(w - fx, fw), min(h - fy, fh) if fw <= 0 or fh <= 0: continue cv2.rectangle(display_frame, (fx, fy), (fx+fw, fy+fh), (255, 200, 0), 2) cv2.putText(display_frame, f"DET:{face.det_score:.2f}", (fx, fy - 25), cv2.FONT_HERSHEY_SIMPLEX, 0.45, (255, 200, 0), 1) if (tiempo_actual - ultimo_saludo) <= COOLDOWN_TIME: continue # FILTRO DE TAMAÑO (Para evitar reconocer píxeles lejanos) if fw < 20 or fh < 20: cv2.putText(display_frame, "muy pequeno", (fx, fy-10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 100, 255), 1) continue # SIMETRÍA MATEMÁTICA INMEDIATA emb = np.array(face.normed_embedding, dtype=np.float32) mejor_match, max_sim = buscar_mejor_match(emb, base_datos) estado = " IDENTIFICADO" if max_sim > UMBRAL_SIM else "DESCONOCIDO" nombre_d = mejor_match.split('_')[0] if mejor_match else "nadie" n_bloques = int(max_sim * 20) barra = "█" * n_bloques + "░" * (20 - n_bloques) print(f"[REGISTRO] {estado} | {nombre_d:<14} | {barra} | " f"{max_sim*100:.1f}% (umbral: {UMBRAL_SIM*100:.0f}%)") if max_sim > UMBRAL_SIM and mejor_match: color = (0, 255, 0) texto = f"{mejor_match.split('_')[0]} ({max_sim:.2f})" if mejor_match == persona_actual: confirmaciones += 1 else: persona_actual, confirmaciones = mejor_match, 1 if confirmaciones >= 2: cv2.rectangle(display_frame, (fx, fy), (fx+fw, fy+fh), (0, 255, 0), 3) # ⚡ GÉNERO INMEDIATO SIN SOBRECARGA DE CPU genero = "Man" if face.sex == 1 else "Woman" threading.Thread( target=hilo_bienvenida, args=(mejor_match, genero), daemon=True ).start() ultimo_saludo = tiempo_actual confirmaciones = 0 else: color = (0, 0, 255) texto = f"? ({max_sim:.2f})" confirmaciones = max(0, confirmaciones - 1) cv2.putText(display_frame, texto, (fx, fy - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.7, color, 2) cv2.imshow("Módulo de Registro", display_frame) key = cv2.waitKey(1) & 0xFF if key == ord('q'): break elif key == ord('r'): if faces_ultimo_frame: # Tomamos la cara más grande detectada por InsightFace face_registro = max(faces_ultimo_frame, key=lambda f: (f.bbox[2]-f.bbox[0]) * (f.bbox[3]-f.bbox[1])) box = face_registro.bbox.astype(int) fx, fy, fw, fh = box[0], box[1], box[2]-box[0], box[3]-box[1] # Margen estético para guardar la foto m_x, m_y = int(fw * 0.30), int(fh * 0.30) y1, y2 = max(0, fy-m_y), min(h, fy+fh+m_y) x1, x2 = max(0, fx-m_x), min(w, fx+fw+m_x) face_roi = frame[y1:y2, x1:x2] if face_roi.size > 0: nom = input("\nNombre de la persona: ").strip() if nom: # Extraemos el género automáticamente gracias a InsightFace gen_str = "Man" if face_registro.sex == 1 else "Woman" # Actualizamos JSON directo sin preguntar ruta_generos = os.path.join(CACHE_PATH, "generos.json") dic_generos = {} if os.path.exists(ruta_generos): with open(ruta_generos, 'r') as f: dic_generos = json.load(f) dic_generos[nom] = gen_str with open(ruta_generos, 'w') as f: json.dump(dic_generos, f) # Guardado de imagen y sincronización foto_path = os.path.join(DB_PATH, f"{nom}.jpg") cv2.imwrite(foto_path, face_roi) print(f"[OK] Rostro de '{nom}' guardado como {gen_str}. Sincronizando...") base_datos = gestionar_vectores(actualizar=True) else: print("[!] Registro cancelado.") else: print("[!] Recorte vacío. Intenta de nuevo.") else: print("\n[!] No se detectó rostro. Acércate más o mira a la lente.") cap.release() cv2.destroyAllWindows() if __name__ == "__main__": sistema_interactivo() ####################################################################### fusion import os os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3' os.environ['CUDA_VISIBLE_DEVICES'] = '-1' os.environ["OPENCV_FFMPEG_CAPTURE_OPTIONS"] = "rtsp_transport;tcp|stimeout;3000000" # ⚡ NUEVOS CANDADOS: Silenciamos la burocracia de OpenCV y Qt os.environ['QT_LOGGING_RULES'] = '*=false' os.environ['OPENCV_LOG_LEVEL'] = 'ERROR' import cv2 import numpy as np import time import threading import queue from queue import Queue from ultralytics import YOLO import warnings import json warnings.filterwarnings("ignore") # ────────────────────────────────────────────────────────────────────────────── # 1. IMPORTAMOS NUESTROS MÓDULOS # ────────────────────────────────────────────────────────────────────────────── # Del motor matemático y tracking from seguimiento2 import GlobalMemory, CamManager, SECUENCIA, URLS, FUENTE, similitud_hibrida # ⚡ Del motor de reconocimiento facial from reconocimiento import ( app, gestionar_vectores, buscar_mejor_match, hilo_bienvenida, UMBRAL_SIM, COOLDOWN_TIME ) # ────────────────────────────────────────────────────────────────────────────── # 2. PROTECCIONES MULTIHILO E INICIALIZACIÓN # ────────────────────────────────────────────────────────────────────────────── COLA_ROSTROS = Queue(maxsize=4) IA_LOCK = threading.Lock() print("\nIniciando carga de base de datos...") BASE_DATOS_ROSTROS = gestionar_vectores(actualizar=True) # ────────────────────────────────────────────────────────────────────────────── # 3. MOTOR ASÍNCRONO CON INSIGHTFACE # ────────────────────────────────────────────────────────────────────────────── def procesar_rostro_async(roi_cabeza, gid, cam_id, global_mem, trk): try: if not BASE_DATOS_ROSTROS or roi_cabeza.size == 0: return with IA_LOCK: faces = app.get(roi_cabeza) if len(faces) == 0: return face = max(faces, key=lambda f: (f.bbox[2] - f.bbox[0]) * (f.bbox[3] - f.bbox[1])) w_rostro = face.bbox[2] - face.bbox[0] h_rostro = face.bbox[3] - face.bbox[1] # ⚡ Límite bajo (20px) para reconocer desde más lejos if w_rostro < 20 or h_rostro < 20 or (w_rostro / h_rostro) < 0.35: return emb = np.array(face.normed_embedding, dtype=np.float32) genero_detectado = "Man" if face.sex == "M" else "Woman" mejor_match, max_sim = buscar_mejor_match(emb, BASE_DATOS_ROSTROS) print(f"[DEBUG CAM {cam_id}] InsightFace: {mejor_match} al {max_sim:.2f}") votos_finales = 0 # ────────────────────────────────────────────────────────────────── # ⚡ LÓGICA DE VOTACIÓN PONDERADA INYECTADA # ────────────────────────────────────────────────────────────────── # Aceptamos rostros de CCTV desde 0.38 (ajustado según tu código a 0.35) if max_sim >= 0.35 and mejor_match: nombre_limpio = mejor_match.split('_')[0] # Variable para rastrear quién es el ID definitivo al final del proceso id_definitivo = gid with global_mem.lock: datos_id = global_mem.db.get(gid) if not datos_id: return if datos_id.get('candidato_nombre') == nombre_limpio: datos_id['votos_nombre'] = datos_id.get('votos_nombre', 0) + 1 else: datos_id['candidato_nombre'] = nombre_limpio datos_id['votos_nombre'] = 1 # Sistema de Puntos if max_sim >= 0.50: datos_id['votos_nombre'] += 3 # Pase VIP por buena foto elif max_sim >= 0.40: datos_id['votos_nombre'] += 2 # Voto extra por foto decente votos_finales = datos_id['votos_nombre'] # ⚡ Exigimos 3 votos (según tu condicional) if votos_finales >= 3: nombre_actual = datos_id.get('nombre') # 1. Buscamos si ese nombre ya lo tiene un veterano en la memoria id_veterano = None for gid_mem, data_mem in global_mem.db.items(): if data_mem.get('nombre') == nombre_limpio and gid_mem != gid: id_veterano = gid_mem break # 2. FUSIÓN MÁGICA: Si ya existe, lo fusionamos if id_veterano is not None: print(f"[FUSIÓN FACIAL] ID {gid} es el clon de la espalda. Fusionando con el veterano ID {id_veterano} ({nombre_limpio}).") # Pasamos la memoria de ropa al veterano global_mem.db[id_veterano]['firmas'].extend(datos_id.get('firmas', [])) if len(global_mem.db[id_veterano]['firmas']) > 15: global_mem.db[id_veterano]['firmas'] = global_mem.db[id_veterano]['firmas'][-15:] # Vaciamos al perdedor y redireccionamos datos_id['firmas'] = [] datos_id['fusionado_con'] = id_veterano # Actualizamos el tracker y el ID definitivo trk.gid = id_veterano id_definitivo = id_veterano # 3. BAUTIZO NORMAL: Si no hay clones, procedemos con tu lógica de protección VIP else: if nombre_actual is not None and nombre_actual != nombre_limpio: # Si ya tiene nombre y quieren robárselo, exigimos 0.56 mínimo if max_sim < 0.56: print(f"[RECHAZO VIP] {nombre_actual} protegido de {nombre_limpio} ({max_sim:.2f})") return else: print(f" [CORRECCIÓN VIP] Renombrando a {nombre_limpio} ({max_sim:.2f})") if nombre_actual != nombre_limpio: datos_id['nombre'] = nombre_limpio print(f"[BAUTIZO] ID {gid} confirmado como {nombre_limpio}") # Actualizamos el tiempo de vida del ID ganador global_mem.db[id_definitivo]['ts'] = time.time() # AUDIO DE BIENVENIDA (Funciona tanto para bautizos como para fusiones) if str(cam_id) == "7": ahora = time.time() # Si estás probando, usa 30 segundos. Para producción usa 28800 (8h). TIEMPO_COOLDOWN = 60 with global_mem.lock: # Intentamos obtener info del registro de saludos persistente info_saludo = global_mem.ultimos_saludos.get(nombre_limpio, {}) # Si es un string (vieja versión), lo convertimos ultimo_ts = info_saludo.get('timestamp', 0) if isinstance(info_saludo, dict) else 0 if (ahora - ultimo_ts) > TIEMPO_COOLDOWN: print(f" Disparando saludo para {nombre_limpio}...") # Actualizamos memoria y disco global_mem.guardar_saludo(nombre_limpio) # Género desde el JSON genero_oficial = genero_detectado try: ruta_gen = os.path.join("cache_nombres", "generos.json") if os.path.exists(ruta_gen): with open(ruta_gen, 'r') as f: genero_oficial = json.load(f).get(nombre_limpio, genero_detectado) except Exception: pass threading.Thread(target=hilo_bienvenida, args=(nombre_limpio, genero_oficial), daemon=True).start() else: print(f"Saludo a {nombre_limpio} omitido (Cooldown activo: {int(ahora - ultimo_ts)}s)") # Blindaje OSNet fuera del lock (Usando el id_definitivo por si hubo fusión) if max_sim > 0.50 and votos_finales >= 3: global_mem.confirmar_firma_vip(id_definitivo, time.time()) except Exception as e: print(f"Error en InsightFace asíncrono: {e}") finally: # ⚡ EL LIBERADOR (Vital para que el tracker no se quede bloqueado) trk.procesando_rostro = False def worker_rostros(global_mem): while True: roi_cabeza, gid, cam_id, trk = COLA_ROSTROS.get() procesar_rostro_async(roi_cabeza, gid, cam_id, global_mem, trk) COLA_ROSTROS.task_done() # ────────────────────────────────────────────────────────────────────────────── # 4. LOOP PRINCIPAL DE FUSIÓN # ────────────────────────────────────────────────────────────────────────────── class CamStream: def __init__(self, url): self.url = url self.cap = cv2.VideoCapture(url) self.cap.set(cv2.CAP_PROP_BUFFERSIZE, 1) self.frame = None threading.Thread(target=self._run, daemon=True).start() def _run(self): while True: ret, f = self.cap.read() if ret: self.frame = f time.sleep(0.01) else: time.sleep(2) self.cap.open(self.url) def dibujar_track_fusion(frame_show, trk, global_mem): try: x1, y1, x2, y2 = map(int, trk.box) except Exception: return nombre_str = "" if trk.gid is not None: with global_mem.lock: nombre = global_mem.db.get(trk.gid, {}).get('nombre') if nombre: nombre_str = f" [{nombre}]" if trk.gid is None: color, label = (150, 150, 150), f"?{trk.local_id}" elif nombre_str: color, label = (255, 0, 255), f"ID:{trk.gid}{nombre_str}" elif getattr(trk, 'en_grupo', False): color, label = (0, 0, 255), f"ID:{trk.gid} [grp]" elif getattr(trk, 'aprendiendo', False): color, label = (255, 255, 0), f"ID:{trk.gid} [++]" elif getattr(trk, 'origen_global', False): color, label = (0, 165, 255), f"ID:{trk.gid} [re-id]" else: color, label = (0, 255, 0), f"ID:{trk.gid}" cv2.rectangle(frame_show, (x1, y1), (x2, y2), color, 2) (tw, th), _ = cv2.getTextSize(label, FUENTE, 0.55, 1) cv2.rectangle(frame_show, (x1, y1-th-6), (x1+tw+2, y1), color, -1) cv2.putText(frame_show, label, (x1+1, y1-4), FUENTE, 0.55, (0,0,0), 1) def main(): print("\nIniciando Sistema") model = YOLO("yolov8n.pt") global_mem = GlobalMemory() managers = {str(c): CamManager(c, global_mem) for c in SECUENCIA} cams = [CamStream(u) for u in URLS] threading.Thread(target=worker_rostros, args=(global_mem,), daemon=True).start() cv2.namedWindow("SmartSoft", cv2.WINDOW_AUTOSIZE) idx = 0 while True: now = time.time() tiles = [] cam_ia = idx % len(cams) for i, cam_obj in enumerate(cams): frame = cam_obj.frame; cid = str(SECUENCIA[i]) if frame is None: tiles.append(np.zeros((270, 480, 3), np.uint8)) continue frame_show = cv2.resize(frame.copy(), (480, 270)) boxes = [] turno_activo = (i == cam_ia) if turno_activo: # ⚡ 1. Subimos la confianza de YOLO de 0.50 a 0.60 para matar siluetas dudosas res = model.predict(frame_show, conf=0.60, iou=0.50, classes=[0], verbose=False, imgsz=480) if res[0].boxes: cajas_crudas = res[0].boxes.xyxy.cpu().numpy().tolist() # ⚡ 2. FILTRO BIOMECÁNICO (Anti-Sillas) for b in cajas_crudas: w_caja = b[2] - b[0] h_caja = b[3] - b[1] # Un humano real es al menos 10% más alto que ancho (ratio > 1.1) # Si la caja es muy cuadrada o ancha, la ignoramos y no se la pasamos a Kalman if h_caja > (w_caja * 1.1): boxes.append(b) # ⚡ UPDATE LIMPIO (Sin IDs Globales) tracks = managers[cid].update(boxes, frame_show, frame, now, turno_activo) for trk in tracks: if getattr(trk, 'time_since_update', 1) <= 2: dibujar_track_fusion(frame_show, trk, global_mem) if turno_activo and trk.gid is not None and not getattr(trk, 'procesando_rostro', False): with global_mem.lock: votos_actuales = global_mem.db.get(trk.gid, {}).get('votos_nombre', 0) if votos_actuales >= 2: continue x1, y1, x2, y2 = trk.box h_real, w_real = frame.shape[:2] escala_x = w_real / 480.0 escala_y = h_real / 270.0 h_box = y2 - y1 y_exp = max(0, y1 - h_box * 0.15) y_cab = min(270, y1 + h_box * 0.50) roi_recortado = frame[ int(y_exp * escala_y) : int(y_cab * escala_y), int(max(0, x1) * escala_x) : int(min(480, x2) * escala_x) ].copy() # ⚡ COMPUERTA ÓPTICA (Filtro Anti-Tartamudeo 25x25) if roi_recortado.size > 0 and roi_recortado.shape[0] >= 25 and roi_recortado.shape[1] >= 25: gray = cv2.cvtColor(roi_recortado, cv2.COLOR_BGR2GRAY) nitidez = cv2.Laplacian(gray, cv2.CV_64F).var() # ⚡ BAJAMOS DE 12.0 a 5.0. # Deja pasar rostros un poco borrosos por la velocidad de caminar, # pero sigue bloqueando espaldas lisas. if nitidez > 8.0: if not COLA_ROSTROS.full(): ultimo_envio = getattr(trk, 'ultimo_rostro_enviado', 0) # Solo mandamos la cara si ha pasado medio segundo desde la última vez que lo intentamos if (now - ultimo_envio) > 0.5: # Tu escudo anti-lag previo if COLA_ROSTROS.qsize() < 2: trk.ultimo_rostro_enviado = now # Registramos la hora del envío trk.procesando_rostro = True # Bloqueamos el tracker # Aquí pones TU línea original de la cola: COLA_ROSTROS.put((roi_recortado, trk.gid, cid, trk), block=False) else: trk.ultimo_intento_cara = time.time() - 0.5 """if nitidez > 8.0: if not COLA_ROSTROS.full(): try: if COLA_ROSTROS.qsize() < 2: COLA_ROSTROS.put((roi_recortado, trk.gid, cid, trk), block=False) trk.procesando_rostro = True trk.ultimo_intento_cara = time.time() except queue.Full: pass else: trk.ultimo_intento_cara = time.time() - 0.5""" if turno_activo: cv2.circle(frame_show, (460, 20), 6, (0, 0, 255), -1) con_id = sum(1 for t in tracks if t.gid and getattr(t, 'time_since_update', 1) == 0) cv2.putText(frame_show, f"CAM {cid} [{con_id} ID]", (10, 28), FUENTE, 0.7, (255, 255, 255), 2) tiles.append(frame_show) if len(tiles) == 6: cv2.imshow("SmartSoft Fusion", np.vstack([np.hstack(tiles[0:3]), np.hstack(tiles[3:6])])) idx += 1 key = cv2.waitKey(1) & 0xFF if key == ord('q'): break elif key == ord('r'): print("\n[MODO REGISTRO] Escaneando mosaico para registrar...") mejor_roi = None max_area = 0 face_metadata = None for i, cam_obj in enumerate(cams): if cam_obj.frame is None: continue faces = app.get(cam_obj.frame) for face in faces: fw = face.bbox[2] - face.bbox[0] fh = face.bbox[3] - face.bbox[1] area = fw * fh if area > max_area: max_area = area h_frame, w_frame = cam_obj.frame.shape[:2] m_x, m_y = int(fw * 0.30), int(fh * 0.30) y1 = max(0, int(face.bbox[1]) - m_y) y2 = min(h_frame, int(face.bbox[3]) + m_y) x1 = max(0, int(face.bbox[0]) - m_x) x2 = min(w_frame, int(face.bbox[2]) + m_x) mejor_roi = cam_obj.frame[y1:y2, x1:x2] face_metadata = face if mejor_roi is not None and mejor_roi.size > 0: cv2.imshow("Nueva Persona", mejor_roi) cv2.waitKey(1) nom = input("Escribe el nombre de la persona: ").strip() cv2.destroyWindow("Nueva Persona") if nom: import json genero_guardado = "Man" if face_metadata.sex == "M" else "Woman" ruta_generos = os.path.join("cache_nombres", "generos.json") os.makedirs("cache_nombres", exist_ok=True) dic_generos = {} if os.path.exists(ruta_generos): try: with open(ruta_generos, 'r') as f: dic_generos = json.load(f) except Exception: pass dic_generos[nom] = genero_guardado with open(ruta_generos, 'w') as f: json.dump(dic_generos, f) ruta_db = "db_institucion" os.makedirs(ruta_db, exist_ok=True) cv2.imwrite(os.path.join(ruta_db, f"{nom}.jpg"), mejor_roi) print(f"[OK] Rostro guardado (Género autodetectado: {genero_guardado})") print(" Sincronizando base de datos en caliente...") nuevos_vectores = gestionar_vectores(actualizar=True) with IA_LOCK: BASE_DATOS_ROSTROS.clear() BASE_DATOS_ROSTROS.update(nuevos_vectores) print(" Sistema listo.") else: print("[!] Registro cancelado.") if __name__ == "__main__": main()