830 lines
41 KiB
Python
830 lines
41 KiB
Python
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
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# 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()
|
|
|
|
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.61
|
|
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.66
|
|
|
|
|
|
"""
|
|
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 = 1.0 if estado_actual else 1.0
|
|
factor_y = 0.4 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, trk.area_referencia = candidato_gid, True, area_actual
|
|
print(f" [ADUANA] ID {candidato_gid} validado.")
|
|
|
|
# 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() |