app.py

import gradio as gr
import os
import sys
from pathlib import Path
import random
import string
import time
from queue import Queue
from threading import Thread

text_gen=gr.Interface.load("spaces/trysem/visua")
proc1=gr.Interface.load("models/trysem/dreamlike-photoreal-2.0", live=True)

def restart_script_periodically():
    while True:
        time.sleep(600) # 10 minutes
        os.execl(sys.executable, sys.executable, *sys.argv)

restart_thread = Thread(target=restart_script_periodically, daemon=True)
restart_thread.start()

def reset_queue_periodically():
    start_time = time.time()
    while True:
        if time.time() - start_time >= 120: # 5 minutes
            queue.queue.clear()
            start_time = time.time()

reset_queue_thread = Thread(target=reset_queue_periodically, daemon=True)
reset_queue_thread.start()

queue = Queue()
queue_threshold = 50

def add_random_noise(prompt, noise_level=0.07):
    # Get the percentage of characters to add as noise
    percentage_noise = noise_level * 5
    # Get the number of characters to add as noise
    num_noise_chars = int(len(prompt) * (percentage_noise/100))
    # Get the indices of the characters to add noise to
    noise_indices = random.sample(range(len(prompt)), num_noise_chars)
    # Add noise to the selected characters
    prompt_list = list(prompt)
    noise_chars = string.ascii_letters + string.punctuation + ' '
    for index in noise_indices:
        prompt_list[index] = random.choice(noise_chars)
    return "".join(prompt_list)


def send_it1(inputs, noise_level, proc1=proc1):
    prompt_with_noise = add_random_noise(inputs, noise_level)
    while queue.qsize() >= queue_threshold:
        time.sleep(2)
    queue.put(prompt_with_noise)
    output1 = proc1(queue.get())
    queue.queue.clear() # Clear the queue after the output is generated
    return output1

def send_it2(inputs, noise_level, proc1=proc1):
    prompt_with_noise = add_random_noise(inputs, noise_level)
    while queue.qsize() >= queue_threshold:
        time.sleep(2)
    queue.put(prompt_with_noise)
    output2 = proc1(queue.get())
    queue.queue.clear() # Clear the queue after the output is generated
    return output2

def send_it3(inputs, noise_level, proc1=proc1):
    prompt_with_noise = add_random_noise(inputs, noise_level)
    while queue.qsize() >= queue_threshold:
        time.sleep(2)
    queue.put(prompt_with_noise)
    output3 = proc1(queue.get())
    queue.queue.clear() # Clear the queue after the output is generated
    return output3

def send_it4(inputs, noise_level, proc1=proc1):
    prompt_with_noise = add_random_noise(inputs, noise_level)
    while queue.qsize() >= queue_threshold:
        time.sleep(2)
    queue.put(prompt_with_noise)
    output4 = proc1(queue.get())
    queue.queue.clear() # Clear the queue after the output is generated
    return output4

def send_it5(inputs, noise_level, proc1=proc1):
    prompt_with_noise = add_random_noise(inputs, noise_level)
    while queue.qsize() >= queue_threshold:
        time.sleep(2)
    queue.put(prompt_with_noise)
    output5 = proc1(queue.get())
    queue.queue.clear() # Clear the queue after the output is generated
    return output5

def send_it6(inputs, noise_level, proc1=proc1):
    prompt_with_noise = add_random_noise(inputs, noise_level)
    while queue.qsize() >= queue_threshold:
        time.sleep(2)
    queue.put(prompt_with_noise)
    output6 = proc1(queue.get())
    queue.queue.clear() # Clear the queue after the output is generated
    return output6

def send_it7(inputs, noise_level, proc1=proc1):
    prompt_with_noise = add_random_noise(inputs, noise_level)
    while queue.qsize() >= queue_threshold:
        time.sleep(2)
    queue.put(prompt_with_noise)
    output7 = proc1(queue.get())
    queue.queue.clear() # Clear the queue after the output is generated
    return output7

def send_it8(inputs, noise_level, proc1=proc1):
    prompt_with_noise = add_random_noise(inputs, noise_level)
    while queue.qsize() >= queue_threshold:
        time.sleep(2)
    queue.put(prompt_with_noise)
    output8 = proc1(queue.get())
    queue.queue.clear() # Clear the queue after the output is generated
    return output8


def get_prompts(prompt_text):
    while queue.qsize() >= queue_threshold:
        time.sleep(2)
    queue.put(prompt_text)
    output = text_gen(queue.get())
    queue.queue.clear() # Clear the queue after the output is generated
    return output


with gr.Blocks() as myface:
    with gr.Row():

        input_text=gr.Textbox(label="Short Prompt")
        see_prompts=gr.Button("Magic Prompt")
    with gr.Row():

        prompt=gr.Textbox(label="Enter Prompt")
        noise_level=gr.Slider(minimum=0.1, maximum=3, step=0.1, label="Noise Level: Controls how much randomness is added to the input before it is sent to the model. Higher noise level produces more diverse outputs, while lower noise level produces similar outputs.")
        run=gr.Button("Generate")

    with gr.Row():
        like_message = gr.Button("❤️ excurl realistic photo art generator! ❤️")
    with gr.Row():
        output1=gr.Image(label="Variant 1")
        output2=gr.Image(label="Variant 2")
    with gr.Row():
        output3=gr.Image(label="Variant 3")
        output4=gr.Image(label="Variant 4")
    with gr.Row():
        output5=gr.Image(label="Variant 5")
        output6=gr.Image(label="Variant 6")
    with gr.Row():
        output7=gr.Image(label="Variant 7")
        output8=gr.Image(label="Variant 8")

    
    run.click(send_it1, inputs=[prompt, noise_level], outputs=[output1])
    run.click(send_it2, inputs=[prompt, noise_level], outputs=[output2])
    run.click(send_it3, inputs=[prompt, noise_level], outputs=[output3])
    run.click(send_it4, inputs=[prompt, noise_level], outputs=[output4])
    run.click(send_it5, inputs=[prompt, noise_level], outputs=[output5])
    run.click(send_it6, inputs=[prompt, noise_level], outputs=[output6])
    run.click(send_it7, inputs=[prompt, noise_level], outputs=[output7])
    run.click(send_it8, inputs=[prompt, noise_level], outputs=[output8])
    see_prompts.click(get_prompts, inputs=[input_text], outputs=[prompt])

myface.launch(enable_queue=True, inline=True)
block.queue(concurrency_count=30, max_size=30)
reset_queue_thread.join()