from __future__ import annotations import csv import json import shutil import tempfile import uuid from pathlib import Path import gradio as gr import yaml from bat_tracker.config import load_config from bat_tracker.pipeline import run_pipeline APP_TITLE = "Bat Tracker" APP_DESCRIPTION = ( "Sube un video IR monocromo para ejecutar el pipeline, revisar la region valida " "detectada y consultar `events.csv` en forma de tabla." ) def _prepare_config(config_file: str | None, work_dir: Path) -> Path: cfg = load_config(config_file) cfg.setdefault("output", {}) cfg["output"]["progress_enabled"] = True cfg["output"]["progress_step_percent"] = 2 config_path = work_dir / "space_config.yaml" with config_path.open("w", encoding="utf-8") as handle: yaml.safe_dump(cfg, handle, sort_keys=False, allow_unicode=True) return config_path def _read_csv_table(path: Path) -> tuple[list[str], list[list[str]]]: with path.open("r", encoding="utf-8", newline="") as handle: reader = csv.reader(handle) rows = list(reader) if not rows: return [], [] return rows[0], rows[1:] def _copy_input_file(src: str, dst_dir: Path) -> Path: src_path = Path(src) safe_name = src_path.name or "input_video.mp4" dst_path = dst_dir / safe_name shutil.copy2(src_path, dst_path) return dst_path def process_video(video_file: str | None, config_file: str | None, progress=gr.Progress()): if not video_file: raise gr.Error("Debes subir un video de entrada.") session_root = Path(tempfile.gettempdir()) / "bat_tracker_spaces" session_root.mkdir(parents=True, exist_ok=True) work_dir = session_root / uuid.uuid4().hex output_dir = work_dir / "output" work_dir.mkdir(parents=True, exist_ok=True) try: progress(0, desc="Preparando ejecución") input_path = _copy_input_file(video_file, work_dir) config_path = _prepare_config(config_file, work_dir) def on_pipeline_progress(pct: int, detail: str | None) -> None: description = detail or "Procesando" progress(min(max(pct / 100.0, 0.0), 1.0), desc=description) meta = run_pipeline( input_video=str(input_path), output_dir=str(output_dir), config_path=str(config_path), progress_callback=on_pipeline_progress, ) outputs = meta.get("outputs", {}) events_csv = Path(outputs["events_csv"]) headers, rows = _read_csv_table(events_csv) table_headers = headers or ["Sin resultados"] table_value = rows if headers else [] valid_region_overlay = outputs.get("valid_region_overlay_png") tracks_overlay = outputs.get("tracks_overlay_png") summary = { "video_id": meta.get("video", {}).get("video_id"), "fps": meta.get("video", {}).get("fps"), "frames_processed": meta.get("metrics", {}).get("frames_processed"), "tracks_total": meta.get("metrics", {}).get("tracks_total"), "detections_kept": meta.get("metrics", {}).get("detections_kept"), "selected_device": meta.get("execution", {}).get("selected_device"), "valid_region": { "enabled": meta.get("valid_region", {}).get("enabled"), "x_start": meta.get("valid_region", {}).get("x_start"), "x_end": meta.get("valid_region", {}).get("x_end"), "width": meta.get("valid_region", {}).get("width"), "method": meta.get("valid_region", {}).get("method"), }, } return ( json.dumps(summary, indent=2, ensure_ascii=False), valid_region_overlay if valid_region_overlay and Path(valid_region_overlay).exists() else None, tracks_overlay if tracks_overlay and Path(tracks_overlay).exists() else None, gr.update( headers=table_headers, value=table_value, column_count=(len(table_headers), "fixed"), ), str(events_csv), outputs.get("tracks_csv"), json.dumps(meta, indent=2, ensure_ascii=False), ) except Exception as exc: # pragma: no cover - UI surface raise gr.Error(f"Error ejecutando el pipeline: {exc}") from exc APP_THEME = gr.themes.Soft( primary_hue="yellow", secondary_hue="stone", neutral_hue="slate", ) APP_CSS = """ :root { --panel-bg: linear-gradient(180deg, #fffdf5 0%, #f2eee4 100%); --frame-bg: rgba(255, 252, 240, 0.78); --frame-border: rgba(58, 42, 22, 0.14); --text-strong: #1f1a13; --text-soft: #5d5242; --accent: #9a5b16; } .gradio-container { background: radial-gradient(circle at top left, rgba(199, 145, 58, 0.18), transparent 28%), radial-gradient(circle at top right, rgba(89, 61, 24, 0.12), transparent 24%), linear-gradient(180deg, #f6f0df 0%, #ece5d6 100%); color: var(--text-strong); font-family: "Source Serif 4", Georgia, serif; } .app-shell { max-width: 1180px; margin: 0 auto; padding: 24px 12px 40px; } .hero { background: var(--panel-bg); border: 1px solid var(--frame-border); border-radius: 24px; padding: 24px; box-shadow: 0 18px 50px rgba(53, 39, 17, 0.08); } .hero h1 { margin: 0; font-size: clamp(2.2rem, 4vw, 3.6rem); line-height: 0.95; letter-spacing: -0.04em; text-transform: uppercase; } .hero p { margin: 12px 0 0; max-width: 760px; color: var(--text-soft); font-size: 1rem; } .panel { background: var(--frame-bg); border: 1px solid var(--frame-border); border-radius: 20px; box-shadow: inset 0 1px 0 rgba(255,255,255,0.55); } .panel-title { margin: 0 0 10px; color: var(--accent); font-size: 0.86rem; font-weight: 700; letter-spacing: 0.12em; text-transform: uppercase; } """ with gr.Blocks(title=APP_TITLE) as demo: with gr.Column(elem_classes=["app-shell"]): gr.HTML( """

Bat Tracker Space

Interfaz minima para ejecutar el tracker sobre un video, inspeccionar la region valida detectada y revisar los eventos agregados en forma de tabla.

""" ) with gr.Row(equal_height=False): with gr.Column(scale=1, elem_classes=["panel"]): gr.Markdown("### Entrada") video_input = gr.File( label="Video de entrada", file_types=[".mp4", ".avi", ".mov", ".mkv"], type="filepath", ) config_input = gr.File( label="Configuración YAML opcional", file_types=[".yaml", ".yml"], type="filepath", ) run_button = gr.Button("Procesar", variant="primary") gr.Markdown( "Si no subes configuración, se usarán los valores por defecto del proyecto." ) with gr.Column(scale=1, elem_classes=["panel"]): gr.Markdown("### Resumen") summary_output = gr.Code( label="Resumen de ejecución", language="json", interactive=False, ) meta_output = gr.Code( label="meta.json", language="json", interactive=False, ) with gr.Row(): valid_region_image = gr.Image( label="Región válida", type="filepath", interactive=False, ) tracks_overlay_image = gr.Image( label="Tracks sobre fondo", type="filepath", interactive=False, ) events_table = gr.Dataframe( label="events.csv", headers=["Sin resultados"], value=[], datatype="str", interactive=False, wrap=True, row_count=(0, "dynamic"), column_count=(1, "fixed"), ) with gr.Row(): events_csv_download = gr.File(label="Descargar events.csv") tracks_csv_download = gr.File(label="Descargar tracks.csv") run_button.click( fn=process_video, inputs=[video_input, config_input], outputs=[ summary_output, valid_region_image, tracks_overlay_image, events_table, events_csv_download, tracks_csv_download, meta_output, ], ) if __name__ == "__main__": demo.launch(theme=APP_THEME, css=APP_CSS)