utils/gradio_helpers.py

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