| | 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.config.patch_size[1] * components.config.vae_stride[1], components.config.patch_size[2] * components.config.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
|
| |
|
| |
|
