import gradio as gr
import os
from pathlib import Path
import argparse
import shutil
# from train_dreambooth import run_training
from textual_inversion import run_training
from convertosd import convert
from PIL import Image
from slugify import slugify
import requests
import torch
import zipfile
import tarfile
import urllib.parse
import gc
from diffusers import StableDiffusionPipeline
from huggingface_hub import snapshot_download
is_spaces = True if "SPACE_ID" in os.environ else False
#is_shared_ui = True if "IS_SHARED_UI" in os.environ else False
if(is_spaces):
is_shared_ui = True if ("lvkaokao/textual-inversion-training" in os.environ['SPACE_ID'] or "Intel/textual-inversion-training" in os.environ['SPACE_ID']) else False
else:
is_shared_ui = False
css = '''
.instruction{position: absolute; top: 0;right: 0;margin-top: 0px !important}
.arrow{position: absolute;top: 0;right: -110px;margin-top: -8px !important}
#component-4, #component-3, #component-10{min-height: 0}
.duplicate-button img{margin: 0}
'''
maximum_concepts = 1
#Pre download the files
'''
model_v1_4 = snapshot_download(repo_id="CompVis/stable-diffusion-v1-4")
#model_v1_5 = snapshot_download(repo_id="runwayml/stable-diffusion-v1-5")
model_v1_5 = snapshot_download(repo_id="stabilityai/stable-diffusion-2")
model_v2_512 = snapshot_download(repo_id="stabilityai/stable-diffusion-2-base", revision="fp16")
safety_checker = snapshot_download(repo_id="multimodalart/sd-sc")
'''
model_v1_4 = "CompVis/stable-diffusion-v1-4"
model_v1_5 = "stabilityai/stable-diffusion-2"
model_v2_512 = "stabilityai/stable-diffusion-2-base"
model_to_load = model_v1_4
with zipfile.ZipFile("mix.zip", 'r') as zip_ref:
zip_ref.extractall(".")
def swap_text(option):
mandatory_liability = "You must have the right to do so and you are liable for the images you use, example:"
if(option == "object"):
instance_prompt_example = "cttoy"
freeze_for = 30
return [f"You are going to train `object`(s), upload 5-10 images of each object you are planning on training on from different angles/perspectives. {mandatory_liability}:", '''''', f"You should name your concept with a unique made up word that has low chance of the model already knowing it (e.g.: `{instance_prompt_example}` here). Images will be automatically cropped to 512x512.", freeze_for, gr.update(visible=False)]
elif(option == "person"):
instance_prompt_example = "julcto"
freeze_for = 70
return [f"You are going to train a `person`(s), upload 10-20 images of each person you are planning on training on from different angles/perspectives. {mandatory_liability}:", '''''', f"You should name your concept with a unique made up word that has low chance of the model already knowing it (e.g.: `{instance_prompt_example}` here). Images will be automatically cropped to 512x512.", freeze_for, gr.update(visible=True)]
elif(option == "style"):
instance_prompt_example = "trsldamrl"
freeze_for = 10
return [f"You are going to train a `style`, upload 10-20 images of the style you are planning on training on. Name the files with the words you would like {mandatory_liability}:", '''''', f"You should name your concept with a unique made up word that has low chance of the model already knowing it (e.g.: `{instance_prompt_example}` here). Images will be automatically cropped to 512x512.", freeze_for, gr.update(visible=False)]
def swap_base_model(selected_model):
global model_to_load
if(selected_model == "v1-4"):
model_to_load = model_v1_4
elif(selected_model == "v1-5"):
model_to_load = model_v1_5
else:
model_to_load = model_v2_512
def count_files(*inputs):
file_counter = 0
concept_counter = 0
for i, input in enumerate(inputs):
if(i < maximum_concepts-1):
files = inputs[i]
if(files):
concept_counter+=1
file_counter+=len(files)
uses_custom = inputs[-1]
type_of_thing = inputs[-4]
if(uses_custom):
Training_Steps = int(inputs[-3])
else:
Training_Steps = file_counter*200
if(Training_Steps > 2400):
Training_Steps=2400
elif(Training_Steps < 1400):
Training_Steps=1400
if(is_spaces):
summary_sentence = f'''The training should take around 24 hours for 1000 steps using the default free CPU.
'''
else:
summary_sentence = f'''You are going to train {concept_counter} {type_of_thing}(s), with {file_counter} images for {Training_Steps} steps.
'''
return([gr.update(visible=True), gr.update(visible=True, value=summary_sentence)])
def update_steps(*files_list):
file_counter = 0
for i, files in enumerate(files_list):
if(files):
file_counter+=len(files)
return(gr.update(value=file_counter*200))
def pad_image(image):
w, h = image.size
if w == h:
return image
elif w > h:
new_image = Image.new(image.mode, (w, w), (0, 0, 0))
new_image.paste(image, (0, (w - h) // 2))
return new_image
else:
new_image = Image.new(image.mode, (h, h), (0, 0, 0))
new_image.paste(image, ((h - w) // 2, 0))
return new_image
def train(*inputs):
if is_shared_ui:
raise gr.Error("This Space only works in duplicated instances")
torch.cuda.empty_cache()
if 'pipe' in globals():
global pipe, pipe_is_set
del pipe
pipe_is_set = False
gc.collect()
if os.path.exists("output_model"): shutil.rmtree('output_model')
if os.path.exists("concept_images"): shutil.rmtree('concept_images')
if os.path.exists("diffusers_model.tar"): os.remove("diffusers_model.tar")
if os.path.exists("model.ckpt"): os.remove("model.ckpt")
if os.path.exists("hastrained.success"): os.remove("hastrained.success")
file_counter = 0
print(inputs)
os.makedirs('concept_images', exist_ok=True)
files = inputs[maximum_concepts*3]
init_word = inputs[maximum_concepts*2]
prompt = inputs[maximum_concepts]
if(prompt == "" or prompt == None):
raise gr.Error("You forgot to define your concept prompt")
for j, file_temp in enumerate(files):
file = Image.open(file_temp.name)
image = pad_image(file)
image = image.resize((512, 512))
extension = file_temp.name.split(".")[1]
image = image.convert('RGB')
image.save(f'concept_images/{j+1}.jpg', format="JPEG", quality = 100)
file_counter += 1
os.makedirs('output_model',exist_ok=True)
uses_custom = inputs[-1]
type_of_thing = inputs[-4]
remove_attribution_after = inputs[-6]
experimental_face_improvement = inputs[-9]
which_model = inputs[-10]
if(uses_custom):
Training_Steps = int(inputs[-3])
else:
Training_Steps = 1000
print(os.listdir("concept_images"))
args_general = argparse.Namespace(
pretrained_model_name_or_path = model_to_load,
train_data_dir="concept_images",
learnable_property=type_of_thing,
placeholder_token=prompt,
initializer_token=init_word,
resolution=512,
train_batch_size=1,
gradient_accumulation_steps=2,
use_bf16=True,
max_train_steps=Training_Steps,
learning_rate=5.0e-4,
scale_lr=True,
lr_scheduler="constant",
lr_warmup_steps=0,
output_dir="output_model",
)
print("Starting single training...")
lock_file = open("intraining.lock", "w")
lock_file.close()
run_training(args_general)
gc.collect()
torch.cuda.empty_cache()
if(which_model in ["v1-5"]):
print("Adding Safety Checker to the model...")
shutil.copytree(f"{safety_checker}/feature_extractor", "output_model/feature_extractor")
shutil.copytree(f"{safety_checker}/safety_checker", "output_model/safety_checker")
shutil.copy(f"model_index.json", "output_model/model_index.json")
if(not remove_attribution_after):
print("Archiving model file...")
with tarfile.open("diffusers_model.tar", "w") as tar:
tar.add("output_model", arcname=os.path.basename("output_model"))
if os.path.exists("intraining.lock"): os.remove("intraining.lock")
trained_file = open("hastrained.success", "w")
trained_file.close()
print(os.listdir("output_model"))
print("Training completed!")
return [
gr.update(visible=True, value=["diffusers_model.tar"]), #result
gr.update(visible=True), #try_your_model
gr.update(visible=True), #push_to_hub
gr.update(visible=True), #convert_button
gr.update(visible=False), #training_ongoing
gr.update(visible=True) #completed_training
]
else:
hf_token = inputs[-5]
model_name = inputs[-7]
where_to_upload = inputs[-8]
push(model_name, where_to_upload, hf_token, which_model, True)
hardware_url = f"https://huggingface.co/spaces/{os.environ['SPACE_ID']}/hardware"
headers = { "authorization" : f"Bearer {hf_token}"}
body = {'flavor': 'cpu-basic'}
requests.post(hardware_url, json = body, headers=headers)
import time
pipe_is_set = False
def generate(prompt, steps):
print("prompt: ", prompt)
print("steps: ", steps)
torch.cuda.empty_cache()
from diffusers import StableDiffusionPipeline
global pipe_is_set
if(not pipe_is_set):
global pipe
if torch.cuda.is_available():
pipe = StableDiffusionPipeline.from_pretrained("./output_model", torch_dtype=torch.float16)
pipe = pipe.to("cuda")
else:
pipe = StableDiffusionPipeline.from_pretrained("./output_model", torch_dtype=torch.float)
pipe_is_set = True
start_time = time.time()
image = pipe(prompt, num_inference_steps=steps, guidance_scale=7.5).images[0]
print("cost: ", time.time() - start_time)
return(image)
def push(model_name, where_to_upload, hf_token, which_model, comes_from_automated=False):
if(not os.path.exists("model.ckpt")):
convert("output_model", "model.ckpt")
from huggingface_hub import HfApi, HfFolder, CommitOperationAdd
from huggingface_hub import create_repo
model_name_slug = slugify(model_name)
api = HfApi()
your_username = api.whoami(token=hf_token)["name"]
if(where_to_upload == "My personal profile"):
model_id = f"{your_username}/{model_name_slug}"
else:
model_id = f"sd-dreambooth-library/{model_name_slug}"
headers = {"Authorization" : f"Bearer: {hf_token}", "Content-Type": "application/json"}
response = requests.post("https://huggingface.co/organizations/sd-dreambooth-library/share/SSeOwppVCscfTEzFGQaqpfcjukVeNrKNHX", headers=headers)
images_upload = os.listdir("concept_images")
image_string = ""
instance_prompt_list = []
previous_instance_prompt = ''
for i, image in enumerate(images_upload):
instance_prompt = image.split("_")[0]
if(instance_prompt != previous_instance_prompt):
title_instance_prompt_string = instance_prompt
instance_prompt_list.append(instance_prompt)
else:
title_instance_prompt_string = ''
previous_instance_prompt = instance_prompt
image_string = f'''{title_instance_prompt_string} {"(use that on your prompt)" if title_instance_prompt_string != "" else ""}
{image_string}![{instance_prompt} {i}](https://huggingface.co/{model_id}/resolve/main/concept_images/{urllib.parse.quote(image)})'''
readme_text = f'''---
license: creativeml-openrail-m
tags:
- text-to-image
---
### {model_name} Dreambooth model trained by {api.whoami(token=hf_token)["name"]} with [Hugging Face Dreambooth Training Space](https://huggingface.co/spaces/multimodalart/dreambooth-training) with the {which_model} base model
You run your new concept via `diffusers` [Colab Notebook for Inference](https://colab.research.google.com/github/huggingface/notebooks/blob/main/diffusers/sd_dreambooth_inference.ipynb). Don't forget to use the concept prompts!
Sample pictures of:
{image_string}
'''
#Save the readme to a file
readme_file = open("model.README.md", "w")
readme_file.write(readme_text)
readme_file.close()
#Save the token identifier to a file
text_file = open("token_identifier.txt", "w")
text_file.write(', '.join(instance_prompt_list))
text_file.close()
try:
create_repo(model_id,private=True, token=hf_token)
except:
import time
epoch_time = str(int(time.time()))
create_repo(f"{model_id}-{epoch_time}", private=True,token=hf_token)
operations = [
CommitOperationAdd(path_in_repo="token_identifier.txt", path_or_fileobj="token_identifier.txt"),
CommitOperationAdd(path_in_repo="README.md", path_or_fileobj="model.README.md"),
CommitOperationAdd(path_in_repo=f"model.ckpt",path_or_fileobj="model.ckpt")
]
api.create_commit(
repo_id=model_id,
operations=operations,
commit_message=f"Upload the model {model_name}",
token=hf_token
)
api.upload_folder(
folder_path="output_model",
repo_id=model_id,
token=hf_token
)
api.upload_folder(
folder_path="concept_images",
path_in_repo="concept_images",
repo_id=model_id,
token=hf_token
)
if is_spaces:
if(not comes_from_automated):
extra_message = "Don't forget to remove the GPU attribution after you play with it."
else:
extra_message = "The GPU has been removed automatically as requested, and you can try the model via the model page"
api.create_discussion(repo_id=os.environ['SPACE_ID'], title=f"Your model {model_name} has finished trained from the Dreambooth Train Spaces!", description=f"Your model has been successfully uploaded to: https://huggingface.co/{model_id}. {extra_message}",repo_type="space", token=hf_token)
return [gr.update(visible=True, value=f"Successfully uploaded your model. Access it [here](https://huggingface.co/{model_id})"), gr.update(visible=True, value=["diffusers_model.tar", "model.ckpt"])]
def convert_to_ckpt():
convert("output_model", "model.ckpt")
return gr.update(visible=True, value=["diffusers_model.tar", "model.ckpt"])
def check_status(top_description):
print('=='*20)
print(os.listdir("./"))
if os.path.exists("hastrained.success"):
if is_spaces:
update_top_tag = gr.update(value=f'''
Yay, congratulations on training your model. Scroll down to play with with it, save it (either downloading it or on the Hugging Face Hub). Once you are done, your model is safe, and you don't want to train a new one, go to the settings page and downgrade your Space to a CPU Basic
Yay, congratulations on training your model. Scroll down to play with with it, save it (either downloading it or on the Hugging Face Hub).
You closed the tab while your model was training, but it's all good! It is still training right now. You can click the "Open logs" button above here to check the training status. Once training is done, reload this tab to interact with your model
For it to work, you can either run locally or duplicate the Space and run it on your own profile using the free CPU or a (paid) private T4 GPU for training. CPU training takes a long time while each T4 costs US$0.60/h which should cost < US$1 to train most models using default settings!
If you want to use CPU, it will take a long time to run the training below. If you want to use GPU, please get this ready: attribute a T4 GPU to it (via the Settings tab) and run the training below. You will be billed by the minute from when you activate the GPU until when it is turned it off.
Do a pip install requirements-local.txt