Move ConditioningData and its field classes to their own file. This will allow new conditioning types to be added more cleanly without introducing circular dependencies.

invokeai/app/invocations/compel.py

@@ -4,18 +4,23 @@ from typing import List, Union
 
 import torch
 from compel import Compel, ReturnedEmbeddingsType
-from compel.prompt_parser import Blend, Conjunction, CrossAttentionControlSubstitute, FlattenedPrompt, Fragment
-from invokeai.app.invocations.primitives import ConditioningField, ConditioningOutput
+from compel.prompt_parser import (
+    Blend,
+    Conjunction,
+    CrossAttentionControlSubstitute,
+    FlattenedPrompt,
+    Fragment,
+)
 
-from invokeai.backend.stable_diffusion.diffusion.shared_invokeai_diffusion import (
+from invokeai.app.invocations.primitives import ConditioningField, ConditioningOutput
+from invokeai.backend.stable_diffusion.diffusion.conditioning_data import (
     BasicConditioningInfo,
+    ExtraConditioningInfo,
     SDXLConditioningInfo,
 )
 
-from ...backend.model_management.models import ModelType
 from ...backend.model_management.lora import ModelPatcher
-from ...backend.model_management.models import ModelNotFoundException
-from ...backend.stable_diffusion.diffusion import InvokeAIDiffuserComponent
+from ...backend.model_management.models import ModelNotFoundException, ModelType
 from ...backend.util.devices import torch_dtype
 from .baseinvocation import (
     BaseInvocation,
@@ -100,14 +105,15 @@ class CompelInvocation(BaseInvocation):
                 # print(traceback.format_exc())
                 print(f'Warn: trigger: "{trigger}" not found')
 
-        with ModelPatcher.apply_lora_text_encoder(
-            text_encoder_info.context.model, _lora_loader()
-        ), ModelPatcher.apply_ti(tokenizer_info.context.model, text_encoder_info.context.model, ti_list) as (
-            tokenizer,
-            ti_manager,
-        ), ModelPatcher.apply_clip_skip(
-            text_encoder_info.context.model, self.clip.skipped_layers
-        ), text_encoder_info as text_encoder:
+        with (
+            ModelPatcher.apply_lora_text_encoder(text_encoder_info.context.model, _lora_loader()),
+            ModelPatcher.apply_ti(tokenizer_info.context.model, text_encoder_info.context.model, ti_list) as (
+                tokenizer,
+                ti_manager,
+            ),
+            ModelPatcher.apply_clip_skip(text_encoder_info.context.model, self.clip.skipped_layers),
+            text_encoder_info as text_encoder,
+        ):
             compel = Compel(
                 tokenizer=tokenizer,
                 text_encoder=text_encoder,
@@ -123,7 +129,7 @@ class CompelInvocation(BaseInvocation):
 
             c, options = compel.build_conditioning_tensor_for_conjunction(conjunction)
 
-            ec = InvokeAIDiffuserComponent.ExtraConditioningInfo(
+            ec = ExtraConditioningInfo(
                 tokens_count_including_eos_bos=get_max_token_count(tokenizer, conjunction),
                 cross_attention_control_args=options.get("cross_attention_control", None),
             )
@@ -214,14 +220,15 @@ class SDXLPromptInvocationBase:
                 # print(traceback.format_exc())
                 print(f'Warn: trigger: "{trigger}" not found')
 
-        with ModelPatcher.apply_lora(
-            text_encoder_info.context.model, _lora_loader(), lora_prefix
-        ), ModelPatcher.apply_ti(tokenizer_info.context.model, text_encoder_info.context.model, ti_list) as (
-            tokenizer,
-            ti_manager,
-        ), ModelPatcher.apply_clip_skip(
-            text_encoder_info.context.model, clip_field.skipped_layers
-        ), text_encoder_info as text_encoder:
+        with (
+            ModelPatcher.apply_lora(text_encoder_info.context.model, _lora_loader(), lora_prefix),
+            ModelPatcher.apply_ti(tokenizer_info.context.model, text_encoder_info.context.model, ti_list) as (
+                tokenizer,
+                ti_manager,
+            ),
+            ModelPatcher.apply_clip_skip(text_encoder_info.context.model, clip_field.skipped_layers),
+            text_encoder_info as text_encoder,
+        ):
             compel = Compel(
                 tokenizer=tokenizer,
                 text_encoder=text_encoder,
@@ -245,7 +252,7 @@ class SDXLPromptInvocationBase:
             else:
                 c_pooled = None
 
-            ec = InvokeAIDiffuserComponent.ExtraConditioningInfo(
+            ec = ExtraConditioningInfo(
                 tokens_count_including_eos_bos=get_max_token_count(tokenizer, conjunction),
                 cross_attention_control_args=options.get("cross_attention_control", None),
             )
@@ -437,9 +444,11 @@ def get_tokens_for_prompt_object(tokenizer, parsed_prompt: FlattenedPrompt, trun
         raise ValueError("Blend is not supported here - you need to get tokens for each of its .children")
 
     text_fragments = [
-        x.text
-        if type(x) is Fragment
-        else (" ".join([f.text for f in x.original]) if type(x) is CrossAttentionControlSubstitute else str(x))
+        (
+            x.text
+            if type(x) is Fragment
+            else (" ".join([f.text for f in x.original]) if type(x) is CrossAttentionControlSubstitute else str(x))
+        )
         for x in parsed_prompt.children
     ]
     text = " ".join(text_fragments)

invokeai/app/invocations/latent.py

@@ -33,13 +33,15 @@ from invokeai.app.invocations.primitives import (
 from invokeai.app.util.controlnet_utils import prepare_control_image
 from invokeai.app.util.step_callback import stable_diffusion_step_callback
 from invokeai.backend.model_management.models import ModelType, SilenceWarnings
+from invokeai.backend.stable_diffusion.diffusion.conditioning_data import (
+    ConditioningData,
+)
 
 from ...backend.model_management.lora import ModelPatcher
 from ...backend.model_management.models import BaseModelType
 from ...backend.model_management.seamless import set_seamless
 from ...backend.stable_diffusion import PipelineIntermediateState
 from ...backend.stable_diffusion.diffusers_pipeline import (
-    ConditioningData,
     ControlNetData,
     IPAdapterData,
     StableDiffusionGeneratorPipeline,

invokeai/backend/stable_diffusion/__init__.py

@@ -2,14 +2,8 @@
 Initialization file for the invokeai.backend.stable_diffusion package
 """
 from .diffusers_pipeline import (  # noqa: F401
-    ConditioningData,
     PipelineIntermediateState,
     StableDiffusionGeneratorPipeline,
 )
 from .diffusion import InvokeAIDiffuserComponent  # noqa: F401
 from .diffusion.cross_attention_map_saving import AttentionMapSaver  # noqa: F401
-from .diffusion.shared_invokeai_diffusion import (  # noqa: F401
-    PostprocessingSettings,
-    BasicConditioningInfo,
-    SDXLConditioningInfo,
-)

invokeai/backend/stable_diffusion/diffusers_pipeline.py

@@ -28,14 +28,12 @@ from transformers import CLIPFeatureExtractor, CLIPTextModel, CLIPTokenizer
 
 from invokeai.app.services.config import InvokeAIAppConfig
 from invokeai.backend.ip_adapter.ip_adapter import IPAdapter, IPAdapterPlus, IPAdapterXL
+from invokeai.backend.stable_diffusion.diffusion.conditioning_data import (
+    ConditioningData,
+)
 
 from ..util import auto_detect_slice_size, normalize_device
-from .diffusion import (
-    AttentionMapSaver,
-    BasicConditioningInfo,
-    InvokeAIDiffuserComponent,
-    PostprocessingSettings,
-)
+from .diffusion import AttentionMapSaver, InvokeAIDiffuserComponent
 
 
 @dataclass
@@ -181,42 +179,6 @@ class IPAdapterData:
     weight: float = Field(default=1.0)
 
 
-@dataclass
-class ConditioningData:
-    unconditioned_embeddings: BasicConditioningInfo
-    text_embeddings: BasicConditioningInfo
-    guidance_scale: Union[float, List[float]]
-    """
-    Guidance scale as defined in [Classifier-Free Diffusion Guidance](https://arxiv.org/abs/2207.12598).
-    `guidance_scale` is defined as `w` of equation 2. of [Imagen Paper](https://arxiv.org/pdf/2205.11487.pdf).
-    Guidance scale is enabled by setting `guidance_scale > 1`. Higher guidance scale encourages to generate
-    images that are closely linked to the text `prompt`, usually at the expense of lower image quality.
-    """
-    extra: Optional[InvokeAIDiffuserComponent.ExtraConditioningInfo] = None
-    scheduler_args: dict[str, Any] = field(default_factory=dict)
-    """
-    Additional arguments to pass to invokeai_diffuser.do_latent_postprocessing().
-    """
-    postprocessing_settings: Optional[PostprocessingSettings] = None
-
-    @property
-    def dtype(self):
-        return self.text_embeddings.dtype
-
-    def add_scheduler_args_if_applicable(self, scheduler, **kwargs):
-        scheduler_args = dict(self.scheduler_args)
-        step_method = inspect.signature(scheduler.step)
-        for name, value in kwargs.items():
-            try:
-                step_method.bind_partial(**{name: value})
-            except TypeError:
-                # FIXME: don't silently discard arguments
-                pass  # debug("%s does not accept argument named %r", scheduler, name)
-            else:
-                scheduler_args[name] = value
-        return dataclasses.replace(self, scheduler_args=scheduler_args)
-
-
 @dataclass
 class InvokeAIStableDiffusionPipelineOutput(StableDiffusionPipelineOutput):
     r"""

invokeai/backend/stable_diffusion/diffusion/__init__.py

@@ -3,9 +3,4 @@ Initialization file for invokeai.models.diffusion
 """
 from .cross_attention_control import InvokeAICrossAttentionMixin  # noqa: F401
 from .cross_attention_map_saving import AttentionMapSaver  # noqa: F401
-from .shared_invokeai_diffusion import (  # noqa: F401
-    InvokeAIDiffuserComponent,
-    PostprocessingSettings,
-    BasicConditioningInfo,
-    SDXLConditioningInfo,
-)
+from .shared_invokeai_diffusion import InvokeAIDiffuserComponent  # noqa: F401

invokeai/backend/stable_diffusion/diffusion/conditioning_data.py

@@ -0,0 +1,87 @@
+import dataclasses
+import inspect
+from dataclasses import dataclass, field
+from typing import Any, List, Optional, Union
+
+import torch
+
+from .cross_attention_control import Arguments
+
+
+@dataclass
+class ExtraConditioningInfo:
+    tokens_count_including_eos_bos: int
+    cross_attention_control_args: Optional[Arguments] = None
+
+    @property
+    def wants_cross_attention_control(self):
+        return self.cross_attention_control_args is not None
+
+
+@dataclass
+class BasicConditioningInfo:
+    embeds: torch.Tensor
+    # TODO(ryand): Right now we awkwardly copy the extra conditioning info from here up to `ConditioningData`. This
+    # should only be stored in one place.
+    extra_conditioning: Optional[ExtraConditioningInfo]
+    # weight: float
+    # mode: ConditioningAlgo
+
+    def to(self, device, dtype=None):
+        self.embeds = self.embeds.to(device=device, dtype=dtype)
+        return self
+
+
+@dataclass
+class SDXLConditioningInfo(BasicConditioningInfo):
+    pooled_embeds: torch.Tensor
+    add_time_ids: torch.Tensor
+
+    def to(self, device, dtype=None):
+        self.pooled_embeds = self.pooled_embeds.to(device=device, dtype=dtype)
+        self.add_time_ids = self.add_time_ids.to(device=device, dtype=dtype)
+        return super().to(device=device, dtype=dtype)
+
+
+@dataclass(frozen=True)
+class PostprocessingSettings:
+    threshold: float
+    warmup: float
+    h_symmetry_time_pct: Optional[float]
+    v_symmetry_time_pct: Optional[float]
+
+
+@dataclass
+class ConditioningData:
+    unconditioned_embeddings: BasicConditioningInfo
+    text_embeddings: BasicConditioningInfo
+    guidance_scale: Union[float, List[float]]
+    """
+    Guidance scale as defined in [Classifier-Free Diffusion Guidance](https://arxiv.org/abs/2207.12598).
+    `guidance_scale` is defined as `w` of equation 2. of [Imagen Paper](https://arxiv.org/pdf/2205.11487.pdf).
+    Guidance scale is enabled by setting `guidance_scale > 1`. Higher guidance scale encourages to generate
+    images that are closely linked to the text `prompt`, usually at the expense of lower image quality.
+    """
+    extra: Optional[ExtraConditioningInfo] = None
+    scheduler_args: dict[str, Any] = field(default_factory=dict)
+    """
+    Additional arguments to pass to invokeai_diffuser.do_latent_postprocessing().
+    """
+    postprocessing_settings: Optional[PostprocessingSettings] = None
+
+    @property
+    def dtype(self):
+        return self.text_embeddings.dtype
+
+    def add_scheduler_args_if_applicable(self, scheduler, **kwargs):
+        scheduler_args = dict(self.scheduler_args)
+        step_method = inspect.signature(scheduler.step)
+        for name, value in kwargs.items():
+            try:
+                step_method.bind_partial(**{name: value})
+            except TypeError:
+                # FIXME: don't silently discard arguments
+                pass  # debug("%s does not accept argument named %r", scheduler, name)
+            else:
+                scheduler_args[name] = value
+        return dataclasses.replace(self, scheduler_args=scheduler_args)

invokeai/backend/stable_diffusion/diffusion/shared_invokeai_diffusion.py

@@ -1,8 +1,7 @@
 from __future__ import annotations
 
-from contextlib import contextmanager
-from dataclasses import dataclass
 import math
+from contextlib import contextmanager
 from typing import Any, Callable, Optional, Union
 
 import torch
@@ -10,9 +9,13 @@ from diffusers import UNet2DConditionModel
 from typing_extensions import TypeAlias
 
 from invokeai.app.services.config import InvokeAIAppConfig
+from invokeai.backend.stable_diffusion.diffusion.conditioning_data import (
+    ExtraConditioningInfo,
+    PostprocessingSettings,
+    SDXLConditioningInfo,
+)
 
 from .cross_attention_control import (
-    Arguments,
     Context,
     CrossAttentionType,
     SwapCrossAttnContext,
@@ -31,37 +34,6 @@ ModelForwardCallback: TypeAlias = Union[
 ]
 
 
-@dataclass
-class BasicConditioningInfo:
-    embeds: torch.Tensor
-    extra_conditioning: Optional[InvokeAIDiffuserComponent.ExtraConditioningInfo]
-    # weight: float
-    # mode: ConditioningAlgo
-
-    def to(self, device, dtype=None):
-        self.embeds = self.embeds.to(device=device, dtype=dtype)
-        return self
-
-
-@dataclass
-class SDXLConditioningInfo(BasicConditioningInfo):
-    pooled_embeds: torch.Tensor
-    add_time_ids: torch.Tensor
-
-    def to(self, device, dtype=None):
-        self.pooled_embeds = self.pooled_embeds.to(device=device, dtype=dtype)
-        self.add_time_ids = self.add_time_ids.to(device=device, dtype=dtype)
-        return super().to(device=device, dtype=dtype)
-
-
-@dataclass(frozen=True)
-class PostprocessingSettings:
-    threshold: float
-    warmup: float
-    h_symmetry_time_pct: Optional[float]
-    v_symmetry_time_pct: Optional[float]
-
-
 class InvokeAIDiffuserComponent:
     """
     The aim of this component is to provide a single place for code that can be applied identically to
@@ -75,15 +47,6 @@ class InvokeAIDiffuserComponent:
     debug_thresholding = False
     sequential_guidance = False
 
-    @dataclass
-    class ExtraConditioningInfo:
-        tokens_count_including_eos_bos: int
-        cross_attention_control_args: Optional[Arguments] = None
-
-        @property
-        def wants_cross_attention_control(self):
-            return self.cross_attention_control_args is not None
-
     def __init__(
         self,
         model,