Agregar pipeline de análisis de video

Submuestrea a 5 fps (tope 60 s), detecta personas con RF-DETR, les da
un ID con ByteTrack, marca intrusos por la zona y escribe el video
anotado en H.264 con PyAV para que se reproduzca en el navegador.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app.py

@@ -3,9 +3,23 @@
 Todo corre local, en CPU. El video nunca sale de la máquina.
 """
 
+import tempfile
+from fractions import Fraction
+
+import av
+import cv2
+import gradio as gr
 import numpy as np
+import supervision as sv
 
+FPS = 5  # cuadros analizados por segundo
+MAX_SECONDS = 60  # tope de video analizado
+CONF = 0.4  # confianza mínima para contar una detección
 MAX_GAP_S = 1.0  # si alguien "desaparece" menos que esto, sigue siendo el mismo evento
+PERSON_ID = 1  # id COCO de "person" en RF-DETR
+# ponytail: CPU-only (~0.3 s/cuadro); si es lento, exportar RF-DETR a ONNX + onnxruntime-directml
+
+model = None  # se carga en main(), así los tests no bajan el modelo
 
 
 def feet_in_zone(feet_xy, zone):
@@ -30,3 +44,81 @@ def build_events(frames_in_zone, fps, max_gap_s=MAX_GAP_S):
             rows.append([tid, round(enter, 1), round(leave, 1), round(leave - enter, 1)])
             start = prev = f
     return sorted(rows, key=lambda r: r[1])
+
+
+def zone_from_editor(editor, shape):
+    layers = (editor or {}).get("layers") or []
+    mask = np.any([layer[..., 3] > 0 for layer in layers], axis=0) if layers else None
+    if mask is None or not mask.any():
+        raise gr.Error("Pintá la zona restringida sobre la imagen antes de analizar.")
+    if mask.shape != shape:
+        mask = cv2.resize(mask.astype(np.uint8), (shape[1], shape[0]), interpolation=cv2.INTER_NEAREST) > 0
+    return mask
+
+
+def first_frame(video_path):
+    ok, frame = cv2.VideoCapture(video_path).read()
+    if not ok:
+        raise gr.Error("No pude leer el video.")
+    return {"background": frame[:, :, ::-1], "layers": [], "composite": None}
+
+
+def draw(frame, det, inside, zone):
+    out = frame.copy()
+    out[zone] = (0.6 * out[zone] + 0.4 * np.array([0, 0, 255])).astype(np.uint8)  # rojo (BGR)
+    for mask, color, text in ((~inside, sv.Color.GREEN, "persona #{}"), (inside, sv.Color.RED, "INTRUSO #{}")):
+        d = det[mask]
+        out = sv.BoxAnnotator(color=color, thickness=3).annotate(out, d)
+        out = sv.LabelAnnotator(color=color).annotate(out, d, labels=[text.format(t) for t in d.tracker_id])
+    return out
+
+
+def analyze(video_path, editor, progress=gr.Progress()):
+    from trackers import ByteTrackTracker
+
+    if not video_path:
+        raise gr.Error("Primero subí un video.")
+    cap = cv2.VideoCapture(video_path)
+    src_fps = cap.get(cv2.CAP_PROP_FPS) or 30
+    step = max(1, round(src_fps / FPS))
+    fps = src_fps / step
+    total = int(min(cap.get(cv2.CAP_PROP_FRAME_COUNT) or 1e9, MAX_SECONDS * src_fps))
+    w, h = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) // 2 * 2, int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) // 2 * 2
+    zone = zone_from_editor(editor, (int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)), int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))))
+
+    tracker = ByteTrackTracker(frame_rate=fps, track_activation_threshold=CONF, minimum_consecutive_frames=1)
+    out_path = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False).name
+    container = av.open(out_path, "w")
+    stream = container.add_stream("libx264", rate=Fraction(fps).limit_denominator(1000))
+    stream.width, stream.height, stream.pix_fmt = w, h, "yuv420p"
+
+    in_zone, snapshots, i = {}, [], 0
+    for idx in progress.tqdm(range(total), desc="Analizando"):
+        if idx % step:
+            if not cap.grab():
+                break
+            continue
+        ok, frame = cap.read()
+        if not ok:
+            break
+        det = model.predict(frame[:, :, ::-1].copy(), threshold=CONF)
+        det = tracker.update(det[det.class_id == PERSON_ID])
+        det = det[det.tracker_id != -1]
+        inside = feet_in_zone(det.get_anchors_coordinates(sv.Position.BOTTOM_CENTER), zone)
+        out = draw(frame, det, inside, zone)
+        for tid in det.tracker_id[inside].tolist():
+            if tid not in in_zone:
+                snapshots.append((out[:, :, ::-1], f"Intruso #{tid} — {i / fps:.1f} s"))
+            in_zone.setdefault(tid, []).append(i)
+        for packet in stream.encode(av.VideoFrame.from_ndarray(out[:h, :w], format="bgr24")):
+            container.mux(packet)
+        i += 1
+    for packet in stream.encode():
+        container.mux(packet)
+    container.close()
+    cap.release()
+
+    rows = build_events(in_zone, fps)
+    if not rows:
+        gr.Info("No hubo intrusiones en la zona.")
+    return out_path, rows, snapshots