[Admin maintenance] Migrate grant to ZeroGPU (#113)

- [Admin maintenance] Migrate grant to ZeroGPU (f1408b83b70e1dabef03d3ff7b51a23d40260ab8)

Dockerfile

@@ -1,32 +0,0 @@
-FROM r8.im/fofr/expression-editor@sha256:bf913bc90e1c44ba288ba3942a538693b72e8cc7df576f3beebe56adc0a92b86
-RUN apt-get update && apt-get install -y netcat jq
-
-RUN useradd -m -u 1000 user
-RUN chown -R user:user / || true
-RUN chown -R user:user /src/
-RUN chown -R user:user /root/
-RUN chown -R user:user /var/
-USER user
-ENV HOME=/home/user \
-    PATH=/home/user/.local/bin:$PATH \
-    PYTHONPATH=$HOME/app \
-    PYTHONUNBUFFERED=1 \
-    GRADIO_ALLOW_FLAGGING=never \
-    GRADIO_NUM_PORTS=1 \
-    GRADIO_SERVER_NAME=0.0.0.0 \
-    GRADIO_THEME=huggingface \
-    SYSTEM=spaces 
-
-WORKDIR $HOME/app
-#COPY ./requirements.txt /code/requirements.txt
-
-# create virtual env for Gradio app
-RUN python -m venv $HOME/.venv && \
-    . $HOME/.venv/bin/activate && \ 
-    pip install --no-cache-dir --upgrade pip && \
-    pip install --no-cache-dir gradio==6.12.0 prance
-  
-
-COPY --chown=user . $HOME/app
-RUN chmod +x $HOME/app/run.sh
-CMD ["bash", "-c", "$HOME/app/run.sh"]

README.md

@@ -3,10 +3,13 @@ title: Expression Editor
 emoji: 🐨
 colorFrom: indigo
 colorTo: red
-sdk: docker
+sdk: gradio
+sdk_version: 6.14.0
+python_version: '3.12'
+app_file: app.py
 pinned: true
 disable_embedding: true
 short_description: Quickly edit the expression of a face
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

app.py

@@ -1,266 +1,281 @@
+"""
+Expression Editor on ZeroGPU.
+
+Vendored from fofr/cog-expression-editor's underlying ComfyUI workflow
+(LoadImage -> ExpressionEditor) but executed without ComfyUI: we clone the
+PowerHouseMan/ComfyUI-AdvancedLivePortrait node at startup and stub the
+two ComfyUI internals it imports (`folder_paths` and `comfy.utils`).
+
+Weights auto-download to ./models on first run via the node's own loader
+(Kijai/LivePortrait_safetensors + Bingsu/adetailer for the YOLO bbox).
+"""
+import os
+import sys
+import types
+import subprocess
+
+# ------------------------------------------------------------------
+# 1. Pull the custom node + stub ComfyUI internals BEFORE importing it
+# ------------------------------------------------------------------
+
+# Use a Python-identifier-legal directory name so we can import it as a
+# package (the repo's `nodes.py` uses relative imports like
+# `from .LivePortrait...` which only work inside a real package).
+CUSTOM_NODE_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "advanced_live_portrait")
+if not os.path.exists(CUSTOM_NODE_DIR):
+    subprocess.check_call([
+        "git", "clone", "--depth=1",
+        "https://github.com/PowerHouseMan/ComfyUI-AdvancedLivePortrait.git",
+        CUSTOM_NODE_DIR,
+    ])
+
+# Writable paths the node expects
+MODELS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "models")
+TEMP_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "temp")
+os.makedirs(os.path.join(MODELS_DIR, "liveportrait"), exist_ok=True)
+os.makedirs(os.path.join(MODELS_DIR, "ultralytics"), exist_ok=True)
+os.makedirs(TEMP_DIR, exist_ok=True)
+
+# Minimal `folder_paths` shim
+_fp = types.ModuleType("folder_paths")
+_fp.models_dir = MODELS_DIR
+_fp.output_directory = os.path.join(os.path.dirname(os.path.abspath(__file__)), "outputs")
+os.makedirs(_fp.output_directory, exist_ok=True)
+_fp.get_folder_paths = lambda m: [os.path.join(MODELS_DIR, m)]
+_fp.get_save_image_path = lambda f, d, *a, **k: (d, f, 0, "", f)
+_fp.get_temp_directory = lambda: TEMP_DIR
+_fp.add_model_folder_path = lambda *a, **k: None
+sys.modules["folder_paths"] = _fp
+
+# Minimal `comfy.utils` shim
+import torch
+import safetensors.torch
+
+_comfy = types.ModuleType("comfy")
+_comfy_utils = types.ModuleType("comfy.utils")
+
+def _load_torch_file(ckpt, *args, **kwargs):
+    s = str(ckpt)
+    if s.endswith(".safetensors"):
+        return safetensors.torch.load_file(s)
+    return torch.load(s, map_location="cpu", weights_only=False)
+_comfy_utils.load_torch_file = _load_torch_file
+
+class _ProgressBar:
+    def __init__(self, *a, **k):
+        pass
+    def update(self, *a, **k):
+        pass
+    def update_absolute(self, *a, **k):
+        pass
+_comfy_utils.ProgressBar = _ProgressBar
+
+_comfy.utils = _comfy_utils  # attach as attribute too — `import comfy.utils` then `comfy.utils.X` needs both sys.modules and attribute access
+sys.modules["comfy"] = _comfy
+sys.modules["comfy.utils"] = _comfy_utils
+
+# ------------------------------------------------------------------
+# 2. Now import the node
+# ------------------------------------------------------------------
+
+# Parent dir is already on the path (it's the app's CWD); import as a package
+import spaces
 import gradio as gr
+import numpy as np
 from PIL import Image
-from urllib.parse import urlparse
-import requests
-import time
-import os
 
-from utils.gradio_helpers import parse_outputs, process_outputs
-
-# Function to verify the image file type and resize it if necessary
-def preprocess_image(image_path):
-    # Check if the file exists
-    if not os.path.exists(image_path):
-        raise FileNotFoundError(f"No such file: '{image_path}'")
-
-    # Get the file extension and make sure it's a valid image format
-    valid_extensions = ['jpg', 'jpeg', 'png', 'webp']
-    file_extension = image_path.split('.')[-1].lower()
-
-    if file_extension not in valid_extensions:
-        raise ValueError("Invalid file type. Only JPG, PNG, and WEBP are allowed.")
-    
-    # Open the image
-    with Image.open(image_path) as img:
-        width, height = img.size
-
-        # Check if any dimension exceeds 1024 pixels
-        if width > 1024 or height > 1024:
-            # Calculate the new size while maintaining aspect ratio
-            if width > height:
-                new_width = 1024
-                new_height = int((new_width / width) * height)
-            else:
-                new_height = 1024
-                new_width = int((new_height / height) * width)
-            
-            # Resize the image
-            img_resized = img.resize((new_width, new_height), Image.LANCZOS)
-            print(f"Resized image to {new_width}x{new_height}.")
-
-            # Save the resized image as 'resized_image.jpg'
-            output_path = 'resized_image.jpg'
-            img_resized.save(output_path, 'JPEG')
-            print(f"Resized image saved as {output_path}")
-            return output_path
-        else:
-            print("Image size is within the limit, no resizing needed.")
-            return image_path
-
-
-def display_uploaded_image(image_in):
-    return image_in
+from advanced_live_portrait.nodes import ExpressionEditor, g_engine  # noqa: E402
 
-def reset_parameters():
-    return gr.update(value=0), gr.update(value=0), gr.update(value=0), gr.update(value=0), gr.update(value=0), gr.update(value=0), gr.update(value=0), gr.update(value=0), gr.update(value=0), gr.update(value=0), gr.update(value=0), gr.update(value=0)
-
-names = ['image', 'rotate_pitch', 'rotate_yaw', 'rotate_roll', 'blink', 'eyebrow', 'wink', 'pupil_x', 'pupil_y', 'aaa', 'eee', 'woo', 'smile', 'src_ratio', 'sample_ratio', 'crop_factor', 'output_format', 'output_quality']
-
-def predict(request: gr.Request, *args, progress=gr.Progress(track_tqdm=True)):
-    headers = {'Content-Type': 'application/json'}
-
-    payload = {"input": {}}
-    base_url = "http://0.0.0.0:7860"
-
-    for i, key in enumerate(names):
-        value = args[i]
-        if value and os.path.exists(str(value)):
-            value = f"{base_url}/gradio_api/file=" + value
-        if value is not None and value != "":
-            payload["input"][key] = value
-
-    time.sleep(1.0)
-    response = requests.post("http://0.0.0.0:5000/predictions", headers=headers, json=payload)
-
-    if response.status_code == 201:
-        follow_up_url = response.json()["urls"]["get"]
-
-        # ⏰ Timeout logic
-        max_wait_seconds = 60
-        poll_interval = 1
-        start_time = time.time()
-
-        while True:
-            response = requests.get(follow_up_url, headers=headers)
-            try:
-                response_json = response.json()
-            except ValueError:
-                raise gr.Error("Cog server response is not valid JSON.")
-
-            status = response_json.get("status")
-            if status == "succeeded":
-                break
-            if status == "failed":
-                raise gr.Error("The submission failed.")
-            if time.time() - start_time > max_wait_seconds:
-                raise gr.Error("Prediction timed out after 60 seconds.")
-            time.sleep(poll_interval)
-
-    if response.status_code == 200:
-        json_response = response.json()
-        if outputs[0].get_config()["name"] == "json":
-            return json_response["output"]
-
-        predict_outputs = parse_outputs(json_response["output"])
-        processed_outputs = process_outputs(predict_outputs)
-        print(f"processed_outputs: {processed_outputs}")
-        return tuple(processed_outputs) if len(processed_outputs) > 1 else processed_outputs[0]
+# Preload pipeline + face detector at module scope so they land in the
+# ZeroGPU snapshot. ZeroGPU forks per @spaces.GPU call; with the snapshot,
+# the models are already resident in GPU memory on the worker and inference
+# starts immediately. Loading them lazily inside the decorated function
+# would re-download / re-init on every cold worker.
+print("Preloading LivePortrait pipeline + YOLO detector for ZeroGPU snapshot...")
+g_engine.get_pipeline()       # downloads + .to('cuda') the 5 LivePortrait modules
+g_engine.get_detect_model()   # downloads + loads YOLO face bbox model
+print("Preload done.")
+
+# Single global editor (state cached across calls)
+_editor = ExpressionEditor()
+
+
+def _pil_to_node_tensor(img: Image.Image) -> torch.Tensor:
+    """ComfyUI image tensors are [N, H, W, C] float32 in [0, 1]."""
+    if img.mode != "RGB":
+        img = img.convert("RGB")
+    arr = np.array(img, dtype=np.float32) / 255.0
+    return torch.from_numpy(arr).unsqueeze(0)
+
+
+def _node_tensor_to_pil(t: torch.Tensor) -> Image.Image:
+    arr = (t.squeeze(0).detach().cpu().numpy() * 255).clip(0, 255).astype(np.uint8)
+    return Image.fromarray(arr)
+
+
+@spaces.GPU(duration=60)
+def edit_expression(
+    image, rotate_pitch, rotate_yaw, rotate_roll,
+    blink, eyebrow, wink, pupil_x, pupil_y,
+    aaa, eee, woo, smile,
+    src_ratio, sample_ratio, sample_parts, crop_factor,
+):
+    if image is None:
+        raise gr.Error("Please upload an image.")
+    src_t = _pil_to_node_tensor(image)
+    out = _editor.run(
+        rotate_pitch=float(rotate_pitch),
+        rotate_yaw=float(rotate_yaw),
+        rotate_roll=float(rotate_roll),
+        blink=float(blink),
+        eyebrow=float(eyebrow),
+        wink=float(wink),
+        pupil_x=float(pupil_x),
+        pupil_y=float(pupil_y),
+        aaa=float(aaa),
+        eee=float(eee),
+        woo=float(woo),
+        smile=float(smile),
+        src_ratio=float(src_ratio),
+        sample_ratio=float(sample_ratio),
+        sample_parts=sample_parts,
+        crop_factor=float(crop_factor),
+        src_image=src_t,
+    )
+    # ExpressionEditor.run returns {"ui": {...}, "result": (out_img, motion_link, exp_data)}
+    out_img_t = out["result"][0]
+    return _node_tensor_to_pil(out_img_t)
+
+
+# ------------------------------------------------------------------
+# 3. Image preprocess (mirrors original: resize so max side <= 1024)
+# ------------------------------------------------------------------
+
+def preprocess_image(img: Image.Image):
+    if img is None:
+        return None
+    if img.mode != "RGB":
+        img = img.convert("RGB")
+    w, h = img.size
+    if w <= 1024 and h <= 1024:
+        return img
+    if w >= h:
+        new_w = 1024
+        new_h = int(round(new_w / w * h))
     else:
-        if response.status_code == 409:
-            raise gr.Error("Sorry, the Cog image is still processing. Try again in a bit.")
-        raise gr.Error(f"The submission failed! Error: {response.status_code}")
+        new_h = 1024
+        new_w = int(round(new_h / h * w))
+    return img.resize((new_w, new_h), Image.LANCZOS)
 
 
+def reset_parameters():
+    return (
+        gr.update(value=0), gr.update(value=0), gr.update(value=0),
+        gr.update(value=0), gr.update(value=0), gr.update(value=0),
+        gr.update(value=0), gr.update(value=0),
+        gr.update(value=0), gr.update(value=0), gr.update(value=0), gr.update(value=0),
+    )
+
+
+# ------------------------------------------------------------------
+# 4. Gradio UI (mirrors fffiloni/expression-editor exactly)
+# ------------------------------------------------------------------
 
-css = '''
-#col-container{max-width: 800px;margin: 0 auto;}
-'''
-with gr.Blocks() as demo:
+css = """
+#col-container{max-width: 800px; margin: 0 auto;}
+"""
+
+with gr.Blocks(css=css, title="Expression Editor") as demo:
     with gr.Column(elem_id="col-container"):
         gr.Markdown("# Expression Editor")
-        gr.Markdown("Demo for expression-editor cog image by fofr")
+        gr.Markdown("Edit a face's expression with sliders. Uses the <a href='https://github.com/PowerHouseMan/ComfyUI-AdvancedLivePortrait' target='_blank'>Expression Editor ComfyUI node</a>, originally packaged by <a href='https://replicate.com/fofr' target='_blank'>fofr</a>.")
+
         with gr.Row():
             with gr.Column():
-                image = gr.Image(
+                image_in = gr.Image(
                     label="Input image",
                     sources=["upload"],
-                    type="filepath"
+                    type="pil",
                 )
+
                 with gr.Tab("HEAD"):
                     with gr.Column():
-                        rotate_pitch = gr.Slider(
-                            label="Rotate Up-Down",
-                            value=0,
-                            minimum=-20, maximum=20
-                        )
-                        rotate_yaw = gr.Slider(
-                            label="Rotate Left-Right turn", 
-                            value=0,
-                            minimum=-20, maximum=20
-                        )
-                        rotate_roll = gr.Slider(
-                            label="Rotate Left-Right tilt", value=0,
-                            minimum=-20, maximum=20
-                        )
+                        rotate_pitch = gr.Slider(label="Rotate Up-Down", value=0, minimum=-20, maximum=20)
+                        rotate_yaw = gr.Slider(label="Rotate Left-Right turn", value=0, minimum=-20, maximum=20)
+                        rotate_roll = gr.Slider(label="Rotate Left-Right tilt", value=0, minimum=-20, maximum=20)
+
                 with gr.Tab("EYES"):
                     with gr.Column():
-                        eyebrow = gr.Slider(
-                            label="Eyebrow", value=0,
-                            minimum=-10, maximum=15
-                        )
+                        eyebrow = gr.Slider(label="Eyebrow", value=0, minimum=-10, maximum=15)
                         with gr.Row():
-                            blink = gr.Slider(
-                                label="Blink", value=0,
-                                minimum=-20, maximum=5
-                            )
-                            
-                            wink = gr.Slider(
-                                label="Wink", value=0,
-                                minimum=0, maximum=25
-                            )
+                            blink = gr.Slider(label="Blink", value=0, minimum=-20, maximum=5)
+                            wink = gr.Slider(label="Wink", value=0, minimum=0, maximum=25)
                         with gr.Row():
-                            pupil_x = gr.Slider(
-                                label="Pupil X", value=0,
-                                minimum=-15, maximum=15
-                            )
-                            pupil_y = gr.Slider(
-                                label="Pupil Y", value=0,
-                                minimum=-15, maximum=15
-                            )
+                            pupil_x = gr.Slider(label="Pupil X", value=0, minimum=-15, maximum=15)
+                            pupil_y = gr.Slider(label="Pupil Y", value=0, minimum=-15, maximum=15)
+
                 with gr.Tab("MOUTH"):
                     with gr.Column():
                         with gr.Row():
-                            aaa = gr.Slider(
-                                label="Aaa", value=0,
-                                minimum=-30, maximum=120
-                            )
-                            eee = gr.Slider(
-                                label="Eee", value=0,
-                                minimum=-20, maximum=15
-                            )
-                            woo = gr.Slider(
-                                label="Woo", value=0,
-                                minimum=-20, maximum=15
-                            )
-                        smile = gr.Slider(
-                            label="Smile", value=0,
-                            minimum=-0.3, maximum=1.3
-                        )
+                            aaa = gr.Slider(label="Aaa", value=0, minimum=-30, maximum=120)
+                            eee = gr.Slider(label="Eee", value=0, minimum=-20, maximum=15)
+                            woo = gr.Slider(label="Woo", value=0, minimum=-20, maximum=15)
+                        smile = gr.Slider(label="Smile", value=0, minimum=-0.3, maximum=1.3)
+
                 with gr.Tab("More Settings"):
                     with gr.Column():
-                        src_ratio = gr.Number(
-                            label="Src Ratio", info='''Source ratio''', value=1
-                        )
-                        sample_ratio = gr.Slider(
-                            label="Sample Ratio", info='''Sample ratio''', value=1,
-                            minimum=-0.2, maximum=1.2
-                        )
-                        crop_factor = gr.Slider(
-                            label="Crop Factor", info='''Crop factor''', value=1.7,
-                            minimum=1.5, maximum=2.5
-                        )
-                        output_format = gr.Dropdown(
-                            choices=['webp', 'jpg', 'png'], label="output_format", info='''Format of the output images''', value="webp"
-                        )
-                        output_quality = gr.Number(
-                            label="Output Quality", info='''Quality of the output images, from 0 to 100. 100 is best quality, 0 is lowest quality.''', value=95
+                        src_ratio = gr.Number(label="Src Ratio", info="Source ratio", value=1)
+                        sample_ratio = gr.Slider(label="Sample Ratio", info="Sample ratio", value=1, minimum=-0.2, maximum=1.2)
+                        sample_parts = gr.Dropdown(
+                            choices=["OnlyExpression", "OnlyRotation", "OnlyMouth", "OnlyEyes", "All"],
+                            value="OnlyExpression",
+                            label="Sample parts",
                         )
+                        crop_factor = gr.Slider(label="Crop Factor", info="Crop factor", value=1.7, minimum=1.5, maximum=2.5)
+
                 with gr.Row():
                     reset_btn = gr.Button("Reset")
-                    submit_btn = gr.Button("Submit")
+                    submit_btn = gr.Button("Submit", variant="primary")
+
             with gr.Column():
-                result_image = gr.Image(elem_id="top")
-                gr.HTML("""
-                <div style="display: flex; flex-direction: column;justify-content: center; align-items: center; text-align: center;">
-                    <p style="display: flex;gap: 6px;">
-                         <a href="https://huggingface.co/spaces/fffiloni/expression-editor?duplicate=true">
-                            <img src="https://huggingface.co/datasets/huggingface/badges/resolve/main/duplicate-this-space-lg.svg" alt="Duplicate this Space">
-                        </a>
-                    </p>
-                    <p>to skip the queue and enjoy faster inference on the GPU of your choice </p>
-                </div>
-                """)
-
-    inputs = [image, rotate_pitch, rotate_yaw, rotate_roll, blink, eyebrow, wink, pupil_x, pupil_y, aaa, eee, woo, smile, src_ratio, sample_ratio, crop_factor, output_format, output_quality]
+                result_image = gr.Image(label="Output", elem_id="top")
+
+    inputs = [
+        image_in, rotate_pitch, rotate_yaw, rotate_roll,
+        blink, eyebrow, wink, pupil_x, pupil_y,
+        aaa, eee, woo, smile,
+        src_ratio, sample_ratio, sample_parts, crop_factor,
+    ]
     outputs = [result_image]
 
-    image.upload(
-        fn = preprocess_image,
-        inputs = [image],
-        outputs = [image],
-        queue = False
+    # Resize on upload (matches original 1024-max preprocess)
+    image_in.upload(
+        fn=preprocess_image,
+        inputs=[image_in],
+        outputs=[image_in],
+        queue=False,
     )
 
     reset_btn.click(
-        fn = reset_parameters,
-        inputs = None,
-        outputs = [rotate_pitch, rotate_yaw, rotate_roll, blink, eyebrow, wink, pupil_x, pupil_y, aaa, eee, woo, smile],
-        queue = False
-    ).then(
-        fn=predict,
-        inputs=inputs,
-        outputs=outputs,
-        api_visibility="private"
-    )
-    
-    submit_btn.click(
-        fn=predict,
-        inputs=inputs,
-        outputs=outputs,
-        api_visibility="private"
-    )
+        fn=reset_parameters,
+        inputs=None,
+        outputs=[
+            rotate_pitch, rotate_yaw, rotate_roll,
+            blink, eyebrow, wink, pupil_x, pupil_y,
+            aaa, eee, woo, smile,
+        ],
+        queue=False,
+    ).then(fn=edit_expression, inputs=inputs, outputs=outputs)
+
+    submit_btn.click(fn=edit_expression, inputs=inputs, outputs=outputs)
+
+    # Regenerate on slider release (matches original's live-editing feel)
+    for slider in (
+        rotate_pitch, rotate_yaw, rotate_roll,
+        blink, eyebrow, wink, pupil_x, pupil_y,
+        aaa, eee, woo, smile,
+    ):
+        slider.release(fn=edit_expression, inputs=inputs, outputs=outputs, show_progress="minimal")
+
 
-    rotate_pitch.release(fn=predict, inputs=inputs, outputs=outputs, show_progress="minimal", api_visibility="private")
-    rotate_yaw.release(fn=predict, inputs=inputs, outputs=outputs, show_progress="minimal", api_visibility="private")
-    rotate_roll.release(fn=predict, inputs=inputs, outputs=outputs, show_progress="minimal", api_visibility="private")
-    blink.release(fn=predict, inputs=inputs, outputs=outputs, show_progress="minimal", api_visibility="private")
-    eyebrow.release(fn=predict, inputs=inputs, outputs=outputs, show_progress="minimal", api_visibility="private")
-    wink.release(fn=predict, inputs=inputs, outputs=outputs, show_progress="minimal", api_visibility="private")
-    pupil_x.release(fn=predict, inputs=inputs, outputs=outputs, show_progress="minimal", api_visibility="private")
-    pupil_y.release(fn=predict, inputs=inputs, outputs=outputs, show_progress="minimal", api_visibility="private")
-    aaa.release(fn=predict, inputs=inputs, outputs=outputs, show_progress="minimal", api_visibility="private")
-    eee.release(fn=predict, inputs=inputs, outputs=outputs, show_progress="minimal", api_visibility="private")
-    woo.release(fn=predict, inputs=inputs, outputs=outputs, show_progress="minimal", api_visibility="private")
-    smile.release(fn=predict, inputs=inputs, outputs=outputs, show_progress="minimal", api_visibility="private")
-
-demo.queue(default_concurrency_limit=1, max_size=20).launch(css=css, share=False, show_error=True, ssr_mode=False)
+demo.queue().launch()

requirements.txt

@@ -1,2 +1,13 @@
-gradio
-prance
+numpy>=1.26.4
+opencv-python-headless
+imageio-ffmpeg>=0.5.1
+lmdb>=1.4.1
+timm>=1.0.7
+rich>=13.7.1
+albumentations>=1.4.10
+ultralytics
+tyro
+dill
+PyYAML
+safetensors
+Pillow

run.sh

@@ -1,44 +0,0 @@
-# Start the cog server in the background - Ensure correct path to cog
-cd /src && python3 -m cog.server.http --threads=10 &
-
-# Initialize counter for the first loop
-counter1=0
-
-# Continuous loop for reliably checking cog server's readiness on port 5000
-while true; do
-  if nc -z localhost 5000; then
-    echo "Cog server is running on port 5000."
-    break  # Exit the loop when the server is up
-  fi
-  echo "Waiting for cog server to start on port 5000..."
-  sleep 5
-  ((counter1++))
-  if [ $counter1 -ge 250 ]; then
-    echo "Error: Cog server did not start on port 5000 after 250 attempts."
-    exit 1  # Exit the script with an error status
-  fi
-done
-
-# Initialize counter for the second loop
-counter2=0
-
-# New check: Waiting for the cog server to be fully ready
-while true; do
-  response=$(curl -s http://localhost:5000/health-check) # Replace localhost:5000 with actual hostname and port if necessary
-  status=$(echo $response | jq -r '.status') # Parse status from JSON response
-  if [ "$status" = "READY" ]; then
-    echo "Cog server is fully ready."
-    break # Exit the loop when the server is fully ready
-  else
-    echo "Waiting for cog server (models loading) on port 5000..."
-    sleep 5
-  fi
-  ((counter2++))
-  if [ $counter2 -ge 250 ]; then
-    echo "Error: Cog server did not become fully ready after 250 attempts."
-    exit 1  # Exit the script with an error status
-  fi
-done
-
-# Run the application - only when cog server is fully ready
-cd $HOME/app && . $HOME/.venv/bin/activate && python app.py

utils/gradio_helpers.py

@@ -1,469 +0,0 @@
-import gradio as gr
-from urllib.parse import urlparse
-import requests
-import time
-from PIL import Image
-import base64
-import io
-import uuid
-import os
-
-
-def extract_property_info(prop):
-    combined_prop = {}
-    merge_keywords = ["allOf", "anyOf", "oneOf"]
-
-    for keyword in merge_keywords:
-        if keyword in prop:
-            for subprop in prop[keyword]:
-                combined_prop.update(subprop)
-            del prop[keyword]
-
-    if not combined_prop:
-        combined_prop = prop.copy()
-
-    for key in ["description", "default"]:
-        if key in prop:
-            combined_prop[key] = prop[key]
-
-    return combined_prop
-
-
-def detect_file_type(filename):
-    audio_extensions = [".mp3", ".wav", ".flac", ".aac", ".ogg", ".m4a"]
-    image_extensions = [
-        ".jpg",
-        ".jpeg",
-        ".png",
-        ".gif",
-        ".bmp",
-        ".tiff",
-        ".svg",
-        ".webp",
-    ]
-    video_extensions = [
-        ".mp4",
-        ".mov",
-        ".wmv",
-        ".flv",
-        ".avi",
-        ".avchd",
-        ".mkv",
-        ".webm",
-    ]
-
-    # Extract the file extension
-    if isinstance(filename, str):
-        extension = filename[filename.rfind(".") :].lower()
-
-        # Check the extension against each list
-        if extension in audio_extensions:
-            return "audio"
-        elif extension in image_extensions:
-            return "image"
-        elif extension in video_extensions:
-            return "video"
-        else:
-            return "string"
-    elif isinstance(filename, list):
-        return "list"
-
-
-def build_gradio_inputs(ordered_input_schema, example_inputs=None):
-    inputs = []
-    input_field_strings = """inputs = []\n"""
-    names = []
-    for index, (name, prop) in enumerate(ordered_input_schema):
-        names.append(name)
-        prop = extract_property_info(prop)
-        if "enum" in prop:
-            input_field = gr.Dropdown(
-                choices=prop["enum"],
-                label=prop.get("title"),
-                info=prop.get("description"),
-                value=prop.get("default"),
-            )
-            input_field_string = f"""inputs.append(gr.Dropdown(
-    choices={prop["enum"]}, label="{prop.get("title")}", info={"'''"+prop.get("description")+"'''" if prop.get("description") else 'None'}, value="{prop.get("default")}"
-))\n"""
-        elif prop["type"] == "integer":
-            if prop.get("minimum") and prop.get("maximum"):
-                input_field = gr.Slider(
-                    label=prop.get("title"),
-                    info=prop.get("description"),
-                    value=prop.get("default"),
-                    minimum=prop.get("minimum"),
-                    maximum=prop.get("maximum"),
-                    step=1,
-                )
-                input_field_string = f"""inputs.append(gr.Slider(
-    label="{prop.get("title")}", info={"'''"+prop.get("description")+"'''" if prop.get("description") else 'None'}, value={prop.get("default")},
-    minimum={prop.get("minimum")}, maximum={prop.get("maximum")}, step=1,
-))\n"""
-            else:
-                input_field = gr.Number(
-                    label=prop.get("title"),
-                    info=prop.get("description"),
-                    value=prop.get("default"),
-                )
-                input_field_string = f"""inputs.append(gr.Number(
-    label="{prop.get("title")}", info={"'''"+prop.get("description")+"'''" if prop.get("description") else 'None'}, value={prop.get("default")}
-))\n"""
-        elif prop["type"] == "number":
-            if prop.get("minimum") and prop.get("maximum"):
-                input_field = gr.Slider(
-                    label=prop.get("title"),
-                    info=prop.get("description"),
-                    value=prop.get("default"),
-                    minimum=prop.get("minimum"),
-                    maximum=prop.get("maximum"),
-                )
-                input_field_string = f"""inputs.append(gr.Slider(
-    label="{prop.get("title")}", info={"'''"+prop.get("description")+"'''" if prop.get("description") else 'None'}, value={prop.get("default")},
-    minimum={prop.get("minimum")}, maximum={prop.get("maximum")}
-))\n"""
-            else:
-                input_field = gr.Number(
-                    label=prop.get("title"),
-                    info=prop.get("description"),
-                    value=prop.get("default"),
-                )
-                input_field_string = f"""inputs.append(gr.Number(
-    label="{prop.get("title")}", info={"'''"+prop.get("description")+"'''" if prop.get("description") else 'None'}, value={prop.get("default")}
-))\n"""
-        elif prop["type"] == "boolean":
-            input_field = gr.Checkbox(
-                label=prop.get("title"),
-                info=prop.get("description"),
-                value=prop.get("default"),
-            )
-            input_field_string = f"""inputs.append(gr.Checkbox(
-    label="{prop.get("title")}", info={"'''"+prop.get("description")+"'''" if prop.get("description") else 'None'}, value={prop.get("default")}
-))\n"""
-        elif (
-            prop["type"] == "string" and prop.get("format") == "uri" and example_inputs
-        ):
-            input_type_example = example_inputs.get(name, None)
-            if input_type_example:
-                input_type = detect_file_type(input_type_example)
-            else:
-                input_type = None
-            if input_type == "image":
-                input_field = gr.Image(label=prop.get("title"), type="filepath")
-                input_field_string = f"""inputs.append(gr.Image(
-    label="{prop.get("title")}", type="filepath"
-))\n"""
-            elif input_type == "audio":
-                input_field = gr.Audio(label=prop.get("title"), type="filepath")
-                input_field_string = f"""inputs.append(gr.Audio(
-    label="{prop.get("title")}", type="filepath"
-))\n"""
-            elif input_type == "video":
-                input_field = gr.Video(label=prop.get("title"))
-                input_field_string = f"""inputs.append(gr.Video(
-    label="{prop.get("title")}"
-))\n"""
-            else:
-                input_field = gr.File(label=prop.get("title"))
-                input_field_string = f"""inputs.append(gr.File(
-    label="{prop.get("title")}"
-))\n"""
-        else:
-            input_field = gr.Textbox(
-                label=prop.get("title"),
-                info=prop.get("description"),
-            )
-            input_field_string = f"""inputs.append(gr.Textbox(
-    label="{prop.get("title")}", info={"'''"+prop.get("description")+"'''" if prop.get("description") else 'None'}
-))\n"""
-        inputs.append(input_field)
-        input_field_strings += f"{input_field_string}\n"
-
-    input_field_strings += f"names = {names}\n"
-
-    return inputs, input_field_strings, names
-
-
-def build_gradio_outputs_replicate(output_types):
-    outputs = []
-    output_field_strings = """outputs = []\n"""
-    if output_types:
-        for output in output_types:
-            if output == "image":
-                output_field = gr.Image()
-                output_field_string = "outputs.append(gr.Image())"
-            elif output == "audio":
-                output_field = gr.Audio(type="filepath")
-                output_field_string = "outputs.append(gr.Audio(type='filepath'))"
-            elif output == "video":
-                output_field = gr.Video()
-                output_field_string = "outputs.append(gr.Video())"
-            elif output == "string":
-                output_field = gr.Textbox()
-                output_field_string = "outputs.append(gr.Textbox())"
-            elif output == "json":
-                output_field = gr.JSON()
-                output_field_string = "outputs.append(gr.JSON())"
-            elif output == "list":
-                output_field = gr.JSON()
-                output_field_string = "outputs.append(gr.JSON())"
-            outputs.append(output_field)
-            output_field_strings += f"{output_field_string}\n"
-    else:
-        output_field = gr.JSON()
-        output_field_string = "outputs.append(gr.JSON())"
-        outputs.append(output_field)
-
-    return outputs, output_field_strings
-
-
-def build_gradio_outputs_cog():
-    pass
-
-
-def process_outputs(outputs):
-    output_values = []
-    for output in outputs:
-        if not output:
-            continue
-        if isinstance(output, str):
-            if output.startswith("data:image"):
-                base64_data = output.split(",", 1)[1]
-                image_data = base64.b64decode(base64_data)
-                image_stream = io.BytesIO(image_data)
-                image = Image.open(image_stream)
-                output_values.append(image)
-            elif output.startswith("data:audio"):
-                base64_data = output.split(",", 1)[1]
-                audio_data = base64.b64decode(base64_data)
-                audio_stream = io.BytesIO(audio_data)
-                filename = f"{uuid.uuid4()}.wav"  # Change format as needed
-                with open(filename, "wb") as audio_file:
-                    audio_file.write(audio_stream.getbuffer())
-                output_values.append(filename)
-            elif output.startswith("data:video"):
-                base64_data = output.split(",", 1)[1]
-                video_data = base64.b64decode(base64_data)
-                video_stream = io.BytesIO(video_data)
-                # Here you can save the audio or return the stream for further processing
-                filename = f"{uuid.uuid4()}.mp4"  # Change format as needed
-                with open(filename, "wb") as video_file:
-                    video_file.write(video_stream.getbuffer())
-                output_values.append(filename)
-            else:
-                output_values.append(output)
-        else:
-            output_values.append(output)
-    return output_values
-
-
-def parse_outputs(data):
-    if isinstance(data, dict):
-        # Handle case where data is an object
-        dict_values = []
-        for value in data.values():
-            extracted_values = parse_outputs(value)
-            # For dict, we append instead of extend to maintain list structure within objects
-            if isinstance(value, list):
-                dict_values += [extracted_values]
-            else:
-                dict_values += extracted_values
-        return dict_values
-    elif isinstance(data, list):
-        # Handle case where data is an array
-        list_values = []
-        for item in data:
-            # Here we extend to flatten the list since we're already in an array context
-            list_values += parse_outputs(item)
-        return list_values
-    else:
-        # Handle primitive data types directly
-        return [data]
-
-
-def create_dynamic_gradio_app(
-    inputs,
-    outputs,
-    api_url,
-    api_id=None,
-    replicate_token=None,
-    title="",
-    model_description="",
-    names=[],
-    local_base=False,
-    hostname="0.0.0.0",
-):
-    expected_outputs = len(outputs)
-
-    def predict(request: gr.Request, *args, progress=gr.Progress(track_tqdm=True)):
-        payload = {"input": {}}
-        if api_id:
-            payload["version"] = api_id
-        parsed_url = urlparse(str(request.url))
-        if local_base:
-            base_url = f"http://{hostname}:7860"
-        else:
-            base_url = parsed_url.scheme + "://" + parsed_url.netloc
-        for i, key in enumerate(names):
-            value = args[i]
-            if value and (os.path.exists(str(value))):
-                value = f"{base_url}/file=" + value
-            if value is not None and value != "":
-                payload["input"][key] = value
-        print(payload)
-        headers = {"Content-Type": "application/json"}
-        if replicate_token:
-            headers["Authorization"] = f"Token {replicate_token}"
-        print(headers)
-        response = requests.post(api_url, headers=headers, json=payload)
-        if response.status_code == 201:
-            follow_up_url = response.json()["urls"]["get"]
-            response = requests.get(follow_up_url, headers=headers)
-            while response.json()["status"] != "succeeded":
-                if response.json()["status"] == "failed":
-                    raise gr.Error("The submission failed!")
-                response = requests.get(follow_up_url, headers=headers)
-                time.sleep(1)
-                # TODO: Add a failing mechanism if the API gets stuck
-        if response.status_code == 200:
-            json_response = response.json()
-            # If the output component is JSON return the entire output response
-            if outputs[0].get_config()["name"] == "json":
-                return json_response["output"]
-            predict_outputs = parse_outputs(json_response["output"])
-            processed_outputs = process_outputs(predict_outputs)
-            difference_outputs = expected_outputs - len(processed_outputs)
-            # If less outputs than expected, hide the extra ones
-            if difference_outputs > 0:
-                extra_outputs = [gr.update(visible=False)] * difference_outputs
-                processed_outputs.extend(extra_outputs)
-            # If more outputs than expected, cap the outputs to the expected number if
-            elif difference_outputs < 0:
-                processed_outputs = processed_outputs[:difference_outputs]
-
-            return (
-                tuple(processed_outputs)
-                if len(processed_outputs) > 1
-                else processed_outputs[0]
-            )
-
-        else:
-            if response.status_code == 409:
-                raise gr.Error(
-                    f"Sorry, the Cog image is still processing. Try again in a bit."
-                )
-            raise gr.Error(f"The submission failed! Error: {response.status_code}")
-
-    app = gr.Interface(
-        fn=predict,
-        inputs=inputs,
-        outputs=outputs,
-        title=title,
-        description=model_description,
-        allow_flagging="never",
-    )
-    return app
-
-
-def create_gradio_app_script(
-    inputs_string,
-    outputs_string,
-    api_url,
-    api_id=None,
-    replicate_token=None,
-    title="",
-    model_description="",
-    local_base=False,
-    hostname="0.0.0.0"
-):
-    headers = {"Content-Type": "application/json"}
-    if replicate_token:
-        headers["Authorization"] = f"Token {replicate_token}"
-
-    if local_base:
-        base_url = f'base_url = "http://{hostname}:7860"'
-    else:
-        base_url = """parsed_url = urlparse(str(request.url))
-    base_url = parsed_url.scheme + "://" + parsed_url.netloc"""
-    headers_string = f"""headers = {headers}\n"""
-    api_id_value = f'payload["version"] = "{api_id}"' if api_id is not None else ""
-    definition_string = """expected_outputs = len(outputs)
-def predict(request: gr.Request, *args, progress=gr.Progress(track_tqdm=True)):"""
-    payload_string = f"""payload = {{"input": {{}}}}
-    {api_id_value}
-    
-    {base_url}
-    for i, key in enumerate(names):
-        value = args[i]
-        if value and (os.path.exists(str(value))):
-            value = f"{{base_url}}/file=" + value
-        if value is not None and value != "":
-            payload["input"][key] = value\n"""
-
-    request_string = (
-        f"""response = requests.post("{api_url}", headers=headers, json=payload)\n"""
-    )
-
-    result_string = f"""
-    if response.status_code == 201:
-        follow_up_url = response.json()["urls"]["get"]
-        response = requests.get(follow_up_url, headers=headers)
-        while response.json()["status"] != "succeeded":
-            if response.json()["status"] == "failed":
-                raise gr.Error("The submission failed!")
-            response = requests.get(follow_up_url, headers=headers)
-            time.sleep(1)
-    if response.status_code == 200:
-        json_response = response.json()
-        #If the output component is JSON return the entire output response 
-        if(outputs[0].get_config()["name"] == "json"):
-            return json_response["output"]
-        predict_outputs = parse_outputs(json_response["output"])
-        processed_outputs = process_outputs(predict_outputs)
-        difference_outputs = expected_outputs - len(processed_outputs)
-        # If less outputs than expected, hide the extra ones
-        if difference_outputs > 0:
-            extra_outputs = [gr.update(visible=False)] * difference_outputs
-            processed_outputs.extend(extra_outputs)
-        # If more outputs than expected, cap the outputs to the expected number
-        elif difference_outputs < 0:
-            processed_outputs = processed_outputs[:difference_outputs]
-        
-        return tuple(processed_outputs) if len(processed_outputs) > 1 else processed_outputs[0]
-    else:
-        if(response.status_code == 409):
-            raise gr.Error(f"Sorry, the Cog image is still processing. Try again in a bit.")
-        raise gr.Error(f"The submission failed! Error: {{response.status_code}}")\n"""
-
-    interface_string = f"""title = "{title}"
-model_description = "{model_description}"
-
-app = gr.Interface(
-    fn=predict,
-    inputs=inputs,
-    outputs=outputs,
-    title=title,
-    description=model_description,
-    allow_flagging="never",
-)
-app.launch(share=True)
-"""
-
-    app_string = f"""import gradio as gr
-from urllib.parse import urlparse
-import requests
-import time
-import os
-
-from utils.gradio_helpers import parse_outputs, process_outputs
-
-{inputs_string}
-{outputs_string}
-{definition_string}
-    {headers_string}
-    {payload_string}
-    {request_string}
-    {result_string}
-{interface_string}
-"""
-    return app_string