Update app.py

app.py

@@ -1,114 +1,210 @@
-import torch
-import torchaudio
-from einops import rearrange
+import logging
+import random
+import warnings
+import os
 import gradio as gr
+import numpy as np
 import spaces
-import os
-import uuid
-from transformers import pipeline
-
-# Importing the model-related functions
-from stable_audio_tools import get_pretrained_model
-from stable_audio_tools.inference.generation import generate_diffusion_cond
-
-# Load the model outside of the GPU-decorated function
-def load_model():
-    print("Loading model...")
-    model, model_config = get_pretrained_model("stabilityai/stable-audio-open-1.0")
-    print("Model loaded successfully.")
-    return model, model_config
-
-# 번역 모델 로드
-translator = pipeline("translation", model="Helsinki-NLP/opus-mt-ko-en")
-
-# Function to set up, generate, and process the audio
-@spaces.GPU(duration=120)  # Allocate GPU only when this function is called
-def generate_audio(prompt, seconds_total=30, steps=100, cfg_scale=7):
-    print(f"Original Prompt: {prompt}")
-
-    # 한글 텍스트를 영어로 번역
-    translated_prompt = translator(prompt, max_length=512)[0]['translation_text']
-    print(f"Translated Prompt: {translated_prompt}")
-    
-    device = "cuda" if torch.cuda.is_available() else "cpu"
-    print(f"Using device: {device}")
-
-    # Fetch the Hugging Face token from the environment variable
-    hf_token = os.getenv('HF_TOKEN')
-    print(f"Hugging Face token: {hf_token}")
-
-    # Use pre-loaded model and configuration
-    model, model_config = load_model()
-    sample_rate = model_config["sample_rate"]
-    sample_size = model_config["sample_size"]
-
-    print(f"Sample rate: {sample_rate}, Sample size: {sample_size}")
-
-    model = model.to(device)
-    print("Model moved to device.")
-
-    # Set up text and timing conditioning
-    conditioning = [{
-        "prompt": translated_prompt,
-        "seconds_start": 0,
-        "seconds_total": seconds_total
-    }]
-    print(f"Conditioning: {conditioning}")
-
-    # Generate stereo audio
-    print("Generating audio...")
-    output = generate_diffusion_cond(
-        model,
-        steps=steps,
-        cfg_scale=cfg_scale,
-        conditioning=conditioning,
-        sample_size=sample_size,
-        sigma_min=0.3,
-        sigma_max=500,
-        sampler_type="dpmpp-3m-sde",
-        device=device
-    )
-    print("Audio generated.")
+import torch
+from diffusers import FluxControlNetModel
+from diffusers.pipelines import FluxControlNetPipeline
+from gradio_imageslider import ImageSlider
+from PIL import Image
+from huggingface_hub import snapshot_download
 
-    # Rearrange audio batch to a single sequence
-    output = rearrange(output, "b d n -> d (b n)")
-    print("Audio rearranged.")
+css = """
+#col-container {
+    margin: 0 auto;
+    max-width: 512px;
+}
+"""
 
-    # Peak normalize, clip, convert to int16
-    output = output.to(torch.float32).div(torch.max(torch.abs(output))).clamp(-1, 1).mul(32767).to(torch.int16).cpu()
-    print("Audio normalized and converted.")
+if torch.cuda.is_available():
+    power_device = "GPU"
+    device = "cuda"
+else:
+    power_device = "CPU"
+    device = "cpu"
 
-    # Generate a unique filename for the output
-    unique_filename = f"output_{uuid.uuid4().hex}.wav"
-    print(f"Saving audio to file: {unique_filename}")
 
-    # Save to file
-    torchaudio.save(unique_filename, output, sample_rate)
-    print(f"Audio saved: {unique_filename}")
+huggingface_token = os.getenv("HUGGINFACE_TOKEN")
 
-    # Return the path to the generated audio file
-    return unique_filename
+model_path = snapshot_download(
+    repo_id="black-forest-labs/FLUX.1-dev", 
+    repo_type="model", 
+    ignore_patterns=["*.md", "*..gitattributes"],
+    local_dir="FLUX.1-dev",
+    token=huggingface_token, # type a new token-id.
+)
 
-css = """
-footer {
-    visibility: hidden;
-}
-"""
 
-# Setting up the Gradio Interface
-interface = gr.Interface(theme="Nymbo/Nymbo_Theme", css=css,
-    fn=generate_audio,
-    inputs=[
-        gr.Textbox(label="프롬프트", placeholder="여기에 텍스트 프롬프트를 입력하세요"),
-        gr.Slider(0, 47, value=30, label="오디오 길이 (초)"),
-        gr.Slider(10, 150, value=100, step=10, label="디퓨전 단계 수"),
-        gr.Slider(1, 15, value=7, step=0.1, label="CFG 스케일")
-    ],
-    outputs=gr.Audio(type="filepath", label="생성된 오디오"),
+# Load pipeline
+controlnet = FluxControlNetModel.from_pretrained(
+    "jasperai/Flux.1-dev-Controlnet-Upscaler", torch_dtype=torch.bfloat16
+).to(device)
+pipe = FluxControlNetPipeline.from_pretrained(
+    model_path, controlnet=controlnet, torch_dtype=torch.bfloat16
 )
+pipe.to(device)
+
+MAX_SEED = 1000000
+MAX_PIXEL_BUDGET = 1024 * 1024
+
+
+def process_input(input_image, upscale_factor, **kwargs):
+    w, h = input_image.size
+    w_original, h_original = w, h
+    aspect_ratio = w / h
+
+    was_resized = False
+
+    if w * h * upscale_factor**2 > MAX_PIXEL_BUDGET:
+        warnings.warn(
+            f"Requested output image is too large ({w * upscale_factor}x{h * upscale_factor}). Resizing to ({int(aspect_ratio * MAX_PIXEL_BUDGET ** 0.5 // upscale_factor), int(MAX_PIXEL_BUDGET ** 0.5 // aspect_ratio // upscale_factor)}) pixels."
+        )
+        gr.Info(
+            f"Requested output image is too large ({w * upscale_factor}x{h * upscale_factor}). Resizing input to ({int(aspect_ratio * MAX_PIXEL_BUDGET ** 0.5 // upscale_factor), int(MAX_PIXEL_BUDGET ** 0.5 // aspect_ratio // upscale_factor)}) pixels budget."
+        )
+        input_image = input_image.resize(
+            (
+                int(aspect_ratio * MAX_PIXEL_BUDGET**0.5 // upscale_factor),
+                int(MAX_PIXEL_BUDGET**0.5 // aspect_ratio // upscale_factor),
+            )
+        )
+        was_resized = True
+
+    # resize to multiple of 8
+    w, h = input_image.size
+    w = w - w % 8
+    h = h - h % 8
+
+    return input_image.resize((w, h)), w_original, h_original, was_resized
+
+
+@spaces.GPU#(duration=42)
+def infer(
+    seed,
+    randomize_seed,
+    input_image,
+    num_inference_steps,
+    upscale_factor,
+    controlnet_conditioning_scale,
+    progress=gr.Progress(track_tqdm=True),
+):
+    if randomize_seed:
+        seed = random.randint(0, MAX_SEED)
+    true_input_image = input_image
+    input_image, w_original, h_original, was_resized = process_input(
+        input_image, upscale_factor
+    )
+
+    # rescale with upscale factor
+    w, h = input_image.size
+    control_image = input_image.resize((w * upscale_factor, h * upscale_factor))
+
+    generator = torch.Generator().manual_seed(seed)
+
+    gr.Info("Upscaling image...")
+    image = pipe(
+        prompt="",
+        control_image=control_image,
+        controlnet_conditioning_scale=controlnet_conditioning_scale,
+        num_inference_steps=num_inference_steps,
+        guidance_scale=3.5,
+        height=control_image.size[1],
+        width=control_image.size[0],
+        generator=generator,
+    ).images[0]
+
+    if was_resized:
+        gr.Info(
+            f"Resizing output image to targeted {w_original * upscale_factor}x{h_original * upscale_factor} size."
+        )
+
+    # resize to target desired size
+    image = image.resize((w_original * upscale_factor, h_original * upscale_factor))
+    image.save("output.jpg")
+    # convert to numpy
+    return [true_input_image, image, seed]
+
+
+with gr.Blocks(theme="Yntec/HaleyCH_Theme_Orange", css=css) as demo:
+
+
+    with gr.Row():
+        run_button = gr.Button(value="Run")
+
+    with gr.Row():
+        with gr.Column(scale=4):
+            input_im = gr.Image(label="Input Image", type="pil")
+        with gr.Column(scale=1):
+            num_inference_steps = gr.Slider(
+                label="Number of Inference Steps",
+                minimum=8,
+                maximum=50,
+                step=1,
+                value=28,
+            )
+            upscale_factor = gr.Slider(
+                label="Upscale Factor",
+                minimum=1,
+                maximum=4,
+                step=1,
+                value=4,
+            )
+            controlnet_conditioning_scale = gr.Slider(
+                label="Controlnet Conditioning Scale",
+                minimum=0.1,
+                maximum=1.5,
+                step=0.1,
+                value=0.6,
+            )
+            seed = gr.Slider(
+                label="Seed",
+                minimum=0,
+                maximum=MAX_SEED,
+                step=1,
+                value=42,
+            )
+
+            randomize_seed = gr.Checkbox(label="Randomize seed", value=True)
+
+    with gr.Row():
+        result = ImageSlider(label="Input / Output", type="pil", interactive=True)
+
+    examples = gr.Examples(
+        examples=[
+            [42, False, "z1.webp", 28, 4, 0.6],
+            [42, False, "z2.webp", 28, 4, 0.6],
+
+        ],
+        inputs=[
+            seed,
+            randomize_seed,
+            input_im,
+            num_inference_steps,
+            upscale_factor,
+            controlnet_conditioning_scale,
+        ],
+        fn=infer,
+        outputs=result,
+        cache_examples="lazy",
+    )
 
-# Pre-load the model to avoid multiprocessing issues
-model, model_config = load_model()
+    gr.on(
+        [run_button.click],
+        fn=infer,
+        inputs=[
+            seed,
+            randomize_seed,
+            input_im,
+            num_inference_steps,
+            upscale_factor,
+            controlnet_conditioning_scale,
+        ],
+        outputs=result,
+        show_api=False,
+        # show_progress="minimal",
+    )
 
-# Launch the Interface
-interface.launch()
+demo.queue().launch(share=False)