|
|
from diffusers.modular_pipelines import ( |
|
|
PipelineBlock, |
|
|
InputParam, |
|
|
OutputParam, |
|
|
ConfigSpec, |
|
|
) |
|
|
|
|
|
from diffusers.utils import load_image |
|
|
from PIL import Image |
|
|
from typing import Union, Tuple |
|
|
|
|
|
|
|
|
def best_output_size(w, h, dw, dh, expected_area): |
|
|
|
|
|
ratio = w / h |
|
|
ow = (expected_area * ratio)**0.5 |
|
|
oh = expected_area / ow |
|
|
|
|
|
|
|
|
ow1 = int(ow // dw * dw) |
|
|
oh1 = int(expected_area / ow1 // dh * dh) |
|
|
assert ow1 % dw == 0 and oh1 % dh == 0 and ow1 * oh1 <= expected_area |
|
|
ratio1 = ow1 / oh1 |
|
|
|
|
|
|
|
|
oh2 = int(oh // dh * dh) |
|
|
ow2 = int(expected_area / oh2 // dw * dw) |
|
|
assert oh2 % dh == 0 and ow2 % dw == 0 and ow2 * oh2 <= expected_area |
|
|
ratio2 = ow2 / oh2 |
|
|
|
|
|
|
|
|
if max(ratio / ratio1, ratio1 / ratio) < max(ratio / ratio2, |
|
|
ratio2 / ratio): |
|
|
return ow1, oh1 |
|
|
else: |
|
|
return ow2, oh2 |
|
|
|
|
|
class Wan225BI2VImageProcessor(PipelineBlock): |
|
|
|
|
|
@property |
|
|
def description(self): |
|
|
return "default Image Processor for wan2.2 5b i2v, it resizes the image to the best output size and center-crop it" |
|
|
|
|
|
@property |
|
|
def inputs(self): |
|
|
return [ |
|
|
InputParam(name="image", type_hint=Union[Image.Image, str], description= "the Image to process"), |
|
|
InputParam(name="max_area", type_hint=int, description= "the maximum area of the Image to process") |
|
|
] |
|
|
|
|
|
@property |
|
|
def intermediate_outputs(self): |
|
|
return [ |
|
|
OutputParam(name="processed_image", type_hint=Image.Image, description= "the processed Image"), |
|
|
] |
|
|
|
|
|
@property |
|
|
def expected_configs(self): |
|
|
return [ |
|
|
ConfigSpec(name="patch_size", default=(1, 2, 2)), |
|
|
ConfigSpec(name="vae_stride", default=(4, 16, 16)), |
|
|
] |
|
|
|
|
|
def __call__(self, components, state): |
|
|
|
|
|
block_state = self.get_block_state(state) |
|
|
|
|
|
if isinstance(block_state.image, str): |
|
|
image = load_image(block_state.image).convert("RGB") |
|
|
elif isinstance(block_state.image, Image.Image): |
|
|
image = block_state.image |
|
|
else: |
|
|
raise ValueError(f"Invalid image type: {type(block_state.image)}; only support PIL Image or url string") |
|
|
|
|
|
ih, iw = image.height, image.width |
|
|
dh, dw = components.patch_size[1] * components.vae_stride[1], components.patch_size[2] * components.vae_stride[2] |
|
|
ow, oh = best_output_size(iw, ih, dw, dh, block_state.max_area) |
|
|
|
|
|
scale = max(ow / iw, oh / ih) |
|
|
resized_image = image.resize((round(iw * scale), round(ih * scale)), Image.LANCZOS) |
|
|
|
|
|
|
|
|
x1 = (resized_image.width - ow) // 2 |
|
|
y1 = (resized_image.height - oh) // 2 |
|
|
cropped_image = resized_image.crop((x1, y1, x1 + ow, y1 + oh)) |
|
|
assert cropped_image.width == ow and cropped_image.height == oh |
|
|
|
|
|
block_state.processed_image = cropped_image |
|
|
|
|
|
print(f" initial image size: {image.size}") |
|
|
print(f" processed image size: {cropped_image.size}") |
|
|
|
|
|
|
|
|
self.set_block_state(state, block_state) |
|
|
return components, state |
|
|
|
|
|
|
