File size: 3,323 Bytes
dcd8d9c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 |
from diffusers.modular_pipelines import (
PipelineBlock,
InputParam,
OutputParam,
ConfigSpec,
)
from diffusers.utils import load_image
from PIL import Image
from typing import Union, Tuple
# copied from https://github.com/Wan-Video/Wan2.2/blob/388807310646ed5f318a99f8e8d9ad28c5b65373/wan/utils/utils.py#L136
def best_output_size(w, h, dw, dh, expected_area):
# float output size
ratio = w / h
ow = (expected_area * ratio)**0.5
oh = expected_area / ow
# process width first
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
# process height first
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
# compare ratios
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)
# center-crop
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
|