nami0342's picture
Display public model download percentage add to build log
f40d1d0
raw
history blame
28.1 kB
import spaces
import gradio as gr
from PIL import Image
from src.tryon_pipeline import StableDiffusionXLInpaintPipeline as TryonPipeline
from src.unet_hacked_garmnet import UNet2DConditionModel as UNet2DConditionModel_ref
from src.unet_hacked_tryon import UNet2DConditionModel
from transformers import (
CLIPImageProcessor,
CLIPVisionModelWithProjection,
CLIPTextModel,
CLIPTextModelWithProjection,
)
from diffusers import DDPMScheduler,AutoencoderKL
from typing import List
import torch
import os
import io
import warnings
import requests
from transformers import AutoTokenizer
import numpy as np
from utils_mask import get_mask_location
from torchvision import transforms
import apply_net
from preprocess.humanparsing.run_parsing import Parsing
from preprocess.openpose.run_openpose import OpenPose
from detectron2.data.detection_utils import convert_PIL_to_numpy,_apply_exif_orientation
from torchvision.transforms.functional import to_pil_image
import pillow_heif # HEIC 이미지 처리용 (아이폰 촬영 사진 포맷)
from urllib.parse import urlparse
# SSL 경고 억제
warnings.filterwarnings("ignore", message=".*OpenSSL.*")
warnings.filterwarnings("ignore", category=UserWarning, module="urllib3")
# requests 세션 설정
session = requests.Session()
session.verify = False # SSL 검증 비활성화 (개발 환경용)
def pil_to_binary_mask(pil_image, threshold=0):
np_image = np.array(pil_image)
grayscale_image = Image.fromarray(np_image).convert("L")
binary_mask = np.array(grayscale_image) > threshold
mask = np.zeros(binary_mask.shape, dtype=np.uint8)
for i in range(binary_mask.shape[0]):
for j in range(binary_mask.shape[1]):
if binary_mask[i,j] == True :
mask[i,j] = 1
mask = (mask*255).astype(np.uint8)
output_mask = Image.fromarray(mask)
return output_mask
base_path = 'yisol/IDM-VTON'
example_path = os.path.join(os.path.dirname(__file__), 'example')
unet = UNet2DConditionModel.from_pretrained(
base_path,
subfolder="unet",
torch_dtype=torch.float16,
)
unet.requires_grad_(False)
tokenizer_one = AutoTokenizer.from_pretrained(
base_path,
subfolder="tokenizer",
revision=None,
use_fast=False,
)
tokenizer_two = AutoTokenizer.from_pretrained(
base_path,
subfolder="tokenizer_2",
revision=None,
use_fast=False,
)
noise_scheduler = DDPMScheduler.from_pretrained(base_path, subfolder="scheduler")
text_encoder_one = CLIPTextModel.from_pretrained(
base_path,
subfolder="text_encoder",
torch_dtype=torch.float16,
)
text_encoder_two = CLIPTextModelWithProjection.from_pretrained(
base_path,
subfolder="text_encoder_2",
torch_dtype=torch.float16,
)
image_encoder = CLIPVisionModelWithProjection.from_pretrained(
base_path,
subfolder="image_encoder",
torch_dtype=torch.float16,
)
vae = AutoencoderKL.from_pretrained(base_path,
subfolder="vae",
torch_dtype=torch.float16,
)
# "stabilityai/stable-diffusion-xl-base-1.0",
UNet_Encoder = UNet2DConditionModel_ref.from_pretrained(
base_path,
subfolder="unet_encoder",
torch_dtype=torch.float16,
)
parsing_model = Parsing(0)
openpose_model = OpenPose(0)
UNet_Encoder.requires_grad_(False)
image_encoder.requires_grad_(False)
vae.requires_grad_(False)
unet.requires_grad_(False)
text_encoder_one.requires_grad_(False)
text_encoder_two.requires_grad_(False)
tensor_transfrom = transforms.Compose(
[
transforms.ToTensor(),
transforms.Normalize([0.5], [0.5]),
]
)
pipe = TryonPipeline.from_pretrained(
base_path,
unet=unet,
vae=vae,
feature_extractor= CLIPImageProcessor(),
text_encoder = text_encoder_one,
text_encoder_2 = text_encoder_two,
tokenizer = tokenizer_one,
tokenizer_2 = tokenizer_two,
scheduler = noise_scheduler,
image_encoder=image_encoder,
torch_dtype=torch.float16,
)
pipe.unet_encoder = UNet_Encoder
# 이미지 전처리 함수
def preprocess_image(image):
# HEIC 이미지 처리
if isinstance(image, np.ndarray):
image = Image.fromarray(image)
# HEIC 이미지를 JPEG로 변환 - 이거 안 먹히는 거 같은데....
try:
output = io.BytesIO()
image.convert("RGB").save(output, format="JPEG", quality=95)
output.seek(0)
image = Image.open(output)
except Exception as e:
print(f"Error converting image: {e}")
# 변환 실패 시 원본 이미지 사용
image = image.convert("RGB")
# 이미지 크기 가져오기
width, height = image.size
# 3:4 비율로 중앙 자르기
target_width = int(min(width, height * (3 / 4)))
target_height = int(min(height, width * (4 / 3)))
left = (width - target_width) / 2
top = (height - target_height) / 2
right = (width + target_width) / 2
bottom = (height + target_height) / 2
# 이미지 자르기
cropped_img = image.crop((left, top, right, bottom))
# 768x1024로 리사이징
resized_img = cropped_img.resize((768, 1024), resample=Image.Resampling.LANCZOS)
return resized_img
# URL에서 이미지 가져오기 함수
def load_image_from_url(url):
try:
response = session.get(url, stream=True, timeout=10)
response.raise_for_status() # HTTP 오류 확인
# 이미지 다운로드
img = Image.open(response.raw).convert("RGB")
# JPEG로 변환
output = io.BytesIO()
img.save(output, format="JPEG", quality=95)
output.seek(0)
# 변환된 JPEG 이미지 반환
jpeg_img = Image.open(output)
return jpeg_img
except requests.exceptions.RequestException as e:
print(f"Error downloading image from URL: {e}")
return None
except Exception as e:
print(f"Error processing image from URL: {e}")
return None
def process_url_image(url):
"""Process image from URL and return PIL Image"""
if not url or not url.strip():
return None
# URL 유효성 검사
try:
result = urlparse(url)
if not all([result.scheme, result.netloc]):
print("Invalid URL format")
return None
except Exception as e:
print(f"Error parsing URL: {e}")
return None
img = load_image_from_url(url)
if img is None:
print("Failed to load image from URL")
return None
return preprocess_image(img)
def load_example_for_editor(image_path):
"""Load example image for ImageEditor component"""
if image_path is None:
return None
# ImageEditor는 특정 형식을 기대하므로 딕셔너리 형태로 반환
return {
"background": image_path,
"layers": None,
"composite": None
}
def download_model_file(model_path, urls):
"""Download model file from multiple URLs if it doesn't exist"""
if os.path.exists(model_path):
print(f"Model file already exists: {model_path}")
return True
os.makedirs(os.path.dirname(model_path), exist_ok=True)
for url in urls:
try:
print(f"Downloading from: {url}")
response = requests.get(url, stream=True)
response.raise_for_status()
total_size = int(response.headers.get('content-length', 0))
block_size = 8192
with open(model_path, 'wb') as f:
downloaded = 0
for chunk in response.iter_content(chunk_size=block_size):
if chunk:
f.write(chunk)
downloaded += len(chunk)
if total_size > 0:
percent = (downloaded / total_size) * 100
if(percent % 10 == 0):
print(f"\rDownload progress: {percent:.1f}%", end='', flush=True)
print(f"\nSuccessfully downloaded: {model_path}")
return True
except Exception as e:
print(f"Failed to download from {url}: {e}")
continue
print(f"Failed to download model file from all URLs: {model_path}")
return False
def download_densepose_model():
"""Download DensePose model file"""
model_path = "ckpt/densepose/model_final_162be9.pkl"
urls = [
"https://dl.fbaipublicfiles.com/densepose/densepose_rcnn_R_50_FPN_s1x/165712039/model_final_162be9.pkl",
"https://github.com/facebookresearch/densepose/releases/download/v1.0/model_final_162be9.pkl"
]
return download_model_file(model_path, urls)
def download_openpose_model():
"""Download OpenPose model file"""
model_path = "ckpt/openpose/ckpts/body_pose_model.pth"
urls = [
"https://huggingface.co/lllyasviel/Annotators/resolve/main/body_pose_model.pth"
]
return download_model_file(model_path, urls)
def download_humanparsing_models():
"""Download Human Parsing model files"""
base_url = "https://huggingface.co/Longcat2957/humanparsing-onnx/resolve/main"
models = [
("ckpt/humanparsing/parsing_atr.onnx", f"{base_url}/parsing_atr.onnx"),
("ckpt/humanparsing/parsing_lip.onnx", f"{base_url}/parsing_lip.onnx")
]
success = True
for model_path, url in models:
if os.path.exists(model_path):
print(f"Human parsing model already exists: {model_path}")
continue
print(f"Downloading {model_path} from {url}")
if download_model_file(model_path, [url]):
print(f"Successfully downloaded: {model_path}")
else:
print(f"Failed to download: {model_path}")
success = False
return success
def download_all_models():
"""Download all required model files"""
print("Checking and downloading required model files...")
# Download DensePose model
print("\n=== Downloading DensePose model ===")
densepose_success = download_densepose_model()
# Download OpenPose model
print("\n=== Downloading OpenPose model ===")
openpose_success = download_openpose_model()
# Download Human Parsing models
print("\n=== Downloading Human Parsing models ===")
parsing_success = download_humanparsing_models()
return densepose_success and openpose_success and parsing_success
@spaces.GPU
def start_tryon(dict,garm_img,garment_des,is_checked,denoise_steps,seed, is_checked_crop):
device = "cuda"
openpose_model.preprocessor.body_estimation.model.to(device)
pipe.to(device)
pipe.unet_encoder.to(device)
garm_img= garm_img.convert("RGB").resize((768,1024))
human_img_orig = dict["background"].convert("RGB")
if is_checked_crop:
width, height = human_img_orig.size
target_width = int(min(width, height * (3 / 4)))
target_height = int(min(height, width * (4 / 3)))
left = (width - target_width) / 2
top = (height - target_height) / 2
right = (width + target_width) / 2
bottom = (height + target_height) / 2
cropped_img = human_img_orig.crop((left, top, right, bottom))
crop_size = cropped_img.size
human_img = cropped_img.resize((768,1024))
else:
human_img = human_img_orig.resize((768,1024))
if is_checked:
keypoints = openpose_model(human_img.resize((384,512)))
model_parse, _ = parsing_model(human_img.resize((384,512)))
mask, mask_gray = get_mask_location('hd', "upper_body", model_parse, keypoints)
mask = mask.resize((768,1024))
else:
mask = pil_to_binary_mask(dict['layers'][0].convert("RGB").resize((768, 1024)))
# mask = transforms.ToTensor()(mask)
# mask = mask.unsqueeze(0)
mask_gray = (1-transforms.ToTensor()(mask)) * tensor_transfrom(human_img)
mask_gray = to_pil_image((mask_gray+1.0)/2.0)
human_img_arg = _apply_exif_orientation(human_img.resize((384,512)))
human_img_arg = convert_PIL_to_numpy(human_img_arg, format="BGR")
# DensePose 모델 다운로드 및 경로 설정
densepose_model_path = './ckpt/densepose/model_final_162be9.pkl'
# 모델 파일이 없으면 다운로드 시도
if not os.path.exists(densepose_model_path):
print("DensePose model not found, attempting to download...")
download_success = download_densepose_model()
if not download_success:
print("Failed to download DensePose model")
return None, None
args = apply_net.create_argument_parser().parse_args(('show', './configs/densepose_rcnn_R_50_FPN_s1x.yaml', densepose_model_path, 'dp_segm', '-v', '--opts', 'MODEL.DEVICE', 'cuda'))
# verbosity = getattr(args, "verbosity", None)
pose_img = args.func(args,human_img_arg)
pose_img = pose_img[:,:,::-1]
pose_img = Image.fromarray(pose_img).resize((768,1024))
with torch.no_grad():
# Extract the images
with torch.cuda.amp.autocast():
with torch.no_grad():
prompt = "model is wearing " + garment_des
negative_prompt = "monochrome, lowres, bad anatomy, worst quality, low quality"
with torch.inference_mode():
(
prompt_embeds,
negative_prompt_embeds,
pooled_prompt_embeds,
negative_pooled_prompt_embeds,
) = pipe.encode_prompt(
prompt,
num_images_per_prompt=1,
do_classifier_free_guidance=True,
negative_prompt=negative_prompt,
)
prompt = "a photo of " + garment_des
negative_prompt = "monochrome, lowres, bad anatomy, worst quality, low quality"
if not isinstance(prompt, List):
prompt = [prompt] * 1
if not isinstance(negative_prompt, List):
negative_prompt = [negative_prompt] * 1
with torch.inference_mode():
(
prompt_embeds_c,
_,
_,
_,
) = pipe.encode_prompt(
prompt,
num_images_per_prompt=1,
do_classifier_free_guidance=False,
negative_prompt=negative_prompt,
)
pose_img = tensor_transfrom(pose_img).unsqueeze(0).to(device,torch.float16)
garm_tensor = tensor_transfrom(garm_img).unsqueeze(0).to(device,torch.float16)
generator = torch.Generator(device).manual_seed(seed) if seed is not None else None
images = pipe(
prompt_embeds=prompt_embeds.to(device,torch.float16),
negative_prompt_embeds=negative_prompt_embeds.to(device,torch.float16),
pooled_prompt_embeds=pooled_prompt_embeds.to(device,torch.float16),
negative_pooled_prompt_embeds=negative_pooled_prompt_embeds.to(device,torch.float16),
num_inference_steps=denoise_steps,
generator=generator,
strength = 1.0,
pose_img = pose_img.to(device,torch.float16),
text_embeds_cloth=prompt_embeds_c.to(device,torch.float16),
cloth = garm_tensor.to(device,torch.float16),
mask_image=mask,
image=human_img,
height=1024,
width=768,
ip_adapter_image = garm_img.resize((768,1024)),
guidance_scale=2.0,
)[0]
if is_checked_crop:
out_img = images[0].resize(crop_size)
human_img_orig.paste(out_img, (int(left), int(top)))
return human_img_orig, mask_gray
else:
return images[0], mask_gray
# return images[0], mask_gray
garm_list = os.listdir(os.path.join(example_path,"cloth"))
garm_list_path = [os.path.join(example_path,"cloth",garm) for garm in garm_list]
human_list = os.listdir(os.path.join(example_path,"human"))
human_list_path = [os.path.join(example_path,"human",human) for human in human_list]
# human_ex_list를 단순한 이미지 경로 리스트로 변경 (그리드 표시를 위해)
human_ex_list = human_list_path
##default human
image_blocks = gr.Blocks().queue()
with image_blocks as demo:
# CSS를 추가하여 ImageEditor의 canvas 크기를 고정하여 브러시 포인터 위치 정렬 (PC 및 모바일 대응)
gr.HTML("""
<style>
/* PC 및 모바일 공통: ImageEditor 컨테이너 */
#image-editor-fixed {
position: relative;
display: inline-block;
}
/* PC: 고정 크기 (384x512) */
@media (min-width: 768px) {
#image-editor-fixed canvas {
width: 384px !important;
height: 512px !important;
max-width: 384px !important;
max-height: 512px !important;
object-fit: contain !important;
}
#image-editor-fixed > div,
#image-editor-fixed .image-editor-container,
#image-editor-fixed .image-editor-wrapper,
#image-editor-fixed [class*="editor"],
#image-editor-fixed [class*="canvas"] {
width: 384px !important;
height: 512px !important;
max-width: 384px !important;
max-height: 512px !important;
}
#image-editor-fixed img {
width: 384px !important;
height: 512px !important;
max-width: 384px !important;
max-height: 512px !important;
object-fit: contain !important;
}
}
/* 모바일: 화면 크기에 맞게 스케일링하되 비율 유지 */
@media (max-width: 767px) {
#image-editor-fixed {
width: 100% !important;
max-width: 100% !important;
}
#image-editor-fixed canvas {
width: 100% !important;
max-width: 100% !important;
height: auto !important;
aspect-ratio: 3/4 !important;
object-fit: contain !important;
}
#image-editor-fixed > div,
#image-editor-fixed .image-editor-container,
#image-editor-fixed .image-editor-wrapper {
width: 100% !important;
max-width: 100% !important;
}
#image-editor-fixed img {
width: 100% !important;
max-width: 100% !important;
height: auto !important;
aspect-ratio: 3/4 !important;
object-fit: contain !important;
}
}
/* 출력 이미지 크기 고정: Masked image output 및 Output */
/* PC: 고정 크기 (384x512) */
@media (min-width: 768px) {
#masked-img img,
#output-img img {
width: 384px !important;
height: 512px !important;
max-width: 384px !important;
max-height: 512px !important;
object-fit: contain !important;
}
#masked-img,
#output-img {
max-width: 384px !important;
}
}
/* 모바일: 화면 크기에 맞게 스케일링하되 비율 유지 (최대 384px) */
@media (max-width: 767px) {
#masked-img img,
#output-img img {
width: 100% !important;
max-width: 384px !important;
height: auto !important;
aspect-ratio: 3/4 !important;
object-fit: contain !important;
}
#masked-img,
#output-img {
width: 100% !important;
max-width: 384px !important;
}
}
</style>
<script>
// PC 및 모바일 모두에서 canvas의 실제 크기를 고정하여 브러시 위치 정확도 유지
(function() {
const CANVAS_WIDTH = 384;
const CANVAS_HEIGHT = 512;
function fixImageEditorCanvas() {
const editor = document.querySelector('#image-editor-fixed');
if (!editor) return;
const canvas = editor.querySelector('canvas');
if (!canvas) return;
// canvas의 실제 픽셀 크기는 항상 고정 (브러시 위치 정확도 유지)
if (canvas.width !== CANVAS_WIDTH || canvas.height !== CANVAS_HEIGHT) {
canvas.width = CANVAS_WIDTH;
canvas.height = CANVAS_HEIGHT;
}
// 모바일에서는 CSS로 표시 크기만 조정, PC에서는 고정 크기
const isMobile = window.innerWidth <= 767;
if (isMobile) {
// 모바일: 화면 크기에 맞게 표시하되 비율 유지
const container = editor.closest('.gradio-column') || editor.parentElement;
if (container) {
const maxWidth = Math.min(container.offsetWidth - 20, 384);
const maxHeight = (maxWidth * 4) / 3;
canvas.style.width = maxWidth + 'px';
canvas.style.height = maxHeight + 'px';
canvas.style.maxWidth = '100%';
canvas.style.maxHeight = 'none';
}
} else {
// PC: 고정 크기
canvas.style.width = CANVAS_WIDTH + 'px';
canvas.style.height = CANVAS_HEIGHT + 'px';
canvas.style.maxWidth = CANVAS_WIDTH + 'px';
canvas.style.maxHeight = CANVAS_HEIGHT + 'px';
}
}
// 초기 실행
if (document.readyState === 'loading') {
document.addEventListener('DOMContentLoaded', fixImageEditorCanvas);
} else {
fixImageEditorCanvas();
}
// 이미지 로드 후 재적용
window.addEventListener('load', function() {
fixImageEditorCanvas();
setTimeout(fixImageEditorCanvas, 500);
setTimeout(fixImageEditorCanvas, 1500);
});
// 리사이즈 시 재적용 (모바일 회전 등)
let resizeTimeout;
window.addEventListener('resize', function() {
clearTimeout(resizeTimeout);
resizeTimeout = setTimeout(fixImageEditorCanvas, 300);
});
// MutationObserver로 동적 변경 감지
const observer = new MutationObserver(function(mutations) {
fixImageEditorCanvas();
});
// ImageEditor가 로드된 후 observer 시작
setTimeout(function() {
const editor = document.querySelector('#image-editor-fixed');
if (editor) {
observer.observe(editor, {
childList: true,
subtree: true,
attributes: true,
attributeFilter: ['style', 'class']
});
}
}, 1000);
})();
</script>
""")
gr.Markdown("## DXCO : GENAI-VTON")
gr.Markdown("임성남, 윤지영, 조민주 based on IDM-VTON")
gr.Markdown("이미지는 3:4비율(384x512 또는 768x1024)로 올려주세요")
with gr.Row():
with gr.Column():
imgs = gr.ImageEditor(sources='upload', type="pil", label='대상 이미지', interactive=True, height=512, elem_id="image-editor-fixed")
img_url_input = gr.Textbox(label="대상 이미지 URL", placeholder="예) https://example.com/human_image.jpg")
with gr.Row():
with gr.Row():
is_checked = gr.Checkbox(label="Yes", info="자동 마스킹",value=True)
example = gr.Examples(
inputs=imgs,
examples_per_page=8,
examples=human_ex_list,
)
with gr.Column():
garm_img = gr.Image(label="의상 이미지", sources='upload', type="pil")
garm_url_input = gr.Textbox(label="의상 이미지 URL", placeholder="예) https://example.com/garment.jpg")
with gr.Row(elem_id="prompt-container"):
with gr.Row():
prompt = gr.Textbox(placeholder="Description of garment ex) Short Sleeve Round Neck T-shirts", show_label=False, elem_id="prompt")
example = gr.Examples(
inputs=garm_img,
examples_per_page=8,
examples=garm_list_path)
with gr.Row():
try_button = gr.Button(value="Try-on")
with gr.Row():
with gr.Column():
masked_img = gr.Image(label="Masked image output", elem_id="masked-img", show_share_button=False, height=512)
with gr.Column():
image_out = gr.Image(label="Output", elem_id="output-img", show_share_button=False, height=512)
# with gr.Accordion(label="Advanced Settings", open=False):
# with gr.Row():
# denoise_steps = gr.Number(label="Denoising Steps", minimum=20, maximum=40, value=30, step=1)
# seed = gr.Number(label="Seed", minimum=-1, maximum=2147483647, step=1, value=42)
# is_checked = gr.Number(value=True)
# 이미지 업로드 시 전처리
# imgs.upload(
# fn=preprocess_image,
# inputs=imgs,
# outputs=imgs, # 전처리된 이미지를 ImageEditor에 다시 표시
# )
# 대상 이미지: URL 입력 처리
img_url_input.change(
fn=lambda url: process_url_image(url),
inputs=img_url_input,
outputs=imgs,
)
# 의상 이미지: URL 입력 처리
garm_url_input.change(
fn=lambda url: process_url_image(url),
inputs=garm_url_input,
outputs=garm_img,
)
is_checked_crop = True
denoise_steps = 30
seed = 42
try_button.click(
fn=lambda *args: start_tryon(*args, is_checked_crop=is_checked_crop, denoise_steps=denoise_steps, seed=seed),
inputs=[imgs, garm_img, prompt, is_checked],
outputs=[image_out, masked_img],
api_name='tryon'
)
# DensePose 모델 다운로드
print("Initializing DensePose model...")
try:
download_all_models()
print("All model files downloaded successfully.")
except Exception as e:
print(f"Warning: Could not download all model files: {e}")
print("The models will be downloaded when needed during inference.")
# 앱 실행
if __name__ == "__main__":
try:
print("Starting GENAI-VTON application...")
image_blocks.launch(server_name="0.0.0.0", server_port=7860, share=False)
except Exception as e:
print(f"Error starting the application: {e}")
print("Please check if all required dependencies are installed.")