Add inpainting to CogView4DenoiseInvocation.

invokeai/app/invocations/cogview4_denoise.py

@@ -1,13 +1,16 @@
 from typing import Callable, Optional
 
 import torch
+import torchvision.transforms as tv_transforms
 from diffusers import CogView4Transformer2DModel
+from torchvision.transforms.functional import resize as tv_resize
 from tqdm import tqdm
 
 from invokeai.app.invocations.baseinvocation import BaseInvocation, Classification, invocation
 from invokeai.app.invocations.constants import LATENT_SCALE_FACTOR
 from invokeai.app.invocations.fields import (
     CogView4ConditioningField,
+    DenoiseMaskField,
     FieldDescriptions,
     Input,
     InputField,
@@ -19,6 +22,7 @@ from invokeai.app.invocations.model import TransformerField
 from invokeai.app.invocations.primitives import LatentsOutput
 from invokeai.app.services.shared.invocation_context import InvocationContext
 from invokeai.backend.flux.sampling_utils import clip_timestep_schedule_fractional
+from invokeai.backend.rectified_flow.rectified_flow_inpaint_extension import RectifiedFlowInpaintExtension
 from invokeai.backend.stable_diffusion.diffusers_pipeline import PipelineIntermediateState
 from invokeai.backend.stable_diffusion.diffusion.conditioning_data import CogView4ConditioningInfo
 from invokeai.backend.util.devices import TorchDevice
@@ -39,6 +43,10 @@ class CogView4DenoiseInvocation(BaseInvocation, WithMetadata, WithBoard):
     latents: Optional[LatentsField] = InputField(
         default=None, description=FieldDescriptions.latents, input=Input.Connection
     )
+    # denoise_mask is used for image-to-image inpainting. Only the masked region is modified.
+    denoise_mask: Optional[DenoiseMaskField] = InputField(
+        default=None, description=FieldDescriptions.denoise_mask, input=Input.Connection
+    )
     denoising_start: float = InputField(default=0.0, ge=0, le=1, description=FieldDescriptions.denoising_start)
     denoising_end: float = InputField(default=1.0, ge=0, le=1, description=FieldDescriptions.denoising_end)
     transformer: TransformerField = InputField(
@@ -64,6 +72,41 @@ class CogView4DenoiseInvocation(BaseInvocation, WithMetadata, WithBoard):
         name = context.tensors.save(tensor=latents)
         return LatentsOutput.build(latents_name=name, latents=latents, seed=None)
 
+    def _prep_inpaint_mask(self, context: InvocationContext, latents: torch.Tensor) -> torch.Tensor | None:
+        """Prepare the inpaint mask.
+        - Loads the mask
+        - Resizes if necessary
+        - Casts to same device/dtype as latents
+
+        Args:
+            context (InvocationContext): The invocation context, for loading the inpaint mask.
+            latents (torch.Tensor): A latent image tensor. Used to determine the target shape, device, and dtype for the
+                inpaint mask.
+
+        Returns:
+            torch.Tensor | None: Inpaint mask. Values of 0.0 represent the regions to be fully denoised, and 1.0
+                represent the regions to be preserved.
+        """
+        if self.denoise_mask is None:
+            return None
+        mask = context.tensors.load(self.denoise_mask.mask_name)
+
+        # The input denoise_mask contains values in [0, 1], where 0.0 represents the regions to be fully denoised, and
+        # 1.0 represents the regions to be preserved.
+        # We invert the mask so that the regions to be preserved are 0.0 and the regions to be denoised are 1.0.
+        mask = 1.0 - mask
+
+        _, _, latent_height, latent_width = latents.shape
+        mask = tv_resize(
+            img=mask,
+            size=[latent_height, latent_width],
+            interpolation=tv_transforms.InterpolationMode.BILINEAR,
+            antialias=False,
+        )
+
+        mask = mask.to(device=latents.device, dtype=latents.dtype)
+        return mask
+
     def _load_text_conditioning(
         self,
         context: InvocationContext,
@@ -227,6 +270,17 @@ class CogView4DenoiseInvocation(BaseInvocation, WithMetadata, WithBoard):
         if len(timesteps) <= 1:
             return latents
 
+        # Prepare inpaint extension.
+        inpaint_mask = self._prep_inpaint_mask(context, latents)
+        inpaint_extension: RectifiedFlowInpaintExtension | None = None
+        if inpaint_mask is not None:
+            assert init_latents is not None
+            inpaint_extension = RectifiedFlowInpaintExtension(
+                init_latents=init_latents,
+                inpaint_mask=inpaint_mask,
+                noise=noise,
+            )
+
         step_callback = self._build_step_callback(context)
 
         step_callback(
@@ -286,6 +340,9 @@ class CogView4DenoiseInvocation(BaseInvocation, WithMetadata, WithBoard):
                 latents = latents + (sigma_prev - sigma_curr) * noise_pred
                 latents = latents.to(dtype=latents_dtype)
 
+                if inpaint_extension is not None:
+                    latents = inpaint_extension.merge_intermediate_latents_with_init_latents(latents, sigma_prev)
+
                 step_callback(
                     PipelineIntermediateState(
                         step=step_idx + 1,