app_vince.py

#!/usr/bin/env python3
"""
VINCIE Service UI (Gradio) — in-process

- Importa diretamente o servidor in-process (vince_server).
- UI minimalista: galeria e vídeo por aba.
- Opções avançadas: steps, cfg_scale, aspect_ratio, resolution.
- Logs e 'ls' de diagnóstico apenas no terminal.

Pré-requisitos garantidos pelo start.sh:
- Repo VINCIE íntegro e no PYTHONPATH (VINCIE_DIR).
- Symlink ckpt/VINCIE-3B -> snapshot (contém dit.pth, vae.pth, llm14b).
- Apex/Flash e demais deps instaladas pelo builder.
"""

import os
from pathlib import Path
from typing import List, Tuple, Optional

import gradio as gr

# Servidor persistente (pipeline aquecida)
from services.vince_server import server


# -------- util --------

def _list_media(out_dir: Path, max_images: int = 24) -> Tuple[List[str], Optional[str]]:
    img_globs = ("*.png", "*.jpg", "*.jpeg", "*.webp")
    images: List[Path] = []
    for pat in img_globs:
        images += list(out_dir.rglob(pat))
    images = sorted(images, key=lambda p: p.stat().st_mtime) if images else []
    image_paths = [str(p) for p in images[-max_images:]] if images else []
    videos = sorted(out_dir.rglob("*.mp4"), key=lambda p: p.stat().st_mtime)
    video_path = str(videos[-1]) if videos else None
    return image_paths, video_path


def _debug_ls(paths: List[Path], depth: int = 2) -> None:
    try:
        for base in paths:
            try:
                p = Path(base)
                print(f"[debug-ls] >>> {p}")
                if p.is_dir():
                    entries = sorted(p.iterdir(), key=lambda x: x.name.lower())
                    for e in entries:
                        print(f"  - {e.name}{'/' if e.is_dir() else ''}")
                    if depth > 1:
                        for e in entries:
                            if e.is_dir():
                                try:
                                    sub = sorted(e.iterdir(), key=lambda x: x.name.lower())
                                    print(f"    [{e.name}/]")
                                    for s in sub:
                                        print(f"      - {s.name}{'/' if s.is_dir() else ''}")
                                except Exception as se:
                                    print(f"    (erro listando {e}): {se}")
                elif p.exists():
                    print(f"  - arquivo: {p.name}")
                else:
                    print("  - (inexistente)")
            except Exception as ie:
                print(f"[debug-ls] erro listando {base}: {ie}")
    except Exception as e:
        print(f"[debug-ls] falha geral: {e}")


# -------- UI handlers --------

def ui_multi_turn(
    input_image: Optional[str],
    turns_text: Optional[str],
    steps_input: int,
    cfg_scale_input: float,
    aspect_ratio_input: str,
    resolution_input: int,
):
    if not input_image or not str(input_image).strip():
        print("[multi_turn] input_image ausente")
        _debug_ls([Path("/app"), Path("/data")], depth=1)
        return [], None
    if not turns_text or not turns_text.strip():
        print("[multi_turn] turns ausentes")
        return [], None

    turns = [ln.strip() for ln in turns_text.splitlines() if ln.strip()]
    try:
        out_dir = server.generate_multi_turn(
            image_path=input_image,
            turns=turns,
            cfg_scale=float(cfg_scale_input) if cfg_scale_input is not None else None,
            aspect_ratio_input=str(aspect_ratio_input) if aspect_ratio_input else None,
            resolution_input=int(resolution_input) if resolution_input is not None else None,
            steps=int(steps_input) if steps_input is not None else None,
        )
    except Exception as e:
        print(f"[multi_turn] erro na geração: {e}")
        _debug_ls([Path("/app/outputs"), Path("/app/VINCIE"), Path("/data")], depth=2)
        return [], None

    out_path = Path(out_dir)
    if not out_path.exists():
        print(f"[multi_turn] saída inexistente: {out_path}")
        _debug_ls([out_path.parent, Path("/app/outputs")], depth=2)
        return [], None

    imgs, vid = _list_media(out_path)
    if not imgs and not vid:
        print(f"[multi_turn] sem mídias em {out_path}")
        _debug_ls([out_path], depth=2)
    return imgs, vid


def ui_multi_concept(
    files: Optional[List[str]],
    descs_text: Optional[str],
    final_prompt: Optional[str],
    steps_input: int,
    cfg_scale_input: float,
    aspect_ratio_input: str,
    resolution_input: int,
):
    if not files:
        print("[multi_concept] files ausentes")
        _debug_ls([Path("/data"), Path("/app")], depth=1)
        return [], None
    if not descs_text or not descs_text.strip():
        print("[multi_concept] descs ausentes")
        return [], None
    if not final_prompt or not final_prompt.strip():
        print("[multi_concept] final_prompt ausente")
        return [], None

    descs = [ln.strip() for ln in descs_text.splitlines() if ln.strip()]
    if len(descs) != len(files):
        print(f"[multi_concept] mismatch descs({len(descs)}) vs files({len(files)})")
        _debug_ls([Path("/data")], depth=1)
        return [], None

    try:
        out_dir = server.generate_multi_concept(
            concept_images=files,
            concept_prompts=descs,
            final_prompt=final_prompt,
            cfg_scale=float(cfg_scale_input) if cfg_scale_input is not None else None,
            aspect_ratio_input=str(aspect_ratio_input) if aspect_ratio_input else None,
            resolution_input=int(resolution_input) if resolution_input is not None else None,
            steps=int(steps_input) if steps_input is not None else None,
        )
    except Exception as e:
        print(f"[multi_concept] erro na geração: {e}")
        _debug_ls([Path("/app/outputs"), Path("/app/VINCIE"), Path("/data")], depth=2)
        return [], None

    out_path = Path(out_dir)
    if not out_path.exists():
        print(f"[multi_concept] saída inexistente: {out_path}")
        _debug_ls([out_path.parent, Path("/app/outputs")], depth=2)
        return [], None

    imgs, vid = _list_media(out_path)
    if not imgs and not vid:
        print(f"[multi_concept] sem mídias em {out_path}")
        _debug_ls([out_path], depth=2)
    return imgs, vid


# -------- UI --------

with gr.Blocks(title="VINCIE Service (in-process)") as demo:
    gr.Markdown(
        "\n".join(
            [
                "# VINCIE — Edição Multi-Turn e Composição Multi-Concept (in-process)",
                "- Pipeline carregada uma única vez; pesos permanecem em VRAM entre requisições.",
                "- UI exibe apenas imagens e vídeo; logs no terminal.",
            ]
        )
    )

    # Opções avançadas (compartilhadas entre abas)
    with gr.Accordion("Opções Avançadas", open=False):
        steps_input = gr.Slider(label="Passos de Inferência", minimum=10, maximum=100, step=1, value=50)
        cfg_scale_input = gr.Slider(label="Escala de Orientação (CFG)", minimum=1.0, maximum=15.0, step=0.5, value=7.5)
        aspect_ratio_input = gr.Dropdown(
            label="Aspect Ratio",
            choices=["keep_ratio", "1:1", "16:9", "9:16", "4:3", "3:4"],
            value="keep_ratio",
        )
        resolution_input = gr.Slider(label="Resolução (lado menor)", minimum=256, maximum=1024, step=64, value=512)

    # Aba 1 — Multi-turn Editing
    with gr.Tab("Multi-turn Editing"):
        with gr.Row():
            img = gr.Image(type="filepath", label="Imagem de entrada")
            turns = gr.Textbox(lines=8, label="Instruções (uma por linha)")
        run1 = gr.Button("Gerar")
        out_gallery = gr.Gallery(label="Imagens", columns=4, height="auto")
        out_video = gr.Video(label="Vídeo (se houver)")
        run1.click(
            ui_multi_turn,
            inputs=[img, turns, steps_input, cfg_scale_input, aspect_ratio_input, resolution_input],
            outputs=[out_gallery, out_video],
        )

    # Aba 2 — Multi-concept Composition
    with gr.Tab("Multi-concept Composition"):
        files = gr.File(file_count="multiple", type="filepath", label="Imagens de conceito")
        descs = gr.Textbox(lines=8, label="Descrições (uma por linha, mesma ordem das imagens)")
        final_prompt = gr.Textbox(lines=2, label="Prompt final")
        run2 = gr.Button("Gerar")
        out_gallery2 = gr.Gallery(label="Imagens", columns=4, height="auto")
        out_video2 = gr.Video(label="Vídeo (se houver)")
        run2.click(
            ui_multi_concept,
            inputs=[files, descs, final_prompt, steps_input, cfg_scale_input, aspect_ratio_input, resolution_input],
            outputs=[out_gallery2, out_video2],
        )

if __name__ == "__main__":
    if os.getenv("GRADIO_ENABLE_QUEUE", "True").lower() == "true" and hasattr(demo, "queue"):
        demo.queue(
            concurrency_count=int(os.getenv("GRADIO_QUEUE_CONCURRENCY", "2")),
            max_size=int(os.getenv("GRADIO_QUEUE_MAX", "32")),
        )
    demo.launch(
        server_name=os.getenv("GRADIO_SERVER_NAME", "0.0.0.0"),
        server_port=int(os.getenv("GRADIO_SERVER_PORT", os.getenv("PORT", "7860"))),
        allowed_paths=["/app/outputs", "/app/ckpt"],
    )