import os import gradio as gr from random import randint from operator import itemgetter import bisect from all_models import tags_plus_models,models,models_plus_tags from datetime import datetime from externalmod import gr_Interface_load import asyncio import os from threading import RLock lock = RLock() HF_TOKEN = os.environ.get("HF_TOKEN") if os.environ.get("HF_TOKEN") else None # If private or gated models aren't used, ENV setting is unnecessary. now2 = 0 inference_timeout = 300 MAX_SEED = 2**32-1 nb_rep=2 nb_mod_dif=20 nb_models=nb_mod_dif*nb_rep cache_image={} cache_image_actu={} def split_models(models,nb_models): models_temp=[] models_lis_temp=[] i=0 for m in models: models_temp.append(m) i=i+1 if i%nb_models==0: models_lis_temp.append(models_temp) models_temp=[] if len(models_temp)>1: models_lis_temp.append(models_temp) return models_lis_temp def split_models_axb(models,a,b): models_temp=[] models_lis_temp=[] i=0 nb_models=b for m in models: for j in range(a): models_temp.append(m) i=i+1 if i%nb_models==0: models_lis_temp.append(models_temp) models_temp=[] if len(models_temp)>1: models_lis_temp.append(models_temp) return models_lis_temp def split_models_8x3(models,nb_models): models_temp=[] models_lis_temp=[] i=0 nb_models_x3=8 for m in models: models_temp.append(m) i=i+1 if i%nb_models_x3==0: models_lis_temp.append(models_temp+models_temp+models_temp) models_temp=[] if len(models_temp)>1: models_lis_temp.append(models_temp+models_temp+models_temp) return models_lis_temp def construct_list_models(tags_plus_models,nb_rep,nb_mod_dif): list_temp=[] output=[] for tag_plus_models in tags_plus_models: list_temp=split_models_axb(tag_plus_models[2],nb_rep,nb_mod_dif) list_temp2=[] i=0 for elem in list_temp: list_temp2.append([tag_plus_models[0]+"_"+str(i)+" : "+elem[0]+" - "+elem[len(elem)-1] ,elem]) i+=1 output.append([tag_plus_models[0] + " (" + str(tag_plus_models[1]) + ")",list_temp2]) return output models_test = [] models_test = construct_list_models(tags_plus_models,nb_rep,nb_mod_dif) def get_current_time(): now = datetime.now() now2 = now current_time = now2.strftime("%Y-%m-%d %H:%M:%S") kii = "" # ? ki = f'{kii} {current_time}' return ki def load_fn_original(models): global models_load global num_models global default_models models_load = {} num_models = len(models) if num_models!=0: default_models = models[:num_models] else: default_models = {} for model in models: if model not in models_load.keys(): try: m = gr.load(f'models/{model}') except Exception as error: m = gr.Interface(lambda txt: None, ['text'], ['image']) print(error) models_load.update({model: m}) def load_fn(models): global models_load global num_models global default_models models_load = {} num_models = len(models) i=0 if num_models!=0: default_models = models[:num_models] else: default_models = {} for model in models: i+=1 if i%50==0: print("\n\n\n-------"+str(i)+'/'+str(len(models))+"-------\n\n\n") if model not in models_load.keys(): try: m = gr_Interface_load(f'models/{model}', hf_token=HF_TOKEN) except Exception as error: m = gr.Interface(lambda txt: None, ['text'], ['image']) print(error) models_load.update({model: m}) """models = models_test[1]""" #load_fn_original load_fn(models) """models = {} load_fn(models)""" def extend_choices(choices): return choices + (nb_models - len(choices)) * ['NA'] """return choices + (num_models - len(choices)) * ['NA']""" def extend_choices_b(choices): choices_plus = extend_choices(choices) return [gr.Textbox(m, visible=False) for m in choices_plus] def update_imgbox(choices): choices_plus = extend_choices(choices) return [gr.Image(None, label=m,interactive=False, visible=(m != 'NA')) for m in choices_plus] def choice_group_a(group_model_choice): return group_model_choice def choice_group_b(group_model_choice): choiceTemp =choice_group_a(group_model_choice) choiceTemp = extend_choices(choiceTemp) """return [gr.Image(label=m, min_width=170, height=170) for m in choice]""" return [gr.Image(None, label=m,interactive=False, visible=(m != 'NA')) for m in choiceTemp] def choice_group_c(group_model_choice): choiceTemp=choice_group_a(group_model_choice) choiceTemp = extend_choices(choiceTemp) return [gr.Textbox(m, visible=False) for m in choiceTemp] def cutStrg(longStrg,start,end): shortStrg='' for i in range(end-start): shortStrg+=longStrg[start+i] return shortStrg def aff_models_perso(txt_list_perso,nb_models=nb_models,models=models): list_perso=[] t1=True start=txt_list_perso.find('\"') if start!=-1: while t1: start+=1 end=txt_list_perso.find('\"',start) if end != -1: txtTemp=cutStrg(txt_list_perso,start,end) if txtTemp in models: list_perso.append(cutStrg(txt_list_perso,start,end)) else : t1=False start=txt_list_perso.find('\"',end+1) if start==-1: t1=False if len(list_perso)>=nb_models: t1=False return list_perso def aff_models_perso_b(txt_list_perso): return choice_group_b(aff_models_perso(txt_list_perso)) def aff_models_perso_c(txt_list_perso): return choice_group_c(aff_models_perso(txt_list_perso)) def tag_choice(group_tag_choice): return gr.Dropdown(label="List of Models with the chosen Tag", show_label=True, choices=list(group_tag_choice) , interactive = True , filterable = False) def test_pass(test): if test==os.getenv('p'): print("ok") return gr.Dropdown(label="Lists Tags", show_label=True, choices=list(models_test) , interactive = True) else: print("nop") return gr.Dropdown(label="Lists Tags", show_label=True, choices=list([]) , interactive = True) def test_pass_aff(test): if test==os.getenv('p'): return gr.Accordion( open=True, visible=True) else: return gr.Accordion( open=True, visible=False) # https://huggingface.co/docs/api-inference/detailed_parameters # https://huggingface.co/docs/huggingface_hub/package_reference/inference_client async def infer(model_str, prompt, nprompt="", height=None, width=None, steps=None, cfg=None, seed=-1, timeout=inference_timeout): from pathlib import Path kwargs = {} if height is not None and height >= 256: kwargs["height"] = height if width is not None and width >= 256: kwargs["width"] = width if steps is not None and steps >= 1: kwargs["num_inference_steps"] = steps if cfg is not None and cfg > 0: cfg = kwargs["guidance_scale"] = cfg noise = "" if seed >= 0: kwargs["seed"] = seed else: rand = randint(1, 500) for i in range(rand): noise += " " task = asyncio.create_task(asyncio.to_thread(models_load[model_str].fn, prompt=f'{prompt} {noise}', negative_prompt=nprompt, **kwargs, token=HF_TOKEN)) await asyncio.sleep(0) try: result = await asyncio.wait_for(task, timeout=timeout) except (Exception, asyncio.TimeoutError) as e: print(e) print(f"Task timed out: {model_str}") if not task.done(): task.cancel() result = None if task.done() and result is not None: with lock: png_path = "image.png" result.save(png_path) image = str(Path(png_path).resolve()) return image return None def gen_fn(model_str, prompt, nprompt="", height=None, width=None, steps=None, cfg=None, seed=-1): if model_str == 'NA': return None try: loop = asyncio.new_event_loop() result = loop.run_until_complete(infer(model_str, prompt, nprompt, height, width, steps, cfg, seed, inference_timeout)) except (Exception, asyncio.CancelledError) as e: print(e) print(f"Task aborted: {model_str}") result = None finally: loop.close() return result def gen_fn_original(model_str, prompt): if model_str == 'NA': return None noise = str(randint(0, 9999)) try : m=models_load[model_str](f'{prompt} {noise}') except Exception as error : print("error : " + model_str) print(error) m=False return m def add_gallery(image, model_str, gallery): if gallery is None: gallery = [] #with lock: if image is not None: gallery.append((image, model_str)) return gallery def reset_gallery(gallery): return add_gallery(None,"",[]) def load_gallery(gallery,id): gallery = reset_gallery(gallery) for c in cache_image[f"{id}"]: gallery=add_gallery(c[0],c[1],gallery) return gallery def load_gallery_actu(gallery,id): gallery = reset_gallery(gallery) #for c in cache_image_actu: for c in cache_image_actu[f"{id}"]: gallery=add_gallery(c[0],c[1],gallery) return gallery def add_cache_image(image, model_str,id,cache_image=cache_image): if image is not None: cache_image[f"{id}"].append((image,model_str)) #cache_image=sorted(cache_image, key=itemgetter(1)) return def add_cache_image_actu(image, model_str,id,cache_image_actu=cache_image_actu): if image is not None: bisect.insort(cache_image_actu[f"{id}"],(image, model_str), key=itemgetter(1)) #cache_image_actu=sorted(cache_image_actu, key=itemgetter(1)) return def reset_cache_image(id,cache_image=cache_image): cache_image[f"{id}"].clear() return def reset_cache_image_actu(id,cache_image_actu=cache_image_actu): cache_image_actu[f"{id}"].clear() return def reset_cache_image_all_sessions(cache_image=cache_image): for key, listT in cache_image.items(): listT.clear() return def set_session(id): if id==0: randTemp=randint(1,MAX_SEED) cache_image[f"{randTemp}"]=[] cache_image_actu[f"{randTemp}"]=[] return gr.Number(visible=False,value=randTemp) else : return id def print_info_sessions(): lenTot=0 print("numbre of sessions : "+str(len(cache_image))) for key, listT in cache_image.items(): print(key+str(len(listT))) lenTot+=len(listT) print("images total = "+str(lenTot)) return def disp_models(group_model_choice,nb_rep=nb_rep): listTemp=[] strTemp='\n' i=0 for m in group_model_choice: if m not in listTemp: listTemp.append(m) for m in listTemp: i+=1 strTemp+="\"" + m + "\",\n" if i%(8/nb_rep)==0: strTemp+="\n" return gr.Textbox(label="models",value=strTemp) def search_models(str_search,tags_plus_models=tags_plus_models): output1="\n" output2="" for m in tags_plus_models[0][2]: if m.find(str_search)!=-1: output1+="\"" + m + "\",\n" outputPlus="\n From tags : \n\n" for tag_plus_models in tags_plus_models: if str_search.lower() == tag_plus_models[0].lower() and str_search!="": for m in tag_plus_models[2]: output2+="\"" + m + "\",\n" if output2 != "": output=output1+outputPlus+output2 else : output=output1 return gr.Textbox(label="out",value=output) def search_info(txt_search_info,models_plus_tags=models_plus_tags): outputList=[] if txt_search_info.find("\"")!=-1: start=txt_search_info.find("\"")+1 end=txt_search_info.find("\"",start) m_name=cutStrg(txt_search_info,start,end) else : m_name = txt_search_info for m in models_plus_tags: if m_name == m[0]: outputList=m[1] if len(outputList)==0: outputList.append("Model Not Find") return gr.Textbox(label="out",value=outputList) def ratio_chosen(choice_ratio,width,height): if choice_ratio == [None,None]: return width , height else : return gr.Slider(label="Width", info="If 0, the default value is used.", maximum=2024, step=32, value=choice_ratio[0]), gr.Slider(label="Height", info="If 0, the default value is used.", maximum=2024, step=32, value=choice_ratio[1]) list_ratios=[["None",[None,None]], ["4:1 (2048 x 512)",[2048,512]], ["12:5 (1536 x 640)",[1536,640]], ["~16:9 (1344 x 768)",[1344,768]], ["~3:2 (1216 x 832)",[1216,832]], ["~4:3 (1152 x 896)",[1152,896]], ["1:1 (1024 x 1024)",[1024,1024]], ["~3:4 (896 x 1152)",[896,1152]], ["~2:3 (832 x 1216)",[832,1216]], ["~9:16 (768 x 1344)",[768,1344]], ["5:12 (640 x 1536)",[640,1536]], ["1:4 (512 x 2048)",[512,2048]]] def make_me(): # with gr.Tab('The Dream'): with gr.Row(): #txt_input = gr.Textbox(lines=3, width=300, max_height=100) #txt_input = gr.Textbox(label='Your prompt:', lines=3, width=300, max_height=100) with gr.Column(scale=4): with gr.Group(): txt_input = gr.Textbox(label='Your prompt:', lines=3) with gr.Accordion("Advanced", open=False, visible=True): neg_input = gr.Textbox(label='Negative prompt:', lines=1) with gr.Row(): width = gr.Slider(label="Width", info="If 0, the default value is used.", maximum=1216, step=32, value=0) height = gr.Slider(label="Height", info="If 0, the default value is used.", maximum=1216, step=32, value=0) with gr.Row(): choice_ratio = gr.Dropdown(label="Ratio Width/Height", info="OverWrite Width and Height (W*H<1024*1024)", show_label=True, choices=list(list_ratios) , interactive = True, value=list_ratios[0]) choice_ratio.change(ratio_chosen,[choice_ratio,width,height],[width,height]) with gr.Row(): steps = gr.Slider(label="Number of inference steps", info="If 0, the default value is used.", maximum=100, step=1, value=0) cfg = gr.Slider(label="Guidance scale", info="If 0, the default value is used.", maximum=30.0, step=0.1, value=0) seed = gr.Slider(label="Seed", info="Randomize Seed if -1.", minimum=-1, maximum=MAX_SEED, step=1, value=-1) #gen_button = gr.Button('Generate images', width=150, height=30) #stop_button = gr.Button('Stop', variant='secondary', interactive=False, width=150, height=30) gen_button = gr.Button('Generate images', scale=3) stop_button = gr.Button('Stop', variant='secondary', interactive=False, scale=1) gen_button.click(lambda: gr.update(interactive=True), None, stop_button) #gr.HTML(""" #