herramientas.py

import os
import socket
import random
import globales
import time
import tempfile
import gradio_client

def obtenAccesoHF():
    if local_check():
        print("Estoy en entorno Local...")        
        import bridges
        llave = bridges.llave
    else:
        print("Estoy en entorno Remoto...")
        llave = os.getenv("llave")
        print("Ésto es llave:", llave)
    return llave

def local_check():
    hostname = socket.gethostname()
    #r-moibe-nowme
    print("Dentro de local_check... , el hostname es: ", hostname)

    #Estoy usando el nombre de la app para identificar que estoy corriendola en HF.
    if globales.hostname in hostname:
        print("Ejecutando api en el servidor.")
        return False
    else:
        print("Ejecutando api en local.")
        return True

async def imageToTemp(image):
    print("Estoy en imageToTemp...")

    try:
        with tempfile.NamedTemporaryFile(delete=False, suffix=f"_{image.filename}") as tmp_file:
            contents = await image.read()
            tmp_file.write(contents)
            temp_file_path = tmp_file.name
        
        print(f"Archivo temporal guardado en: {temp_file_path}")
        return temp_file_path

    except Exception as e:
        print(f"Error al procesar el archivo: {e}")
        return {"error": "Error al procesar la imagen"}


def borraTemp(archivo):
    try:
        os.remove(archivo)
        print(f"Archivo temporal eliminado: {archivo}")
        return True
    except FileNotFoundError:
        print(f"Advertencia: Archivo no encontrado para eliminar: {archivo}")
    except Exception as e:
        print(f"Error al eliminar el archivo {archivo}: {e}")

def getPosition(carpeta_positions):
    """
    Regresa una posición del cuerpo humano para ser utilizada por el proceso de Stable Diffusion.

    Parameters:
    dataframe (dataframe): El dataframe en el que estuvimos trabajando.

    Returns:
    bool: True si se guardó el archivo correctamente.
    """
    #Con carpeta específica.
    ruta_carpeta = os.path.join("images", "positions", carpeta_positions)
    #O una carpeta específica para todas.
    ruta_carpeta = os.path.join("images", "positions", "all")
        
    try: 
        lista_archivos = os.listdir(ruta_carpeta)
        posicion_aleatoria = random.choice(lista_archivos)
        ruta_posicion = os.path.join(ruta_carpeta, posicion_aleatoria)
        print("Ésto es ruta posición: ", ruta_posicion)
        return ruta_posicion     
    except Exception as e: 
        print("No hay carpeta de posiciones:", e)
        return e 
    
async def exeCute(endpoint, image, prompt):    

    temp_image = await imageToTemp(image)
    try:        
        posicion = getPosition(endpoint)
        imagenSource = gradio_client.handle_file(temp_image)     
        imagenPosition = gradio_client.handle_file(posicion)

        client = gradio_client.Client(globales.api, hf_token=obtenAccesoHF())
        result = client.predict(
                    imagenSource,
                    imagenPosition,
                    prompt=prompt,
                    #negative_prompt="(lowres, low quality, worst quality:1.2), (text:1.2), watermark, (frame:1.2), deformed, ugly, deformed eyes, blur, out of focus, blurry, deformed cat, deformed, photo, anthropomorphic cat, monochrome, pet collar, gun, weapon, 3d, drones, drone, buildings in background",
                    style_name="(No style)", #ver lista en styles.txt
                    num_steps=30,
                    identitynet_strength_ratio=0.8,
                    adapter_strength_ratio=0.8,
                    #pose_strength=0.4,
                    canny_strength=0.4,
                    depth_strength=0.4,
                    controlnet_selection=["depth"], #pueden ser ['pose', 'canny', 'depth'] #Al parecer pose ya no.
                    guidance_scale=5,
                    seed=random.randint(0, 2147483647), 
                    scheduler="EulerDiscreteScheduler",
                    enable_LCM=False,
                    enhance_face_region=True,
                    api_name="/generate_image"
            ) 
        return result[0] 

    except Exception as e:
        print(f"Error en exeCute: {e}")        
        print("E es del tipo : ", type(e))
        if "PAUSED" in str(e): 
            return {"error": "Space Paused"}
        elif "timed out" in str(e):
            return {"error": "Network error try again."}
        else: 
            return {"error": "Face not detected, please try again or with other pic."} 
    finally: 
        borrado = borraTemp(temp_image) 
        print("El resultado de borrarTemp fue: ", borrado)

def formaFilename(prompt):

    timestamp_segundos = int(time.time())
    print(f"Printed at: {timestamp_segundos}.") 
    
    return f"{prompt}-{timestamp_segundos}.jpg"