"""---------------------------------------------version 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 import math import queue from scipy.spatial.distance import cosine # ────────────────────────────────────────────────────────────────────────────── # CONFIGURACIÓN # ────────────────────────────────────────────────────────────────────────────── USUARIO, PASSWORD, IP_DVR = "admin", "TCA200503", "192.168.1.244" SECUENCIA = [1, 7, 5, 8, 3, 6] 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_5_msmt17_batch1.onnx" VECINOS = { "1": ["7"], "7": ["1", "5"], "5": ["7", "8"], "8": ["5", "3"], "3": ["8", "6"], "6": ["3"] } TIEMPO_MAX_AUSENCIA = 900.0 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 # ────────────────────────────────────────────────────────────────────────────── # OSNET # ────────────────────────────────────────────────────────────────────────────── print("Cargando OSNet...") try: ort_session = ort.InferenceSession(ONNX_MODEL_PATH, providers=['CPUExecutionProvider']) input_name = ort_session.get_inputs()[0].name print("OSNet listo para CPU.") except Exception as e: print(f"ERROR FATAL: {e}"); exit() MEAN = np.array([0.485, 0.456, 0.406], dtype=np.float32).reshape(3, 1, 1) STD = np.array([0.229, 0.224, 0.225], dtype=np.float32).reshape(3, 1, 1) # ────────────────────────────────────────────────────────────────────────────── # EXTRACCIÓN DE FIRMAS MULTI-DESCRIPTOR # ────────────────────────────────────────────────────────────────────────────── def extraer_features_osnet_seguro(rois): #Procesamiento secuencial 1-a-1 de grado comercial. Garantiza 0% de probabilidad de crasheo C++ en ONNX. if not rois: return [] features_norm = [] for roi in rois: # 1. Preprocesamiento matemático exacto para OSNet img = cv2.resize(roi, (128, 256)) img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) img = img.transpose(2, 0, 1).astype(np.float32) / 255.0 # Expandimos la dimensión para crear un Batch estricto de tamaño 1: (1, 3, 256, 128) blob = np.expand_dims((img - MEAN) / STD, 0) # 2. INFERENCIA AISLADA (El motor ONNX no sufrirá estrés de memoria) # El [0][0] extrae el vector ignorando la dimensión del batch df = ort_session.run(None, {input_name: blob})[0][0].flatten() # 3. Normalización L2 (Vital para que funcione la distancia Coseno) n = np.linalg.norm(df) if n > 0: df /= n features_norm.append(df) return features_norm def calcular_nitidez(img): if img is None or img.size == 0: return 0 gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) return cv2.Laplacian(gray, cv2.CV_64F).var() def preprocess_onnx(roi): img = cv2.resize(roi, (128, 256)) img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) img = img.transpose(2, 0, 1).astype(np.float32) / 255.0 return np.expand_dims((img - MEAN) / STD, 0) SOPORTA_BATCH = None # Variable global para detectar compatibilidad ONNX def procesar_batch_osnet(rois): #Procesa los recortes de forma segura y secuencial para evitar abortos C++ en ONNX if not rois: return [] features_norm = [] for roi in rois: # Preprocesamiento individual img = cv2.resize(roi, (128, 256)) img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) img = img.transpose(2, 0, 1).astype(np.float32) / 255.0 blob = np.expand_dims((img - MEAN) / STD, 0) # ⚡ Ejecución segura 1 a 1 (Garantiza que ONNX no colapse) df = ort_session.run(None, {input_name: blob})[0][0].flatten() n = np.linalg.norm(df) if n > 0: df /= n features_norm.append(df) return features_norm def extraer_color_zonas(roi): if roi is None or roi.size == 0 or roi.shape[0] < 30 or roi.shape[1] < 10: return np.zeros(512 * 3, dtype=np.float32) h, w = roi.shape[:2] # ⚡ ALUCÍN OPTIMIZADO: LA MÁSCARA ELÍPTICA # Creamos un lienzo negro y dibujamos un óvalo blanco en el centro. # El histograma IGNORARÁ todo lo negro (la pared/fondo) y solo leerá lo blanco (el cuerpo). mask = np.zeros((h, w), dtype=np.uint8) centro_x, centro_y = int(w/2), int(h/2) eje_x, eje_y = int(w * 0.35), int(h * 0.48) # Óvalo ajustado a proporciones humanas cv2.ellipse(mask, (centro_x, centro_y), (eje_x, eje_y), 0, 0, 360, 255, -1) # ⚡ REGRESO AL HSV PURO (Sin CLAHE que adultere la luz) hsv = cv2.cvtColor(roi, cv2.COLOR_BGR2HSV) t1 = int(h * 0.30) t2 = int(h * 0.65) # Cortamos la imagen Y la máscara en las 3 zonas (Cabeza, Torso, Piernas) z1_img, z1_mask = hsv[:t1, :], mask[:t1, :] z2_img, z2_mask = hsv[t1:t2, :], mask[t1:t2, :] z3_img, z3_mask = hsv[t2:, :], mask[t2:, :] def calc_hist(img_part, mask_part): if img_part.size == 0: return np.zeros(512, dtype=np.float32) # Calculamos el color 3D (H,S,V) PERO pasándole la máscara para ignorar el fondo hist = cv2.calcHist([img_part], [0, 1, 2], mask_part, [8, 8, 8], [0, 180, 0, 256, 0, 256]) cv2.normalize(hist, hist, alpha=1.0, norm_type=cv2.NORM_L1) return hist.flatten() h1 = calc_hist(z1_img, z1_mask) h2 = calc_hist(z2_img, z2_mask) h3 = calc_hist(z3_img, z3_mask) # Retorna 1536 dimensiones de color PURO y libre de paredes return np.concatenate([h1, h2, h3]).astype(np.float32) def extraer_proporciones_anatomicas(kpts): if kpts is None or len(kpts) < 17: return None kpts = np.array(kpts) if kpts.shape[0] < 17: return None puntos = {i: kpts[i] for i in range(17)} if puntos[0][2] < 0.30: return None if puntos[5][2] < 0.30 and puntos[6][2] < 0.30: return None if puntos[11][2] < 0.30 and puntos[12][2] < 0.30: return None def dist(i, j): if puntos[i][2] < 0.30 or puntos[j][2] < 0.30: return 0.0 return float(np.linalg.norm(puntos[i][:2] - puntos[j][:2])) hombros = dist(5, 6) caderas = dist(11, 12) torso = max(dist(5, 11), dist(6, 12)) pierna_izq = dist(11, 13) + dist(13, 15) if puntos[13][2] > 0.30 and puntos[15][2] > 0.30 else 0 pierna_der = dist(12, 14) + dist(14, 16) if puntos[14][2] > 0.30 and puntos[16][2] > 0.30 else 0 piernas = max(pierna_izq, pierna_der) brazo_izq = dist(5, 7) + dist(7, 9) if puntos[7][2] > 0.30 and puntos[9][2] > 0.30 else 0 brazo_der = dist(6, 8) + dist(8, 10) if puntos[8][2] > 0.30 and puntos[10][2] > 0.30 else 0 brazos = max(brazo_izq, brazo_der) altura = torso + piernas if altura < 15.0: return None feats = np.array([ hombros / max(altura, 1e-6), caderas / max(altura, 1e-6), torso / max(altura, 1e-6), piernas / max(altura, 1e-6), brazos / max(altura, 1e-6), hombros / max(caderas, 1e-6), piernas / max(torso, 1e-6), ], dtype=np.float32) feats = np.clip(feats, 0.0, 3.0) n = np.linalg.norm(feats) if n > 0: feats /= n return feats def es_humano_valido(kpts, box, conf): if conf < 0.45: return False if kpts is None or len(kpts) < 17: return False x1, y1, x2, y2 = box w, h = x2 - x1, y2 - y1 if h / max(w, 1) < 0.50 or h < 50: return False cabeza_segura = any(p[2] > 0.55 for p in kpts[:5]) if not cabeza_segura: return False torso_seguro = any(p[2] > 0.40 for p in kpts[5:13]) if not torso_seguro: return False return True def extraer_firma_hibrida(frame_hd, box_480, kpts=None, deep_feat=None): try: h_hd, w_hd = frame_hd.shape[:2] x1, y1, x2, y2 = box_480 x1_c = max(0, int(x1 * (w_hd/480.0))) y1_c = max(0, int(y1 * (h_hd/270.0))) x2_c = min(w_hd, int(x2 * (w_hd/480.0))) y2_c = min(h_hd, int(y2 * (h_hd/270.0))) 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 score_nitidez = calcular_nitidez(roi) # ⚡ SOPORTE BATCH: Si no se entregó deep_feat masivo, lo calculamos individualmente (Fallback) if deep_feat is None: blob = preprocess_onnx(roi) deep_feat = ort_session.run(None, {input_name: blob})[0][0].flatten() n = np.linalg.norm(deep_feat) if n > 0: deep_feat /= n color_f = extraer_color_zonas(roi) anat_f = extraer_proporciones_anatomicas(kpts) ratio_hw = roi.shape[0] / max(roi.shape[1], 1) calidad = (x2_c - x1_c) * (y2_c - y1_c) return { 'deep': deep_feat, 'color': color_f, # ⚡ ELIMINADO: Textura LBP 'anatomia': anat_f, 'ratio_hw': ratio_hw, 'calidad': calidad, 'nitidez': score_nitidez } except Exception as e: return None def similitud_hibrida(f1, f2, cross_cam=False, confianza_fisica=1.0): if f1 is None or f2 is None: return 0.0 from scipy.spatial.distance import cosine import numpy as np deep1, deep2 = f1.get('deep'), f2.get('deep') if deep1 is None or deep2 is None: return 0.0 sim_deep = max(0.0, 1.0 - cosine(deep1, deep2)) # ───────────────────────────────────────────────────────────── # ⚡ ECUALIZADOR CROSS-CAM (Domain Gap Fix) # Compensamos la pérdida de puntaje causada por el cambio de iluminación. # Evita que IDs viejos de la misma cámara ganen injustamente. # ───────────────────────────────────────────────────────────── if cross_cam and sim_deep > 0.50: sim_deep += 0.05 # 1. ZONA DE CERTEZA ABSOLUTA if sim_deep >= 0.70: return min(1.0, sim_deep + 0.15) # 2. VÍA DE RECHAZO (No hay forma de que sea él) if sim_deep < 0.45: return sim_deep # 3. TRIBUNAL DE TEXTURAS Y ANATOMÍA resultado_final = sim_deep + 0.05 a1, a2 = f1.get("anatomia"), f2.get("anatomia") if a1 is not None and a2 is not None and hasattr(a1, "shape") and a1.shape == a2.shape: sim_anat = max(0.0, 1.0 - np.linalg.norm(a1 - a2)) peso_anat = 0.12 * confianza_fisica if sim_anat > 0.88: resultado_final += peso_anat if sim_anat >= 0.95: resultado_final += 0.04 elif sim_anat < 0.35: resultado_final -= peso_anat else: # PUNTOS CIEGOS (Solapamiento / Espaldas) if sim_deep >= 0.50: resultado_final += 0.12 return float(max(0.0, min(1.0, resultado_final))) def desglose_similitud(f1, f2, cross_cam=False): # ------------------------------------------------- # 1. DEEP (OSNET) # ------------------------------------------------- sim_deep = 0.0 deep1, deep2 = f1.get('deep'), f2.get('deep') if deep1 is not None and deep2 is not None and np.sum(deep1) > 0 and np.sum(deep2) > 0: sim_deep = max(0.0, 1.0 - cosine(deep1, deep2)) # ------------------------------------------------- # 2. COLOR (2D HS INMUNE A SOMBRAS) # ------------------------------------------------- sim_color = -1.0 c1, c2 = f1.get("color"), f2.get("color") # ⚡ CÓDIGO RESTAURADO A 1536 PARA LEER EL HSV PURO Y ENMASCARADO if c1 is not None and c2 is not None and c1.shape == c2.shape and len(c1) == 1536: if np.sum(c1) > 0.1 and np.sum(c2) > 0.1: dist1 = cv2.compareHist(c1[:512].astype(np.float32), c2[:512].astype(np.float32), cv2.HISTCMP_BHATTACHARYYA) dist2 = cv2.compareHist(c1[512:1024].astype(np.float32), c2[512:1024].astype(np.float32), cv2.HISTCMP_BHATTACHARYYA) dist3 = cv2.compareHist(c1[1024:].astype(np.float32), c2[1024:].astype(np.float32), cv2.HISTCMP_BHATTACHARYYA) s1 = max(0.0, 1.0 - float(dist1)) s2 = max(0.0, 1.0 - float(dist2)) s3 = max(0.0, 1.0 - float(dist3)) # Escudo chamarra abierta if cross_cam and s3 > 0.60 and s2 < 0.35: s2 = max(s2, s3 * 0.80) sim_color = (s1 * 0.20) + (s2 * 0.50) + (s3 * 0.30) # ------------------------------------------------- # 3. ANATOMIA (7 Puntos) # ------------------------------------------------- sim_anat = -1.0 a1, a2 = f1.get("anatomia"), f2.get("anatomia") if a1 is not None and a2 is not None and hasattr(a1, "shape") and hasattr(a2, "shape") and a1.shape == a2.shape: dist_a = np.linalg.norm(a1 - a2) sim_anat = max(0.0, 1.0 - dist_a) return sim_deep, sim_color, sim_anat # ────────────────────────────────────────────────────────────────────────────── # KALMAN TRACKER # ────────────────────────────────────────────────────────────────────────────── class KalmanTrack: _count = 0 def __init__(self, box, now): kf = cv2.KalmanFilter(7, 4) 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) kf.transitionMatrix = np.eye(7, dtype=np.float32) kf.transitionMatrix[0,4] = kf.transitionMatrix[1,5] = kf.transitionMatrix[2,6] = 1 kf.processNoiseCov = np.diag([2.0, 4.0, 15.0, 0.01, 1.0, 2.0, 1.0]).astype(np.float32) kf.measurementNoiseCov = np.diag([10.0, 10.0, 50.0, 0.10]).astype(np.float32) kf.errorCovPost = np.diag([10., 10., 25., 1., 500., 500., 100.]).astype(np.float32) kf.statePost = np.zeros((7,1), np.float32) kf.statePost[:4] = self._z(box) self.kf = kf self.local_id = KalmanTrack._count; KalmanTrack._count += 1 self.gid, self.origen_global, self.aprendiendo = None, False, False self.box = list(box) self.ts_creacion = 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.firma_pre_grupo = None self.ts_salio_grupo = 0.0 self.fallos_post_grupo = 0 self.ultimo_aprendizaje = 0.0 self.frames_continuos = 0 self.frames_observados = 0 self.firma_ema = None self.muestras_capturadas = 0 self.ultimo_cambio_id = 0.0 def _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/max(h,1e-6)]]).astype(np.float32) def _bbox(self, x): cx, cy = float(x[0].item()), float(x[1].item()) s, r = float(x[2].item()), float(x[3].item()) w = np.sqrt(max(s * r, 1e-6)); h = s / (w + 1e-6) return [cx-w/2., cy-h/2., cx+w/2., cy+h/2.] @property def confianza_fisica(self): return min(1.0, self.frames_continuos / 20.0) def calcular_score_calidad(self, firma): if firma is None: return 0.0 score_nitidez = min(firma.get('nitidez', 0) / 100.0, 1.0) * 50.0 score_area = min(firma.get('calidad', 0) / 12000.0, 1.0) * 50.0 return score_nitidez + score_area def actualizar_ema(self, firma): if firma is None: return self.muestras_capturadas = getattr(self, 'muestras_capturadas', 0) + 1 score_nuevo = self.calcular_score_calidad(firma) score_viejo = getattr(self, 'score_ema', 0.0) if self.firma_ema is None: # Primera impresión (puede ser mala, pero es lo único que tenemos) self.firma_ema = {k: v.copy() if isinstance(v, np.ndarray) else v for k, v in firma.items()} self.score_ema = score_nuevo return # ⚡ PROTECCIÓN DE MEMORIA (Anti-polución) if self.muestras_capturadas > 4 and score_nuevo < (score_viejo * 0.60): return # ───────────────────────────────────────────────────────────── # ⚡ ASIMILACIÓN DINÁMICA (La cura a la mala primera impresión) # ───────────────────────────────────────────────────────────── # Si estamos en los primeros 5 frames, O la nueva foto es notoriamente mejor # que nuestro mejor recuerdo (>20% mejor), somos una "esponja". if self.muestras_capturadas <= 5 or score_nuevo > (score_viejo * 1.20): # Alpha bajo = Sobrescribe agresivamente el pasado alpha_deep = 0.20 # 20% vieja, 80% NUEVA alpha_color = 0.20 alpha_geo = 0.50 self.score_ema = score_nuevo # Actualizamos nuestro estándar de calidad else: # Estado de madurez: La firma ya es excelente y estable. # Los cambios deben ser muy lentos para no arruinarla. alpha_deep = 0.85 # 85% vieja, 15% nueva alpha_color = 0.50 alpha_geo = 0.80 self.score_ema = max(score_nuevo, score_viejo) ema = self.firma_ema if 'deep' in firma: ema['deep'] = alpha_deep * ema['deep'] + (1-alpha_deep) * firma['deep'] n = np.linalg.norm(ema['deep']) if n > 0: ema['deep'] /= n # Ya sin la textura LBP que eliminamos por rendimiento for key in ['color']: if key in firma and firma[key] is not None: ema[key] = alpha_color * ema[key] + (1-alpha_color) * firma[key] normed = cv2.normalize(ema[key], None, alpha=1.0, norm_type=cv2.NORM_L1) ema[key] = normed.flatten() if 'ratio_hw' in firma: ema['ratio_hw'] = alpha_geo * ema['ratio_hw'] + (1-alpha_geo) * firma['ratio_hw'] if 'anatomia' in firma and firma['anatomia'] is not None: if 'anatomia' not in ema or ema['anatomia'] is None: ema['anatomia'] = np.array(firma['anatomia'], dtype=np.float32) else: ema['anatomia'] = alpha_geo * np.asarray(ema['anatomia']) + (1-alpha_geo) * np.asarray(firma['anatomia']) if 'calidad' in firma: ema['calidad'] = max(ema.get('calidad', 0), firma['calidad']) def predict(self, turno_activo=True): if self.time_since_update > 30: return None if getattr(self, 'en_grupo', False): 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._bbox(self.kf.statePre) return self.box if self.time_since_update > 0: self.kf.statePost[4] *= 0.5 self.kf.statePost[5] *= 0.5 self.kf.statePost[6] = 0.0 self.frames_continuos = max(0, self.frames_continuos - 1) 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._bbox(self.kf.statePre) return self.box def update(self, box, en_grupo, now, kpts=None, firma_actual=None): self.ts_ultima_deteccion = now self.time_since_update, self.en_grupo = 0, en_grupo self.box = list(box) self.kf.correct(self._z(box)) if not en_grupo: self.frames_observados += 1 self.frames_continuos += 1 self.listo_para_id = True # ────────────────────────────────────────────────────────────────────────────── # MEMORIA GLOBAL CON PERSISTENCIA # ────────────────────────────────────────────────────────────────────────────── class GlobalMemory: def __init__(self): self.db = {} self.next_gid = 100 self.lock = threading.RLock() self.ruta_memoria = os.path.join("cache_nombres", "memoria_ids.json") self.ruta_movimientos = os.path.join("cache_nombres", "registro_movimientos.csv") os.makedirs("cache_nombres", exist_ok=True) self.nombres_activos = {} self.ultimos_saludos = {} self._votos_nombre = {} self.fecha_actual = datetime.now().date() self._cargar_memoria() self.reserva_temporal = {} self.id_locks = {} self.ultimas_detecciones = {} def consolidar_clones(self): #Rutina de Auto-Merge comercial con Fusión Forzada por Biometría Facial. from scipy.spatial.distance import cosine with self.lock: ids_activos = list(self.db.keys()) for i in range(len(ids_activos)): for j in range(i + 1, len(ids_activos)): id1, id2 = ids_activos[i], ids_activos[j] if id1 not in self.db or id2 not in self.db: continue ema1 = self.db[id1].get('ema') ema2 = self.db[id2].get('ema') if not ema1 or not ema2: continue deep1, deep2 = ema1.get('deep'), ema2.get('deep') if deep1 is None or deep2 is None: continue nom1 = self.db[id1].get('nombre') nom2 = self.db[id2].get('nombre') fusion_forzada = False # ⚡ AUTORIDAD FACIAL if nom1 and nom2 and nom1 == nom2: fusion_forzada = True # InsightFace los bautizó igual. Fusionamos sin importar la ropa. elif nom1 and nom2 and nom1 != nom2: continue # Nombres distintos = Personas distintas. Bloquear fusión. similitud = 1.0 - cosine(deep1, deep2) # Se fusionan si OSNet está seguro O si InsightFace forzó la orden if similitud > 0.76 or fusion_forzada: id_ganador = min(id1, id2) id_clon = max(id1, id2) if self.db[id_clon].get('nombre'): self.db[id_ganador]['nombre'] = self.db[id_clon]['nombre'] # Promediamos las firmas para que OSNet aprenda de ambos ángulos self.db[id_ganador]['ema']['deep'] = (self.db[id_ganador]['ema']['deep'] * 0.7) + (self.db[id_clon]['ema']['deep'] * 0.3) self.db[id_clon] = {'fusionado_con': id_ganador, 'ts': time.time()} print(f" [AUTO-MERGE] Fragmentación corregida. Clon ID {id_clon} absorbido por Original ID {id_ganador} (Sim: {similitud:.2f})") razon = "FUSIÓN FACIAL FORZADA" if fusion_forzada else "Fragmentación corregida" print(f" [AUTO-MERGE] {razon}. Clon ID {id_clon} absorbido por Original ID {id_ganador} (Sim: {similitud:.2f})") # Método 1: Registrar detección global def registrar_deteccion_global(self, gid, cam_id, firma, now): #Registra que un ID fue detectado en una cámara para coordinación solapada. if firma is None or 'deep' not in firma: return # Hash simple de la firma deep para comparación rápida firma_hash = hash(firma['deep'].tobytes()) % 10000 self.ultimas_detecciones[gid] = { 'cam': str(cam_id), 'ts': now, 'firma_hash': firma_hash } # Limpiar detecciones antiguas (>10 segundos) antiguas = [g for g, d in self.ultimas_detecciones.items() if now - d['ts'] > 10.0] for g in antiguas: del self.ultimas_detecciones[g] # Método 2: Buscar coincidencia solapada def buscar_coincidencia_solapada(self, firma, cam_id, now, umbral_sim=0.85): #Busca si esta detección podría ser un ID ya detectado en cámara vecina recientemente. if firma is None or 'deep' not in firma: return None firma_hash = hash(firma['deep'].tobytes()) % 10000 vecinos = VECINOS.get(str(cam_id), []) for gid, info in self.ultimas_detecciones.items(): # Solo considerar si fue detectado en cámara vecina hace <5s if info['cam'] not in vecinos: continue if now - info['ts'] > 5.0: continue # Si el hash coincide exactamente, es muy probable que sea el mismo if info['firma_hash'] == firma_hash: return gid # Si no, hacer comparación completa (más costosa) ema = self.db.get(gid, {}).get('ema') if ema is not None: sim = similitud_hibrida(firma, ema, cross_cam=True) if sim >= umbral_sim: return gid return None def lock_id_for_camera(self, gid, cam_id, now, duration=15.0): self.id_locks[gid] = { 'cam': str(cam_id), 'unlock_ts': now + duration, 'nombre': self.db.get(gid, {}).get('nombre') } print(f" [ID LOCK] ID {gid} bloqueado en Cam{cam_id} por {duration}s") def is_id_locked(self, gid, cam_id, now): if gid not in self.id_locks: return False lock = self.id_locks[gid] if lock['cam'] != str(cam_id): return False if now < lock['unlock_ts']: return True else: del self.id_locks[gid] return False def limpiar_locks_vencidos(self, now): vencidos = [gid for gid, lock in self.id_locks.items() if now >= lock['unlock_ts']] for gid in vencidos: del self.id_locks[gid] def registrar_salida(self, gid, cam_salida, now): if gid not in self.db: return vecinas = VECINOS.get(str(cam_salida), []) self.reserva_temporal[gid] = { 'cam_salida': str(cam_salida), 'ts_salida': now, 'cam_esperadas': vecinas, 'nombre': self.db[gid].get('nombre') } reservas_a_limpiar = [] for gid_reserva, info in self.reserva_temporal.items(): if now - info['ts_salida'] > 30.0: reservas_a_limpiar.append(gid_reserva) for gid_limpiar in reservas_a_limpiar: del self.reserva_temporal[gid_limpiar] def _bonus_reserva(self, gid_evaluado, cam_actual, now): return 0.0 def _firma_a_dict(self, firma): if not firma: return None res = {} for k, v in firma.items(): if isinstance(v, np.ndarray): res[k] = v.tolist() elif k == 'anatomia' and v is not None: res[k] = v.tolist() if isinstance(v, np.ndarray) else list(v) else: res[k] = v return res def _dict_a_firma(self, d): if not d: return None res = {} for k, v in d.items(): if isinstance(v, list) and k in ['deep', 'color', 'textura', 'anatomia']: res[k] = np.array(v, dtype=np.float32) else: res[k] = v return res def _cargar_memoria(self): self.fecha_actual = datetime.now().date() if os.path.exists(self.ruta_memoria): # ⚡ RESET DIARIO EN INICIO: Si el archivo es de un día anterior, iniciar limpio. fecha_archivo = datetime.fromtimestamp(os.path.getmtime(self.ruta_memoria)).date() if fecha_archivo != self.fecha_actual: print(f"[Memoria] Archivo de ayer ({fecha_archivo}). Iniciando memoria limpia desde ID 100.") return try: with open(self.ruta_memoria, 'r') as f: datos = json.load(f) ahora = time.time() cargados_validos = 0 for gid_str, info in datos.items(): ts_guardado = info.get('ts', 0) # ⚡ PERSISTENCIA COMERCIAL es_vip = info.get('nombre') is not None # Los VIP duran 3 horas (10800s), los desconocidos 1 hora (3600s) para no saturar RAM if es_vip and (ahora - ts_guardado > 10800.0): continue if not es_vip and (ahora - ts_guardado > 3600.0): continue gid = int(gid_str) self.db[gid] = { 'ema': self._dict_a_firma(info.get('ema')), 'last_cam': info.get('last_cam', '1'), 'ts': ts_guardado, 'nombre': info.get('nombre'), 'actualizaciones_globales': info.get('actualizaciones_globales', 0), 'firmas_altas': [self._dict_a_firma(f) for f in info.get('firmas_altas', []) if f], } cargados_validos += 1 if cargados_validos > 0: self.next_gid = max([int(k) for k in datos.keys()] + [99]) + 1 print(f"[Memoria] {cargados_validos} IDs persistentes válidos cargados.") except Exception as e: print(f"[Memoria] Error cargando: {e}") # ⚡ FIX: Cargar los saludos previos para evitar spam auditivo ruta_saludos = os.path.join("cache_nombres", "registro_saludos.json") if os.path.exists(ruta_saludos): try: with open(ruta_saludos, 'r') as f: self.ultimos_saludos = json.load(f) except Exception: pass def guardar_memoria(self): try: datos = {} for gid, info in self.db.items(): datos[str(gid)] = { 'ema': self._firma_a_dict(info.get('ema')), 'last_cam': info.get('last_cam', '1'), 'nombre': info.get('nombre'), 'ts': info.get('ts', time.time()), 'actualizaciones_globales': info.get('actualizaciones_globales', 0), 'firmas_altas': [self._firma_a_dict(f) for f in info.get('firmas_altas', [])], } with open(self.ruta_memoria, 'w') as f: json.dump(datos, f) except Exception as e: print(f"[Memoria] Error guardando: {e}") def registrar_movimiento(self, nombre, cam_origen, cam_destino, duracion_seg): nuevo = not os.path.exists(self.ruta_movimientos) try: with open(self.ruta_movimientos, 'a', newline='', encoding='utf-8') as f: w = csv.writer(f) if nuevo: w.writerow(["Fecha_Hora", "Nombre", "Origen", "Destino", "Seg_en_Origen"]) w.writerow([datetime.now().strftime("%Y-%m-%d %H:%M:%S"), nombre, cam_origen, cam_destino, round(duracion_seg, 2)]) except: pass def asignar_nombre(self, gid, nombre, confianza, es_fusion=False): VOTOS_NOMBRE_MIN = 3 UMBRAL_VOTO = 0.52 UMBRAL_CONFIRMACION = 0.62 if confianza < (0.50 if es_fusion else UMBRAL_VOTO): return False with self.lock: rec = self.db.get(gid) if not rec: return False nombre_viejo = rec.get('nombre') if not hasattr(self, 'nombres_activos'): self.nombres_activos = {} gid_ocupante = self.nombres_activos.get(nombre) if gid_ocupante is not None and gid_ocupante != gid: if gid_ocupante in self.db: ts_ultimo = self.db[gid_ocupante].get('ts', 0) if (time.time() - ts_ultimo) < 10.0: return False conf_actual = rec.get('confianza_nombre', 0.0) if nombre_viejo == nombre and conf_actual > confianza: return False if not hasattr(self, '_votos_nombre'): self._votos_nombre = {} if gid not in self._votos_nombre: self._votos_nombre[gid] = {} votos_gid = self._votos_nombre[gid] if any(n != nombre for n in votos_gid.keys()): votos_gid.clear() if nombre not in votos_gid: votos_gid[nombre] = [] votos_gid[nombre].append(confianza) votos_gid[nombre] = votos_gid[nombre][-6:] n_votos = len(votos_gid[nombre]) conf_media = sum(votos_gid[nombre]) / n_votos if n_votos < VOTOS_NOMBRE_MIN or conf_media < UMBRAL_CONFIRMACION: return False if nombre_viejo is not None and nombre_viejo != nombre: if confianza < 0.70: print(f" [PROTECCIÓN] Rechazando cambio de {nombre_viejo} a {nombre} (conf={confianza:.2f} < 0.70)") return False # Rechazar cambio automático # Si confianza >= 0.70, permitir cambio pero con advertencia print(f" [CAMBIO VIP] ID {gid} cambió de {nombre_viejo} a {nombre} (conf={confianza:.2f} >= 0.70)") rec['nombre'] = nombre rec['confianza_nombre'] = conf_media self.nombres_activos[nombre] = gid votos_gid.clear() tipo = "FUSIÓN" if es_fusion else "BAUTIZO" print(f" [{tipo}] ID {gid} confirmado como {nombre} | Votos: {n_votos} | Conf. Media: {conf_media:.2f}") return True def confirmar_firma_vip(self, gid, ts=None): with self.lock: rec = self.db.get(gid) if rec and rec.get('ema') is not None: # ⚡ FIX P2: Respetar la cristalización if rec.get('cristalizado', False): print(f" [MEMORIA] ID {gid} está cristalizado. No se añaden más firmas VIP.") return firma_actual = rec['ema'] if 'firmas_altas' not in rec: rec['firmas_altas'] = [] rec['firmas_altas'].append({ k: v.copy() if isinstance(v, np.ndarray) else v for k, v in firma_actual.items() }) if ts is not None: rec['ts'] = ts print(f" [MEMORIA] Firma de ID {gid} bloqueada y protegida como VIP.") def guardar_saludo(self, nombre): #Guarda el saludo en memoria y lo escribe en el disco duro. ahora = time.time() fecha_hoy = datetime.fromtimestamp(ahora).strftime("%Y-%m-%d") self.ultimos_saludos[nombre] = { 'fecha': fecha_hoy, 'timestamp': ahora } # Guardar físicamente en el disco ruta_saludos = os.path.join("cache_nombres", "registro_saludos.json") try: with open(ruta_saludos, 'w') as f: json.dump(self.ultimos_saludos, f, indent=4) except Exception as e: print(f" [MEMORIA] Error guardando registro de saludos: {e}") def _actualizar_sin_lock(self, gid, firma, cam_id, now): if gid not in self.db: self.db[gid] = { 'ema': firma, 'last_cam': cam_id, 'ts': now, 'nombre': None, 'actualizaciones_globales': 1, 'cristalizado': False } return rec = self.db[gid] # ⚡ CRISTALIZACIÓN: Después de 15 actualizaciones, CONGELAR EMA if not rec.get('cristalizado', False) and rec.get('actualizaciones_globales', 0) >= 15: rec['cristalizado'] = True print(f" [CRISTALIZADO] ID {gid} ha alcanzado madurez. EMA congelado permanentemente.") # ⚡ BLINDAJE DE CRISTALIZACIÓN # Si el ID ya es maduro, actualizamos dónde fue visto por última vez, # pero NUNCA permitimos que una cámara modifique su firma maestra perfecta. if rec.get('cristalizado', False): rec['last_cam'] = cam_id rec['ts'] = now rec['actualizaciones_globales'] += 1 return ema = rec['ema'] rec['actualizaciones_globales'] += 1 if ema is None: rec['ema'] = firma else: alpha_deep = 0.85 alpha_color = 0.50 alpha_geo = 0.80 if 'deep' in firma: ema['deep'] = alpha_deep * ema['deep'] + (1-alpha_deep) * firma['deep'] n = np.linalg.norm(ema['deep']) if n > 0: ema['deep'] /= n for key in ['color']: if key in firma and firma[key] is not None: ema[key] = alpha_color * ema[key] + (1-alpha_color) * firma[key] normed = cv2.normalize(ema[key], None, alpha=1.0, norm_type=cv2.NORM_L1) ema[key] = normed.flatten() if 'ratio_hw' in firma: ema['ratio_hw'] = alpha_geo * ema['ratio_hw'] + (1-alpha_geo) * firma['ratio_hw'] if 'anatomia' in firma and firma['anatomia'] is not None: if 'anatomia' not in ema or ema['anatomia'] is None: ema['anatomia'] = np.array(firma['anatomia'], dtype=np.float32) else: ema['anatomia'] = 0.80 * np.asarray(ema['anatomia']) + 0.20 * np.asarray(firma['anatomia']) if 'calidad' in firma: ema['calidad'] = firma['calidad'] rec['last_cam'] = cam_id rec['ts'] = now def _sim_contra_firma(self, firma_nueva, firma_guardada, cross_cam, confianza_fisica): if firma_guardada is None: return 0.0 return similitud_hibrida(firma_nueva, firma_guardada, cross_cam, confianza_fisica) def _sim_robusta(self, firma_nueva, gid_data, cross_cam, confianza_fisica=0.0): ema = gid_data.get('ema') if not ema: return 0.0 sim_ema = self._sim_contra_firma(firma_nueva, ema, cross_cam, confianza_fisica) if sim_ema > 0.75: return sim_ema sim_anclas = [sim_ema] for ancla in gid_data.get('firmas_altas', []): sim_anclas.append(self._sim_contra_firma(firma_nueva, ancla, cross_cam, confianza_fisica)) mejor_ancla = max(sim_anclas[1:]) if len(sim_anclas) > 1 else sim_ema return sim_ema * 0.60 + mejor_ancla * 0.40 def identificar_candidato(self, firma, cam_id, now, active_gids, en_borde=False, firma_ema_local=None, confianza_fisica=0.0): self.limpiar_fantasmas() with self.lock: candidatos = [] for gid, data in self.db.items(): dt = now - data.get('ts', now) distancia = self._distancia_topologica(str(data['last_cam']), str(cam_id)) misma_cam = (distancia == 0) es_vecino = (distancia == 1) es_salto = (distancia == 2) es_cross = not misma_cam if gid in active_gids: if misma_cam or distancia >= 3: continue if dt > TIEMPO_MAX_AUSENCIA or data.get('ema') is None: continue if self.is_id_locked(gid, cam_id, now): continue if not misma_cam: if (es_vecino and dt < 0.3) or (es_salto and dt < 2.0) or (distancia >= 3 and dt < 8.0): continue # ⚡ CÁLCULO OBLIGATORIO DE SIMILITUD ANTES DE CUALQUIER FILTRO sim = self._sim_robusta(firma, data, es_cross, confianza_fisica) sim_deep, sim_color, sim_anat = -1.0, -1.0, -1.0 ema = data.get("ema") if ema is not None: try: sim_deep, sim_color, sim_anat = desglose_similitud(firma, ema, cross_cam=es_cross) except: pass # Penalizaciones y ajustes if not misma_cam: # NUNCA bloqueamos cámaras vecinas (distancia 1) por tiempo, # porque al estar solapadas, pueden ver a la persona en el mismo milisegundo (dt=0.0). if es_salto and dt < 1.5: continue # Mínimo 1.5s para saltar una cámara entera if distancia >= 3 and dt < 5.0: continue # Mínimo 5.0s para cruzar a la otra punta del edificio if firma_ema_local is not None: sim_local = self._sim_robusta(firma_ema_local, data, es_cross, confianza_fisica) sim = 0.60 * sim + 0.40 * sim_local # ⚡ VALIDACIONES VIP es_vip = data.get('nombre') is not None tiene_firmas_altas = len(data.get('firmas_altas', [])) > 0 if es_vip and tiene_firmas_altas and sim < 0.78: continue elif es_vip and not misma_cam and sim < 0.75: continue # ───────────────────────────────────────────────────────────── # ⚡ UMBRALES CORREGIDOS (Ni parálisis, ni robo de identidad) # ───────────────────────────────────────────────────────────── penal_multitud = min(0.04, max(0, len(active_gids)-10) * 0.002) if misma_cam: umbral = 0.68 elif es_vecino: umbral = 0.72 elif es_salto: umbral = 0.75 else: umbral = 0.78 # AHORA SÍ: sim_anat y sim_deep existen if sim_anat > 0.90 and sim_deep > 0.65: umbral -= 0.04 if en_borde and not misma_cam: umbral -= 0.04 umbral = max(0.65, umbral + penal_multitud) bonus_aplicado = min(self._bonus_reserva(gid, cam_id, now), 0.03) sim += bonus_aplicado # ⚡ RESTAURAR LOGS DE EVAL PARA DIAGNÓSTICO if sim >= umbral - 0.10 or len(candidatos) < 3: estado = "ACEPTADO" if sim >= umbral else "RECHAZADO" faltante = umbral - sim if sim < umbral else 0.0 info = f"(Faltó {faltante:.2f})" if faltante > 0 else "" bonus_str = f" [+{bonus_aplicado:.2f}]" if bonus_aplicado > 0 else "" # ⚡ FIX P0: Eliminada la llamada doble a desglose_similitud. # Usamos sim_deep, sim_color y sim_anat pre-calculados. color_str = f"{sim_color:.2f}" if sim_color >= 0 else "N/A" anat_str = f"{sim_anat:.2f}" if sim_anat >= 0 else "N/A" print( f"[EVAL] Cam{cam_id} evalúa ID{gid} " f"(Cam{data['last_cam']}, dt={dt:.1f}s) | " f"Deep={sim_deep:.2f} | Color={color_str} | Anat={anat_str} | " f"Final={sim:.2f}{bonus_str} | Umb={umbral:.2f} {info} -> {estado}" ) if sim >= umbral: candidatos.append((sim, sim, gid)) firma_guardar = firma_ema_local if firma_ema_local is not None else firma if not candidatos: nid = self.next_gid self.next_gid += 1 self._actualizar_sin_lock(nid, firma_guardar, cam_id, now) return nid, False candidatos.sort(reverse=True) best_sim, _, best_gid = candidatos[0] # ⚡ ANTI-GEMELOS RECALIBRADO: Solo duda si son casi idénticos (<0.02) if len(candidatos) >= 2: _, segunda_sim, segundo_gid = candidatos[1] if abs(best_sim - segunda_sim) < 0.02 and best_sim < 0.75: nid = self.next_gid self.next_gid += 1 self._actualizar_sin_lock(nid, firma_guardar, cam_id, now) return nid, False if best_gid in self.reserva_temporal: del self.reserva_temporal[best_gid] # ───────────────────────────────────────────────────────────── # ⚡ FIX: REGISTRO DE MOVIMIENTOS EN CSV # Antes de actualizar, revisamos si viene de OTRA cámara y si tiene nombre # ───────────────────────────────────────────────────────────── cam_anterior = self.db[best_gid].get('last_cam') nombre_registrado = self.db[best_gid].get('nombre') if cam_anterior and str(cam_anterior) != str(cam_id) and nombre_registrado: tiempo_origen = now - self.db[best_gid].get('ts', now) # Lanzamos el registro en un hilo para no frenar el motor de tracking threading.Thread( target=self.registrar_movimiento, args=(nombre_registrado, cam_anterior, cam_id, tiempo_origen), daemon=True ).start() # Ahora sí, actualizamos la memoria con la nueva cámara self._actualizar_sin_lock(best_gid, firma_guardar, cam_id, now) return best_gid, True def limpiar_fantasmas(self): ahora = time.time() fecha_hoy = datetime.fromtimestamp(ahora).date() with self.lock: # ⚡ RESET DIARIO EN CALIENTE (Si el programa corre 24/7) if fecha_hoy != self.fecha_actual: print(f"\n[SISTEMA] ¡Cambio de día! Purgando memoria y reiniciando contador a 100.\n") self.db.clear() self.nombres_activos.clear() self.next_gid = 100 self.fecha_actual = fecha_hoy self.guardar_memoria() return # ⚡ PURGA NORMAL (3 hrs VIP, 1 hr Desconocidos) muertos = [] for g, d in self.db.items(): if d.get('nombre') is None and (ahora - d.get('ts', ahora)) > 3600: muertos.append(g) elif d.get('nombre') is not None and (ahora - d.get('ts', ahora)) > 10800: muertos.append(g) for g in muertos: nombre_perdido = self.db[g].get('nombre') if nombre_perdido and nombre_perdido in self.nombres_activos: del self.nombres_activos[nombre_perdido] del self.db[g] if muertos: self.guardar_memoria() def _distancia_topologica(self, cam_o, cam_d): if cam_o == cam_d: return 0 vecinos_directos = VECINOS.get(cam_o, []) if cam_d in vecinos_directos: return 1 for v in vecinos_directos: if cam_d in VECINOS.get(v, []): return 2 return 3 # ────────────────────────────────────────────────────────────────────────────── # GESTOR LOCAL # ────────────────────────────────────────────────────────────────────────────── def iou_overlap(A, B): xA,yA = max(A[0],B[0]),max(A[1],B[1]) xB,yB = min(A[2],B[2]),min(A[3],B[3]) inter = max(0,xB-xA)*max(0,yB-yA) return inter/((A[2]-A[0])*(A[3]-A[1])+(B[2]-B[0])*(B[3]-B[1])-inter+1e-6) class CamManager: def __init__(self, cam_id, global_mem): self.cam_id, self.global_mem, self.trackers = cam_id, global_mem, [] self.ultimas_detecciones_globales = {} self.tracks_post_cruce = {} def _limpiar_tracks_post_cruce(self, now): #Limpia registros de post-cruce antiguos. antiguos = [idx for idx, info in self.tracks_post_cruce.items() if now - info['ts_cruce'] > 10.0] for idx in antiguos: del self.tracks_post_cruce[idx] def _detectar_grupo(self, trk, box, todos): x1,y1,x2,y2 = box w,h = x2-x1, y2-y1 cx,cy = (x1+x2)/2, (y1+y2)/2 fx, fy = (1.1, 0.45) if getattr(trk,'en_grupo',False) else (0.9, 0.35) for o in todos: if o is trk or not hasattr(o,'box') or o.box is None or o.time_since_update > 2: continue if iou_overlap(box, o.box) > 0.05: return True ocx,ocy = (o.box[0]+o.box[2])/2, (o.box[1]+o.box[3])/2 if abs(cx-ocx)= 0.48: trk.fallos_post_grupo = 0; trk.firma_pre_grupo = None with self.global_mem.lock: if trk.gid in self.global_mem.db: self.global_mem.db[trk.gid]['ema'] = fa else: trk.fallos_post_grupo += 1 if trk.fallos_post_grupo >= 3: trk.gid = trk.firma_pre_grupo = None; trk.fallos_post_grupo = 0 def _asignar(self, boxes, confidences, frame_hd, now, keypoints_list): n_trk, n_det = len(self.trackers), len(boxes) firmas = [None] * n_det # ───────────────────────────────────────────────────────────── # ⚡ OPTIMIZACIÓN BATCH: Extraemos OSNet para todas las detecciones de un golpe # ───────────────────────────────────────────────────────────── if n_det > 0: rois_validos = [] mapa_rois = {} h_hd, w_hd = frame_hd.shape[:2] # 1. Recolectar todos los recortes (ROIs) de la imagen for d_idx, box in enumerate(boxes): x1, y1, x2, y2 = box x1_c = max(0, int(x1 * (w_hd/480.0))) y1_c = max(0, int(y1 * (h_hd/270.0))) x2_c = min(w_hd, int(x2 * (w_hd/480.0))) y2_c = min(h_hd, int(y2 * (h_hd/270.0))) roi = frame_hd[y1_c:y2_c, x1_c:x2_c] # Descartamos basura óptica de inmediato if roi.size > 0 and roi.shape[0] >= 40 and roi.shape[1] >= 20: mapa_rois[d_idx] = len(rois_validos) rois_validos.append(roi) # 2. Ejecutar la red neuronal para todos SIMULTÁNEAMENTE if rois_validos: # ⚡ Llamamos a la función segura que procesa 1 a 1 rápidamente deep_feats_seguros = extraer_features_osnet_seguro(rois_validos) # Ensamblar las firmas for d_idx, box in enumerate(boxes): if d_idx in mapa_rois: df_persona = deep_feats_seguros[mapa_rois[d_idx]] kpts = keypoints_list[d_idx] if keypoints_list and d_idx < len(keypoints_list) else None # Extraemos el resto de la firma (Color enmascarado, Anatomía, etc.) firmas[d_idx] = extraer_firma_hibrida(frame_hd, box, kpts, deep_feat=df_persona) def obtener_firma(d_idx): return firmas[d_idx] if n_trk == 0: return [], list(range(n_det)), [], firmas if n_det == 0: return [], [], list(range(n_trk)), firmas # ... (A partir de aquí, el código de "alta" y "baja" confidence se queda EXACTAMENTE como ya lo tienes) ... alta = [(d,boxes[d]) for d,c in enumerate(confidences) if c >= 0.45] baja = [(d,boxes[d]) for d,c in enumerate(confidences) if c < 0.45] matched, sin_match_trk = [], list(range(n_trk)) if alta: C = np.full((n_trk, len(alta)), 100.0, np.float32) for t, trk in enumerate(self.trackers): dt_oculto = now - trk.ts_ultima_deteccion radio = min(400., max(150., 300.*math.sqrt(max(0.1,dt_oculto)))) ref_firma = trk.firma_ema or (self.global_mem.db.get(trk.gid, {}).get('ema') if trk.gid else None) for idx, (d, det) in enumerate(alta): iou = iou_overlap(trk.box, det) dist = math.sqrt(((trk.box[0]+trk.box[2])/2-(det[0]+det[2])/2)**2 + ((trk.box[1]+trk.box[3])/2-(det[1]+det[3])/2)**2) if dist > radio: continue a_trk = (trk.box[2]-trk.box[0])*(trk.box[3]-trk.box[1]) a_det = (det[2]-det[0])*(det[3]-det[1]) costo = (1.0*(1-iou)) + (0.3*(dist/radio)) + ((max(a_trk,a_det)/(min(a_trk,a_det)+1e-6)-1)*0.1) costo += (abs((trk.box[3]-trk.box[1])/(trk.box[2]-trk.box[0]+1e-6) - (det[3]-det[1])/(det[2]-det[0]+1e-6))*0.1) costo += min(0.3, dt_oculto*0.1) if ref_firma is not None and obtener_firma(d) is not None: sim_ap = similitud_hibrida(ref_firma, firmas[d]) if sim_ap < 0.25: costo += 50.0 else: costo += (1.0 - sim_ap) * 2.0 nombre_vip = self.global_mem.db.get(trk.gid, {}).get('nombre') umbral_ap = 0.55 if nombre_vip else 0.40 if sim_ap < umbral_ap: costo += (umbral_ap - sim_ap) * 3.0 C[t,idx] = costo for r,c in zip(*linear_sum_assignment(C)): if C[r,c] <= 6.5: matched.append((r, alta[c][0])); sin_match_trk.remove(r) if sin_match_trk and baja: C2 = np.ones((len(sin_match_trk), len(baja)), np.float32) for ti, trk_i in enumerate(sin_match_trk): for dj, (d,det) in enumerate(baja): C2[ti,dj] = 1.0 - iou_overlap(self.trackers[trk_i].box, det) nuevos_sin = [] for r,c in zip(*linear_sum_assignment(C2)): if C2[r,c] < 0.72: matched.append((sin_match_trk[r], baja[c][0])) else: nuevos_sin.append(sin_match_trk[r]) sin_match_trk = nuevos_sin return matched, [d for d in range(n_det) if d not in [m[1] for m in matched] and confidences[d] >= 0.52], sin_match_trk, firmas def update(self, boxes, confidences, keypoints, frame_show, frame_hd, now, turno_activo): # ───────────────────────────────────────────────────────────── # 1. Resolver fusiones globales # ───────────────────────────────────────────────────────────── for trk in self.trackers: if trk.gid: with self.global_mem.lock: d = self.global_mem.db.get(trk.gid) if d and 'fusionado_con' in d: trk.gid = d['fusionado_con'] # ───────────────────────────────────────────────────────────── # 2. Predict y filtrar tracks muertos por Kalman # ───────────────────────────────────────────────────────────── self.trackers = [t for t in self.trackers if t.predict(turno_activo) is not None] if not turno_activo: return self.trackers # ───────────────────────────────────────────────────────────── # 3. Asignar detecciones a tracks existentes # ───────────────────────────────────────────────────────────── matched, new_dets, _, firmas = self._asignar(boxes, confidences, frame_hd, now, keypoints) active_gids = {t.gid for t in self.trackers if t.gid is not None} fh, fw = frame_hd.shape[:2] # ───────────────────────────────────────────────────────────── # 4. DETECCIÓN DE CRUCES + Registro para recuperación post-cruce # ───────────────────────────────────────────────────────────── tracks_en_cruce_locals = set() for i in range(len(self.trackers)): for j in range(i + 1, len(self.trackers)): trk_i, trk_j = self.trackers[i], self.trackers[j] if trk_i.gid is None or trk_j.gid is None or trk_i.box is None or trk_j.box is None: continue iou = iou_overlap(trk_i.box, trk_j.box) if iou > 0.30: tracks_en_cruce_locals.add(trk_i.local_id) tracks_en_cruce_locals.add(trk_j.local_id) if not hasattr(self, 'tracks_post_cruce'): self.tracks_post_cruce = {} self.tracks_post_cruce[trk_i.local_id] = {'gid_antes': trk_i.gid, 'ts_cruce': now} self.tracks_post_cruce[trk_j.local_id] = {'gid_antes': trk_j.gid, 'ts_cruce': now} # ───────────────────────────────────────────────────────────── # 5. PROCESAR MATCHES # ───────────────────────────────────────────────────────────── for t_idx, d_idx in matched: trk, box = self.trackers[t_idx], boxes[d_idx] # Asegurar firma visual REAL (mitigación del bug de contaminación en _asignar) if firmas[d_idx] is None: kpts = keypoints[d_idx] if keypoints and d_idx < len(keypoints) else None firmas[d_idx] = extraer_firma_hibrida(frame_hd, box, kpts) firma_det = firmas[d_idx] # Debug if trk.gid is None: print(f" [DEBUG] LOCAL Trk {trk.local_id}: listo={trk.listo_para_id}, cap_locales={getattr(trk, 'muestras_capturadas', 0)}") else: with self.global_mem.lock: db_act = self.global_mem.db.get(trk.gid, {}).get('actualizaciones_globales', 0) print(f" [MEMORIA] ID {trk.gid} ha sido actualizado {db_act} veces en la red global.") # Detección de grupo es_grupo = self._detectar_grupo(trk, box, self.trackers) if not trk.en_grupo and es_grupo: with self.global_mem.lock: trk.firma_pre_grupo = self.global_mem.db.get(trk.gid, {}).get('ema') trk.ts_salio_grupo = 0.0 elif trk.en_grupo and not es_grupo: trk.ts_salio_grupo = now trk.en_grupo = es_grupo # Update Kalman trk.update(box, es_grupo, now, kpts=(keypoints[d_idx] if keypoints else None), firma_actual=firma_det) # Geometría de frame margen_x = fw * 0.05 esta_en_centro = (box[0] > margen_x and box[2] < (fw - margen_x)) # ───────────────────────────────────────────────────────── # 5a. APRENDIZAJE EMA (zona segura, centro del frame) # ───────────────────────────────────────────────────────── if firma_det is not None and not es_grupo and esta_en_centro: trk.actualizar_ema(firma_det) # Nutrir memoria global cada 2 segundos if trk.gid is not None: if not hasattr(trk, 'ultimo_update_global'): trk.ultimo_update_global = 0 if (now - trk.ultimo_update_global) > 2.0: with self.global_mem.lock: if trk.gid in self.global_mem.db: self.global_mem.db[trk.gid]['ema'] = trk.firma_ema.copy() self.global_mem.db[trk.gid]['ts'] = now self.global_mem.db[trk.gid]['actualizaciones_globales'] = \ self.global_mem.db[trk.gid].get('actualizaciones_globales', 0) + 1 trk.ultimo_update_global = now # ⚡ INYECCIÓN EXACTA AQUÍ: Disparar el Auto-Merge self.global_mem.consolidar_clones() # ───────────────────────────────────────────────────────── # 5b. RESCATE DE BORDE (Acelerado) # ───────────────────────────────────────────────────────── elif firma_det is not None and not es_grupo and not esta_en_centro: muestras_act = getattr(trk, 'muestras_capturadas', 0) if muestras_act < 3: trk.actualizar_ema(firma_det) elif muestras_act < 5 and trk.frames_observados % 3 == 0: trk.actualizar_ema(firma_det) # ───────────────────────────────────────────────────────── # 5c. IDENTIFICACIÓN CON PROTECCIÓN # ───────────────────────────────────────────────────────── if trk.gid is None and trk.listo_para_id and firma_det is not None: if getattr(trk, 'muestras_capturadas', 0) >= 4: if trk.local_id in tracks_en_cruce_locals: print(f" [CRUCE] Track {trk.local_id} en cruce físico. Bloqueando identificación.") else: tiempo_enfriamiento = now - getattr(trk, 'ultimo_cambio_id', 0) if tiempo_enfriamiento < 0.6: print(f" [HYSTERESIS] Track {trk.local_id} en enfriamiento ({tiempo_enfriamiento:.2f}s)") else: gid_c, es_reid = self.global_mem.identificar_candidato( firma_det, self.cam_id, now, active_gids, en_borde=not esta_en_centro, firma_ema_local=trk.firma_ema, confianza_fisica=trk.confianza_fisica ) if gid_c: active_gids.add(gid_c) trk.gid = gid_c trk.origen_global = es_reid trk.ultimo_cambio_id = now # ───────────────────────────────────────────────────────────── # 6. VERIFICACIÓN POST-CRUCE (recuperar ID original tras separarse) # ───────────────────────────────────────────────────────────── if hasattr(self, 'tracks_post_cruce') and self.tracks_post_cruce: tracks_post_cruce_nuevos = {} # ⚡ CAMBIO CRÍTICO: Iterar por local_id, no por índice for local_id, info in list(self.tracks_post_cruce.items()): # Buscar el track con este local_id (único e inmutable) trk = next((t for t in self.trackers if t.local_id == local_id), None) if trk is None: continue # El track ya murió, ignorar # Solo actuar si ya pasaron >2 segundos desde el cruce if now - info['ts_cruce'] > 2.0: # Si el ID cambió durante el cruce (o se asignó uno nuevo incorrecto) if trk.gid is not None and trk.gid != info['gid_antes']: if trk.firma_ema is not None and info['gid_antes'] in self.global_mem.db: ema_original = self.global_mem.db[info['gid_antes']].get('ema') if ema_original is not None: sim_recuperacion = similitud_hibrida( trk.firma_ema, ema_original, cross_cam=False ) if sim_recuperacion > 0.85: # Verificar que el ID original no esté activo en otro track if info['gid_antes'] not in active_gids: print(f" [POST-CRUCE] Track {trk.local_id} recuperó ID {info['gid_antes']} (sim={sim_recuperacion:.2f})") active_gids.discard(trk.gid) trk.gid = info['gid_antes'] active_gids.add(trk.gid) # No guardamos este registro más (ya cumplió su ciclo) continue # Aún no pasan 2 segundos, mantener registro (usando local_id) tracks_post_cruce_nuevos[local_id] = info self.tracks_post_cruce = tracks_post_cruce_nuevos # ───────────────────────────────────────────────────────────── # 7. CREAR NUEVOS TRACKERS para detecciones sin match # ───────────────────────────────────────────────────────────── for d_idx in new_dets: nt = KalmanTrack(boxes[d_idx], now) f = firmas[d_idx] if f: nt.actualizar_ema(f) self.trackers.append(nt) # ───────────────────────────────────────────────────────────── # 8. LIMPIEZA: matar tracks sin detección por >3 segundos # ───────────────────────────────────────────────────────────── tracks_vivos = [] for t in self.trackers: if (now - t.ts_ultima_deteccion) < 3.0: tracks_vivos.append(t) self.trackers = tracks_vivos # ⚡ FIX P0: Purga de memory leak en tracks_post_cruce if hasattr(self, 'tracks_post_cruce'): local_ids_vivos = {t.local_id for t in self.trackers} self.tracks_post_cruce = {lid: info for lid, info in self.tracks_post_cruce.items() if lid in local_ids_vivos} # ───────────────────────────────────────────────────────────── # 9. Gestión post-grupo (recuperación de firma tras salir de multitud) # ───────────────────────────────────────────────────────────── self._gestionar_post_grupo(now, frame_hd) return self.trackers class CamStream: def __init__(self, url): self.url = url; self.cap = cv2.VideoCapture(url) self.q = queue.Queue(maxsize=1); self.stopped = False threading.Thread(target=self._run, daemon=True).start() def _run(self): while not self.stopped: ret, f = self.cap.read() if not ret: time.sleep(2); self.cap.open(self.url); continue if self.q.full(): try: self.q.get_nowait() except queue.Empty: pass self.q.put(f) @property def frame(self): try: return self.q.get(timeout=0.5) except queue.Empty: return None def stop(self): self.stopped = True; self.cap.release() def dibujar_track(frame_show, trk): try: x1,y1,x2,y2 = map(int,trk.box) except: return if trk.gid is None: color,label = C_CANDIDATO, f"?{trk.local_id}" elif getattr(trk,'en_grupo',False): color,label = C_GRUPO, f"ID:{trk.gid}[G]" elif getattr(trk,'aprendiendo',False): color,label = C_APRENDIZAJE, f"ID:{trk.gid}[+]" elif getattr(trk,'origen_global',False): color,label = C_GLOBAL, f"ID:{trk.gid}[R]" 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) # ────────────────────────────────────────────────────────────────────────────── # MAIN # ────────────────────────────────────────────────────────────────────────────── def main(): print("\n=== Tracker Autónomo Definitivo ===") print("Persistencia activa. Los IDs se recuerdan entre reinicios.") print("Tecla [q] para salir (la memoria se guarda automáticamente).\n") model = YOLO("yolov8n-pose.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 ultimo_guardado = time.time() try: while True: now, tiles = time.time(), [] 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, confs, kpts = [], [], [] 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: raw_boxes = res[0].boxes.xyxy.cpu().numpy().tolist() raw_confs = res[0].boxes.conf.cpu().numpy().tolist() raw_kpts = res[0].keypoints.data.cpu().numpy().tolist() if res[0].keypoints is not None else [] for d_idx, (box, conf) in enumerate(zip(raw_boxes, raw_confs)): kpt_persona = raw_kpts[d_idx] if d_idx < len(raw_kpts) else None if es_humano_valido(kpt_persona, box, conf): boxes.append(box) confs.append(conf) kpts.append(kpt_persona) ESQUELETO = [(0,1),(0,2),(1,3),(2,4),(5,6),(5,7),(7,9),(6,8),(8,10),(5,11),(6,12),(11,12),(11,13),(13,15),(12,14),(14,16)] for pk in kpts: if pk is None: continue for a,b in ESQUELETO: if pk[a][2]>0.40 and pk[b][2]>0.40: cv2.line(frame_show, (int(pk[a][0]),int(pk[a][1])), (int(pk[b][0]),int(pk[b][1])), (255,0,255), 2) for kx, ky, kc in pk: if kc > 0.40: cv2.circle(frame_show, (int(kx), int(ky)), 3, (0, 255, 255), -1) tracks = managers[cid].update(boxes, confs, kpts, 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 getattr(t,'gid',None) 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[:3]),np.hstack(tiles[3:])])) idx += 1 if now - ultimo_guardado > 30: global_mem.guardar_memoria() ultimo_guardado = now if cv2.waitKey(1) == ord('q'): break finally: global_mem.guardar_memoria() print("[Memoria] Guardada antes de salir.") cv2.destroyAllWindows() if __name__ == "__main__": main() """