app.py

# pylint: disable=no-name-in-module
# pylint: disable=no-member

"""
Author : Bastien GUILLAUME
Version : 0.0.1
Date : 2023-03-16

Title : Inference With Gradio running an onnxruntime backend
"""

import inspect
import os
import shutil
from pathlib import Path

import gradio as gr
import requests

import sys
sys.path.append(".")

from config_parser import *
from inferencer import *

gr.close_all()


def format_examples(task_number, product, product_example):
    response = requests.get(product_example)
    examples_folder = Path(f"examples/{product}")
    os.makedirs(examples_folder, exist_ok=True)
    filepath = Path(examples_folder / f'{product_example.split("/")[-1]}')
    if filepath.exists():
        pass
    else:
        with open(filepath, "wb") as f:
            f.write(response.content)
    # return [f"task{task_number+1}", product, filepath]
    return [filepath]


def generate_parralel_interface(task_number, product):
    logging.log(level=logging.INFO, msg=f"Entering generate_parralel_interface")
    generate_parralel_interface_list = []
    number_of_model = len(config["tasks"][f"task{task_number+1}"]["models"][product])
    task_info = config["tasks"][f"task{task_number+1}"]
    product_examples = (
        [
            format_examples(task_number, product, product_example)
            for product_example in task_info["examples"][product]
        ]
        if "examples" in task_info and product in task_info["examples"]
        else []
    )
    # product_examples = [test(task_number, product,product_example) for product_example in product_examples]
    # print(product_examples)
    # product_examples = []
    logging.log(level=logging.INFO, msg=f"Real Number of model: {number_of_model}")
    logging.log(level=logging.DEBUG, msg=f"List of examples : {product_examples}")
    for model_number in range(0, number_of_model):
        logging.log(level=logging.INFO, msg=f"Building Interface for {product}")
        logging.log(
            level=logging.INFO, msg=f"Building Interface for Model n°{model_number+1}"
        )
        generate_parralel_interface_list.append(
            create_interface(task_number, product, model_number)
        )
    generated_parralel_interface = gr.Parallel(
        *generate_parralel_interface_list,
        title=tasks[task_number],
        description=tasks_description[task_number],
        allow_flagging="never",
        css="footer {visibility: hidden} body}, .gradio-container {background-color: white}",
        examples=product_examples,
    )
    logging.log(level=logging.INFO, msg=f"Leaving generate_parralel_interface")
    return generated_parralel_interface


def create_interface(task_number, product, model_number):
    logging.log(level=logging.INFO, msg=f"Entering create_interface")
    title = tasks[task_number]
    description = tasks_description[task_number]
    product_lists = tasks_products[task_number]
    model_info = config["tasks"][f"task{task_number+1}"]["models"][product][
        model_number
    ]
    model_name = (
        f'Résultst of {model_info["name"]}' if "name" in model_info else "Résultats"
    )
    model_uuid = model_info["path"].split("/")[-2:-1][0]
    task_info = config["tasks"][f"task{task_number+1}"]
    product_examples = (
        [
            format_examples(task_number, product, product_example)
            for product_example in task_info["examples"][product]
        ]
        if "examples" in task_info
        and product in task_info["examples"]
        and inspect.currentframe().f_back.f_code.co_name == "interface_builder"
        else []
    )
    logging.log(level=logging.INFO, msg=f"Title : {title}")
    logging.log(level=logging.INFO, msg=f"Description : {description}")
    logging.log(level=logging.INFO, msg=f"Product list : {product_lists}")
    logging.log(level=logging.INFO, msg=f"Model Number: {model_number}")
    logging.log(level=logging.DEBUG, msg=f"List of examples : {product_examples}")

    created_interface = gr.Interface(
        title=title,
        description=description,
        fn=inferencer_arr[f"task{task_number+1}"][product][str(model_number)][model_uuid][
            "function"
        ],
        # fn=lambda x: x,
        allow_flagging="never",
        css="footer {visibility: hidden} body}, .gradio-container {background-color: white}",
        inputs=[
            # gr.Textbox(
            #     value=f"task{task_number+1}",
            #     label="Tasks",
            #     visible=False,
            #     interactive=False,
            # ),
            # gr.Dropdown(
            #     tasks_products[task_number],
            #     type="value",
            #     value=product,
            #     label="Choix",
            #     # visible=True if len(tasks_products[task_number]) > 1 else False,
            #     visible=False,
            #     info="Sur quel type de produit, voulez vous lancer l'analyse ?",
            # ),
            gr.Image(
                label="Image à analyser",
                shape=None,
                image_mode="RGB",
                invert_colors=False,
                source="upload",
                tool="editor",
                type="numpy",
            ),
        ],
        outputs=gr.Label(
            label=model_name,
        ),
        examples=product_examples,
    )
    logging.log(level=logging.INFO, msg=f"Leaving create_interface")
    return created_interface


def interface_builder():
    interface_builder_list = []
    interface_builder_dict = {}
    logging.log(level=logging.INFO, msg=f"Building Interfaces")
    logging.log(level=logging.INFO, msg=f"Number of task(s) : {len(tasks)}")
    for task_number in range(0, len(tasks)):
        logging.log(level=logging.INFO, msg=f"Treating task n°{task_number+1}")
        logging.log(level=logging.INFO, msg=f"Building Interface n°{task_number+1}")
        interface_builder_dict[tasks[task_number]] = {}
        product_list = list(config["tasks"][f"task{task_number+1}"]["models"].keys())
        logging.log(level=logging.DEBUG, msg=f"Products : {product_list}")
        interface_builder_product_level_list = []
        for product in product_list:
            logging.log(level=logging.INFO, msg=f"Product : {product}")
            interface_builder_dict[tasks[task_number]][product] = []
            if len(config["tasks"][f"task{task_number+1}"]["models"][product]) > 1:
                generated_parralel_interface = generate_parralel_interface(
                    task_number, product
                )
                interface_builder_dict[tasks[task_number]][product].append(
                    generated_parralel_interface
                )
                interface_builder_product_level_list.append(
                    generated_parralel_interface
                )
            else:
                generated_interface = create_interface(
                    task_number=task_number, product=product, model_number=0
                )
                interface_builder_dict[tasks[task_number]][product].append(
                    generated_interface
                )
                interface_builder_product_level_list.append(generated_interface)
        interface_builder_list.append(
            gr.TabbedInterface(
                interface_list=interface_builder_product_level_list,
                tab_names=product_list,
            )
        )
    logging.log(level=logging.INFO, msg=f"Interfaces ready\n")
    # logging.log(level=logging.INFO, msg=f"Interfaces Dict {interface_dict}")
    logging.log(level=logging.DEBUG, msg=f"Interfaces List {interface_builder_list}")
    return interface_builder_list


generated_interface = interface_builder()

iface_generated = gr.TabbedInterface(
    interface_list=generated_interface,
    tab_names=tasks_name,
    title=f"{title} - {description}",
    css="footer {visibility: hidden} body}, .gradio-container {background-color: white}",
)

if __name__ == "__main__":
    logging.log(level=logging.INFO, msg="Starting the Gradio server...")
    iface_generated.launch(
        server_name="0.0.0.0"
    )
    # shutil.rmtree("examples")
    logging.log(level=logging.INFO, msg="Stopping the Gradio server...")