Mostrar a los intrusos en una tabla con su foto a medida que se analizan

La página es para jefes no técnicos: se explica sola, sin instrucciones de uso.
Los conceptos (qué es intruso, cuadros por segundo, límites, que el video va a
Hugging Face) quedan en "Sobre esta demo", plegado.

- Una sola tabla con la foto recortada de cada intruso en vez de tabla + galería
de cuadros enteros, que parecían todos la misma foto.
- analyze es un generador: las filas aparecen mientras se analiza; salida y
duración se completan al final.
- El video aparece abajo y recién al terminar: vacío parecía roto.
- Un solo indicador de avance, en el botón. Las barras de Gradio iban una por
resultado, y la de la tabla tapaba las filas nuevas.

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

camguard/pipeline.py

@@ -1,12 +1,13 @@
 """El análisis completo de un video: lee, detecta, decide quién es intruso, dibuja y resume.
 
-No depende de la interfaz: informa errores con ``ValueError`` y el progreso con un callback,
-así se puede usar desde Gradio, desde un script o desde los tests (con un tracker falso).
+No depende de la interfaz: informa errores con ``ValueError`` y va devolviendo resultados
+parciales a medida que avanza, así se puede usar desde Gradio, desde un script o desde los tests
+(con un tracker falso).
 """
 
 import math
 import tempfile
-from collections.abc import Callable
+from collections.abc import Callable, Iterator
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Protocol
@@ -30,12 +31,16 @@ class Tracker(Protocol):
         ...
 
 
+# Aire que se deja alrededor de una persona al recortarla, como fracción de su alto y ancho.
+CROP_MARGIN = 0.1
+
+
 @dataclass(frozen=True)
 class Snapshot:
-    """Captura del momento en que una persona aparece por primera vez dentro de la zona.
+    """Foto de una persona en el momento en que aparece por primera vez dentro de la zona.
 
     Attributes:
-        image: Cuadro anotado, en RGB (listo para mostrar).
+        image: La persona recortada del cuadro original (sin dibujos encima), en RGB.
         track_id: Número de seguimiento de la persona.
         time_s: Segundo del video.
     """
@@ -47,41 +52,54 @@ class Snapshot:
 
 @dataclass(frozen=True)
 class AnalysisResult:
-    """Resultado de analizar un video.
+    """Resultado de analizar un video, parcial mientras el análisis avanza.
 
     Attributes:
-        video_path: mp4 anotado (H.264), en una carpeta temporal.
-        events: Cada estadía de una persona en la zona, ordenadas por entrada.
-        snapshots: Una captura por persona: la de su primera entrada.
+        progress: Avance, entre 0 y 1.
+        events: Cada estadía de una persona en la zona, ordenadas por entrada. Mientras se
+            analiza, la salida de quien sigue adentro todavía no es la definitiva.
+        snapshots: Una foto por persona: la de su primera entrada.
+        video_path: mp4 anotado (H.264), en una carpeta temporal. ``None`` hasta terminar.
     """
 
-    video_path: Path
+    progress: float
     events: list[Event]
     snapshots: list[Snapshot]
+    video_path: Path | None = None
+
+
+def crop_person(frame: np.ndarray, box: np.ndarray) -> np.ndarray:
+    """Recorta a una persona del cuadro, con algo de aire alrededor."""
+    x1, y1, x2, y2 = box
+    mx, my = (x2 - x1) * CROP_MARGIN, (y2 - y1) * CROP_MARGIN
+    h, w = frame.shape[:2]
+    top, bottom = max(0, int(y1 - my)), min(h, int(y2 + my))
+    left, right = max(0, int(x1 - mx)), min(w, int(x2 + mx))
+    return frame[top:bottom, left:right]
 
 
 def analyze(
     video_path: str | Path,
     zone_layers: list[np.ndarray],
     make_tracker: Callable[[float], Tracker] = PersonTracker,
-    progress: Callable[[float], None] | None = None,
     out_dir: Path | None = None,
-) -> AnalysisResult:
+) -> Iterator[AnalysisResult]:
     """Analiza un video y detecta quién entra a la zona restringida.
 
     Se analizan ``FPS`` cuadros por segundo y hasta ``MAX_SECONDS`` segundos (ver ``config``).
+    Es un generador: devuelve el resultado parcial después de cada cuadro, para mostrar a los
+    intrusos a medida que aparecen, y al final el completo, con el video.
 
     Args:
         video_path: Video a analizar.
         zone_layers: Capas RGBA pintadas por el usuario (ver :func:`zone_from_layers`).
         make_tracker: Crea el tracker a partir de los fps analizados. Por defecto el real
             (RF-DETR + ByteTrack); los tests pasan uno falso.
-        progress: Se llama después de cada cuadro con el avance, entre 0 y 1.
         out_dir: Dónde guardar el video anotado (en una subcarpeta propia). Por defecto, la
             carpeta temporal del sistema. La app pasa una que vacía al arrancar.
 
-    Returns:
-        El video anotado, los eventos y las capturas.
+    Yields:
+        Los resultados parciales y, último, el completo (con ``video_path``).
 
     Raises:
         ValueError: Si el video no se puede leer o no hay zona pintada.
@@ -109,12 +127,15 @@ def analyze(
             centers = detections.get_anchors_coordinates(sv.Position.CENTER)
             inside = points_in_zone(centers, zone)
             annotated = annotate(frame, detections, inside, zone)
-            for track_id in detections.tracker_id[inside].tolist():
+            for box, track_id in zip(
+                detections.xyxy[inside], detections.tracker_id[inside], strict=True
+            ):
                 if track_id not in frames_in_zone:
-                    snapshots.append(Snapshot(annotated[:, :, ::-1], track_id, i / fps))
-                frames_in_zone.setdefault(track_id, []).append(i)
+                    photo = crop_person(frame, box)[:, :, ::-1]
+                    snapshots.append(Snapshot(photo, int(track_id), i / fps))
+                frames_in_zone.setdefault(int(track_id), []).append(i)
             writer.write(annotated)
-            if progress:
-                progress(min(1.0, (i + 1) / expected))
+            progress = min(1.0, (i + 1) / expected)
+            yield AnalysisResult(progress, build_events(frames_in_zone, fps), list(snapshots))
 
-    return AnalysisResult(out_path, build_events(frames_in_zone, fps), snapshots)
+    yield AnalysisResult(1.0, build_events(frames_in_zone, fps), snapshots, out_path)

camguard/ui.py

@@ -1,19 +1,21 @@
 """La página web (Gradio): subir un video, pintar la zona, analizar y ver el resultado.
 
 Es la única parte que conoce Gradio: traduce los ``ValueError`` del pipeline en avisos para
-el usuario y los resultados en tabla y galería.
+el usuario y los resultados en una tabla (con la foto de cada intruso) y el video anotado.
 """
 
+import base64
 import os
 import shutil
 
+import cv2
 import gradio as gr
 import numpy as np
 import spaces
 
 from camguard.config import FPS, MAX_SECONDS, MAX_UPLOAD, TEMP_DIR
 from camguard.detection import load_model
-from camguard.pipeline import analyze
+from camguard.pipeline import AnalysisResult, analyze
 from camguard.video import first_frame
 
 # Grosor del pincel, en píxeles del video (1920 de ancho en el ejemplo).
@@ -48,31 +50,22 @@ REFIT_EDITOR_JS = """() => {
 # corre entera y el trazo cae fuera de la imagen. Recortar lo que sobresale lo evita.
 EDITOR_CSS = ".pixi-target, .pixi-target-crop { overflow: hidden; }"
 
+# La página se explica sola: esto son conceptos, plegados para quien quiera saber más.
 # Hugging Face define SPACE_ID en todo Space; en la PC no existe.
-WHERE = (
-    "> 🔒 **El video se procesa en servidores de Hugging Face.** No subas grabaciones reales o "
-    "sensibles."
-    if os.getenv("SPACE_ID")
-    else "Todo corre en esta PC."
-)
-
-INSTRUCTIONS = f"""\
-# CamGuard — Intruso en zona
-1. Ya viene cargado un video de ejemplo. Si querés usar uno propio, tocá la ✕ del video y \
-subí el tuyo.
-2. Pintá encima de la imagen de la derecha el área restringida: el pincel ya está activo, \
-arrastrá directamente sobre la imagen. Si te equivocás, tocá **Borrar zona** y volvé a pintar.
-3. Tocá **Analizar**.
-
-Se analizan {FPS} cuadros por segundo y hasta {MAX_SECONDS} s de video.
-
-{WHERE}
-
-> ⚠️ **Es una demo.** Puede no detectar a algunas personas (lejanas, tapadas, agachadas o con \
-poca luz) y no reemplaza la vigilancia humana. Una entrada a los 0,2 s puede significar que la \
-persona ya estaba en la zona cuando empezó el video.
+ABOUT = f"""\
+- Es intruso quien tiene el centro del cuerpo sobre la zona.
+- Se analizan {FPS} cuadros por segundo y hasta {MAX_SECONDS} s de video.
+- Si alguien entró en el segundo 0, ya estaba en la zona cuando empezó el video.
+- Puede no detectar a personas lejanas, tapadas, agachadas o con poca luz: no reemplaza la \
+vigilancia humana.
+{"- El video se procesa en servidores de Hugging Face." if os.getenv("SPACE_ID") else ""}
 """
 
+HEADERS = ["ID", "Entró (s)", "Salió (s)", "Duración (s)", "Foto"]
+
+# Alto con que se ve la foto de cada intruso en la tabla.
+PHOTO_PX = 120
+
 
 def editor_value(background: np.ndarray, layers: list[np.ndarray] | None = None) -> dict:
     """Arma el valor del editor de Gradio: imagen de fondo y capas pintadas."""
@@ -124,69 +117,111 @@ def blank_zone(video_path: str | None):
         return gr.skip()
 
 
+def photo_html(image_rgb: np.ndarray) -> str:
+    """Convierte la foto de un intruso en un <img> embebido, para la columna "Foto"."""
+    _, jpg = cv2.imencode(".jpg", image_rgb[:, :, ::-1], [cv2.IMWRITE_JPEG_QUALITY, 85])
+    b64 = base64.b64encode(jpg).decode()
+    return f'<img src="data:image/jpeg;base64,{b64}" style="height:{PHOTO_PX}px;width:auto">'
+
+
+def table_rows(result: AnalysisResult) -> list[list]:
+    """Una fila por entrada a la zona. Salida y duración quedan vacías hasta terminar."""
+    photos = {s.track_id: photo_html(s.image) for s in result.snapshots}
+    done = result.video_path is not None
+    return [
+        [
+            e.track_id,
+            e.enter_s,
+            e.leave_s if done else "",
+            e.duration_s if done else "",
+            photos[e.track_id],
+        ]
+        for e in result.events
+    ]
+
+
 # En el Space de Hugging Face pide una GPU (ZeroGPU) mientras dura el análisis; en la PC no hace
 # nada. Va sobre on_analyze y no sobre analyze porque ZeroGPU lo corre en otro proceso y convierte
 # cualquier excepción que no sea gr.Error en un "ZeroGPU worker error": acá los ValueError ya
 # se volvieron avisos.
 @spaces.GPU
-def on_analyze(video_path: str | None, editor: dict | None, progress=gr.Progress()):  # noqa: B008
-    """Corre el análisis y lo convierte en lo que muestran los componentes de la página.
+def on_analyze(video_path: str | None, editor: dict | None):
+    """Corre el análisis y va mostrando sus resultados en la página.
 
-    Returns:
-        (video anotado, filas de la tabla de eventos, imágenes de la galería con su leyenda), o
-        ``gr.skip()`` en las tres (deja los resultados como estaban) si falta el video o la zona.
+    Yields:
+        (filas de la tabla, video anotado, texto del botón): al empezar (y se oculta el video
+        anterior), cada vez que entra alguien o avanza un 1%, y al final con el video. Si falta
+        el video o la zona, una sola vez ``gr.skip()`` en los tres: deja todo como estaba.
     """
     if not video_path:
         warn("Primero subí un video.")
-        return (gr.skip(),) * 3
+        yield (gr.skip(),) * 3
+        return
     layers = (editor or {}).get("layers") or []
+    results = analyze(video_path, layers, out_dir=TEMP_DIR / "resultados")
     try:
-        result = analyze(
-            video_path,
-            layers,
-            progress=lambda p: progress(p, desc="Analizando"),
-            out_dir=TEMP_DIR / "resultados",
-        )
+        result = next(results)  # el primer paso valida el video y la zona
     except ValueError as e:  # video ilegible o sin zona pintada
         warn(str(e))
-        return (gr.skip(),) * 3
+        yield (gr.skip(),) * 3
+        return
+
+    yield table_rows(result), gr.Video(value=None, visible=False), gr.skip()
+    shown, label = len(result.events), ""
+    for result in results:
+        if len(result.events) != shown:
+            shown = len(result.events)
+            yield table_rows(result), gr.skip(), gr.skip()
+        text = f"Analizando… {result.progress:.0%}"
+        if text != label:
+            label = text
+            yield gr.skip(), gr.skip(), label
 
     if not result.events:
         gr.Info("No hubo intrusiones en la zona.")
-    rows = [[e.track_id, e.enter_s, e.leave_s, e.duration_s] for e in result.events]
-    gallery = [(s.image, f"Intruso #{s.track_id} — {s.time_s:.1f} s") for s in result.snapshots]
-    return str(result.video_path), rows, gallery
+    yield table_rows(result), gr.Video(value=str(result.video_path), visible=True), gr.skip()
 
 
 def build_demo() -> gr.Blocks:
-    """Arma la página: video y editor lado a lado, el botón, y los tres resultados."""
+    """Arma la página: video y editor lado a lado, el botón, la tabla de intrusos y el video."""
     example_video, example_editor = load_example()
     # analytics_enabled=False: Gradio manda telemetría de uso por defecto (no los videos, pero
     # una app que maneja grabaciones de seguridad no debería hablar con nadie).
     with gr.Blocks(title="CamGuard", analytics_enabled=False, delete_cache=CACHE_CLEANUP) as demo:
-        gr.Markdown(INSTRUCTIONS)
+        gr.Markdown("# CamGuard — Intruso en zona")
+        with gr.Accordion("Sobre esta demo", open=False):
+            gr.Markdown(ABOUT)
         with gr.Row():
             video = gr.Video(value=example_video, label="Video", sources=["upload"])
             with gr.Column():
                 editor = gr.ImageEditor(
                     value=example_editor,
-                    label="Zona restringida (pintá encima)",
+                    label="Zona restringida",
                     type="numpy",
                     sources=[],  # la imagen siempre es el primer cuadro del video
                     brush=gr.Brush(default_size=BRUSH_SIZE, colors=["#ff0000"], color_mode="fixed"),
                 )
                 clear = gr.Button("Borrar zona", size="sm")
         run = gr.Button("Analizar", variant="primary")
-        result = gr.Video(label="Resultado")
         table = gr.Dataframe(
-            headers=["ID", "Entró (s)", "Salió (s)", "Duración (s)"], label="Eventos"
+            headers=HEADERS,
+            datatype=["number", "number", "number", "number", "html"],
+            label="Intrusos",
+            interactive=False,
+            wrap=True,
+            max_height="none",  # la página crece con la tabla, en vez de un scroll adentro
         )
-        gallery = gr.Gallery(label="Momento de entrada de cada intruso", columns=4)
+        # Oculto hasta que termina el análisis: vacío parecía roto.
+        result = gr.Video(label="Video analizado", visible=False)
 
         demo.load(None, js=REFIT_EDITOR_JS)
         video.upload(blank_zone, video, editor).then(None, js=REFIT_EDITOR_JS)
         clear.click(blank_zone, video, editor).then(None, js=REFIT_EDITOR_JS)
-        run.click(on_analyze, [video, editor], [result, table, gallery])
+        # El avance se ve en el botón, que queda deshabilitado mientras analiza. Las barras de
+        # progreso de Gradio iban una por resultado, y la de la tabla tapaba las filas nuevas.
+        run.click(lambda: gr.Button("Analizando…", interactive=False), None, run, queue=False).then(
+            on_analyze, [video, editor], [table, result, run], show_progress="hidden"
+        ).then(lambda: gr.Button("Analizar", interactive=True), None, run, queue=False)
     return demo
 
 

tests/test_pipeline.py

@@ -10,6 +10,12 @@ from camguard.video import H264Writer, probe
 SIZE = 100  # video de 100x100; la zona es el cuarto inferior derecho
 
 
+def last(results):
+    """El resultado final de analyze, que va devolviendo parciales."""
+    *_, result = results
+    return result
+
+
 class FakeTracker:
     """Reemplaza al modelo: una persona (#1) camina afuera, entra en los cuadros 2 a 5 y sale."""
 
@@ -45,15 +51,21 @@ def zone_layers():
 
 
 def test_analyze_detects_the_intrusion(video, zone_layers):
-    progress = []
-    result = analyze(video, zone_layers, make_tracker=FakeTracker, progress=progress.append)
+    *partials, result = analyze(video, zone_layers, make_tracker=FakeTracker)
 
     assert result.events == [Event(track_id=1, enter_s=0.4, leave_s=1.2)]
     assert [(s.track_id, s.time_s) for s in result.snapshots] == [(1, 0.4)]
-    assert result.snapshots[0].image.shape == (SIZE, SIZE, 3)
     assert probe(result.video_path).frame_count == 10
-    assert len(progress) == 10
-    assert progress[-1] == 1.0
+    # Un parcial por cuadro, sin video; el intruso aparece desde el cuadro en que entra.
+    assert [p.progress for p in partials] == [(i + 1) / 10 for i in range(10)]
+    assert all(p.video_path is None for p in partials)
+    assert [len(p.events) for p in partials] == [0, 0] + [1] * 8
+
+
+def test_snapshot_is_the_person_cropped(video, zone_layers):
+    photo = last(analyze(video, zone_layers, make_tracker=FakeTracker)).snapshots[0].image
+    # Recuadro de 10x20 en (70, 55), con 10% de aire: 1 px a los costados, 2 arriba y abajo.
+    assert photo.shape == (24, 12, 3)
 
 
 class StraddlingTracker:
@@ -72,23 +84,23 @@ class StraddlingTracker:
 
 def test_person_counts_by_body_center_not_feet(video, zone_layers):
     # En la imagen se la ve afuera de la zona, que es lo que importa a quien mira el video.
-    result = analyze(video, zone_layers, make_tracker=StraddlingTracker)
+    result = last(analyze(video, zone_layers, make_tracker=StraddlingTracker))
     assert result.events == []
 
 
 def test_analyze_writes_the_result_inside_out_dir(video, zone_layers, tmp_path):
     # Así la app puede borrar todos los videos generados (son grabaciones reales).
     out_dir = tmp_path / "resultados"
-    result = analyze(video, zone_layers, make_tracker=FakeTracker, out_dir=out_dir)
+    result = last(analyze(video, zone_layers, make_tracker=FakeTracker, out_dir=out_dir))
     assert out_dir in result.video_path.parents
 
 
 def test_analyze_respects_max_seconds(video, zone_layers, monkeypatch):
     monkeypatch.setattr("camguard.pipeline.MAX_SECONDS", 1)
-    result = analyze(video, zone_layers, make_tracker=FakeTracker)
+    result = last(analyze(video, zone_layers, make_tracker=FakeTracker))
     assert probe(result.video_path).frame_count == 5
 
 
 def test_analyze_without_zone_raises(video):
     with pytest.raises(ValueError, match="Pintá la zona"):
-        analyze(video, [], make_tracker=FakeTracker)
+        next(analyze(video, [], make_tracker=FakeTracker))

tests/test_ui.py

@@ -22,12 +22,12 @@ def video(tmp_path):
 
 def test_analyze_without_video_warns_and_skips_outputs():
     with pytest.warns(UserWarning, match="Primero subí un video"):
-        assert on_analyze(None, None) == (gr.skip(),) * 3
+        assert list(on_analyze(None, None)) == [(gr.skip(),) * 3]
 
 
 def test_analyze_without_zone_warns_and_skips_outputs(video):
     with pytest.warns(UserWarning, match="Pintá la zona"):
-        assert on_analyze(video, {"layers": []}) == (gr.skip(),) * 3
+        assert list(on_analyze(video, {"layers": []})) == [(gr.skip(),) * 3]
 
 
 def test_clear_zone_without_video_warns_and_skips_editor():