app.py

import os
import sys

import gradio as gr

sys.path.append("./ctm")
from ctm.ctms.ctm_base import BaseConsciousnessTuringMachine
from PIL import Image
import io
import base64

ctm = BaseConsciousnessTuringMachine()
ctm.add_supervisor("gpt4_supervisor")

DEPLOYED = os.getenv("DEPLOYED", "true").lower() == "true"



def convert_base64(image_array):
    if image_array is None:
        return None
    image = Image.fromarray(image_array)
    buffer = io.BytesIO()
    image.save(buffer, format="PNG")
    byte_data = buffer.getvalue()
    base64_string = base64.b64encode(byte_data).decode("utf-8")
    return base64_string



def add_processor(processor_name, display_name, state):
    print("add processor ", processor_name)
    ctm.add_processor(processor_name)
    print(ctm.processor_group_map)
    print(len(ctm.processor_list))
    return gr.Button(
        value=display_name,
        elem_id="selected"
    )


def processor_tab():
    # Categorized model names
    text_processors = [
        "gpt4_text_emotion_processor",
        "gpt4_text_summary_processor",
        "gpt4_speaker_intent_processor",
        "roberta_text_sentiment_processor",
    ]
    vision_processors = [
        "gpt4v_cloth_fashion_processor",
        "gpt4v_face_emotion_processor",
        "gpt4v_ocr_processor",
        "gpt4v_posture_processor",
        "gpt4v_scene_location_processor",
    ]

    with gr.Accordion('Select your processors here.'):
        with gr.Row():
            with gr.Blocks():
                for model_name in text_processors:
                    display_name = (
                        model_name.replace("processor", "")
                        .replace("_", " ")
                        .title()
                    )

                    button = gr.Button(
                        value=display_name,
                        elem_id="unselected"
                    )
                    processor_name = gr.Textbox(
                        value=model_name, visible=False
                    )
                    display_name = gr.Textbox(
                        value=display_name, visible=False
                    )
                    button.click(
                        fn=add_processor,
                        inputs=[processor_name, display_name, gr.State()],
                        outputs=[button],
                    )

                for model_name in vision_processors:
                    display_name = (
                        model_name.replace("processor", "")
                        .replace("_", " ")
                        .title()
                    )

                    button = gr.Button(
                        value=display_name,
                        elem_id="unselected"
                    )
                    processor_name = gr.Textbox(
                        value=model_name, visible=False
                    )
                    display_name = gr.Textbox(
                        value=display_name, visible=False
                    )
                    button.click(
                        fn=add_processor,
                        inputs=[processor_name, display_name, gr.State()],
                        outputs=[button],
                    )


def forward(query, content, image, state):
    image = convert_base64(image)
    state["question"] = query
    ask_processors_output_info, state = ask_processors(
        query, content, image, state
    )
    uptree_competition_output_info, state = uptree_competition(state)
    ask_supervisor_output_info, state = ask_supervisor(state)

    ctm.downtree_broadcast(state["winning_output"])
    ctm.link_form(state["processor_output"])
    return (
        ask_processors_output_info,
        uptree_competition_output_info,
        ask_supervisor_output_info,
        state,
        gr.Button(
            value="Update CTM",
            elem_id="selected-ctm",
        )
    )


def ask_processors(query, text, image, state):
    # Simulate processing here
    processor_output = ctm.ask_processors(
        query=query,
        text=text,
        image=image,
    )
    output_info = ""
    for name, info in processor_output.items():
        gist = info["gist"].replace("\n", "").strip()
        output_info += f"<{name}>\n{gist}\n\n"
    state["processor_output"] = processor_output
    return output_info, state


def uptree_competition(state):
    winning_output = ctm.uptree_competition(state["processor_output"])
    state["winning_output"] = winning_output
    output_info = (
        "<{}>\n{}".format(
            winning_output["name"], winning_output["gist"].replace("\n", "").strip()
        )
    )
    return output_info, state


def ask_supervisor(state):
    question = state["question"]
    winning_output = state["winning_output"]
    answer, score = ctm.ask_supervisor(question, winning_output)
    output_info = answer
    state["answer"] = answer
    state["score"] = score
    return output_info, state


def input_tab():
    state = gr.State({})  # State to hold and pass values

    with gr.Accordion("Enter your input here."):
        with gr.Row():
            query = gr.Textbox(label="Query", placeholder="Type your query here", lines=3)

        with gr.Row():
            text = gr.Textbox(label="Text Input", placeholder="Input text data", lines=11)
            image = gr.Image(label="Image Input")
    return query, text, image, state

def output_tab(query, text, image, state):

    with gr.Accordion("Check your outputs here."):
        processors_output = gr.Textbox(label="STM Chunks", visible=True, lines=5)
        competition_output = gr.Textbox(label="Winning Chunk", visible=True, lines=3)
        supervisor_output = gr.Textbox(label="Answer", visible=True, lines=2)
        forward_button = gr.Button("Launch CTM")

    forward_button.click(
        fn=forward,
        inputs=[query, text, image, state],
        outputs=[
            processors_output,
            competition_output,
            supervisor_output,
            state,
            forward_button,
        ],
    )



def main():
    with gr.Blocks(
        theme="gradio/monochrome",
        css="""#chat_container {height: 820px; width: 1000px; margin-left: auto; margin-right: auto;}
                #chatbot {height: 600px; overflow: auto;}
                #create_container {height: 750px; margin-left: 0px; margin-right: 0px;}
                #tokenizer_renderer span {white-space: pre-wrap}
                #selected {background-color: orange; width: 180px}
                #unselected {width: 180px}
                #selected-ctm {background-color: orange;}
                """,
    ) as demo:
        gr.Image("images/banner.jpg", elem_id="banner-image", show_label=False)
        with gr.Row():
            with gr.Column():
                processor_tab()
                query, text, image, state = input_tab()
            with gr.Column():
                output_tab(query, text, image, state)
    return demo


def start_demo():
    demo = main()
    if DEPLOYED:
        demo.queue(api_open=False).launch(show_api=False)
    else:
        demo.queue()
        demo.launch(share=False, server_name="0.0.0.0")


if __name__ == "__main__":
    start_demo()