app.py

import spaces
import argparse
import json
import numpy as np
import os

import gradio as gr
import requests
from openai import OpenAI
from func_timeout import FunctionTimedOut, func_timeout
from tqdm import tqdm

HUGGINGFACE=True
MOCK = False
TEST_FOLDER = "c4f5"

if HUGGINGFACE:
    MODEL_NAME="xu3kev/deepseekcoder-7b-logo-pbe"
    import torch
    from transformers import AutoModelForCausalLM, AutoTokenizer
    hug_model = AutoModelForCausalLM.from_pretrained(MODEL_NAME, torch_dtype=torch.float16,).to('cuda')
    hug_tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)

INPUT_STRUCTION_TEMPLATE = """Here is a gray scale images representing with integer values 0-9.
{image_str}
Please write a Python program that generates the image using our own custom turtle module"""

PROMPT_TEMPLATE = "### Instruction:\n{input_struction}\n### Response:\n"

TEST_IMAGE_STR ="00000000000000000000000000000000\n00000000000000000000000000000000\n00000000000000000000000000000000\n00000000000000000000000000000000\n00000000000000000000000000000000\n00000000000000000000000000000000\n00000000000000000000000000000000\n00000000000000000000000000000000\n00000000000000000000000000000000\n00000000000000000000000000000000\n00000000000000000000000000000000\n00000000000000000000000000000000\n00000000000000000000000000000000\n00000000000001222222000000000000\n00000000000002000002000000000000\n00000000000002022202000000000000\n00000000000002020202000000000000\n00000000000002020002000000000000\n00000000000002022223000000000000\n00000000000002000000000000000000\n00000000000002000000000000000000\n00000000000000000000000000000000\n00000000000000000000000000000000\n00000000000000000000000000000000\n00000000000000000000000000000000\n00000000000000000000000000000000\n00000000000000000000000000000000\n00000000000000000000000000000000\n00000000000000000000000000000000\n00000000000000000000000000000000\n00000000000000000000000000000000\n00000000000000000000000000000000"

MOCK_RESPONSE = [
"""for i in range(7):
    with fork_state():
        for j in range(4):
            forward(2*i)
            left(90.0)
"""
] * 16

LOGO_HEADER = """from myturtle_cv import Turtle
from myturtle import HALF_INF, INF, EPS_DIST, EPS_ANGLE

turtle = Turtle()
def forward(dist):
    turtle.forward(dist)
def left(angle):
    turtle.left(angle)
def right(angle):   
    turtle.right(angle)
def teleport(x, y, theta):
    turtle.teleport(x, y, theta)
def penup():
    turtle.penup()
def pendown():
    turtle.pendown()
def position():
    return turtle.x, turtle.y
def heading():
    return turtle.heading
def isdown():
    return turtle.is_down
def fork_state():
    \"\"\"
    Fork the current state of the turtle.

    Usage:
    with fork_state():
        forward(100)
        left(90)
        forward(100)
    \"\"\"
    return turtle._TurtleState(turtle)"""


def invert_colors(image):
    """
    Inverts the colors of the input image.
    Args:
    - image (dict): Input image dictionary from Sketchpad.

    Returns:
    - numpy array: Color-inverted image array.
    """
    # Extract image data from the dictionary and convert to NumPy array
    image_data = image['layers'][0]
    image_array = np.array(image_data)
    
    
    # Invert colors
    inverted_image = 255 - image_array
    return inverted_image

def crop_image_to_center(image, target_height=512, target_width=512, detect_cropping_non_white=False):
    # Calculate the center of the original image
    h, w = image.shape
    center_y, center_x = h // 2, w // 2

    # Calculate the top-left corner of the crop area
    start_x = max(center_x - target_width // 2, 0)
    start_y = max(center_y - target_height // 2, 0)

    # Ensure the crop area does not exceed the image boundaries
    end_x = min(start_x + target_width, w)
    end_y = min(start_y + target_height, h)

    # Crop the image
    cropped_image = image[start_y:end_y, start_x:end_x]
    if detect_cropping_non_white:
        cropping_non_white = False
        all_black_pixel_count = np.sum(image < 50)
        cropped_black_pixel_count = np.sum(cropped_image < 50)
        if cropped_black_pixel_count < all_black_pixel_count:
            cropping_non_white = True
    
    # If the cropped image is smaller than the target, pad it to the required size
    if cropped_image.shape[0] < target_height or cropped_image.shape[1] < target_width:
        pad_height = target_height - cropped_image.shape[0]
        pad_width = target_width - cropped_image.shape[1]
        cropped_image = cv2.copyMakeBorder(cropped_image, 0, pad_height, 0, pad_width, cv2.BORDER_CONSTANT, value=255) # Using white padding

    if detect_cropping_non_white:
        if cropping_non_white:
            return None
        else:
            return cropped_image
    else:
        return cropped_image

def downscale_image(image, block_size=8, black_threshold=50, gray_level=10, return_level=False):
    # Calculate the size of the output image
    h, w = image.shape
    new_h, new_w = h // block_size, w // block_size
    
    # Initialize the output image
    downscaled = np.zeros((new_h, new_w), dtype=np.uint8)
    image_with_level = np.zeros((new_h, new_w), dtype=np.uint8)    
    for i in range(0, h, block_size):
        for j in range(0, w, block_size):
            # Extract the block
            block = image[i:i+block_size, j:j+block_size]
            
            # Calculate the proportion of black pixels
            black_pixels = np.sum(block < black_threshold)
            total_pixels = block_size * block_size
            proportion_of_black = black_pixels / total_pixels
            discrete_gray_step = 1 / gray_level
            if proportion_of_black >= 0.95:
                proportion_of_black = 0.94
            proportion_of_black = round (proportion_of_black / discrete_gray_step) * discrete_gray_step
            # check that gray level is descretize to 0 ~ gray_level-1
            try:
                assert 0 <= round(proportion_of_black / discrete_gray_step) < gray_level
            except:
                breakpoint()
            
            # Assign the new grayscale value (inverse proportion if needed)
            grayscale_value = int(proportion_of_black * 255)
            
            # Assign to the downscaled image
            downscaled[i // block_size, j // block_size] = grayscale_value
            image_with_level[i // block_size, j // block_size] = int(proportion_of_black // discrete_gray_step)
    if return_level:
        return downscaled, image_with_level 
    else:
        return downscaled


PORT = 8008
MODEL_NAME="./axolotl/lora-logo_fix_full_deepseek33b_ds33i_epoch3_lr_0.0002_alpha_512_r_512_merged"
MODEL_NAME="./axolotl/lora-logo_fix_full_deepseek7b_ds33i_lr_0.0002_alpha_512_r_512_merged"


def generate_grid_images(gif_results):
    import matplotlib.patches as patches
    import matplotlib.pyplot as plt
    num_rows, num_cols = 8,8
    fig, axes = plt.subplots(num_rows, num_cols, figsize=(12, 12))
    fig.tight_layout(pad=0)

    # Plot each image with its AST count as a caption
    # load all jpg images in the folder
    import glob
    import os
    print(f"load file path")
    image_files = glob.glob(os.path.join(folder, "*.jpg"))
    print(f"load file path done")

    images = []
    for idx, image_file in enumerate(image_files):
        img = load_img(image_file)
        images.append(img)
    
    print(f"Loaded {len(images)} images")

    for idx, img in tqdm(enumerate(images)):
        if idx >= num_rows * num_cols:
            break
        row, col = divmod(idx, num_cols)
        ax = axes[row, col]
        if img is None:
            ax.axis('off')
            continue
        try:
            ax.imshow(img, cmap='gray')
        except:
            breakpoint()
        ax.axis('off')

    # Hide remaining empty subplots
    for idx in range(len(images), num_rows * num_cols):
        row, col = divmod(idx, num_cols)
        axes[row, col].axis('off')

    # convert fig to numpy return image array
    fig.canvas.draw()
    image_array = np.array(fig.canvas.renderer.buffer_rgba())
    plt.close(fig)
    return image_array

@spaces.GPU
def llm_call(question_prompt, model_name, 
    temperature=1, max_tokens=320, 
    top_p=1, n_samples=64, stop=None):
    if HUGGINGFACE:
        model_inputs = hug_tokenizer([question_prompt], return_tensors="pt").to('cuda')
        generated_ids = hug_model.generate(**model_inputs, max_length=1400, temperature=1, num_return_sequences=9, do_sample=True)
        responses = hug_tokenizer.batch_decode(generated_ids, skip_special_tokens=True)
        codes = []
        for response in responses:
            codes.append(response[len(question_prompt):].strip()+'\n')
        return codes

    else:
        client = OpenAI(base_url=f"http://localhost:{PORT}/v1", api_key="empty")

        response = client.completions.create(
            prompt=question_prompt,
            model=model_name,
            temperature=temperature,
            max_tokens=max_tokens,
            top_p=top_p,
            frequency_penalty=0,
            presence_penalty=0,
            n=n_samples,
            stop=stop
        )
        codes = []
        for i, choice in enumerate(response.choices):
            print(f"Choice {i}: {choice.text}")
            codes.append(choice.text)

        return codes


import cv2
def load_img(path):
    img = cv2.imread(path, cv2.IMREAD_GRAYSCALE)
    
    # Threshold the image to create a binary image (white background, black object)
    _, thresh = cv2.threshold(img, 240, 255, cv2.THRESH_BINARY)
    
    # Invert the binary image
    thresh_inv = cv2.bitwise_not(thresh)
    
    # Find the bounding box of the non-white area
    x, y, w, h = cv2.boundingRect(thresh_inv)
    
    # Extract the ROI (region of interest) of the non-white area
    roi = img[y:y+h, x:x+w]
    
    # If the ROI is larger than 200x200, resize it
    if w > 256 or h > 256:
        scale = min(256 / w, 256 / h)
        new_w = int(w * scale)
        new_h = int(h * scale)
        roi = cv2.resize(roi, (new_w, new_h), interpolation=cv2.INTER_AREA)
        w, h = new_w, new_h

    # Create a new 200x200 white image
    centered_img = np.ones((256, 256), dtype=np.uint8) * 255
    
    # Calculate the position to center the ROI in the 200x200 image
    start_x = max(0, (256 - w) // 2)
    start_y = max(0, (256 - h) // 2)
    
    # Place the ROI in the centered position
    centered_img[start_y:start_y+h, start_x:start_x+w] = roi
    
    return centered_img


def run_code(new_folder, counter, code):
    import matplotlib
    fname = f"{new_folder}/logo_{counter}_.jpg"
    counter += 1
    code_with_header_and_save= f"""
{LOGO_HEADER}
{code}
# turtle.save('{fname}')
gif = turtle.save_gif('')
"""
    try:
        results = {}
        func_timeout(30, exec, args=(code_with_header_and_save, results))
        # exec(code_with_header_and_save, globals())
        if 'gif' in results:
            return results['gif']
    except FunctionTimedOut:
        print("Timeout")
    except Exception as e:
        print(e)

def run(img_str):
    prompt = PROMPT_TEMPLATE.format(input_struction=INPUT_STRUCTION_TEMPLATE.format(image_str=img_str))
    if not MOCK:
        responses = llm_call(prompt, MODEL_NAME)
        print(responses)
        codes = responses
    else:
        codes = MOCK_RESPONSE
    
    gradio_test_images_folder = "gradio_test_images"
    # import os
    # os.makedirs(gradio_test_images_folder, exist_ok=True)

    counter = 0
    # generate a random hash id
    import hashlib
    import random
    random_id = hashlib.md5(str(random.random()).encode()).hexdigest()[0:4]
    new_folder = os.path.join(gradio_test_images_folder, random_id)
    # os.makedirs(new_folder, exist_ok=True)



    for code in tqdm(codes):
        pass

    from concurrent.futures import ProcessPoolExecutor
    from concurrent.futures import as_completed
    gif_results = []
    with ProcessPoolExecutor(max_workers=8) as executor:
        futures = [executor.submit(run_code, new_folder, i, code) for i, code in enumerate(codes)]
        for future in as_completed(futures):
            try:
                gif_results.append(future.result())
            except Exception as exc:
                print(f'Generated an exception: {exc}')

        # with open("temp.py", 'w') as f: 
        #     f.write(code_with_header_and_save)

        # p = subprocess.Popen(["python", "temp.py"], stderr=subprocess.PIPE, stdout=subprocess.PIPE, env=my_env)
        # out, errs = p.communicate()
        # out, errs, = out.decode(), errs.decode()
    # render
    print(random_id)
    folder_path = f"gradio_test_images/{random_id}"
    return gif_results, codes


def test_gen_img_wrapper(_):
    return generate_grid_images(f"gradio_test_images/{TEST_FOLDER}")

def int_img_to_str(integer_img):
    lines = [] 
    for row in integer_img:
        print("".join([str(x) for x in row]))
        lines.append("".join([str(x) for x in row]))
    image_str = "\n".join(lines)
    return image_str


def create_tmp_folder():
    import hashlib
    import random
    random_id = hashlib.md5(str(random.random()).encode()).hexdigest()[0:4]
    folder_name = f"generated_images_{random_id}"
    os.makedirs(folder_name, exist_ok=True)
    return folder_name


def img_to_code_img(sketchpad_img):
    img = sketchpad_img['layers'][0]
    image_array = np.array(img)
    image_array = 255 - image_array[:,:,3]

    # height, width = image_array.shape
    # output_size = 512
    # block_size = max(height, width) // output_size
    
    # # Create new downscaled image array
    # new_image_array = np.zeros((output_size, output_size), dtype=np.uint8)
    # # Process each block
    # for i in range(output_size):
    #     for j in range(output_size):
    #         # Define the block
    #         block = image_array[i*block_size:(i+1)*block_size, j*block_size:(j+1)*block_size]
    #         # Calculate the number of pixels set to 255 in the block
    #         white_pixels = np.sum(block == 255)
    #         # Set the new pixel value
    #         if white_pixels >= (block_size * block_size) / 2:
    #             new_image_array[i, j] = 255
    new_image_array= image_array

    _, int_img = downscale_image(new_image_array, block_size=16, return_level=True)

    if int_img is not None:
        img_str = int_img_to_str(int_img)
        print(img_str)

    gif_results, codes = run(img_str)

    # generated_grid_img = generate_grid_images()

    # return generated_grid_img[0] 
    
    folder = create_tmp_folder()
    for i in range(len(gif_results)):
        if gif_results[i]:
            with open(f"{folder}/img{i}.gif", "wb") as f:
                f.write(gif_results[i])
    return [f"{folder}/img{i}.gif" for i in range(len(gif_results))]


def main():
    """
    Sets up and launches the Gradio demo.
    """
    import gradio as gr
    from gradio import Brush
    theme = gr.themes.Default().set(
    ) 
    with gr.Blocks(theme=theme) as demo:
        gr.Markdown('# Visual Program Synthesis with LLM')
        gr.Markdown("""LOGO/Turtle graphics Programming-by-Example problems aims to synthesize a program that generates the given target image, where the program uses drawing library similar to Python Turtle.""")
        gr.Markdown("""Here we can draw a target image using the sketchpad, and see what kinds of graphics program LLM generates. To allow the LLM to visually perceive the input image, we convert the image to ASCII strings.""")
        gr.Markdown("## Draw logo")
        with gr.Row():
            with gr.Column():
                canvas = gr.Sketchpad(canvas_size=(512,512), brush=Brush(colors=["black"], default_size=2, color_mode='fixed'))
                submit_button = gr.Button("Submit")
            output_gallery = gr.Gallery(
        label="Generated images", show_label=False, elem_id="gallery"
    , columns=[3], rows=[3], object_fit="contain", height="auto")
            # output_image = gr.Image(label="output")
        
        submit_button.click(img_to_code_img, inputs=canvas, outputs=output_gallery)
        demo.load(
        None,
        None,
        js="""
  () => {
  const params = new URLSearchParams(window.location.search);
  if (!params.has('__theme')) {
    params.set('__theme', 'light');
    window.location.search = params.toString();
  }
  }""",
        )

    demo.launch(share=True)

if __name__ == "__main__":
    # parser = argparse.ArgumentParser()
    # parser.add_argument("--host", type=str, default=None)
    # parser.add_argument("--port", type=int, default=8001)
    # parser.add_argument("--model-url",
    #                     type=str,
    #                     default="http://localhost:8000/generate")
    # args = parser.parse_args()

    # main()
    # run()
    
    # demo = build_demo()
    # demo.queue().launch(server_name=args.host,
    #                     server_port=args.port,
    #                     share=True)
    main()