refactor img2img processing

This commit is contained in:
Vladimir Mandic
2024-01-04 14:46:11 -05:00
parent d4196b8185
commit 43d68ca784
5 changed files with 82 additions and 114 deletions
+6 -8
View File
@@ -67,7 +67,7 @@ def control_run(units: List[unit.Unit], inputs, inits, mask, unit_type: str, is_
cfg_scale, clip_skip, image_cfg_scale, diffusers_guidance_rescale, sag_scale, full_quality, restore_faces, tiling,
hdr_clamp, hdr_boundary, hdr_threshold, hdr_center, hdr_channel_shift, hdr_full_shift, hdr_maximize, hdr_max_center, hdr_max_boundry,
resize_mode, resize_name, width, height, scale_by, selected_scale_tab, resize_time,
denoising_strength, batch_count, batch_size,
denoising_strength, batch_count, batch_size, mask_blur, mask_overlap,
video_skip_frames, video_type, video_duration, video_loop, video_pad, video_interpolate,
ip_adapter, ip_scale, ip_image, ip_type,
):
@@ -123,6 +123,7 @@ def control_run(units: List[unit.Unit], inputs, inits, mask, unit_type: str, is_
denoising_strength = denoising_strength,
n_iter = batch_count,
batch_size = batch_size,
mask_blur=mask_blur,
outpath_samples=shared.opts.outdir_samples or shared.opts.outdir_control_samples,
outpath_grids=shared.opts.outdir_grids or shared.opts.outdir_control_grids,
)
@@ -434,13 +435,12 @@ def control_run(units: List[unit.Unit], inputs, inits, mask, unit_type: str, is_
if pipe is not None:
if not has_models and (unit_type == 'controlnet' or unit_type == 'adapter' or unit_type == 'xs' or unit_type == 'lite'): # run in txt2img/img2img/inpaint mode
if mask is not None:
p.task_args['image'] = input_image
p.task_args['mask_image'] = mask
p.task_args['strength'] = denoising_strength
p.image_mask = mask
p.mask = mask
p.inpaint_full_res = False
p.init_images = [input_image]
# if mask_overlap > 0:
# p.task_args['padding_mask_crop'] = mask_overlap # TODO enable once fixed in diffusers
shared.sd_model = sd_models.set_diffuser_pipe(shared.sd_model, sd_models.DiffusersTaskType.INPAINTING)
elif processed_image is not None:
p.init_images = [processed_image]
@@ -453,13 +453,11 @@ def control_run(units: List[unit.Unit], inputs, inits, mask, unit_type: str, is_
else: # actual control
p.is_control = True
if mask is not None:
# p.task_args['image'] = p.image
p.task_args['mask_image'] = mask
p.task_args['strength'] = denoising_strength
p.task_args['padding_mask_crop'] = 64 # should be configurable based on ui
p.image_mask = mask
p.mask = mask
p.inpaint_full_res = False
# if mask_overlap > 0:
# p.task_args['padding_mask_crop'] = mask_overlap # TODO enable once fixed in diffusers
shared.sd_model = sd_models.set_diffuser_pipe(shared.sd_model, sd_models.DiffusersTaskType.INPAINTING) # only controlnet supports inpaint
elif 'control_image' in p.task_args:
shared.sd_model = sd_models.set_diffuser_pipe(shared.sd_model, sd_models.DiffusersTaskType.IMAGE_2_IMAGE) # only controlnet supports img2img
-1
View File
@@ -46,7 +46,6 @@ def make_noise_disk(H, W, C, F):
def nms(x, t, s):
x = cv2.GaussianBlur(x.astype(np.float32), (0, 0), s)
f1 = np.array([[0, 0, 0], [1, 1, 1], [0, 0, 0]], dtype=np.uint8)
f2 = np.array([[0, 1, 0], [0, 1, 0], [0, 1, 0]], dtype=np.uint8)
f3 = np.array([[1, 0, 0], [0, 1, 0], [0, 0, 1]], dtype=np.uint8)
+39 -59
View File
@@ -49,19 +49,20 @@ debug('Trace: PROCESS')
def setup_color_correction(image):
shared.log.debug("Calibrating color correction.")
debug("Calibrating color correction")
correction_target = cv2.cvtColor(np.asarray(image.copy()), cv2.COLOR_RGB2LAB)
return correction_target
def apply_color_correction(correction, original_image):
shared.log.debug("Applying color correction.")
shared.log.debug(f"Applying color correction: correction={correction} image={original_image}")
image = Image.fromarray(cv2.cvtColor(exposure.match_histograms(cv2.cvtColor(np.asarray(original_image), cv2.COLOR_RGB2LAB), correction, channel_axis=2), cv2.COLOR_LAB2RGB).astype("uint8"))
image = blendLayers(image, original_image, BlendType.LUMINOSITY)
return image
def apply_overlay(image: Image, paste_loc, index, overlays):
debug(f'Apply overlay: image={image} loc={paste_loc} index={index} overlays={overlays}')
if overlays is None or index >= len(overlays):
return image
overlay = overlays[index]
@@ -1240,8 +1241,8 @@ class StableDiffusionProcessingImg2Img(StableDiffusionProcessing):
self.image_mask = mask
self.latent_mask = None
self.mask_for_overlay = None
self.mask_blur_x: int = 4
self.mask_blur_y: int = 4
self.mask_blur_x = mask_blur # a1111 compatibility item
self.mask_blur_y = mask_blur # a1111 compatibility item
self.mask_blur = mask_blur
self.inpainting_fill = inpainting_fill
self.inpaint_full_res = inpaint_full_res
@@ -1262,16 +1263,6 @@ class StableDiffusionProcessingImg2Img(StableDiffusionProcessing):
self.scripts = None
self.script_args = []
@property
def mask_blur(self):
mask_blur = max(self.mask_blur_x, self.mask_blur_y)
return mask_blur
@mask_blur.setter
def mask_blur(self, value):
self.mask_blur_x = value
self.mask_blur_y = value
def init(self, all_prompts, all_seeds, all_subseeds):
if shared.backend == shared.Backend.DIFFUSERS and self.image_mask is not None and not self.is_control:
shared.sd_model = modules.sd_models.set_diffuser_pipe(self.sd_model, modules.sd_models.DiffusersTaskType.INPAINTING)
@@ -1290,46 +1281,40 @@ class StableDiffusionProcessingImg2Img(StableDiffusionProcessing):
self.ops.append('img2img')
crop_region = None
image_mask = self.image_mask
if image_mask is not None:
if type(image_mask) == list:
image_mask = image_mask[0]
image_mask = create_binary_mask(image_mask)
if self.image_mask is not None:
if type(self.image_mask) == list:
self.image_mask = self.image_mask[0]
self.image_mask = create_binary_mask(self.image_mask)
if self.inpainting_mask_invert:
image_mask = ImageOps.invert(image_mask)
if self.mask_blur_x > 0:
np_mask = np.array(image_mask)
kernel_size = 2 * int(2.5 * self.mask_blur_x + 0.5) + 1
np_mask = cv2.GaussianBlur(np_mask, (kernel_size, 1), self.mask_blur_x)
image_mask = Image.fromarray(np_mask)
if self.mask_blur_y > 0:
np_mask = np.array(image_mask)
kernel_size = 2 * int(2.5 * self.mask_blur_y + 0.5) + 1
np_mask = cv2.GaussianBlur(np_mask, (1, kernel_size), self.mask_blur_y)
image_mask = Image.fromarray(np_mask)
self.image_mask = ImageOps.invert(self.image_mask)
if self.mask_blur > 0:
np_mask = np.array(self.image_mask)
kernel_size = 2 * int(2.5 * self.mask_blur + 0.5) + 1
np_mask = cv2.GaussianBlur(np_mask, (kernel_size, 1), self.mask_blur)
np_mask = cv2.GaussianBlur(np_mask, (1, kernel_size), self.mask_blur)
self.image_mask = Image.fromarray(np_mask)
if self.inpaint_full_res:
self.mask_for_overlay = image_mask
mask = image_mask.convert('L')
self.mask_for_overlay = self.image_mask
mask = self.image_mask.convert('L')
crop_region = modules.masking.get_crop_region(np.array(mask), self.inpaint_full_res_padding)
crop_region = modules.masking.expand_crop_region(crop_region, self.width, self.height, mask.width, mask.height)
x1, y1, x2, y2 = crop_region
mask = mask.crop(crop_region)
image_mask = images.resize_image(2, mask, self.width, self.height)
self.image_mask = images.resize_image(2, mask, self.width, self.height)
self.paste_to = (x1, y1, x2-x1, y2-y1)
else:
image_mask = images.resize_image(self.resize_mode, image_mask, self.width, self.height)
np_mask = np.array(image_mask)
self.image_mask = images.resize_image(self.resize_mode, self.image_mask, self.width, self.height)
np_mask = np.array(self.image_mask)
np_mask = np.clip((np_mask.astype(np.float32)) * 2, 0, 255).astype(np.uint8)
self.mask_for_overlay = Image.fromarray(np_mask)
self.overlay_images = []
latent_mask = self.latent_mask if self.latent_mask is not None else image_mask
latent_mask = self.latent_mask if self.latent_mask is not None else self.image_mask
add_color_corrections = shared.opts.img2img_color_correction and self.color_corrections is None
if add_color_corrections:
self.color_corrections = []
imgs = []
unprocessed = []
processed = []
if getattr(self, 'init_images', None) is None:
return
if not isinstance(self.init_images, list):
@@ -1349,7 +1334,7 @@ class StableDiffusionProcessingImg2Img(StableDiffusionProcessing):
image = images.resize_image(self.resize_mode, image, self.width, self.height, self.resize_name)
self.width = image.width
self.height = image.height
if image_mask is not None:
if self.image_mask is not None:
try:
image_masked = Image.new('RGBa', (image.width, image.height))
image_to_paste = image.convert("RGBA").convert("RGBa")
@@ -1363,37 +1348,32 @@ class StableDiffusionProcessingImg2Img(StableDiffusionProcessing):
image = image.crop(crop_region)
if image.width != self.width or image.height != self.height:
image = images.resize_image(3, image, self.width, self.height, self.resize_name)
if image_mask is not None and self.inpainting_fill != 1:
if self.image_mask is not None and self.inpainting_fill != 1:
image = modules.masking.fill(image, latent_mask)
if add_color_corrections:
self.color_corrections.append(setup_color_correction(image))
if shared.backend == shared.Backend.DIFFUSERS:
unprocessed.append(image) # assign early for diffusers
image = np.array(image).astype(np.float32) / 255.0
image = np.moveaxis(image, 2, 0)
imgs.append(image)
self.init_images = unprocessed if shared.backend == shared.Backend.DIFFUSERS else imgs
if len(imgs) == 1:
batch_images = np.expand_dims(imgs[0], axis=0).repeat(self.batch_size, axis=0)
if self.overlay_images is not None:
self.overlay_images = self.overlay_images * self.batch_size
if self.color_corrections is not None and len(self.color_corrections) == 1:
self.color_corrections = self.color_corrections * self.batch_size
elif len(imgs) <= self.batch_size:
self.batch_size = len(imgs)
batch_images = np.array(imgs)
else:
raise RuntimeError(f"Incorrect number of of images={len(imgs)} expected={self.batch_size} or less")
processed.append(image)
self.init_images = processed
self.batch_size = len(self.init_images)
if self.overlay_images is not None:
self.overlay_images = self.overlay_images * self.batch_size
if self.color_corrections is not None and len(self.color_corrections) == 1:
self.color_corrections = self.color_corrections * self.batch_size
if shared.backend == shared.Backend.DIFFUSERS:
return # we've already set self.init_images and self.mask and we dont need any more processing
self.init_images = [np.moveaxis((np.array(image).astype(np.float32) / 255.0), 2, 0) for image in self.init_images]
if len(self.init_images) == 1:
batch_images = np.expand_dims(self.init_images[0], axis=0).repeat(self.batch_size, axis=0)
elif len(self.init_images) <= self.batch_size:
batch_images = np.array(self.init_images)
image = torch.from_numpy(batch_images)
image = 2. * image - 1.
image = image.to(device=shared.device, dtype=devices.dtype_vae)
self.init_latent = self.sd_model.get_first_stage_encoding(self.sd_model.encode_first_stage(image))
if self.resize_mode == 4:
self.init_latent = torch.nn.functional.interpolate(self.init_latent, size=(self.height // 8, self.width // 8), mode="bilinear")
if image_mask is not None:
if self.image_mask is not None:
init_mask = latent_mask
latmask = init_mask.convert('RGB').resize((self.init_latent.shape[3], self.init_latent.shape[2]))
latmask = np.moveaxis(np.array(latmask, dtype=np.float32), 2, 0) / 255
@@ -1406,7 +1386,7 @@ class StableDiffusionProcessingImg2Img(StableDiffusionProcessing):
self.init_latent = self.init_latent * self.mask + create_random_tensors(self.init_latent.shape[1:], all_seeds[0:self.init_latent.shape[0]]) * self.nmask
elif self.inpainting_fill == 3:
self.init_latent = self.init_latent * self.mask
self.image_conditioning = self.img2img_image_conditioning(image, self.init_latent, image_mask)
self.image_conditioning = self.img2img_image_conditioning(image, self.init_latent, self.image_mask)
def sample(self, conditioning, unconditional_conditioning, seeds, subseeds, subseed_strength, prompts):
hypertile_set(self)
+25 -24
View File
@@ -35,18 +35,21 @@ def process_diffusers(p: StableDiffusionProcessing):
def is_refiner_enabled():
return p.enable_hr and p.refiner_steps > 0 and p.refiner_start > 0 and p.refiner_start < 1 and shared.sd_refiner is not None
if getattr(p, 'init_images', None) is not None and len(p.init_images) > 0:
tgt_width, tgt_height = 8 * math.ceil(p.init_images[0].width / 8), 8 * math.ceil(p.init_images[0].height / 8)
if p.init_images[0].width != tgt_width or p.init_images[0].height != tgt_height:
shared.log.debug(f'Resizing init images: original={p.init_images[0].width}x{p.init_images[0].height} target={tgt_width}x{tgt_height}')
p.init_images = [images.resize_image(1, image, tgt_width, tgt_height, upscaler_name=None) for image in p.init_images]
p.height = tgt_height
p.width = tgt_width
hypertile_set(p)
if getattr(p, 'mask', None) is not None and p.mask.size != (tgt_width, tgt_height):
p.mask = images.resize_image(1, p.mask, tgt_width, tgt_height, upscaler_name=None)
if getattr(p, 'mask_for_overlay', None) is not None and p.mask_for_overlay.size != (tgt_width, tgt_height):
p.mask_for_overlay = images.resize_image(1, p.mask_for_overlay, tgt_width, tgt_height, upscaler_name=None)
def resize_images():
if getattr(p, 'init_images', None) is not None and len(p.init_images) > 0:
tgt_width, tgt_height = 8 * math.ceil(p.init_images[0].width / 8), 8 * math.ceil(p.init_images[0].height / 8)
if p.init_images[0].size != (tgt_width, tgt_height):
shared.log.debug(f'Resizing init images: original={p.init_images[0].width}x{p.init_images[0].height} target={tgt_width}x{tgt_height}')
p.init_images = [images.resize_image(1, image, tgt_width, tgt_height, upscaler_name=None) for image in p.init_images]
p.height = tgt_height
p.width = tgt_width
hypertile_set(p)
if getattr(p, 'mask', None) is not None and p.mask.size != (tgt_width, tgt_height):
p.mask = images.resize_image(1, p.mask, tgt_width, tgt_height, upscaler_name=None)
if getattr(p, 'mask_for_overlay', None) is not None and p.mask_for_overlay.size != (tgt_width, tgt_height):
p.mask_for_overlay = images.resize_image(1, p.mask_for_overlay, tgt_width, tgt_height, upscaler_name=None)
return tgt_width, tgt_height
return p.width, p.height
def hires_resize(latents): # input=latents output=pil
if not torch.is_tensor(latents):
@@ -165,13 +168,15 @@ def process_diffusers(p: StableDiffusionProcessing):
}
elif (sd_models.get_diffusers_task(model) == sd_models.DiffusersTaskType.INPAINTING or is_img2img_model) and len(getattr(p, 'init_images' ,[])) > 0:
p.ops.append('inpaint')
if getattr(p, 'mask', None) is None:
if getattr(p, 'image_mask', None) is not None:
p.mask = p.image_mask
else:
p.mask = TF.to_pil_image(torch.ones_like(TF.to_tensor(p.init_images[0]))).convert("L")
width = 8 * math.ceil(p.init_images[0].width / 8)
height = 8 * math.ceil(p.init_images[0].height / 8)
if p.task_args.get('mask_image', None) is not None: # provided as override by a module
p.mask = shared.sd_model.mask_processor.blur(p.task_args['mask_image'], blur_factor=p.mask_blur) if p.mask_blur > 0 else p.task_args['mask_image']
elif getattr(p, 'image_mask', None) is not None: # standard
p.mask = p.image_mask
elif getattr(p, 'mask', None) is not None: # backward compatibility
pass
else: # fallback
p.mask = TF.to_pil_image(torch.ones_like(TF.to_tensor(p.init_images[0]))).convert("L")
width, height = resize_images()
task_args = {
'image': p.init_images,
'mask_image': p.mask,
@@ -180,10 +185,6 @@ def process_diffusers(p: StableDiffusionProcessing):
'width': width,
# 'padding_mask_crop': p.inpaint_full_res_padding # done back in main processing method
}
if p.task_args.get('mask_image', None) is None:
if p.mask_blur > 0:
p.mask = shared.sd_model.mask_processor.blur(p.mask, blur_factor=p.mask_blur)
task_args['mask_image'] = p.mask
if model.__class__.__name__ == 'LatentConsistencyModelPipeline' and hasattr(p, 'init_images') and len(p.init_images) > 0:
p.ops.append('lcm')
init_latents = [vae_encode(image, model=shared.sd_model, full_quality=p.full_quality).squeeze(dim=0) for image in p.init_images]
@@ -465,7 +466,7 @@ def process_diffusers(p: StableDiffusionProcessing):
try:
t0 = time.time()
output = shared.sd_model(**base_args) # pylint: disable=not-callable
downcast_openvino(op="base")
downcast_openvino(op="base") # only executes on compiled vino models
if shared.cmd_opts.profile:
t1 = time.time()
shared.log.debug(f'Profile: pipeline call: {t1-t0:.2f}')
+12 -22
View File
@@ -117,8 +117,7 @@ def get_video(filepath: str):
return msg
def select_mask(image: Image.Image, blur: int = 0, negative: bool = False):
import cv2
def select_mask(image: Image.Image, negative: bool = False):
if image is None:
return image
image_mask = image.convert("L")
@@ -126,36 +125,27 @@ def select_mask(image: Image.Image, blur: int = 0, negative: bool = False):
image_mask = image_mask.point(lambda x: 255 if x < 4 else 0)
else:
image_mask = image_mask.point(lambda x: 255 if x > 127 else 0)
if blur > 0:
kernel_size = 2 * int(2.5 * blur + 0.5) + 1
np_mask = np.array(image_mask)
np_mask = cv2.GaussianBlur(np_mask, (kernel_size, 1), blur)
np_mask = cv2.GaussianBlur(np_mask, (1, kernel_size), blur)
image_mask = Image.fromarray(np_mask.astype(np.uint8))
return image_mask
def expand_mask(image: Image.Image, blur: int = 0, erode: int = 3, dilate: int = 16, iterations: int = 8, threshold: int = 4):
def expand_mask(image: Image.Image, expand: int = 64):
import cv2
if image is None:
return image
pil_mask = image.convert("L")
np_mask = np.array(pil_mask)
erode, dilate, threshold = 3, 8, 4
if threshold > 0:
_thres, np_mask = cv2.threshold(np_mask, threshold, 255, cv2.THRESH_BINARY_INV) # create mask
if erode > 0:
np_mask = cv2.erode(np_mask, np.ones((erode, erode), np.uint8), iterations=iterations) # remove noise
np_mask = cv2.erode(np_mask, np.ones((erode, erode), np.uint8), iterations=expand//dilate) # remove noise
if dilate > 0:
np_mask = cv2.dilate(np_mask, np.ones((dilate, dilate), np.uint8), iterations=iterations) # expand area
if blur > 0:
blur_size = 2 * int(2.5 * blur + 0.5) + 1
np_mask = cv2.GaussianBlur(np_mask, (blur_size, 1), blur) # blur x-axis
np_mask = cv2.GaussianBlur(np_mask, (1, blur_size), blur) # blur y-axis
np_mask = cv2.dilate(np_mask, np.ones((dilate, dilate), np.uint8), iterations=expand//dilate) # expand area
image_mask = Image.fromarray(np_mask.astype(np.uint8))
return image_mask
def select_input(input_mode, input_image, selected_init, init_type, input_resize, input_inpaint, input_video, input_batch, input_folder, mask_blur, mask_overlap):
def select_input(input_mode, input_image, selected_init, init_type, input_resize, input_inpaint, input_video, input_batch, input_folder, _mask_blur, mask_overlap):
global busy, input_source, input_init, input_mask # pylint: disable=global-statement
busy = True
if input_mode == 'Select':
@@ -183,14 +173,14 @@ def select_input(input_mode, input_image, selected_init, init_type, input_resize
# control inputs
if isinstance(selected_input, Image.Image): # image via upload -> image
if input_mode == 'Outpaint':
input_mask = expand_mask(image=selected_input, blur=mask_blur, iterations=mask_overlap)
input_mask = expand_mask(image=selected_input, expand=mask_overlap)
input_source = [selected_input]
input_type = 'PIL.Image'
shared.log.debug(f'Control input: type={input_type} input={input_source}')
status = f'Control input | Image | Size {selected_input.width}x{selected_input.height} | Mode {selected_input.mode}'
res = [gr.Tabs.update(selected='out-gallery'), status]
elif isinstance(selected_input, dict): # inpaint -> dict image+mask
input_mask = select_mask(image=selected_input['mask'], blur=mask_blur, negative=False)
input_mask = select_mask(image=selected_input['mask'], negative=False)
selected_input = selected_input['image']
input_source = [selected_input]
input_type = 'PIL.Image'
@@ -228,7 +218,7 @@ def select_input(input_mode, input_image, selected_init, init_type, input_resize
elif init_type == 2: # Separate init image
if isinstance(selected_init, Image.Image): # image via upload -> image
if input_mode == 'Outpaint':
input_mask = expand_mask(image=selected_init, blur=mask_blur, iterations=mask_overlap)
input_mask = expand_mask(image=selected_init, expand=mask_overlap)
input_source = [selected_init]
input_init = [selected_init]
input_type = 'PIL.Image'
@@ -236,7 +226,7 @@ def select_input(input_mode, input_image, selected_init, init_type, input_resize
status = f'Control input | Image | Size {selected_init.width}x{selected_init.height} | Mode {selected_init.mode}'
res = [gr.Tabs.update(selected='out-gallery'), status]
elif isinstance(selected_init, dict): # inpaint -> dict image+mask
input_mask = select_mask(image=selected_init['mask'], blur=mask_blur)
input_mask = select_mask(image=selected_init['mask'])
input_init = selected_init['image']
input_source = [selected_init]
input_type = 'PIL.Image'
@@ -327,7 +317,7 @@ def create_ui(_blocks: gr.Blocks=None):
denoising_strength = gr.Slider(minimum=0.01, maximum=1.0, step=0.01, label='Denoising strength', value=0.50, elem_id="control_denoising_strength")
with gr.Row():
mask_blur = gr.Slider(minimum=0, maximum=100, step=1, label='Blur', value=8, elem_id="control_mask_blur")
mask_overlap = gr.Slider(minimum=0, maximum=100, step=1, label='Overlap', value=8, elem_id="control_mask_overlap")
mask_overlap = gr.Slider(minimum=0, maximum=100, step=1, label='Overlap', value=64, elem_id="control_mask_overlap")
resize_mode, resize_name, width, height, scale_by, selected_scale_tab, resize_time = ui_sections.create_resize_inputs('control', [], time_selector=True, scale_visible=False, mode='Fixed')
@@ -709,7 +699,7 @@ def create_ui(_blocks: gr.Blocks=None):
seed, subseed, subseed_strength, seed_resize_from_h, seed_resize_from_w,
cfg_scale, clip_skip, image_cfg_scale, diffusers_guidance_rescale, sag_scale, full_quality, restore_faces, tiling, hdr_clamp, hdr_boundary, hdr_threshold, hdr_center, hdr_channel_shift, hdr_full_shift, hdr_maximize, hdr_max_center, hdr_max_boundry,
resize_mode, resize_name, width, height, scale_by, selected_scale_tab, resize_time,
denoising_strength, batch_count, batch_size,
denoising_strength, batch_count, batch_size, mask_blur, mask_overlap,
video_skip_frames, video_type, video_duration, video_loop, video_pad, video_interpolate,
ip_adapter, ip_scale, ip_image, ip_type,
]