""" Tracker Multi-Cámara Definitivo (SmartSoft IA) Características: 1. OSNet dominante con EMA fluida. 2. Color LAB con Veto Cruzado Letal (anti-robos de identidad). 3. Huella ANATÓMICA por keypoints (anti-uniformes). 4. Sistema Inmune (Hard Negatives): Aprende de tus correcciones (tecla 'n'). 5. Memoria Persistente: No olvida las correcciones al reiniciar. """ 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 # ────────────────────────────────────────────────────────────────────────────── # 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_dinamico_int8.onnx" VECINOS = { "1": ["7"], "7": ["1", "5"], "5": ["7", "8"], "8": ["5", "3"], "3": ["8", "6"], "6": ["3"] } # ⚡ TIEMPOS REALES: Bloqueo estricto de teletransportaciones TIEMPO_MIN_POR_DISTANCIA = { 0: 0.0, 1: 1.0, # Vecino directo (casi instantáneo) 2: 15.0, # Un salto intermedio -> Mínimo 15 seg. 3: 40.0, # Dos o más saltos -> Mínimo 40 seg. } TIEMPO_MAX_AUSENCIA = 800.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) C_FEEDBACK = (255, 0, 255) 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) # ────────────────────────────────────────────────────────────────────────────── # HUELLA ANATÓMICA (Anti-uniformes) # ────────────────────────────────────────────────────────────────────────────── 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)} nariz_ok = puntos[0][2] > 0.30 hombro_izq = puntos[5][2] > 0.30 hombro_der = puntos[6][2] > 0.30 cadera_izq = puntos[11][2] > 0.30 cadera_der = puntos[12][2] > 0.30 if not nariz_ok or (not hombro_izq and not hombro_der) or (not cadera_izq and not cadera_der): return None puntos_validos = sum(1 for p in puntos.values() if p[2] > 0.30) if puntos_validos < 7: 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_izq = dist(5, 11) torso_der = 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 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 altura_approx = max(torso_izq, torso_der) + max(pierna_izq, pierna_der) if altura_approx < 10.0: return None feats = np.array([ hombros / altura_approx, caderas / altura_approx, max(torso_izq, torso_der) / altura_approx, max(pierna_izq, pierna_der) / altura_approx, max(brazo_izq, brazo_der) / altura_approx, hombros / max(caderas, 1e-6), max(pierna_izq, pierna_der) / max(max(torso_izq, torso_der), 1e-6), (brazo_izq + brazo_der) / max(max(torso_izq, torso_der) * 2, 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 # ────────────────────────────────────────────────────────────────────────────── # EXTRACCIÓN DE FIRMAS # ────────────────────────────────────────────────────────────────────────────── def analizar_calidad(box, kpts=None, estricto=False): x1, y1, x2, y2 = box w, h = x2-x1, y2-y1 if w <= 0 or h <= 0: return False ratio = h / float(w) puntos_buenos = sum(1 for kx, ky, conf in kpts if conf > 0.40) if kpts is not None else 0 if puntos_buenos >= 4: ratio_min = 0.60 else: ratio_min = 1.25 if estricto else 1.20 ratio_max = 4.0 if estricto else 5.0 area_min = 800 if estricto else 400 return (ratio_min < ratio < ratio_max) and (w*h > area_min) 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) def extraer_color_zonas(img): if img is None or img.size == 0 or img.shape[0] < 5 or img.shape[1] < 5: return np.zeros(512 * 3, dtype=np.float32) h, w = img.shape[:2] t1, t2 = int(h * 0.15), int(h * 0.55) m_w = int(w * 0.20) core_img = img[:, m_w:(w - m_w)] if core_img.size == 0 or core_img.shape[1] < 4: core_img = img lab = cv2.cvtColor(core_img, cv2.COLOR_BGR2LAB) l, a, b = cv2.split(lab) if l.shape[0] >= 8 and l.shape[1] >= 8: l = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8,8)).apply(l) lab_eq = cv2.merge((l, a, b)) def hzone(z): if z is None or z.size == 0 or z.shape[0] < 2 or z.shape[1] < 2: return np.zeros(512, dtype=np.float32) hist = cv2.calcHist([z], [0,1,2], None, [8,8,8], [0,256, 0,256, 0,256]) cv2.normalize(hist, hist, alpha=1.0, norm_type=cv2.NORM_L1) return hist.flatten() return np.concatenate([hzone(lab_eq[:t1,:]), hzone(lab_eq[t1:t2,:]), hzone(lab_eq[t2:,:])]) def calidad_real(roi_shape, x1_c, y1_c, x2_c, y2_c, w_hd, h_hd): area = (x2_c - x1_c) * (y2_c - y1_c) ratio = roi_shape[0] / max(roi_shape[1], 1) factor_ratio = 1.0 if 1.8 < ratio < 3.2 else 0.5 en_borde = (x1_c < 20 or y1_c < 20 or x2_c > w_hd-20 or y2_c > h_hd-20) factor_borde = 0.3 if en_borde else 1.0 return area * factor_ratio * factor_borde def extraer_firma_hibrida(frame_hd, box_480, kpts=None): try: h_hd, w_hd = frame_hd.shape[:2] x1, y1, x2, y2 = box_480 w, h = x2-x1, y2-y1 px, py = w*0.08, h*0.05 x1_c = max(0, int((x1-px)*(w_hd/480.0))) y1_c = max(0, int((y1-py)*(h_hd/270.0))) x2_c = min(w_hd, int((x2+px)*(w_hd/480.0))) y2_c = min(h_hd, int((y2+py)*(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 cal = calidad_real(roi.shape, x1_c, y1_c, x2_c, y2_c, w_hd, h_hd) blob = preprocess_onnx(roi) deep_f = ort_session.run(None, {input_name: blob})[0][0].flatten() n = np.linalg.norm(deep_f) if n > 0: deep_f /= n color_f = extraer_color_zonas(roi) anat_f = extraer_proporciones_anatomicas(kpts) return { 'deep': deep_f, 'color': color_f, 'anatomia': anat_f, 'ratio_hw': roi.shape[0] / max(roi.shape[1], 1), 'calidad': cal, } except Exception as e: print(f"[Firma Error] {e}") return None def similitud_hibrida(f1, f2, cross_cam=False, confianza_fisica=0.0): if f1 is None or f2 is None: return 0.0 sim_deep = max(0.0, 1.0 - cosine(f1['deep'], f2['deep'])) sim_head = sim_torso = sim_legs = 1.0 if f1['color'].shape == f2['color'].shape and f1['color'].size > 1: L = len(f1['color'])//3 def hcmp(a,b): if np.sum(a) == 0 or np.sum(b) == 0: return 1.0 dist = cv2.compareHist(a.astype(np.float32), b.astype(np.float32), cv2.HISTCMP_BHATTACHARYYA) return max(0.0, 1.0 - float(dist)) sim_head = hcmp(f1['color'][:L], f2['color'][:L]) sim_torso = hcmp(f1['color'][L:2*L], f2['color'][L:2*L]) sim_legs = hcmp(f1['color'][2*L:], f2['color'][2*L:]) sim_color = (sim_head * 0.15) + (sim_torso * 0.50) + (sim_legs * 0.35) sim_anat = 1.0 if 'anatomia' in f1 and 'anatomia' in f2: a1, a2 = f1['anatomia'], f2['anatomia'] if a1 is not None and a2 is not None: sim_anat = max(0.0, float(np.dot(a1, a2))) penal_sil = 0.0 if 'ratio_hw' in f1 and 'ratio_hw' in f2: diff = abs(f1['ratio_hw'] - f2['ratio_hw']) tol = 0.70 if cross_cam else (0.40 + confianza_fisica * 0.35) if diff > tol: penal_sil = min(0.20, (diff-tol) * 0.25) * (1.0 - confianza_fisica) if cross_cam: # ⚡ VETO CRUZADO LETAL: Evita que alguien de claro le robe el ID a alguien oscuro veto_cruzado = 0.0 msg_veto = "" if sim_torso < 0.20: veto_cruzado = 0.60 msg_veto = "[VETO LETAL COLOR -0.60]" elif sim_torso < 0.38: # ⚡ Subimos el límite para atrapar el 0.34 de tu log veto_cruzado = 0.35 msg_veto = "[VETO FUERTE COLOR -0.35]" elif sim_torso < 0.48: # ⚡ Nuevo nivel de advertencia veto_cruzado = 0.15 msg_veto = "[VETO LEVE COLOR -0.15]" # Veto Biométrico Anti-Uniformes if sim_anat < 0.70: veto_cruzado += 0.20 msg_veto += " [VETO BIOMÉTRICO]" resultado = max(0.0, sim_deep - veto_cruzado - penal_sil) if sim_deep > 0.40: print(f" ↳ [CROSS-CAM] Deep:{sim_deep:.2f} Anat:{sim_anat:.2f} Col_Torso:{sim_torso:.2f} {msg_veto} Sil:{-penal_sil:.2f} => FINAL:{resultado:.2f}") else: resultado = max(0.0, (sim_deep * 0.80) + (sim_color * 0.10) + (sim_anat * 0.10) - penal_sil) return resultado # ────────────────────────────────────────────────────────────────────────────── # 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, self.en_grupo, self.frames_buena_calidad = 0, False, 0 self.listo_para_id = False self.firma_pre_grupo, self.ts_salio_grupo, self.fallos_post_grupo = None, 0.0, 0 self.ultimo_aprendizaje, self.frames_continuos, self.frames_observados = 0.0, 0, 0 self.firma_ema = None 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 actualizar_ema(self, firma, alpha=0.82): if firma is None: return if self.firma_ema is None: self.firma_ema = {k: v.copy() if isinstance(v, np.ndarray) else v for k, v in firma.items()} if 'anatomia' in self.firma_ema and self.firma_ema['anatomia'] is not None: self.firma_ema['anatomia'] = self.firma_ema['anatomia'].copy() return for key in ['deep','color']: if key in firma: self.firma_ema[key] = alpha*self.firma_ema[key] + (1-alpha)*firma[key] n = np.linalg.norm(self.firma_ema[key]) if n > 0: self.firma_ema[key] /= n self.firma_ema['ratio_hw'] = alpha*self.firma_ema['ratio_hw'] + (1-alpha)*firma['ratio_hw'] if 'anatomia' in firma and firma['anatomia'] is not None: if 'anatomia' not in self.firma_ema or self.firma_ema['anatomia'] is None: self.firma_ema['anatomia'] = firma['anatomia'].copy() else: self.firma_ema['anatomia'] = alpha*self.firma_ema['anatomia'] + (1-alpha)*firma['anatomia'] n = np.linalg.norm(self.firma_ema['anatomia']) if n > 0: self.firma_ema['anatomia'] /= n def predict(self, turno_activo=True): if self.time_since_update > 4: 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 = 0 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): 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 if analizar_calidad(box, kpts=kpts, estricto=False) and not en_grupo: self.frames_buena_calidad += 1 if self.frames_buena_calidad >= 3: self.listo_para_id = True elif self.gid is None: self.frames_buena_calidad = max(0, self.frames_buena_calidad-1) # ────────────────────────────────────────────────────────────────────────────── # MEMORIA GLOBAL # ────────────────────────────────────────────────────────────────────────────── class GlobalMemory: def __init__(self): self.db = {} self.next_gid = 100 self.lock = threading.RLock() self.ruta_memoria = os.path.join("cache_nombres", "memoria_ia.json") os.makedirs("cache_nombres", exist_ok=True) self._cargar_memoria_inmune() def _guardar_memoria_inmune(self): """Guarda los anticuerpos (Hard Negatives) en el disco""" datos = {} for gid, data in self.db.items(): if data.get('hard_negatives'): datos[str(gid)] = [hn.tolist() for hn in data['hard_negatives']] try: with open(self.ruta_memoria, 'w') as f: json.dump(datos, f) except Exception as e: print(f"[Persistencia] Error guardando: {e}") def _cargar_memoria_inmune(self): """Carga los anticuerpos al iniciar el sistema""" if os.path.exists(self.ruta_memoria): try: with open(self.ruta_memoria, 'r') as f: datos = json.load(f) for gid_str, hn_list in datos.items(): gid = int(gid_str) if gid not in self.db: self.db[gid] = {'ema': None, 'last_cam': '1', 'ts': time.time(), 'hard_negatives': []} self.db[gid]['hard_negatives'] = [np.array(hn, dtype=np.float32) for hn in hn_list] print(f"[Persistencia] Memoria Inmune cargada para {len(datos)} IDs.") except Exception as e: print(f"[Persistencia] Error cargando: {e}") def _actualizar_sin_lock(self, gid, firma, cam_id, now): ALPHA = 0.75 if gid not in self.db: self.db[gid] = { 'ema': firma, 'last_cam': cam_id, 'ts': now, 'nombre': None, 'hard_negatives': [] # ⚡ Lista negra de firmas usurpadoras } return rec = self.db[gid] ema = rec.get('ema') if ema is None: rec['ema'] = firma else: for key in ['deep','color']: if key in firma: ema[key] = ALPHA*ema[key] + (1-ALPHA)*firma[key] n = np.linalg.norm(ema[key]) if n > 0: ema[key] /= n if 'ratio_hw' in firma and 'ratio_hw' in ema: ema['ratio_hw'] = ALPHA*ema['ratio_hw'] + (1-ALPHA)*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'] = firma['anatomia'].copy() else: ema['anatomia'] = ALPHA*ema['anatomia'] + (1-ALPHA)*firma['anatomia'] n = np.linalg.norm(ema['anatomia']) if n > 0: ema['anatomia'] /= n ema['calidad'] = firma['calidad'] rec['last_cam'] = cam_id rec['ts'] = now 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 return similitud_hibrida(firma_nueva, ema, cross_cam, confianza_fisica) 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(): 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 dt = now - data.get('ts', now) # ⚡ Validación Estricta de Tiempo Mínimo (Evita Teletransportes) tiempo_min = TIEMPO_MIN_POR_DISTANCIA.get(distancia, 40.0) if not misma_cam and dt < tiempo_min: continue if dt > TIEMPO_MAX_AUSENCIA: continue if data.get('ema') is None: continue sim = self._sim_robusta(firma, data, es_cross, confianza_fisica) # ⚡ SISTEMA INMUNE: Comparamos contra la Lista Negra del ID anticuerpos = data.get('hard_negatives', []) for anticuerpo_deep in anticuerpos: sim_contra_error = max(0.0, 1.0 - cosine(firma['deep'], anticuerpo_deep)) if sim_contra_error > 0.85: sim -= 0.40 # Veto letal por parecerse al usurpador conocido print(f" [INMUNE] ID {gid} protegido. Usurpador detectado (Sim: {sim_contra_error:.2f})") break 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 penal_multitud = min(0.04, max(0, len(active_gids)-10) * 0.002) if misma_cam: tipo_distancia = "MISMA_CAM" umbral = 0.60 if dt < 5.0 else (0.65 if dt < 30.0 else 0.68) elif es_vecino: tipo_distancia = "VECINO" umbral = 0.60 if dt < 10.0 else 0.75 umbral -= confianza_fisica * 0.05 elif es_salto: tipo_distancia = "SALTO(2)" umbral = 0.82 else: tipo_distancia = "MURO_LEJANO(3+)" # ⚡ LÍNEA FALTANTE umbral = 0.94 piso_seguridad = 0.58 if (es_vecino and dt < 10.0) else 0.65 umbral = max(piso_seguridad, umbral + penal_multitud - (0.03 if data.get('nombre') and not misma_cam else 0.0)) if sim > 0.40 and not misma_cam: estado = "ACEPTADO" if sim > umbral else "RECHAZADO" faltante = umbral - sim if sim <= umbral else 0.0 info_faltante = f"(Faltó {faltante:.2f})" if faltante > 0 else "" print(f" [EVAL] Cam{cam_id} evalúa ID{gid}(Viene de Cam{data['last_cam']} hace {dt:.1f}s) | Tipo: {tipo_distancia}") print(f" -> Sim_Final={sim:.2f} vs Umbral={umbral:.2f} {info_faltante} -> {estado}") 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, cam_id, now) return nid, False candidatos.sort(reverse=True) best_sim, best_gid, best_misma, best_vecino = candidatos[0] self._actualizar_sin_lock(best_gid, firma, cam_id, now) return best_gid, True def limpiar_fantasmas(self): with self.lock: ahora = time.time() muertos = [g for g,d in self.db.items() if (d.get('nombre') is None and ahora-d.get('ts',ahora)>600) or (d.get('nombre') is not None and ahora-d.get('ts',ahora)>900)] for g in muertos: # Solo borrar si no tiene anticuerpos importantes if not self.db[g].get('hard_negatives'): del self.db[g] 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 # ────────────────────────────────────────────────────────────────────────────── # FEEDBACK EN VIVO (Sistema de Aprendizaje) # ────────────────────────────────────────────────────────────────────────────── class FeedbackCorrector: def __init__(self, global_mem, managers): self.global_mem = global_mem self.managers = managers self.modo_fusion = False self.gid_fusion_origen = None def _tracker_bajo_mouse(self, tile_idx, mx, my): if tile_idx < 0 or tile_idx >= len(SECUENCIA): return None, None cid = str(SECUENCIA[tile_idx]) mgr = self.managers.get(cid) if not mgr: return None, None for trk in mgr.trackers: x1, y1, x2, y2 = trk.box if x1 <= mx <= x2 and y1 <= my <= y2: return trk, cid return None, None def procesar_tecla(self, key, mouse_state): tile_idx = mouse_state.get('tile_idx', -1) mx = mouse_state.get('tile_x', 0) my = mouse_state.get('tile_y', 0) if key == ord('n'): # ⚡ CREAR NUEVO ID + INYECTAR ANTICUERPO AL ORIGINAL trk, cid = self._tracker_bajo_mouse(tile_idx, mx, my) if trk and trk.gid is not None and trk.firma_ema is not None: viejo_gid = trk.gid with self.global_mem.lock: # Inyectar el error en la lista negra del ID legítimo if 'hard_negatives' not in self.global_mem.db[viejo_gid]: self.global_mem.db[viejo_gid]['hard_negatives'] = [] self.global_mem.db[viejo_gid]['hard_negatives'].append(trk.firma_ema['deep'].copy()) self.global_mem._guardar_memoria_inmune() # Guardar en disco # Nace el nuevo ID limpio nid = self.global_mem.next_gid self.global_mem.next_gid += 1 self.global_mem._actualizar_sin_lock(nid, trk.firma_ema, cid, time.time()) trk.gid = nid trk.origen_global = False print(f"\n[APRENDIZAJE ACTIVO] ¡Error corregido!") print(f" -> La firma intrusa se inyectó como Anticuerpo en el ID {viejo_gid}.") print(f" -> Nace el ID {nid} de forma independiente.") return True elif key == ord('m'): # FUSIONAR IDs trk, cid = self._tracker_bajo_mouse(tile_idx, mx, my) if trk and trk.gid is not None: if not self.modo_fusion: self.modo_fusion = True self.gid_fusion_origen = trk.gid print(f"\n[FUSIÓN] Seleccionaste ID {trk.gid} como ORIGEN. Ahora clickea el DESTINO y presiona 'm'.") else: gid_destino = trk.gid if gid_destino != self.gid_fusion_origen: with self.global_mem.lock: if gid_destino in self.global_mem.db and self.gid_fusion_origen in self.global_mem.db: origen_firmas = self.global_mem.db[self.gid_fusion_origen].get('ema') if origen_firmas: self.global_mem._actualizar_sin_lock(gid_destino, origen_firmas, cid, time.time()) self.global_mem.db[self.gid_fusion_origen]['fusionado_con'] = gid_destino for m in self.managers.values(): for t in m.trackers: if t.gid == self.gid_fusion_origen: t.gid = gid_destino print(f"\n[FUSIÓN] Éxito: ID {self.gid_fusion_origen} absorbido por ID {gid_destino}") self.modo_fusion = False self.gid_fusion_origen = None return True elif key == ord('c'): if self.modo_fusion: print("\n[FUSIÓN] Modo fusión cancelado.") self.modo_fusion = False self.gid_fusion_origen = None return True return False # ────────────────────────────────────────────────────────────────────────────── # GESTOR LOCAL Y RESTO DEL CÓDIGO # ────────────────────────────────────────────────────────────────────────────── 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, [] 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.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.5*(1-iou)) + (0.5*(dist/radio)) + ((max(a_trk,a_det)/(min(a_trk,a_det)+1e-6)-1)*0.2) + (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.2) + min(0.5, dt_oculto*0.2) 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 = 100.0 if trk.frames_observados >= 5 else costo + 1.2 elif sim_ap < 0.45: costo += (0.45-sim_ap)*(3.5 if trk.frames_observados >= 5 else 1.5) 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): 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'] self.trackers = [t for t in self.trackers if t.predict(turno_activo) is not None] if not turno_activo: return self.trackers 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] for t_idx, d_idx in matched: trk, box = self.trackers[t_idx], boxes[d_idx] 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] es_grupo = self._detectar_grupo(trk, box, self.trackers) if not trk.en_grupo and es_grupo: if trk.gid: 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 kpts_persona = keypoints[d_idx] if keypoints and d_idx < len(keypoints) else None trk.update(box, es_grupo, now, kpts=kpts_persona) if firma_det is not None and not es_grupo and (trk.gid is None or trk.frames_observados < 30 or (box[0]fw*.85 or box[3]>fh*.85)): trk.actualizar_ema(firma_det) if not trk.en_grupo and not ((now - getattr(trk,'ts_salio_grupo',0) < 2.0) and getattr(trk,'ts_salio_grupo',0) > 0): if trk.gid is None and trk.listo_para_id and firma_det is not None and not (box[0]<5 or box[1]<5 or box[2]>fw-5 or box[3]>fh-5): gid_c, es_reid = self.global_mem.identificar_candidato( firma_det, self.cam_id, now, active_gids, en_borde=(box[0]<35 or box[1]<35 or box[2]>fw-35 or box[3]>fh-35), 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 elif trk.gid is not None and now-trk.ultimo_aprendizaje > 1.5 and (box[0]fw*.85 or box[3]>fh*.85) and analizar_calidad(box, kpts=kpts_persona, estricto=True) and firma_det is not None and not (box[0]<15 or box[1]<15 or box[2]>fw-15 or box[3]>fh-15): with self.global_mem.lock: rec = self.global_mem.db.get(trk.gid) if rec and rec.get('ema'): sim_ema = similitud_hibrida(firma_det, rec['ema'], confianza_fisica=trk.confianza_fisica) if sim_ema < 0.28 and trk.confianza_fisica < 0.30: trk.gid = None; trk.listo_para_id = False; trk.frames_buena_calidad = 0; continue if sim_ema > max(0.30, 0.48 - trk.confianza_fisica*0.15): self.global_mem._actualizar_sin_lock(trk.gid, firma_det, self.cam_id, now) trk.ultimo_aprendizaje = now; trk.aprendiendo = True for d_idx in new_dets: nt = KalmanTrack(boxes[d_idx], now) kpts = keypoints[d_idx] if keypoints and d_idx < len(keypoints) else None if firmas[d_idx] is None: firmas[d_idx] = extraer_firma_hibrida(frame_hd, boxes[d_idx], kpts) if firmas[d_idx]: nt.actualizar_ema(firmas[d_idx]) self.trackers.append(nt) self.trackers = [t for t in self.trackers if not (t.gid is None and (now - t.ts_creacion) > 4.0) and (now - t.ts_ultima_deteccion) < ((1.0 if (t.box[0]<20 or t.box[1]<20 or t.box[2]>fw-20 or t.box[3]>fh-20) else 2.5) if t.gid is None else (3.0 if (t.box[0]<20 or t.box[1]<20 or t.box[2]>fw-20 or t.box[3]>fh-20) else 8.0))] 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, seleccionado=False): try: x1,y1,x2,y2 = map(int,trk.box) except: return if seleccionado: color, label = C_FEEDBACK, f"SEL:{trk.gid if trk.gid is not None else trk.local_id}" elif 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--- Iniciando Sistema SmartSoft (Versión Anti-Uniformes e Inmunológica) ---") print("Controles de Aprendizaje en Vivo (Apunta con el Mouse):") print(" [n] - ¡Te equivocaste! Separar este ID y marcarlo como usurpador (Anticuerpo).") print(" [m] - ¡Son el mismo! Fusionar IDs (click origen -> click destino -> 'm').") print(" [c] - Cancelar fusión.") print(" [q] - Salir.\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] feedback = FeedbackCorrector(global_mem, managers) cv2.namedWindow("SmartSoft", cv2.WINDOW_AUTOSIZE) mouse_state = {'x': 0, 'y': 0, 'tile_idx': -1, 'tile_x': 0, 'tile_y': 0} def on_mouse(event, x, y, flags, param): mouse_state['x'] = x; mouse_state['y'] = y col, row = x // 480, y // 270 if 0 <= col < 3 and 0 <= row < 2: mouse_state['tile_idx'] = row * 3 + col mouse_state['tile_x'] = x % 480 mouse_state['tile_y'] = y % 270 else: mouse_state['tile_idx'] = -1 cv2.setMouseCallback("SmartSoft", on_mouse) idx = 0 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 es_humano = False if kpt_persona is not None: p_validos = sum(1 for p in kpt_persona if p[2] > 0.30) nariz_ok = kpt_persona[0][2] > 0.30 hombro_ok = kpt_persona[5][2] > 0.30 or kpt_persona[6][2] > 0.30 cadera_ok = kpt_persona[11][2] > 0.30 or kpt_persona[12][2] > 0.30 if p_validos >= 7 and nariz_ok and hombro_ok and cadera_ok: es_humano = True else: if analizar_calidad(box, kpts=None, estricto=True): es_humano = True if es_humano: 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) sel_trk = None if mouse_state['tile_idx'] == i: for trk in tracks: x1,y1,x2,y2 = trk.box if x1<=mouse_state['tile_x']<=x2 and y1<=mouse_state['tile_y']<=y2: sel_trk = trk; break for trk in tracks: if trk.time_since_update <= 1: dibujar_track(frame_show, trk, seleccionado=(trk is sel_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) if feedback.modo_fusion and mouse_state['tile_idx'] == i: cv2.putText(frame_show, "MODO FUSION", (10, 55), FUENTE, 0.6, (0, 0, 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 key = cv2.waitKey(1) & 0xFF if key in [ord('n'), ord('m'), ord('c')]: feedback.procesar_tecla(key, mouse_state) elif key == ord('q'): break cv2.destroyAllWindows() if __name__ == "__main__": main()