Separar la demo en el paquete camguard por responsabilidad

config (perillas), zone (máscara y regla de pies), events (Event y
armado de eventos), video (lectura submuestreada y escritura H.264),
detection (RF-DETR + ByteTrack), annotate (dibujo), pipeline (análisis
sin depender de Gradio) y ui (página). app.py queda como punto de
entrada. Mismo comportamiento: el video de ejemplo da las mismas 18
capturas que antes.

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

app.py

@@ -1,177 +1,6 @@
-"""CamGuard: detecta personas que entran a una zona restringida de un video.
-
-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.5 s/cuadro en 1080p); 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):
-    """True para cada punto (x, y) que cae dentro de la máscara de zona."""
-    h, w = zone.shape
-    x = np.clip(feet_xy[:, 0].astype(int), 0, w - 1)
-    y = np.clip(feet_xy[:, 1].astype(int), 0, h - 1)
-    return zone[y, x]
-
-
-def build_events(frames_in_zone, fps, max_gap_s=MAX_GAP_S):
-    """{id: [índices de cuadro dentro de la zona]} -> filas [id, entró, salió, duración] en segundos."""
-    max_gap = max(1, round(max_gap_s * fps))
-    rows = []
-    for tid, frames in frames_in_zone.items():
-        start = prev = frames[0]
-        for f in frames[1:] + [None]:
-            if f is not None and f - prev <= max_gap:
-                prev = f
-                continue
-            enter, leave = start / fps, (prev + 1) / fps
-            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 example():
-    """Video de ejemplo (se baja la primera vez) con una zona ya pintada abajo a la derecha."""
-    from supervision.assets import VideoAssets, download_assets
-
-    try:
-        path = download_assets(VideoAssets.PEOPLE_WALKING, directory="examples")
-    except Exception:  # sin internet: arranca sin ejemplo
-        return None, None
-    editor = first_frame(path)
-    h, w = editor["background"].shape[:2]
-    layer = np.zeros((h, w, 4), np.uint8)
-    layer[int(h * 0.55) :, w // 2 :] = (255, 0, 0, 255)
-    editor["layers"] = [layer]
-    return path, editor
-
-
-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
-
-
-def main():
-    global model
-    from rfdetr import RFDETRSmall
-
-    model = RFDETRSmall()
-    example_video, example_zone = example()
-    with gr.Blocks(title="CamGuard") as demo:
-        gr.Markdown(
-            "# CamGuard — Intruso en zona\n"
-            "**Para probar rápido:** ya viene cargado un video de ejemplo con la zona marcada. Tocá **Analizar**.\n\n"
-            "**Con tu propio video:** tocá la ✕ del video, subí el tuyo, pintá encima de la imagen el área "
-            "restringida y tocá **Analizar**.\n\n"
-            f"Se analizan {FPS} cuadros por segundo y hasta {MAX_SECONDS} s de video. Todo corre en esta PC."
-        )
-        with gr.Row():
-            video = gr.Video(value=example_video, label="Video", sources=["upload"])
-            editor = gr.ImageEditor(
-                value=example_zone,
-                label="Zona restringida (pintá encima)",
-                type="numpy",
-                sources=[],
-                brush=gr.Brush(default_size=120, colors=["#ff0000"], color_mode="fixed"),
-            )
-        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")
-        gallery = gr.Gallery(label="Momento de entrada de cada intruso", columns=4)
-
-        video.upload(first_frame, video, editor)
-        run.click(analyze, [video, editor], [result, table, gallery])
-    demo.launch(inbrowser=True)
+"""Punto de entrada de CamGuard: ``uv run python app.py`` (es lo que ejecutan run.bat y run.sh)."""
 
+from camguard.ui import main
 
 if __name__ == "__main__":
     main()

camguard/__init__.py

@@ -0,0 +1,14 @@
+"""CamGuard: detecta personas que entran a una zona restringida de un video.
+
+Todo corre local, en CPU: el video nunca sale de la máquina.
+
+Módulos:
+    config: perillas ajustables (fps analizados, tope de duración, confianza).
+    zone: de lo que pinta el usuario a una máscara, y quién la pisa.
+    events: de cuadros sueltos a eventos de entrada y salida.
+    video: lectura submuestreada y escritura H.264.
+    detection: detección (RF-DETR) y seguimiento (ByteTrack) de personas.
+    annotate: dibujo de la zona y los recuadros sobre cada cuadro.
+    pipeline: orquesta todo el análisis; no depende de la interfaz.
+    ui: la página web (Gradio).
+"""

camguard/annotate.py

@@ -0,0 +1,36 @@
+"""Dibujo sobre cada cuadro: la zona en rojo translúcido y un recuadro por persona."""
+
+import numpy as np
+import supervision as sv
+
+ZONE_COLOR_BGR = np.array([0, 0, 255])  # rojo (OpenCV usa BGR)
+ZONE_OPACITY = 0.4
+
+
+def annotate(
+    frame_bgr: np.ndarray, detections: sv.Detections, inside: np.ndarray, zone: np.ndarray
+) -> np.ndarray:
+    """Dibuja la zona, las personas de afuera en verde y los intrusos en rojo.
+
+    Args:
+        frame_bgr: Cuadro original en BGR. No se modifica.
+        detections: Personas detectadas, con ``tracker_id``.
+        inside: Array booleano: True para las personas con los pies en la zona.
+        zone: Máscara booleana de la zona.
+
+    Returns:
+        Una copia del cuadro, anotada, en BGR.
+    """
+    out = frame_bgr.copy()
+    # Mezcla en vez de pintar sólido, para que se vea lo que hay debajo de la zona.
+    out[zone] = ((1 - ZONE_OPACITY) * out[zone] + ZONE_OPACITY * ZONE_COLOR_BGR).astype(np.uint8)
+    groups = (
+        (~inside, sv.Color.GREEN, "persona #{}"),
+        (inside, sv.Color.RED, "INTRUSO #{}"),
+    )
+    for mask, color, label in groups:
+        group = detections[mask]
+        out = sv.BoxAnnotator(color=color, thickness=3).annotate(out, group)
+        labels = [label.format(track_id) for track_id in group.tracker_id]
+        out = sv.LabelAnnotator(color=color).annotate(out, group, labels=labels)
+    return out

camguard/config.py

@@ -0,0 +1,21 @@
+"""Perillas de la demo: valores que se pueden ajustar sin tocar la lógica."""
+
+# Cuadros analizados por segundo. Una persona caminando no cruza una zona en menos de 0,2 s,
+# así que alcanza y es 5-6 veces más rápido que analizar los 25-30 cuadros del video.
+# ponytail: CPU-only (~0.5 s/cuadro en 1080p); si es lento, exportar RF-DETR a ONNX +
+# onnxruntime-directml para usar la GPU AMD.
+FPS = 5
+
+# Segundos de video que se analizan como máximo, para que en CPU no tarde una eternidad.
+MAX_SECONDS = 60
+
+# Confianza mínima para contar una detección. Más bajo aparecen "personas" falsas;
+# más alto se pierden las que están lejos.
+CONF = 0.4
+
+# Si alguien "desaparece" de la zona menos que esto (el detector lo pierde un cuadro, alguien
+# le pasa por delante), sigue siendo el mismo evento en vez de partirse en varios.
+MAX_GAP_S = 1.0
+
+# Id de "person" entre las 80 clases COCO que reconoce RF-DETR.
+PERSON_ID = 1

camguard/detection.py

@@ -0,0 +1,63 @@
+"""Detección y seguimiento de personas.
+
+RF-DETR (Roboflow, Apache-2.0) encuentra personas en cada cuadro; ByteTrack les asigna un número
+que se mantiene entre cuadros, para poder decir "el #3 entró a los 12 s y salió a los 18 s".
+Este es el único módulo que conoce el modelo: cambiarlo (por ejemplo a una versión ONNX) no toca
+el resto del código.
+"""
+
+from functools import cache
+
+import numpy as np
+import supervision as sv
+from trackers import ByteTrackTracker
+
+from camguard.config import CONF, PERSON_ID
+
+
+@cache
+def load_model():
+    """Carga RF-DETR Small una sola vez por proceso.
+
+    La primera vez descarga los pesos (~368 MB) a ``~/.roboflow/models``.
+    """
+    # Import diferido: rfdetr arrastra PyTorch (~14 s de import) y los tests no lo necesitan.
+    from rfdetr import RFDETRSmall
+
+    return RFDETRSmall()
+
+
+class PersonTracker:
+    """Detecta personas en cada cuadro y les asigna un número de seguimiento estable.
+
+    Se crea uno nuevo por cada video analizado, para que la numeración arranque de cero.
+
+    Args:
+        fps: Cuadros por segundo que va a recibir (los analizados, no los del archivo). ByteTrack
+            lo usa para saber cuánto tiempo "recordar" a una persona que dejó de verse.
+        conf: Confianza mínima de detección.
+    """
+
+    def __init__(self, fps: float, conf: float = CONF):
+        self._model = load_model()
+        self._conf = conf
+        # El umbral de activación por defecto (0.7) tardaría en "confirmar" a las personas,
+        # y minimum_consecutive_frames=1 les da número desde el primer cuadro en que aparecen.
+        self._tracker = ByteTrackTracker(
+            frame_rate=fps, track_activation_threshold=conf, minimum_consecutive_frames=1
+        )
+
+    def update(self, frame_bgr: np.ndarray) -> sv.Detections:
+        """Procesa el siguiente cuadro del video.
+
+        Args:
+            frame_bgr: Cuadro en BGR, como lo entrega OpenCV.
+
+        Returns:
+            Las personas detectadas, con ``tracker_id`` asignado. Las que el tracker todavía
+            no confirmó (``tracker_id == -1``) se descartan.
+        """
+        # El modelo espera RGB; .copy() porque la vista invertida tiene strides negativos.
+        detections = self._model.predict(frame_bgr[:, :, ::-1].copy(), threshold=self._conf)
+        detections = self._tracker.update(detections[detections.class_id == PERSON_ID])
+        return detections[detections.tracker_id != -1]

camguard/events.py

@@ -0,0 +1,57 @@
+"""Eventos de intrusión: de "en qué cuadros estuvo cada persona en la zona" a entradas y salidas."""
+
+from dataclasses import dataclass
+
+from camguard.config import MAX_GAP_S
+
+
+@dataclass(frozen=True)
+class Event:
+    """Una estadía continua de una persona dentro de la zona.
+
+    Attributes:
+        track_id: Número de seguimiento de la persona (el "#N" del video).
+        enter_s: Segundo del video en que entró (o en que se la vio adentro por primera vez).
+        leave_s: Segundo del video en que salió.
+    """
+
+    track_id: int
+    enter_s: float
+    leave_s: float
+
+    @property
+    def duration_s(self) -> float:
+        """Segundos que estuvo dentro de la zona."""
+        return round(self.leave_s - self.enter_s, 1)
+
+
+def build_events(
+    frames_in_zone: dict[int, list[int]], fps: float, max_gap_s: float = MAX_GAP_S
+) -> list[Event]:
+    """Agrupa los cuadros en que cada persona estuvo en la zona en eventos de entrada y salida.
+
+    Mientras el salto entre dos cuadros consecutivos sea de ``max_gap_s`` o menos, es el mismo
+    evento: así un cuadro en que el detector "pierde" a la persona no parte una entrada en varias.
+    Si el salto es mayor, la persona salió y volvió: son dos eventos.
+
+    Args:
+        frames_in_zone: Para cada persona, los índices (crecientes) de los cuadros analizados
+            en que tuvo los pies en la zona.
+        fps: Cuadros analizados por segundo, para pasar de índices a segundos.
+        max_gap_s: Tolerancia en segundos antes de considerar que la persona salió.
+
+    Returns:
+        Los eventos, ordenados por segundo de entrada.
+    """
+    max_gap = max(1, round(max_gap_s * fps))
+    events = []
+    for track_id, frames in frames_in_zone.items():
+        start = prev = frames[0]
+        for frame in [*frames[1:], None]:  # el None final cierra el último evento
+            if frame is not None and frame - prev <= max_gap:
+                prev = frame
+                continue
+            # prev + 1: si estuvo un solo cuadro, igual estuvo 1/fps segundos, no 0.
+            events.append(Event(track_id, round(start / fps, 1), round((prev + 1) / fps, 1)))
+            start = prev = frame
+    return sorted(events, key=lambda e: e.enter_s)

camguard/pipeline.py

@@ -0,0 +1,112 @@
+"""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).
+"""
+
+import math
+import tempfile
+from collections.abc import Callable
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Protocol
+
+import numpy as np
+import supervision as sv
+
+from camguard.annotate import annotate
+from camguard.config import FPS, MAX_SECONDS
+from camguard.detection import PersonTracker
+from camguard.events import Event, build_events
+from camguard.video import H264Writer, frame_limit, probe, sample_frames, sampling_step
+from camguard.zone import feet_in_zone, zone_from_layers
+
+
+class Tracker(Protocol):
+    """Lo que el pipeline necesita de un tracker: personas con número, cuadro a cuadro."""
+
+    def update(self, frame_bgr: np.ndarray) -> sv.Detections:
+        """Devuelve las personas del cuadro, con ``tracker_id``."""
+        ...
+
+
+@dataclass(frozen=True)
+class Snapshot:
+    """Captura del momento en que una persona aparece por primera vez dentro de la zona.
+
+    Attributes:
+        image: Cuadro anotado, en RGB (listo para mostrar).
+        track_id: Número de seguimiento de la persona.
+        time_s: Segundo del video.
+    """
+
+    image: np.ndarray
+    track_id: int
+    time_s: float
+
+
+@dataclass(frozen=True)
+class AnalysisResult:
+    """Resultado de analizar un video.
+
+    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.
+    """
+
+    video_path: Path
+    events: list[Event]
+    snapshots: list[Snapshot]
+
+
+def analyze(
+    video_path: str | Path,
+    zone_layers: list[np.ndarray],
+    make_tracker: Callable[[float], Tracker] = PersonTracker,
+    progress: Callable[[float], None] | None = None,
+) -> 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``).
+
+    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.
+
+    Returns:
+        El video anotado, los eventos y las capturas.
+
+    Raises:
+        ValueError: Si el video no se puede leer o no hay zona pintada.
+    """
+    info = probe(video_path)
+    zone = zone_from_layers(zone_layers, (info.height, info.width))
+    step = sampling_step(info.fps, FPS)
+    fps = info.fps / step  # cuadros analizados por segundo
+    max_frames = frame_limit(info, MAX_SECONDS)
+    expected = max(1, math.ceil(max_frames / step))
+
+    tracker = make_tracker(fps)
+    out_path = Path(tempfile.NamedTemporaryFile(suffix=".mp4", delete=False).name)
+    frames_in_zone: dict[int, list[int]] = {}
+    snapshots: list[Snapshot] = []
+
+    with H264Writer(out_path, fps, info.width, info.height) as writer:
+        for i, frame in enumerate(sample_frames(video_path, step, max_frames)):
+            detections = tracker.update(frame)
+            feet = detections.get_anchors_coordinates(sv.Position.BOTTOM_CENTER)
+            inside = feet_in_zone(feet, zone)
+            annotated = annotate(frame, detections, inside, zone)
+            for track_id in detections.tracker_id[inside].tolist():
+                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)
+            writer.write(annotated)
+            if progress:
+                progress(min(1.0, (i + 1) / expected))
+
+    return AnalysisResult(out_path, build_events(frames_in_zone, fps), snapshots)

camguard/ui.py

@@ -0,0 +1,120 @@
+"""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.
+"""
+
+import gradio as gr
+import numpy as np
+
+from camguard.config import FPS, MAX_SECONDS
+from camguard.detection import load_model
+from camguard.pipeline import analyze
+from camguard.video import first_frame
+
+# Zona del video de ejemplo, en fracción de la imagen: de la mitad hacia la derecha
+# y del 55 % de la altura hacia abajo.
+EXAMPLE_ZONE_LEFT = 0.5
+EXAMPLE_ZONE_TOP = 0.55
+
+INSTRUCTIONS = f"""\
+# CamGuard — Intruso en zona
+**Para probar rápido:** ya viene cargado un video de ejemplo con la zona marcada. \
+Tocá **Analizar**.
+
+**Con tu propio video:** tocá la ✕ del video, subí el tuyo, pintá encima de la imagen \
+el área restringida y tocá **Analizar**.
+
+Se analizan {FPS} cuadros por segundo y hasta {MAX_SECONDS} s de video. Todo corre en esta PC.
+"""
+
+
+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."""
+    return {"background": background, "layers": layers or [], "composite": None}
+
+
+def load_example() -> tuple[str | None, dict | None]:
+    """Descarga el video de ejemplo (solo la primera vez) y le pinta una zona por defecto.
+
+    El video es de Roboflow y no tiene una licencia clara, por eso se descarga en vez de
+    incluirlo en el repo.
+
+    Returns:
+        (ruta del video, valor del editor), o (None, None) si no hay internet.
+    """
+    from supervision.assets import VideoAssets, download_assets
+
+    try:
+        path = download_assets(VideoAssets.PEOPLE_WALKING, directory="examples")
+    except Exception:  # sin internet: la app arranca igual, sin ejemplo
+        return None, None
+    background = first_frame(path)
+    h, w = background.shape[:2]
+    layer = np.zeros((h, w, 4), np.uint8)
+    layer[int(h * EXAMPLE_ZONE_TOP) :, int(w * EXAMPLE_ZONE_LEFT) :] = (255, 0, 0, 255)
+    return path, editor_value(background, [layer])
+
+
+def on_upload(video_path: str) -> dict:
+    """Al subir un video, pone su primer cuadro en el editor para pintar la zona."""
+    try:
+        return editor_value(first_frame(video_path))
+    except ValueError as e:
+        raise gr.Error(str(e)) from e
+
+
+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.
+
+    Returns:
+        (video anotado, filas de la tabla de eventos, imágenes de la galería con su leyenda).
+    """
+    if not video_path:
+        raise gr.Error("Primero subí un video.")
+    layers = (editor or {}).get("layers") or []
+    try:
+        result = analyze(video_path, layers, progress=lambda p: progress(p, desc="Analizando"))
+    except ValueError as e:
+        raise gr.Error(str(e)) from e
+
+    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
+
+
+def build_demo() -> gr.Blocks:
+    """Arma la página: video y editor lado a lado, el botón, y los tres resultados."""
+    example_video, example_editor = load_example()
+    with gr.Blocks(title="CamGuard") as demo:
+        gr.Markdown(INSTRUCTIONS)
+        with gr.Row():
+            video = gr.Video(value=example_video, label="Video", sources=["upload"])
+            editor = gr.ImageEditor(
+                value=example_editor,
+                label="Zona restringida (pintá encima)",
+                type="numpy",
+                sources=[],  # la imagen siempre es el primer cuadro del video
+                brush=gr.Brush(default_size=120, colors=["#ff0000"], color_mode="fixed"),
+            )
+        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"
+        )
+        gallery = gr.Gallery(label="Momento de entrada de cada intruso", columns=4)
+
+        video.upload(on_upload, video, editor)
+        run.click(on_analyze, [video, editor], [result, table, gallery])
+    return demo
+
+
+def main() -> None:
+    """Carga el modelo, arma la página y la abre en el navegador.
+
+    Solo escucha en 127.0.0.1: nadie de la red ni de internet puede entrar (son videos reales).
+    """
+    load_model()  # al arrancar, para que el primer análisis no espere la carga
+    build_demo().launch(inbrowser=True)

camguard/video.py

@@ -0,0 +1,146 @@
+"""Lectura y escritura de video.
+
+Junta los detalles de bajo nivel en un solo lugar: OpenCV entrega los cuadros en BGR (no RGB),
+H.264 exige ancho y alto pares, y saltear cuadros con ``grab()`` es más rápido que decodificarlos.
+Si mañana se lee de una cámara en vivo (RTSP), este es el único módulo que cambia.
+"""
+
+import math
+from collections.abc import Iterator
+from dataclasses import dataclass
+from fractions import Fraction
+from pathlib import Path
+
+import av
+import cv2
+import numpy as np
+
+UNREADABLE_MSG = "No pude leer el video."
+
+
+@dataclass(frozen=True)
+class VideoInfo:
+    """Datos básicos de un archivo de video.
+
+    Attributes:
+        fps: Cuadros por segundo del archivo.
+        width: Ancho en píxeles.
+        height: Alto en píxeles.
+        frame_count: Cantidad de cuadros (0 si el archivo no lo informa).
+    """
+
+    fps: float
+    width: int
+    height: int
+    frame_count: int
+
+
+def probe(path: str | Path) -> VideoInfo:
+    """Lee los datos básicos de un video.
+
+    Raises:
+        ValueError: Si el archivo no se puede abrir como video.
+    """
+    cap = cv2.VideoCapture(str(path))
+    try:
+        if not cap.isOpened():
+            raise ValueError(UNREADABLE_MSG)
+        return VideoInfo(
+            fps=cap.get(cv2.CAP_PROP_FPS) or 30.0,
+            width=int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)),
+            height=int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)),
+            frame_count=int(cap.get(cv2.CAP_PROP_FRAME_COUNT)),
+        )
+    finally:
+        cap.release()
+
+
+def first_frame(path: str | Path) -> np.ndarray:
+    """Devuelve el primer cuadro del video en RGB (el orden que espera el navegador).
+
+    Raises:
+        ValueError: Si no se puede leer ningún cuadro.
+    """
+    cap = cv2.VideoCapture(str(path))
+    ok, frame = cap.read()
+    cap.release()
+    if not ok:
+        raise ValueError(UNREADABLE_MSG)
+    return frame[:, :, ::-1]
+
+
+def sampling_step(source_fps: float, target_fps: float) -> int:
+    """Cada cuántos cuadros del archivo hay que analizar uno para acercarse a ``target_fps``."""
+    return max(1, round(source_fps / target_fps))
+
+
+def frame_limit(info: VideoInfo, max_seconds: float) -> int:
+    """Cuántos cuadros del archivo recorrer como máximo para no pasar de ``max_seconds``."""
+    limit = math.floor(max_seconds * info.fps)
+    return min(info.frame_count, limit) if info.frame_count > 0 else limit
+
+
+def sample_frames(path: str | Path, step: int, max_frames: int) -> Iterator[np.ndarray]:
+    """Recorre el video y devuelve uno de cada ``step`` cuadros, en BGR.
+
+    Args:
+        path: Archivo de video.
+        step: Se devuelve el cuadro 0, el ``step``, el ``2*step``... (ver :func:`sampling_step`).
+        max_frames: Cuadros del archivo a recorrer como máximo (ver :func:`frame_limit`).
+
+    Yields:
+        Cuadros BGR (alto, ancho, 3).
+    """
+    cap = cv2.VideoCapture(str(path))
+    try:
+        for idx in range(max_frames):
+            if idx % step:
+                # grab() avanza sin decodificar la imagen completa: mucho más rápido que read().
+                if not cap.grab():
+                    return
+                continue
+            ok, frame = cap.read()
+            if not ok:
+                return
+            yield frame
+    finally:
+        cap.release()
+
+
+class H264Writer:
+    """Escribe cuadros BGR en un mp4 H.264, el formato que reproducen todos los navegadores.
+
+    El codec por defecto de OpenCV muchas veces no se reproduce en el navegador; PyAV trae
+    libx264 incluido, así que no hace falta instalar ffmpeg. Se usa con ``with``::
+
+        with H264Writer("salida.mp4", fps=5, width=1920, height=1080) as writer:
+            writer.write(frame)
+    """
+
+    def __init__(self, path: str | Path, fps: float, width: int, height: int):
+        # El formato de color yuv420p exige ancho y alto pares: si son impares se recorta 1 px.
+        self._width, self._height = width // 2 * 2, height // 2 * 2
+        self._container = av.open(str(path), "w")
+        self._stream = self._container.add_stream(
+            "libx264", rate=Fraction(fps).limit_denominator(1000)
+        )
+        self._stream.width, self._stream.height = self._width, self._height
+        self._stream.pix_fmt = "yuv420p"
+
+    def write(self, frame_bgr: np.ndarray) -> None:
+        """Agrega un cuadro BGR al video."""
+        frame = av.VideoFrame.from_ndarray(
+            frame_bgr[: self._height, : self._width], format="bgr24"
+        )
+        self._container.mux(self._stream.encode(frame))
+
+    def close(self) -> None:
+        """Vacía los cuadros que el encoder tiene en buffer y cierra el archivo."""
+        self._container.mux(self._stream.encode())
+        self._container.close()
+
+    def __enter__(self) -> "H264Writer":
+        return self
+
+    def __exit__(self, *exc_info: object) -> None:
+        self.close()

camguard/zone.py

@@ -0,0 +1,56 @@
+"""Zona restringida: de lo que pinta el usuario a una máscara, y quién la pisa.
+
+La zona es una máscara booleana del tamaño del video (True = restringido) en vez de un
+polígono, porque el usuario la pinta a mano: puede tener cualquier forma o ser varias zonas.
+"""
+
+import cv2
+import numpy as np
+
+NO_ZONE_MSG = "Pintá la zona restringida sobre la imagen antes de analizar."
+
+
+def zone_from_layers(layers: list[np.ndarray], shape: tuple[int, int]) -> np.ndarray:
+    """Combina las capas pintadas en el editor en una sola máscara del tamaño del video.
+
+    Args:
+        layers: Capas RGBA del editor. Un píxel es zona si su canal alfa es mayor que 0.
+        shape: (alto, ancho) del video.
+
+    Returns:
+        Máscara booleana de forma ``shape``.
+
+    Raises:
+        ValueError: Si no hay nada pintado.
+    """
+    if not layers:
+        raise ValueError(NO_ZONE_MSG)
+    mask = np.any([layer[..., 3] > 0 for layer in layers], axis=0)
+    if not mask.any():
+        raise ValueError(NO_ZONE_MSG)
+    if mask.shape != shape:
+        # INTER_NEAREST mantiene los bordes nítidos (sí/no) en vez de difuminarlos.
+        size = (shape[1], shape[0])  # OpenCV pide (ancho, alto)
+        mask = cv2.resize(mask.astype(np.uint8), size, interpolation=cv2.INTER_NEAREST) > 0
+    return mask
+
+
+def feet_in_zone(feet_xy: np.ndarray, zone: np.ndarray) -> np.ndarray:
+    """Indica qué personas tienen los pies dentro de la zona.
+
+    Se usan los pies (centro del borde inferior del recuadro) y no el recuadro entero porque,
+    con la cámara en perspectiva, alguien que pasa delante de una puerta la tapa con el cuerpo
+    sin estar en ella: los pies dicen dónde está parado de verdad.
+
+    Args:
+        feet_xy: Array (N, 2) con las coordenadas (x, y) de los pies de cada persona.
+        zone: Máscara booleana (alto, ancho).
+
+    Returns:
+        Array booleano (N,): True si los pies de esa persona caen en la zona.
+    """
+    h, w = zone.shape
+    # clip: un punto justo en el borde (x == ancho) se saldría del array al indexar.
+    x = np.clip(feet_xy[:, 0].astype(int), 0, w - 1)
+    y = np.clip(feet_xy[:, 1].astype(int), 0, h - 1)
+    return zone[y, x]

test_zone.py

@@ -1,6 +1,7 @@
 import numpy as np
 
-from app import build_events, feet_in_zone
+from camguard.events import Event, build_events
+from camguard.zone import feet_in_zone
 
 zone = np.zeros((10, 10), bool)
 zone[5:, 5:] = True
@@ -9,6 +10,6 @@ assert feet_in_zone(feet, zone).tolist() == [True, False, True, True]
 
 # fps=5: #1 entra en 0 s y sale en 1 s; #2 entra, sale 2 s y vuelve (dos eventos); un hueco de 1 cuadro no corta
 events = build_events({1: [0, 1, 2, 3, 4], 2: [10, 12, 30]}, fps=5)
-assert events == [[1, 0.0, 1.0, 1.0], [2, 2.0, 2.6, 0.6], [2, 6.0, 6.2, 0.2]], events
+assert events == [Event(1, 0.0, 1.0), Event(2, 2.0, 2.6), Event(2, 6.0, 6.2)], events
 assert build_events({}, fps=5) == []
 print("ok")