Spaces:
Build error
Build error
File size: 3,904 Bytes
481a6ac 2b49a40 481a6ac 2b49a40 481a6ac 2b49a40 481a6ac |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 |
import os
import gradio as gr
import logging
from datetime import datetime
from dotenv import load_dotenv
# Cargar variables de entorno
load_dotenv()
# Configuración inicial
logging.basicConfig(filename='art_transformer.log', level=logging.INFO)
logger = logging.getLogger(__name__)
# Importaciones internas
from image_processor.analyzer import ImageAnalyzer
from image_processor.framer import FrameGenerator
from image_processor.priority_queue import PriorityQueue
from image_processor.google_auth import GooglePhotosAuth
from image_processor.monitoring import ResourceMonitor
class ArtTransformer:
def __init__(self):
self.openai_key = os.getenv("OPENAI_API_KEY")
self.analyzer = ImageAnalyzer()
self.framer = FrameGenerator(self.openai_key)
self.auth = GooglePhotosAuth()
self.queue = PriorityQueue()
self.monitor = ResourceMonitor(interval=15)
self.current_job = None
def process_album(self, image_paths, priority=2, progress=gr.Progress()):
try:
self.monitor.start()
# Autenticación con Google Photos
credentials = self.auth.authenticate()
# Configurar cola de prioridad
for idx, path in enumerate(image_paths):
self.queue.add_task(priority + idx/1000, path)
results = []
total = len(image_paths)
while not self.queue.is_empty():
if self.current_job and self.current_job.is_alive():
yield gr.Gallery(visible=True), f"⏳ Procesando {self.current_job.name}..."
next_path = self.queue.get_next_task()
self.current_job = threading.Thread(
target=self._process_single_image,
args=(next_path, results, progress),
name=f"Job-{datetime.now().strftime('%H%M%S')}"
)
self.current_job.start()
self.current_job.join()
return self._build_output(results), "✅ Proceso completado!"
except Exception as e:
logger.error(f"Error general: {str(e)}")
return None, f"❌ Error crítico: {str(e)}"
finally:
self.monitor.stop()
def create_interface():
with gr.Blocks(title="Art Transformer", theme=gr.themes.Soft()) as demo:
gr.Markdown("# 🎨 Transformador Artístico Automático")
with gr.Row():
with gr.Column(scale=1):
api_key = gr.Textbox(
label="OpenAI API Key",
type="password",
placeholder="sk-...",
info="Necesario para generar marcos"
)
priority = gr.Dropdown(
label="Prioridad",
choices=[("Alta", 0), ("Media", 1), ("Baja", 2)],
value=1
)
btn_run = gr.Button("Iniciar Procesamiento", variant="primary")
with gr.Column(scale=3):
gallery = gr.Gallery(
label="Resultados",
columns=4,
height="auto",
object_fit="contain"
)
status = gr.Textbox(label="Estado", interactive=False)
inputs = gr.File(
file_count="multiple",
file_types=["image"],
label="Subir imágenes"
)
btn_run.click(
fn=ArtTransformer().process_album,
inputs=[inputs, priority],
outputs=[gallery, status]
)
return demo
if __name__ == "__main__":
demo = create_interface()
demo.queue(concurrency_count=2, max_size=10)
demo.launch(server_name="0.0.0.0", server_port=7860) |