serve/vote_utils.py

import datetime
import time
import json
import uuid
import gradio as gr
from pathlib import Path
from .utils import *
from .log_utils import build_logger
from .constants import IMAGE_DIR

t2s_logger = build_logger("gradio_web_server_text2shape", "gr_web_text2shape.log") # t2s = image generation, loggers for single model direct chat
t2s_multi_logger = build_logger("gradio_web_server_text2shape_multi", "gr_web_text2shape_multi.log") # t2s_multi = image generation multi, loggers for side-by-side and battle
i2s_logger = build_logger("gradio_web_server_image2shape", "gr_web_image2shape.log") # i2s = image editing, loggers for single model direct chat
i2s_multi_logger = build_logger("gradio_web_server_image2shape_multi", "gr_web_image2shape_multi.log") # i2s_multi = image editing multi, loggers for side-by-side and battle

def vote_last_response_t2s(state, vote_type, model_selector, request: gr.Request):
    with open(get_conv_log_filename(), "a") as fout:
        data = {
            "tstamp": round(time.time(), 4),
            "type": vote_type,
            "model": model_selector,
            "state": state.dict(),
            "ip": get_ip(request),
        }
        fout.write(json.dumps(data) + "\n")
        append_json_item_on_log_server(data, get_conv_log_filename())
        
def vote_last_response_t2s_multi(states, vote_type, model_selectors, request: gr.Request):
    with open(get_conv_log_filename(), "a") as fout:
        data = {
            "tstamp": round(time.time(), 4),
            "type": vote_type,
            "models": [x for x in model_selectors],
            "states": [x.dict() for x in states],
            "ip": get_ip(request),
        }
        fout.write(json.dumps(data) + "\n")
        append_json_item_on_log_server(data, get_conv_log_filename())
    # for state in states:
    #     output_file = f'{IMAGE_DIR}/text2shape/{state.conv_id}.png'
    #     with open(output_file, 'w') as f:
    #         state.output.save(f, 'PNG')
    #     save_image_file_on_log_server(output_file)

def vote_last_response_i2s(state, vote_type, model_selector, request: gr.Request):
    with open(get_conv_log_filename(), "a") as fout:
        data = {
            "tstamp": round(time.time(), 4),
            "type": vote_type,
            "model": model_selector,
            "state": state.dict(),
            "ip": get_ip(request),
        }
        fout.write(json.dumps(data) + "\n")
        append_json_item_on_log_server(data, get_conv_log_filename())
    # output_file = f'{IMAGE_DIR}/image2shape/{state.conv_id}.png'
    # source_file = f'{IMAGE_DIR}/image2shape/{state.conv_id}_source.png'
    # with open(output_file, 'w') as f:
    #     state.output.save(f, 'PNG')
    # with open(source_file, 'w') as sf:
    #     state.source_image.save(sf, 'PNG')
    # save_image_file_on_log_server(output_file)
    # save_image_file_on_log_server(source_file)
        
def vote_last_response_i2s_multi(states, vote_type, model_selectors, request: gr.Request):
    with open(get_conv_log_filename(), "a") as fout:
        data = {
            "tstamp": round(time.time(), 4),
            "type": vote_type,
            "models": [x for x in model_selectors],
            "states": [x.dict() for x in states],
            "ip": get_ip(request),
        }
        fout.write(json.dumps(data) + "\n")
        append_json_item_on_log_server(data, get_conv_log_filename())
    # for state in states:
    #     output_file = f'{IMAGE_DIR}/image2shape/{state.conv_id}.png'
    #     source_file = f'{IMAGE_DIR}/image2shape/{state.conv_id}_source.png'
    #     with open(output_file, 'w') as f:
    #         state.output.save(f, 'PNG')
    #     with open(source_file, 'w') as sf:
    #         state.source_image.save(sf, 'PNG')
    #     save_image_file_on_log_server(output_file)
    #     save_image_file_on_log_server(source_file)
            

## Text-to-Shape Generation (t2s) Single Model Direct Chat
def upvote_last_response_t2s(state, model_selector, request: gr.Request):
    ip = get_ip(request)
    t2s_logger.info(f"upvote. ip: {ip}")
    vote_last_response_t2s(state, "upvote", model_selector, request)
    return ("",) + (disable_btn,) * 3

def downvote_last_response_t2s(state, model_selector, request: gr.Request):
    ip = get_ip(request)
    t2s_logger.info(f"downvote. ip: {ip}")
    vote_last_response_t2s(state, "downvote", model_selector, request)
    return ("",) + (disable_btn,) * 3

def flag_last_response_t2s(state, model_selector, request: gr.Request):
    ip = get_ip(request)
    t2s_logger.info(f"flag. ip: {ip}")
    vote_last_response_t2s(state, "flag", model_selector, request)
    return ("",) + (disable_btn,) * 3


## Text-to-Shape Generation Multi (t2s_multi) Side-by-Side and Battle
def leftvote_last_response_t2s_named(
    state0, state1, model_selector0, model_selector1, request: gr.Request
):
    t2s_multi_logger.info(f"leftvote (named). ip: {get_ip(request)}")
    vote_last_response_t2s_multi(
        [state0, state1], "leftvote", [model_selector0, model_selector1], request
    )
    return ("",) + (disable_btn,) * 4

def rightvote_last_response_t2s_named(
    state0, state1, model_selector0, model_selector1, request: gr.Request
):
    t2s_multi_logger.info(f"rightvote (named). ip: {get_ip(request)}")
    vote_last_response_t2s_multi(
        [state0, state1], "rightvote", [model_selector0, model_selector1], request
    )
    return ("",) + (disable_btn,) * 4

def tievote_last_response_t2s_named(
    state0, state1, model_selector0, model_selector1, request: gr.Request
):
    t2s_multi_logger.info(f"tievote (named). ip: {get_ip(request)}")
    vote_last_response_t2s_multi(
        [state0, state1], "tievote", [model_selector0, model_selector1], request
    )
    return ("",) + (disable_btn,) * 4

def bothbad_vote_last_response_t2s_named(
    state0, state1, model_selector0, model_selector1, request: gr.Request
):
    t2s_multi_logger.info(f"bothbad_vote (named). ip: {get_ip(request)}")
    vote_last_response_t2s_multi(
        [state0, state1], "bothbad_vote", [model_selector0, model_selector1], request
    )
    return ("",) + (disable_btn,) * 4 


def leftvote_last_response_t2s_anony(
    state0, state1, model_selector0, model_selector1, request: gr.Request
):
    t2s_multi_logger.info(f"leftvote (named). ip: {get_ip(request)}")
    vote_last_response_t2s_multi(
        [state0, state1], "leftvote", [model_selector0, model_selector1], request
    )
    names = (gr.Markdown(f"### Model A: {state0.model_name}", visible=True), gr.Markdown(f"### Model B: {state1.model_name}", visible=True))
    return ("",) + (disable_btn,) * 4 + names

def rightvote_last_response_t2s_anony(
    state0, state1, model_selector0, model_selector1, request: gr.Request
):
    t2s_multi_logger.info(f"rightvote (named). ip: {get_ip(request)}")
    vote_last_response_t2s_multi(
        [state0, state1], "rightvote", [model_selector0, model_selector1], request
    )
    names = (gr.Markdown(f"### Model A: {state0.model_name}", visible=True), gr.Markdown(f"### Model B: {state1.model_name}", visible=True))
    return ("",) + (disable_btn,) * 4 + names

def tievote_last_response_t2s_anony(
    state0, state1, model_selector0, model_selector1, request: gr.Request
):
    t2s_multi_logger.info(f"tievote (named). ip: {get_ip(request)}")
    vote_last_response_t2s_multi(
        [state0, state1], "tievote", [model_selector0, model_selector1], request
    )
    names = (gr.Markdown(f"### Model A: {state0.model_name}", visible=True), gr.Markdown(f"### Model B: {state1.model_name}", visible=True))
    return ("",) + (disable_btn,) * 4 + names

def bothbad_vote_last_response_t2s_anony(
    state0, state1, model_selector0, model_selector1, request: gr.Request
):
    t2s_multi_logger.info(f"bothbad_vote (named). ip: {get_ip(request)}")
    vote_last_response_t2s_multi(
        [state0, state1], "bothbad_vote", [model_selector0, model_selector1], request
    )
    names = (gr.Markdown(f"### Model A: {state0.model_name}", visible=True), gr.Markdown(f"### Model B: {state1.model_name}", visible=True))
    return ("",) + (disable_btn,) * 4 + names

## Image-to-Shape (i2s) Single Model Direct Chat
def upvote_last_response_i2s(state, model_selector, request: gr.Request):
    ip = get_ip(request)
    i2s_logger.info(f"upvote. ip: {ip}")
    vote_last_response_i2s(state, "upvote", model_selector, request)
    return ("", "", gr.Image(height=512, width=512, type="pil"), "",) + (disable_btn,) * 3

def downvote_last_response_i2s(state, model_selector, request: gr.Request):
    ip = get_ip(request)
    i2s_logger.info(f"downvote. ip: {ip}")
    vote_last_response_i2s(state, "downvote", model_selector, request)
    return ("", "", gr.Image(height=512, width=512, type="pil"), "",) + (disable_btn,) * 3

def flag_last_response_i2s(state, model_selector, request: gr.Request):
    ip = get_ip(request)
    i2s_logger.info(f"flag. ip: {ip}")
    vote_last_response_i2s(state, "flag", model_selector, request)
    return ("", "", gr.Image(height=512, width=512, type="pil"), "",) + (disable_btn,) * 3


## Image-to-Shape Multi (i2s_multi) Side-by-Side and Battle
def leftvote_last_response_i2s_named(
    state0, state1, model_selector0, model_selector1, request: gr.Request
):
    i2s_multi_logger.info(f"leftvote (anony). ip: {get_ip(request)}")
    vote_last_response_i2s_multi(
        [state0, state1], "leftvote", [model_selector0, model_selector1], request
    )
    return gr.Image(height=512, width=512, type="pil") + (disable_btn,) * 4
        
def rightvote_last_response_i2s_named(
    state0, state1, model_selector0, model_selector1, request: gr.Request
):
    i2s_multi_logger.info(f"rightvote (anony). ip: {get_ip(request)}")
    vote_last_response_i2s_multi(
        [state0, state1], "rightvote", [model_selector0, model_selector1], request
    )
    return gr.Image(height=512, width=512, type="pil") + (disable_btn,) * 4

def tievote_last_response_i2s_named(
    state0, state1, model_selector0, model_selector1, request: gr.Request
):
    i2s_multi_logger.info(f"tievote (anony). ip: {get_ip(request)}")
    vote_last_response_i2s_multi(
        [state0, state1], "tievote", [model_selector0, model_selector1], request
    )
    return gr.Image(height=512, width=512, type="pil") + (disable_btn,) * 4

def bothbad_vote_last_response_i2s_named(
    state0, state1, model_selector0, model_selector1, request: gr.Request
):
    i2s_multi_logger.info(f"bothbad_vote (anony). ip: {get_ip(request)}")
    vote_last_response_i2s_multi(
        [state0, state1], "bothbad_vote", [model_selector0, model_selector1], request
    )
    return gr.Image(height=512, width=512, type="pil") + (disable_btn,) * 4


def leftvote_last_response_i2s_anony(
    state0, state1, model_selector0, model_selector1, request: gr.Request
):
    i2s_multi_logger.info(f"leftvote (anony). ip: {get_ip(request)}")
    vote_last_response_i2s_multi(
        [state0, state1], "leftvote", [model_selector0, model_selector1], request
    )
    # names = (
    #     "### Model A: " + state0.model_name,
    #     "### Model B: " + state1.model_name,
    # )
    # names = (state0.model_name, state1.model_name)
    names = (gr.Markdown(f"### Model A: {state0.model_name}", visible=True), gr.Markdown(f"### Model B: {state1.model_name}", visible=True))
    return gr.Image(height=512, width=512, type="pil") + (disable_btn,) * 4 + names
        
def rightvote_last_response_i2s_anony(
    state0, state1, model_selector0, model_selector1, request: gr.Request
):
    i2s_multi_logger.info(f"rightvote (anony). ip: {get_ip(request)}")
    vote_last_response_i2s_multi(
        [state0, state1], "rightvote", [model_selector0, model_selector1], request
    )
    # names = (
    #     "### Model A: " + state0.model_name,
    #     "### Model B: " + state1.model_name,
    # )
    names = (gr.Markdown(f"### Model A: {state0.model_name}", visible=True), gr.Markdown(f"### Model B: {state1.model_name}", visible=True))
    return gr.Image(height=512, width=512, type="pil") + (disable_btn,) * 4 + names

def tievote_last_response_i2s_anony(
    state0, state1, model_selector0, model_selector1, request: gr.Request
):
    i2s_multi_logger.info(f"tievote (anony). ip: {get_ip(request)}")
    vote_last_response_i2s_multi(
        [state0, state1], "tievote", [model_selector0, model_selector1], request
    )
    names = (gr.Markdown(f"### Model A: {state0.model_name}", visible=True), gr.Markdown(f"### Model B: {state1.model_name}", visible=True))
    return gr.Image(height=512, width=512, type="pil") + (disable_btn,) * 4 + names

def bothbad_vote_last_response_i2s_anony(
    state0, state1, model_selector0, model_selector1, request: gr.Request
):
    i2s_multi_logger.info(f"bothbad_vote (anony). ip: {get_ip(request)}")
    vote_last_response_i2s_multi(
        [state0, state1], "bothbad_vote", [model_selector0, model_selector1], request
    )
    names = (gr.Markdown(f"### Model A: {state0.model_name}", visible=True), gr.Markdown(f"### Model B: {state1.model_name}", visible=True))
    return gr.Image(height=512, width=512, type="pil") + (disable_btn,) * 4 + names


share_js = """
function (a, b, c, d) {
    const captureElement = document.querySelector('#share-region-named');
    html2canvas(captureElement)
        .then(canvas => {
            canvas.style.display = 'none'
            document.body.appendChild(canvas)
            return canvas
        })
        .then(canvas => {
            const image = canvas.toDataURL('image/png')
            const a = document.createElement('a')
            a.setAttribute('download', 'chatbot-arena.png')
            a.setAttribute('href', image)
            a.click()
            canvas.remove()
        });
    return [a, b, c, d];
}
"""
def share_click_t2s_multi(state0, state1, model_selector0, model_selector1, request: gr.Request):
    t2s_multi_logger.info(f"share (anony). ip: {get_ip(request)}")
    if state0 is not None and state1 is not None:
        vote_last_response_t2s_multi(
            [state0, state1], "share", [model_selector0, model_selector1], request
        )

def share_click_i2s_multi(state0, state1, model_selector0, model_selector1, request: gr.Request):
    i2s_multi_logger.info(f"share (anony). ip: {get_ip(request)}")
    if state0 is not None and state1 is not None:
        vote_last_response_i2s_multi(
            [state0, state1], "share", [model_selector0, model_selector1], request
        )