Upload z-patch_prompt-fusion-extension using SD-Hub

z-patch_prompt-fusion-extension/.gitignore

@@ -0,0 +1,3 @@
+/venv/
+/.idea/
+__pycache__/

z-patch_prompt-fusion-extension/LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2003
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

z-patch_prompt-fusion-extension/lib_prompt_fusion/ast_nodes.py

@@ -0,0 +1,307 @@
+import math
+from lib_prompt_fusion import interpolation_functions
+from lib_prompt_fusion.t_scaler import scale_t
+from lib_prompt_fusion import interpolation_tensor
+
+
+class ListExpression:
+    def __init__(self, expressions):
+        self.__expressions = expressions
+
+    def extend_tensor(self, tensor_builder, steps_range, total_steps, context, is_hires, use_old_scheduling):
+        if not self.__expressions:
+            return
+
+        def expr_extend_tensor(expr):
+            expr.extend_tensor(tensor_builder, steps_range, total_steps, context, is_hires, use_old_scheduling)
+
+        expr_extend_tensor(self.__expressions[0])
+        for expression in self.__expressions[1:]:
+            tensor_builder.append(' ')
+            expr_extend_tensor(expression)
+
+
+class InterpolationExpression:
+    @staticmethod
+    def create(exprs, steps, function_name):
+        if function_name == "mean":
+            return AverageExpression(exprs, steps)
+
+        max_len = min(len(exprs), len(steps))
+        exprs = exprs[:max_len]
+        steps = steps[:max_len]
+
+        return InterpolationExpression(exprs, steps, function_name)
+
+    def __init__(self, expressions, steps, function_name=None):
+        assert len(expressions) >= 2
+        assert len(steps) == len(expressions), 'the number of steps must be the same as the number of expressions'
+        self.__expressions = expressions
+        self.__steps = steps
+        self.__function_name = function_name if function_name is not None else 'linear'
+
+    def extend_tensor(self, tensor_builder, steps_range, total_steps, context, is_hires, use_old_scheduling):
+        def tensor_updater(expr):
+            return lambda t: expr.extend_tensor(t, steps_range, total_steps, context, is_hires, use_old_scheduling)
+
+        tensor_builder.extrude(
+            [tensor_updater(expr) for expr in self.__expressions],
+            self.get_interpolation_function(steps_range, total_steps, context, is_hires, use_old_scheduling))
+
+    def get_interpolation_function(self, steps_range, total_steps, context, is_hires, use_old_scheduling):
+        steps = list(self.__steps)
+        if steps[0] is None:
+            steps[0] = LiftExpression(str(steps_range[0] - 1))
+        if steps[-1] is None:
+            steps[-1] = LiftExpression(str(steps_range[1] - 1))
+
+        for i, step in enumerate(steps):
+            if step is None:
+                continue
+
+            step = _eval_int_or_float(step, steps_range, total_steps, context, is_hires, use_old_scheduling)
+
+            if use_old_scheduling and 0 < step < 1:
+                step *= total_steps
+            elif not use_old_scheduling and isinstance(step, float):
+                step = (step - int(is_hires)) * total_steps
+            else:
+                step += 1
+
+            steps[i] = int(step)
+
+        i = 1
+        while i < len(steps):
+            none_len = 0
+            while steps[i + none_len] is None:
+                none_len += 1
+
+            min_step, max_step = steps[i - 1], steps[i + none_len]
+
+            for j in range(none_len):
+                steps[i + j] = min_step + (max_step - min_step) * (j + 1) / (none_len + 1)
+
+            i += 1 + none_len
+
+        interpolation_function = {
+            'linear': interpolation_functions.compute_linear,
+            'bezier': interpolation_functions.compute_bezier,
+            'catmull': interpolation_functions.compute_catmull,
+        }[self.__function_name]
+
+        def steps_scale_t(conds, params: interpolation_tensor.InterpolationParams):
+            scaled_t = (params.t * total_steps - steps[0]) / max(1, steps[-1] - steps[0])
+            scaled_t = scale_t(scaled_t, steps)
+
+            new_params = interpolation_tensor.InterpolationParams(scaled_t, *params[1:])
+            return interpolation_function(conds, new_params)
+
+        return steps_scale_t
+
+
+class AverageExpression:
+    def __init__(self, expressions, weights):
+        if len(expressions) < len(weights):
+            raise ValueError
+
+        self.__expressions = expressions
+        self.__weights = weights
+
+    def extend_tensor(self, tensor_builder, steps_range, total_steps, context, is_hires, use_old_scheduling):
+        def tensor_updater(expr):
+            return lambda t: expr.extend_tensor(t, steps_range, total_steps, context, is_hires, use_old_scheduling)
+
+        tensor_builder.extrude(
+            [tensor_updater(expr) for expr in self.__expressions],
+            self.get_interpolation_function(steps_range, total_steps, context, is_hires, use_old_scheduling))
+
+    def get_interpolation_function(self, steps_range, total_steps, context, is_hires, use_old_scheduling):
+        weights = [
+            _eval_int_or_float(weight, steps_range, total_steps, context, is_hires, use_old_scheduling) if weight is not None else None
+            for weight in self.__weights
+        ]
+        explicit_weights = [weight for weight in weights if weight is not None]
+        weights = [
+            weight / sum(explicit_weights) * len(explicit_weights) / len(self.__expressions)
+            if weight is not None
+            else 1 / len(self.__expressions)
+            for weight in weights
+        ]
+        weights.extend(1 / len(self.__expressions) for _ in range(len(self.__expressions) - len(weights)))
+
+        def interpolation_function(conds, _params):
+            total = None
+            for cond, weight in zip(conds, weights):
+                cond *= weight
+                if total is None:
+                    total = cond
+                else:
+                    total += cond
+
+            return total
+
+        return interpolation_function
+
+
+class AlternationExpression:
+    def __init__(self, expressions, speed):
+        self.__expressions = expressions
+        self.__speed = speed
+
+    def extend_tensor(self, tensor_builder, steps_range, total_steps, context, is_hires, use_old_scheduling):
+        if self.__speed is None:
+            speed = None
+        else:
+            speed = _eval_int_or_float(self.__speed, steps_range, total_steps, context, is_hires, use_old_scheduling)
+
+        if speed is None:
+            tensor_builder.append('[')
+            for expr_i, expr in enumerate(self.__expressions):
+                if expr_i >= 1:
+                    tensor_builder.append('|')
+                expr.extend_tensor(tensor_builder, steps_range, total_steps, context, is_hires, use_old_scheduling)
+            tensor_builder.append(']')
+            return
+
+        def tensor_updater(expr):
+            return lambda t: expr.extend_tensor(t, steps_range, total_steps, context, is_hires, use_old_scheduling)
+
+        exprs = self.__expressions + [self.__expressions[0]]
+
+        tensor_builder.extrude(
+            [tensor_updater(expr) for expr in exprs],
+            self.get_interpolation_function(speed, exprs, steps_range, total_steps))
+
+    def get_interpolation_function(self, speed, exprs, steps_range, total_steps):
+        def compute_wrap(control_points, params: interpolation_tensor.InterpolationParams):
+            wrapped_t = math.fmod((params.t * total_steps - steps_range[0]) / (len(exprs) - 1) * speed, 1.0)
+            if wrapped_t < 0:
+                wrapped_t = wrapped_t + 1
+            new_params = interpolation_tensor.InterpolationParams(wrapped_t, *params[1:])
+            return interpolation_functions.compute_linear(control_points, new_params)
+
+        return compute_wrap
+
+
+class EditingExpression:
+    def __init__(self, expressions, step):
+        assert 1 <= len(expressions) <= 2
+        self.__expressions = expressions
+        self.__step = step
+
+    def extend_tensor(self, tensor_builder, steps_range, total_steps, context, is_hires, use_old_scheduling):
+        if self.__step is None:
+            tensor_builder.append('[')
+            for expr_i, expr in enumerate(self.__expressions):
+                expr.extend_tensor(tensor_builder, steps_range, total_steps, context, is_hires, use_old_scheduling)
+                tensor_builder.append(':')
+            tensor_builder.append(']')
+            return
+
+        step = _eval_int_or_float(self.__step, steps_range, total_steps, context, is_hires, use_old_scheduling)
+        step_int = step
+        if use_old_scheduling and 0 < step < 1:
+            step_int *= total_steps
+        elif not use_old_scheduling and isinstance(step, float):
+            step_int = (step_int - int(is_hires)) * total_steps
+        else:
+            step_int += 1
+
+        step_int = int(step_int)
+
+        tensor_builder.append('[')
+        for expr_i, expr in enumerate(self.__expressions):
+            expr_steps_range = (steps_range[0], step_int) if expr_i == 0 and len(self.__expressions) >= 2 else (step_int, steps_range[1])
+            expr.extend_tensor(tensor_builder, expr_steps_range, total_steps, context, is_hires, use_old_scheduling)
+            tensor_builder.append(':')
+
+        tensor_builder.append(f'{step}]')
+
+
+class WeightedExpression:
+    def __init__(self, nested, weight=None, positive=True):
+        self.__nested = nested
+        if not positive:
+            assert weight is None
+
+        self.__weight = weight
+        self.__positive = positive
+
+    def extend_tensor(self, tensor_builder, steps_range, total_steps, context, is_hires, use_old_scheduling):
+        open_bracket, close_bracket = ('(', ')') if self.__positive else ('[', ']')
+        tensor_builder.append(open_bracket)
+        self.__nested.extend_tensor(tensor_builder, steps_range, total_steps, context, is_hires, use_old_scheduling)
+
+        if self.__weight is not None:
+            tensor_builder.append(':')
+            self.__weight.extend_tensor(tensor_builder, steps_range, total_steps, context, is_hires, use_old_scheduling)
+
+        tensor_builder.append(close_bracket)
+
+
+class WeightInterpolationExpression:
+    def __init__(self, nested, weight_begin, weight_end):
+        self.__nested = nested
+        self.__weight_begin = weight_begin if weight_begin is not None else LiftExpression(str(1.))
+        self.__weight_end = weight_end if weight_end is not None else LiftExpression(str(1.))
+
+    def extend_tensor(self, tensor_builder, steps_range, total_steps, context, is_hires, use_old_scheduling):
+        steps_range_size = steps_range[1] - steps_range[0]
+
+        weight_begin = _eval_int_or_float(self.__weight_begin, steps_range, total_steps, context, is_hires, use_old_scheduling)
+        weight_end = _eval_int_or_float(self.__weight_end, steps_range, total_steps, context, is_hires, use_old_scheduling)
+
+        for i in range(steps_range_size):
+            step = i + steps_range[0]
+
+            weight = weight_begin + (weight_end - weight_begin) * (i / max(steps_range_size - 1, 1))
+            weight_step_expr = WeightedExpression(self.__nested, LiftExpression(str(weight)))
+            if step > steps_range[0]:
+                weight_step_expr = EditingExpression([weight_step_expr], LiftExpression(str(step - 1)))
+            if step + 1 < steps_range[1]:
+                weight_step_expr = EditingExpression([weight_step_expr, ListExpression([])], LiftExpression(str(step)))
+
+            weight_step_expr.extend_tensor(tensor_builder, steps_range, total_steps, context, is_hires, use_old_scheduling)
+
+
+class DeclarationExpression:
+    def __init__(self, symbol, parameters, value, target):
+        self.__symbol = symbol
+        self.__value = value
+        self.__target = target
+        self.__parameters = parameters
+
+    def extend_tensor(self, tensor_builder, steps_range, total_steps, context, is_hires, use_old_scheduling):
+        updated_context = dict(context)
+        updated_context[self.__symbol] = (self.__value, self.__parameters)
+        self.__target.extend_tensor(tensor_builder, steps_range, total_steps, updated_context, is_hires, use_old_scheduling)
+
+
+class SubstitutionExpression:
+    def __init__(self, symbol, arguments):
+        self.__symbol = symbol
+        self.__arguments = arguments
+
+    def extend_tensor(self, tensor_builder, steps_range, total_steps, context, is_hires, use_old_scheduling):
+        updated_context = dict(context)
+        nested, parameters = context[self.__symbol]
+        for argument, parameter in zip(self.__arguments, parameters):
+            updated_context[parameter] = argument, []
+        nested.extend_tensor(tensor_builder, steps_range, total_steps, updated_context, is_hires, use_old_scheduling)
+
+
+class LiftExpression:
+    def __init__(self, value):
+        self.__value = value
+
+    def extend_tensor(self, tensor_builder, *_args, **_kwargs):
+        tensor_builder.append(self.__value)
+
+
+def _eval_int_or_float(expression, steps_range, total_steps, context, is_hires, use_old_scheduling):
+    mock_database = ['']
+    expression.extend_tensor(interpolation_tensor.InterpolationTensorBuilder(prompt_database=mock_database), steps_range, total_steps, context, is_hires, use_old_scheduling)
+    try:
+        return int(mock_database[0])
+    except ValueError:
+        return float(mock_database[0])

z-patch_prompt-fusion-extension/lib_prompt_fusion/empty_cond.py

@@ -0,0 +1,19 @@
+from lib_prompt_fusion import interpolation_tensor
+
+
+_empty_cond = None
+
+
+def get():
+    return _empty_cond
+
+
+def init(model):
+    global _empty_cond
+    cond = model.get_learned_conditioning([''])
+    if isinstance(cond, dict):
+        cond = interpolation_tensor.DictCondWrapper({k: v[0] for k, v in cond.items()})
+    else:
+        cond = interpolation_tensor.TensorCondWrapper(cond[0])
+
+    _empty_cond = cond

z-patch_prompt-fusion-extension/lib_prompt_fusion/geometries.py

@@ -0,0 +1,33 @@
+import math
+import torch
+from lib_prompt_fusion import interpolation_tensor
+
+
+def slerp_geometry(control_points, params: interpolation_tensor.InterpolationParams):
+    p0, p1 = control_points
+    p0_norm = torch.linalg.norm(p0)
+    p1_norm = torch.linalg.norm(p1)
+
+    similarity = torch.sum((p0 / p0_norm) * (p1 / p1_norm))
+    similarity = min(1., max(-1., float(similarity)))
+    if similarity <= params.slerp_epsilon - 1 or similarity >= 1 - params.slerp_epsilon:
+        return linear_geometry(control_points, params)
+
+    angle = math.acos(float(similarity)) / 2
+
+    slerp_t = angle * (2 * params.t - 1)
+    slerp_t = math.tan(slerp_t) / math.tan(angle)
+    slerp_t = (slerp_t + 1) / 2
+
+    normalized_p1 = p1 / p1_norm * p0_norm
+    slerp_p = p0 + (normalized_p1 - p0) * slerp_t
+    slerp_p = slerp_p / torch.linalg.norm(slerp_p) * (p0_norm + (p1_norm - p0_norm) * params.t)
+
+    lerp_p = linear_geometry(control_points, params)
+    return lerp_p + (slerp_p - lerp_p) * params.slerp_scale
+
+
+def linear_geometry(control_points, params: interpolation_tensor.InterpolationParams):
+    p0, p1 = control_points
+    res = p0 + (p1 - p0) * params.t
+    return res

z-patch_prompt-fusion-extension/lib_prompt_fusion/global_state.py

@@ -0,0 +1,45 @@
+from typing import List, Optional, Literal
+import os
+from modules import shared, prompt_parser
+from lib_prompt_fusion import empty_cond
+
+
+old_webui_is_negative: bool = False
+negative_schedules: Optional[List[prompt_parser.ScheduledPromptConditioning]] = None
+negative_schedules_hires: Optional[List[prompt_parser.ScheduledPromptConditioning]] = None
+
+
+def get_origin_cond_at(step: int, is_hires: bool = False):
+    fallback_schedules = negative_schedules_hires if is_hires else negative_schedules
+    if not fallback_schedules or not shared.opts.data.get('prompt_fusion_slerp_negative_origin', False):
+        return empty_cond.get()
+
+    for schedule in fallback_schedules:
+        if schedule.end_at_step >= step:
+            return schedule.cond
+
+    return empty_cond.get()
+
+
+def get_slerp_scale():
+    return shared.opts.data.get('prompt_fusion_slerp_scale', 0.0)
+
+
+def get_slerp_epsilon():
+    return shared.opts.data.get('prompt_fusion_slerp_epsilon', 0.0001)
+    
+def get_neg_origin_mode() -> Literal["first","linear","spherical"]:
+    """
+    Режим агрегирования origin для негативного prompt'а:
+      - first     : как сейчас (первая подпрампта)
+      - linear    : взвешенное линейное усреднение
+      - spherical : «сферическое» усреднение (норм., сумма, норм.)
+    Источник: сначала UI-настройка (если есть), потом ENV, по умолчанию 'first'.
+    """
+    try:
+        mode = (shared.opts.data.get("prompt_fusion_neg_origin_mode") or "first").lower()
+    except Exception:
+        mode = os.environ.get("PROMPT_FUSION_NEG_ORIGIN_MODE", "first").lower()
+    if mode not in ("first","linear","spherical"):
+        mode = "first"
+    return mode

z-patch_prompt-fusion-extension/lib_prompt_fusion/hijacker.py

@@ -0,0 +1,34 @@
+class ModuleHijacker:
+    def __init__(self, module):
+        self.__module = module
+        self.__original_functions = dict()
+
+    def hijack(self, attribute):
+        if attribute not in self.__original_functions:
+            self.__original_functions[attribute] = getattr(self.__module, attribute)
+
+        def decorator(function):
+            def wrapper(*args, **kwargs):
+                return function(*args, **kwargs, original_function=self.__original_functions[attribute])
+
+            setattr(self.__module, attribute, wrapper)
+            return function
+
+        return decorator
+
+    def reset_module(self):
+        for attribute, original_function in self.__original_functions.items():
+            setattr(self.__module, attribute, original_function)
+
+        self.__original_functions.clear()
+
+    @staticmethod
+    def install_or_get(module, hijacker_attribute, register_uninstall=lambda _callback: None):
+        if not hasattr(module, hijacker_attribute):
+            module_hijacker = ModuleHijacker(module)
+            setattr(module, hijacker_attribute, module_hijacker)
+            register_uninstall(lambda: delattr(module, hijacker_attribute))
+            register_uninstall(module_hijacker.reset_module)
+            return module_hijacker
+        else:
+            return getattr(module, hijacker_attribute)

z-patch_prompt-fusion-extension/lib_prompt_fusion/interpolation_functions.py

@@ -0,0 +1,87 @@
+import copy
+import math
+from lib_prompt_fusion import interpolation_tensor, geometries
+
+
+def compute_linear(control_points, params: interpolation_tensor.InterpolationParams):
+    if len(control_points) <= 2:
+        return geometries.slerp_geometry(control_points, params)
+    else:
+        target_curve = min(int(params.t * (len(control_points) - 1)), len(control_points) - 1)
+        cp0 = control_points[target_curve]
+        cp1 = control_points[target_curve + 1] if target_curve + 1 < len(control_points) else control_points[-1]
+        new_params = interpolation_tensor.InterpolationParams(math.fmod(params.t * (len(control_points) - 1), 1.), *params[1:])
+        return geometries.slerp_geometry([cp0, cp1], new_params)
+
+
+def compute_bezier(control_points, params: interpolation_tensor.InterpolationParams):
+    def compute_casteljau(ps, size):
+        for i in reversed(range(1, size)):
+            for j in range(i):
+                ps[j] = geometries.slerp_geometry([ps[j], ps[j+1]], params)
+
+        return ps[0]
+
+    if len(control_points) == 1:
+        return control_points[0]
+    elif len(control_points) == 2:
+        return geometries.slerp_geometry(control_points, params)
+    copied_control_points = copy.deepcopy(control_points)
+    return compute_casteljau(copied_control_points, len(copied_control_points))
+
+
+def compute_catmull(control_points, params: interpolation_tensor.InterpolationParams):
+    if len(control_points) <= 2:
+        return compute_linear(control_points, params)
+    else:
+        target_curve = min(int(params.t * (len(control_points) - 1)), len(control_points) - 1)
+        g0 = control_points[target_curve - 1] if target_curve > 0 else control_points[0] * 2 - control_points[1]
+        cp0 = control_points[target_curve]
+        cp1 = control_points[target_curve + 1] if target_curve + 1 < len(control_points) else control_points[-1]
+        g1 = control_points[target_curve + 2] if target_curve + 2 < len(control_points) else cp1 * 2 - cp0
+        ip0 = cp0 + (cp1 - g0)/6
+        ip1 = cp1 + (cp0 - g1)/6
+
+        new_params = interpolation_tensor.InterpolationParams(math.fmod(params.t * (len(control_points) - 1), 1.), *params[1:])
+        return compute_bezier([cp0, ip0, ip1, cp1], new_params)
+
+
+if __name__ == '__main__':
+    import turtle as tr
+    import torch
+    size = 60
+    turtle_tool = tr.Turtle()
+    turtle_tool.speed(10)
+    turtle_tool.up()
+
+    points = torch.Tensor([[-2., -2.], [2., 2.]])
+    origin = torch.Tensor([1.5, 1.6])
+
+    def sample(slerp_scale, color):
+        for i in range(size):
+            t = i / size
+            params = interpolation_tensor.InterpolationParams(t, i, size, slerp_scale, 0.0001)
+            point = origin + compute_linear(points - origin, params)
+            try:
+                turtle_tool.goto(tuple(float(p) * 100. for p in point))
+                turtle_tool.dot(5, color)
+                print(point)
+            except ValueError:
+                pass
+
+    sample(0, "black")
+    sample(1, "green")
+    sample(2, "blue")
+    sample(-1, "purple")
+    sample(-2, "orange")
+
+    for point in points:
+        turtle_tool.goto(tuple(float(p) * 100. for p in point))
+        turtle_tool.dot(5, "red")
+
+    turtle_tool.goto(tuple(float(p) * 100. for p in origin))
+    turtle_tool.dot(10, "red")
+
+    turtle_tool.goto(100000, 100000)
+    turtle_tool.dot()
+    tr.done()

z-patch_prompt-fusion-extension/lib_prompt_fusion/interpolation_tensor.py

@@ -0,0 +1,249 @@
+import dataclasses
+import torch
+from modules import prompt_parser
+from typing import NamedTuple, Union
+
+
+class InterpolationParams(NamedTuple):
+    t: float
+    step: int
+    total_steps: int
+    slerp_scale: float
+    slerp_epsilon: float
+
+
+class InterpolationTensor:
+    def __init__(self, sub_tensors, interpolation_function):
+        self.__sub_tensors = sub_tensors
+        self.__interpolation_function = interpolation_function
+
+    def interpolate(self, params: InterpolationParams, origin_cond, empty_cond):
+        cond = self.interpolate_cond_rec(params, origin_cond, empty_cond)
+        if params.slerp_scale != 0:
+            cond = (cond + origin_cond.extend_like(cond, empty_cond)).to(dtype=origin_cond.dtype)
+        return cond
+
+    def interpolate_cond_rec(self, params: InterpolationParams, origin_cond, empty_cond):
+        if self.__interpolation_function is None:
+            return self.get_cond_point(params.step, origin_cond, empty_cond, params.slerp_scale)
+
+        control_points = [
+            sub_tensor.interpolate_cond_rec(params, origin_cond, empty_cond)
+            for sub_tensor in self.__sub_tensors
+        ]
+
+        CondWrapper, control_points_values = conds_to_cp_values(control_points)
+        return CondWrapper.from_cp_values(self.__interpolation_function(control_points, params) for control_points in control_points_values)
+
+    def get_cond_point(self, step, origin_cond, empty_cond, slerp_scale):
+        schedule = None
+        for schedule in self.__sub_tensors:
+            if schedule.end_at_step >= step:
+                break
+
+        res = schedule.cond.extend_like(origin_cond, empty_cond)
+        if slerp_scale != 0:
+            res = res.to(dtype=torch.float) - origin_cond.extend_like(schedule.cond, empty_cond).to(dtype=torch.float)
+        return res
+
+
+def conds_to_cp_values(conds):
+    CondWrapper = type(conds[0])
+    cp_values = [
+        cond.to_cp_values()
+        for cond in conds
+    ]
+    return CondWrapper, [
+        [v[i] for v in cp_values]
+        for i in range(len(cp_values[0]))
+    ]
+
+
+class InterpolationTensorBuilder:
+    def __init__(self, tensor=None, prompt_database=None, interpolation_functions=None):
+        self.__indices_tensor = tensor if tensor is not None else 0
+        self.__prompt_database = prompt_database if prompt_database is not None else ['']
+        self.__interpolation_functions = interpolation_functions if interpolation_functions is not None else []
+
+    def append(self, suffix):
+        for i in range(len(self.__prompt_database)):
+            self.__prompt_database[i] += suffix
+
+    def extrude(self, tensor_updaters, interpolation_function):
+        extruded_indices_tensor = []
+        extruded_prompt_database = []
+        extruded_interpolation_functions = []
+
+        for update_tensor in tensor_updaters:
+            nested_tensor_builder = InterpolationTensorBuilder(
+                self.__indices_tensor,
+                self.__prompt_database[:],
+                interpolation_functions=[])
+
+            update_tensor(nested_tensor_builder)
+
+            extruded_indices_tensor.append(InterpolationTensorBuilder.__offset_tensor(
+                tensor=nested_tensor_builder.__indices_tensor,
+                offset=len(extruded_prompt_database)))
+            extruded_prompt_database.extend(nested_tensor_builder.__prompt_database)
+            extruded_interpolation_functions.append(nested_tensor_builder.__interpolation_functions)
+
+        self.__indices_tensor = extruded_indices_tensor
+        self.__prompt_database[:] = extruded_prompt_database
+        self.__interpolation_functions.insert(0, (interpolation_function, extruded_interpolation_functions))
+
+    def get_prompt_database(self):
+        return self.__prompt_database
+
+    @staticmethod
+    def __offset_tensor(tensor, offset):
+        try:
+            return tensor + offset
+
+        except TypeError:
+            return [InterpolationTensorBuilder.__offset_tensor(e, offset) for e in tensor]
+
+    def build(self, conds, empty_cond):
+        max_cond_size = self.__max_cond_size(conds)
+        conds = self.__resize_uniformly(conds, max_cond_size, empty_cond)
+        return InterpolationTensorBuilder.__build_conditionings_tensor(self.__indices_tensor, self.__interpolation_functions, conds)
+
+    @staticmethod
+    def __build_conditionings_tensor(tensor, int_funcs, conds):
+        if type(tensor) is int:
+            return InterpolationTensor(conds[tensor], None)
+        else:
+            int_func, nested_int_funcs = int_funcs[0]
+            return InterpolationTensor(
+                [
+                    InterpolationTensorBuilder.__build_conditionings_tensor(sub_tensor, nested_int_funcs + int_funcs[1:], conds)
+                    for sub_tensor, nested_int_funcs in zip(tensor, nested_int_funcs)
+                ],
+                int_func,
+            )
+
+    def __resize_uniformly(self, conds, max_cond_size: int, empty_cond):
+        return [
+            [
+                prompt_parser.ScheduledPromptConditioning(
+                    cond=schedule.cond.resize_schedule(max_cond_size, empty_cond),
+                    end_at_step=schedule.end_at_step
+                )
+                for schedule in schedules
+            ]
+            for schedules in conds
+        ]
+
+    @staticmethod
+    def __max_cond_size(conds):
+        return max(schedule.cond.size(0)
+                   for schedules in conds
+                   for schedule in schedules)
+
+
+@dataclasses.dataclass
+class DictCondWrapper:
+    original_cond: dict
+
+    @staticmethod
+    def from_cp_values(cp_values):
+        return DictCondWrapper({
+            k: v
+            for k, v in zip(('crossattn', 'vector'), cp_values)
+        })
+
+    def size(self, *args, **kwargs):
+        return self.original_cond['crossattn'].size(*args, **kwargs)
+
+    def extend_like(self, that, empty):
+        missing_size = max(0, that.size(0) - self.size(0)) // 77
+        extended = DictCondWrapper(self.original_cond.copy())
+        extended.original_cond['crossattn'] = torch.concatenate([self.original_cond['crossattn']] + [empty.original_cond['crossattn']] * missing_size)
+        return extended
+
+    def resize_schedule(self, target_size, empty_cond):
+        cond_missing_size = (target_size - self.size(0)) // 77
+        if cond_missing_size <= 0:
+            return self
+
+        resized_cond = self.original_cond.copy()
+        resized_cond['crossattn'] = torch.concatenate([self.original_cond['crossattn']] + [empty_cond.original_cond['crossattn']] * cond_missing_size)
+        return DictCondWrapper(resized_cond)
+
+    def to_cp_values(self):
+        return list(self.original_cond.values())
+
+    def to(self, dtype: Union[dict, torch.dtype]):
+        if not isinstance(dtype, dict):
+            dtype = {
+                k: dtype
+                for k in self.original_cond.keys()
+            }
+        return DictCondWrapper({
+            k: v.to(dtype=dtype[k])
+            for k, v in self.original_cond.items()
+        })
+
+    @property
+    def dtype(self):
+        return {
+            k: v.dtype
+            for k, v in self.original_cond.items()
+        }
+
+    def __sub__(self, that):
+        return DictCondWrapper({
+            k: v - that.original_cond[k]
+            for k, v in self.original_cond.items()
+        })
+
+    def __add__(self, that):
+        return DictCondWrapper({
+            k: v + that.original_cond[k]
+            for k, v in self.original_cond.items()
+        })
+
+    def __eq__(self, that):
+        return all((self.original_cond[k] == that.original_cond[k]).all() for k in self.original_cond.keys())
+
+
+@dataclasses.dataclass
+class TensorCondWrapper:
+    original_cond: torch.Tensor
+
+    @staticmethod
+    def from_cp_values(cp_values):
+        return TensorCondWrapper(next(iter(cp_values)))
+
+    def size(self, *args, **kwargs):
+        return self.original_cond.size(*args, **kwargs)
+
+    def extend_like(self, that, empty):
+        missing_size = max(0, that.size(0) - self.original_cond.size(0)) // 77
+        return TensorCondWrapper(torch.concatenate([self.original_cond] + [empty.original_cond] * missing_size))
+
+    def resize_schedule(self, target_size, empty_cond):
+        cond_missing_size = (target_size - self.original_cond.size(0)) // 77
+        if cond_missing_size <= 0:
+            return self
+
+        return TensorCondWrapper(torch.concatenate([self.original_cond] + [empty_cond.original_cond] * cond_missing_size))
+
+    def to_cp_values(self):
+        return [self.original_cond]
+
+    def to(self, dtype: torch.dtype):
+        return TensorCondWrapper(self.original_cond.to(dtype=dtype))
+
+    @property
+    def dtype(self):
+        return self.original_cond.dtype
+
+    def __sub__(self, that):
+        return TensorCondWrapper(self.original_cond - that.original_cond)
+
+    def __add__(self, that):
+        return TensorCondWrapper(self.original_cond + that.original_cond)
+
+    def __eq__(self, that):
+        return (self.original_cond == that.original_cond).all()

z-patch_prompt-fusion-extension/lib_prompt_fusion/prompt_parser.py

@@ -0,0 +1,378 @@
+import lib_prompt_fusion.ast_nodes as ast
+from collections import namedtuple
+import re
+
+
+ParseResult = namedtuple('ParseResult', ['prompt', 'expr'])
+
+
+def parse_prompt(prompt):
+    prompt = prompt.lstrip()
+    prompt, list_expr = parse_list_expression(prompt, set())
+    return list_expr
+
+
+def parse_list_expression(prompt, stoppers):
+    exprs = []
+    try:
+        while True:
+            prompt, expr = parse_expression(prompt, stoppers)
+            exprs.append(expr)
+    except ValueError:
+        return ParseResult(prompt=prompt, expr=ast.ListExpression(exprs))
+
+
+def parse_expression(prompt, stoppers):
+    for parse in _parsers():
+        try:
+            return parse(prompt, stoppers)
+        except ValueError:
+            pass
+
+    raise ValueError
+
+
+def _parsers():
+    return (
+        parse_text,
+        parse_declaration,
+        parse_substitution,
+        parse_positive_attention,
+        parse_negative_attention,
+        parse_editing,
+        parse_alternation,
+        parse_interpolation,
+        parse_unrestricted_text,
+    )
+
+
+def parse_text(prompt, stoppers):
+    return parse_unrestricted_text(prompt, set_concat(stoppers, {'[', '(', '$'}))
+
+
+def parse_unrestricted_text(prompt, stoppers):
+    escaped_stoppers = ''.join(re.escape(stopper) for stopper in stoppers)
+    regex = rf'(?:[^{escaped_stoppers}\\\s]|\$(?![a-zA-Z_])|\\.)+'
+    prompt, expr = parse_token(prompt, whitespace_tail_regex(regex, stoppers))
+    return ParseResult(prompt=prompt, expr=ast.LiftExpression(expr))
+
+
+def parse_substitution(prompt, stoppers):
+    prompt, symbol = parse_symbol(prompt, stoppers)
+    prompt, arguments = parse_arguments(prompt, stoppers)
+    return ParseResult(prompt=prompt, expr=ast.SubstitutionExpression(symbol, arguments))
+
+
+def parse_arguments(prompt, stoppers):
+    try:
+        prompt, _ = parse_open_paren(prompt, stoppers)
+        prompt, arguments = parse_inner_arguments(prompt, stoppers)
+        prompt, _ = parse_close_paren(prompt, stoppers)
+    except ValueError:
+        arguments = []
+    return ParseResult(prompt=prompt, expr=arguments)
+
+
+def parse_inner_arguments(prompt, stoppers):
+    arguments = []
+    try:
+        while True:
+            prompt, arg = parse_list_expression(prompt, {',', ')'})
+            arguments.append(arg)
+            prompt, _ = parse_comma(prompt, stoppers)
+    except ValueError:
+        pass
+
+    return ParseResult(prompt=prompt, expr=arguments)
+
+
+def parse_declaration(prompt, stoppers):
+    prompt, symbol = parse_symbol(prompt, stoppers)
+    prompt, parameters = parse_parameters(prompt, stoppers)
+    prompt, _ = parse_equals(prompt, stoppers)
+    prompt, value = parse_list_expression(prompt, set_concat(stoppers, '\n'))
+    prompt, _ = parse_newline(prompt, stoppers)
+    prompt, expr = parse_list_expression(prompt, stoppers)
+    return ParseResult(prompt=prompt, expr=ast.DeclarationExpression(symbol, parameters, value, expr))
+
+
+def parse_parameters(prompt, stoppers):
+    try:
+        prompt, _ = parse_open_paren(prompt, stoppers)
+        prompt, parameters = parse_inner_parameters(prompt, stoppers)
+        prompt, _ = parse_close_paren(prompt, stoppers)
+    except ValueError:
+        parameters = []
+    return ParseResult(prompt=prompt, expr=parameters)
+
+
+def parse_inner_parameters(prompt, stoppers):
+    parameters = []
+    try:
+        while True:
+            prompt, param = parse_symbol(prompt, stoppers)
+            parameters.append(param)
+            prompt, _ = parse_comma(prompt, stoppers)
+    except ValueError:
+        pass
+
+    return ParseResult(prompt=prompt, expr=parameters)
+
+
+def parse_interpolation(prompt, stoppers):
+    prompt, _ = parse_open_square(prompt, stoppers)
+    prompt, exprs = parse_interpolation_exprs(prompt, stoppers)
+    prompt, steps = parse_interpolation_steps(prompt, stoppers)
+    prompt, function_name = parse_interpolation_function_name(prompt, stoppers)
+    prompt, _ = parse_close_square(prompt, stoppers)
+    return ParseResult(prompt=prompt, expr=ast.InterpolationExpression.create(exprs, steps, function_name))
+
+
+def parse_interpolation_exprs(prompt, stoppers):
+    exprs = []
+
+    try:
+        while True:
+            prompt_tmp, expr = parse_list_expression(prompt, {':', ']'})
+            if parse_interpolation_function_name(prompt_tmp, stoppers).expr is not None:
+                raise ValueError
+
+            prompt, _ = parse_colon(prompt_tmp, stoppers)
+            exprs.append(expr)
+    except ValueError:
+        pass
+
+    return ParseResult(prompt=prompt, expr=exprs)
+
+
+def parse_interpolation_function_name(prompt, stoppers):
+    try:
+        prompt, _ = parse_colon(prompt, stoppers)
+        function_names = ('linear', 'catmull', 'bezier', 'mean')
+        return parse_token(prompt, whitespace_tail_regex('|'.join(function_names), stoppers))
+    except ValueError:
+        return ParseResult(prompt=prompt, expr=None)
+
+
+def parse_interpolation_steps(prompt, stoppers):
+    steps = []
+
+    try:
+        while True:
+            prompt, step = parse_interpolation_step(prompt, stoppers)
+            steps.append(step)
+            prompt, _ = parse_comma(prompt, stoppers)
+    except ValueError:
+        pass
+
+    return ParseResult(prompt=prompt, expr=steps)
+
+
+def parse_interpolation_step(prompt, stoppers):
+    try:
+        return parse_step(prompt, stoppers)
+    except ValueError:
+        pass
+
+    if prompt[0] in {',', ':', ']'}:
+        return ParseResult(prompt=prompt, expr=None)
+
+    raise ValueError
+
+
+def parse_alternation(prompt, stoppers):
+    prompt, _ = parse_open_square(prompt, stoppers)
+    prompt, exprs = parse_alternation_exprs(prompt, stoppers)
+    prompt, speed = parse_alternation_speed(prompt, stoppers)
+    prompt, _ = parse_close_square(prompt, stoppers)
+    return ParseResult(prompt=prompt, expr=ast.AlternationExpression(exprs, speed))
+
+
+def parse_alternation_exprs(prompt, stoppers):
+    exprs = []
+
+    try:
+        while True:
+            prompt, expr = parse_list_expression(prompt, {'|', ':', ']'})
+            exprs.append(expr)
+            prompt, _ = parse_vertical_bar(prompt, stoppers)
+    except ValueError:
+        if len(exprs) < 2:
+            raise
+
+    return ParseResult(prompt=prompt, expr=exprs)
+
+
+def parse_alternation_speed(prompt, stoppers):
+    try:
+        prompt, _ = parse_colon(prompt, stoppers)
+        prompt, speed = parse_step(prompt, stoppers)
+        return ParseResult(prompt=prompt, expr=speed)
+    except ValueError:
+        pass
+
+    return ParseResult(prompt=prompt, expr=None)
+
+
+def parse_editing(prompt, stoppers):
+    prompt, _ = parse_open_square(prompt, stoppers)
+    prompt, exprs = parse_editing_exprs(prompt, stoppers)
+    try:
+        prompt, step = parse_step(prompt, stoppers)
+    except ValueError:
+        step = None
+
+    prompt, _ = parse_close_square(prompt, stoppers)
+    return ParseResult(prompt=prompt, expr=ast.EditingExpression(exprs, step))
+
+
+def parse_editing_exprs(prompt, stoppers):
+    exprs = []
+
+    try:
+        for _ in range(2):
+            prompt_tmp, expr = parse_list_expression(prompt, {'|', ':', ']'})
+            prompt, _ = parse_colon(prompt_tmp, stoppers)
+            exprs.append(expr)
+    except ValueError:
+        pass
+
+    return ParseResult(prompt=prompt, expr=exprs)
+
+
+def parse_negative_attention(prompt, stoppers):
+    prompt, _ = parse_open_square(prompt, stoppers)
+    prompt, expr = parse_list_expression(prompt, set_concat(stoppers, {':', ']'}))
+    prompt, _ = parse_close_square(prompt, stoppers)
+    return ParseResult(prompt=prompt, expr=ast.WeightedExpression(expr, positive=False))
+
+
+def parse_positive_attention(prompt, stoppers):
+    prompt, _ = parse_open_paren(prompt, stoppers)
+    prompt, expr = parse_list_expression(prompt, {':', ')'})
+    prompt, weight_exprs = parse_attention_weights(prompt, stoppers)
+    prompt, _ = parse_close_paren(prompt, stoppers)
+    if len(weight_exprs) >= 2:
+        return ParseResult(prompt=prompt, expr=ast.WeightInterpolationExpression(expr, *weight_exprs[:2]))
+    else:
+        return ParseResult(prompt=prompt, expr=ast.WeightedExpression(expr, *weight_exprs[:1]))
+
+
+def parse_attention_weights(prompt, stoppers):
+    weights = []
+    try:
+        prompt, _ = parse_colon(prompt, stoppers)
+    except ValueError:
+        return ParseResult(prompt=prompt, expr=weights)
+
+    while True:
+        try:
+            prompt, weight_expr = parse_weight(prompt, stoppers)
+            weights.append(weight_expr)
+            prompt, _ = parse_comma(prompt, stoppers)
+        except ValueError:
+            return ParseResult(prompt=prompt, expr=weights)
+
+
+def parse_step(prompt, stoppers):
+    try:
+        prompt, step = parse_int_not_float(prompt, stoppers)
+        return ParseResult(prompt=prompt, expr=ast.LiftExpression(step))
+    except ValueError:
+        pass
+    try:
+        prompt, step = parse_float(prompt, stoppers)
+        return ParseResult(prompt=prompt, expr=ast.LiftExpression(step))
+    except ValueError:
+        pass
+
+    return parse_substitution(prompt, stoppers)
+
+
+def parse_weight(prompt, stoppers):
+    try:
+        prompt, step = parse_float(prompt, stoppers)
+        return ParseResult(prompt=prompt, expr=ast.LiftExpression(step))
+    except ValueError:
+        pass
+
+    return parse_substitution(prompt, stoppers)
+
+
+def parse_symbol(prompt, stoppers):
+    prompt, _ = parse_dollar(prompt)
+    return parse_symbol_text(prompt, stoppers)
+
+
+def parse_symbol_text(prompt, stoppers):
+    return parse_token(prompt, whitespace_tail_regex('[a-zA-Z_][a-zA-Z0-9_]*', stoppers))
+
+
+def parse_float(prompt, stoppers):
+    return parse_token(prompt, whitespace_tail_regex(r'[+-]?(?:\d+(?:\.\d*)?|\.\d+)', stoppers))
+
+
+def parse_int_not_float(prompt, stoppers):
+    return parse_token(prompt, whitespace_tail_regex(r'[+-]?\d+(?!\.)', stoppers))
+
+
+def parse_dollar(prompt):
+    dollar_sign = re.escape('$')
+    return parse_token(prompt, f'({dollar_sign})')
+
+
+def parse_equals(prompt, stoppers):
+    return parse_token(prompt, whitespace_tail_regex(re.escape('='), stoppers))
+
+
+def parse_comma(prompt, stoppers):
+    return parse_token(prompt, whitespace_tail_regex(re.escape(','), stoppers))
+
+
+def parse_colon(prompt, stoppers):
+    return parse_token(prompt, whitespace_tail_regex(re.escape(':'), stoppers))
+
+
+def parse_vertical_bar(prompt, stoppers):
+    return parse_token(prompt, whitespace_tail_regex(re.escape('|'), stoppers))
+
+
+def parse_open_square(prompt, stoppers):
+    return parse_token(prompt, whitespace_tail_regex(re.escape('['), stoppers))
+
+
+def parse_close_square(prompt, stoppers):
+    return parse_token(prompt, whitespace_tail_regex(re.escape(']'), stoppers))
+
+
+def parse_open_paren(prompt, stoppers):
+    return parse_token(prompt, whitespace_tail_regex(re.escape('('), stoppers))
+
+
+def parse_close_paren(prompt, stoppers):
+    return parse_token(prompt, whitespace_tail_regex(re.escape(')'), stoppers))
+
+
+def parse_newline(prompt, stoppers):
+    return parse_token(prompt, whitespace_tail_regex('\n|$', stoppers))
+
+
+def parse_token(prompt, regex):
+    match = re.match(regex, prompt)
+    if match is None:
+        raise ValueError
+
+    return ParseResult(prompt=prompt[len(match.group()):], expr=match.groups()[-1])
+
+
+def whitespace_tail_regex(regex, stoppers):
+    if '\n' in stoppers:
+        return rf'({regex})[ \t\f\r]*'
+
+    return rf'({regex})\s*'
+
+
+def set_concat(left, right):
+    result = set(left)
+    result.update(right)
+    return result

z-patch_prompt-fusion-extension/lib_prompt_fusion/t_scaler.py

@@ -0,0 +1,38 @@
+def scale_t(t, positions):
+    if t >= 1.:
+        return 1.
+
+    if t <= 0.:
+        return 0.
+
+    distances = []
+    for i in range(len(positions)-1):
+        distances.append(positions[i+1] - positions[i])
+
+    total_distance = sum(distances)
+    for i in range(len(distances)):
+        distances[i] = distances[i]/total_distance
+
+    for i in range(len(distances)-1):
+        distances[i+1] = distances[i] + distances[i+1]
+
+    distances.insert(0, 0.0)
+
+    spline_index = 0
+    for i, distance in enumerate(distances):
+        if t > distance:
+            spline_index = i
+        else:
+            break
+
+    if spline_index >= len(distances) - 1:
+        return 1
+
+    local_ratio = (t - distances[spline_index]) / (distances[spline_index+1] - distances[spline_index])
+    return (spline_index + local_ratio)/(len(distances)-1)
+
+
+if __name__ == "__main__":
+    total_steps = 20
+    for i in range(total_steps):
+        print(i, scale_t(i/total_steps, [9, 10]))

z-patch_prompt-fusion-extension/metadata.ini

@@ -0,0 +1,2 @@
+[Extension]
+Name = prompt-fusion

z-patch_prompt-fusion-extension/readme.md

@@ -0,0 +1,95 @@
+# Prompt Fusion
+
+Prompt Fusion is an [auto1111 webui extension](https://github.com/AUTOMATIC1111/stable-diffusion-webui/wiki/Developing-extensions) that adds more possibilities to the native prompt syntax. Among other additions, it allows to interpolate between the embeddings of different prompts, continuously:
+
+```
+# linear prompt interpolation
+[night light:magical forest: 5, 15]
+
+# catmull-rom curve prompt interpolation
+[night light:magical forest:norvegian territory: 5, 15, 25:catmull]
+
+# alternation interpolation
+[ufo|a strange sight:0.5]
+
+# linear attention interpolation
+(fire extinguisher: 1.0, 2.0)
+
+# prompt-editing-aware attention interpolation
+[(fire extinguisher: 1.0, 2.0)::5]
+
+# weighted sum of conditions
+[space station : kitchen mixer :: mean]
+
+# define functions and variables to simplify repeating patterns and use a consistent structure
+$prompt($style, $quality, $character, $background) = (
+    A detailed picture in the style of $style,
+    $quality,
+    $character lying back,
+    $background in the background
+:1)
+```
+
+## Features
+
+- [Prompt interpolation using a curve function](https://github.com/ljleb/prompt-fusion-extension/wiki/Prompt-Interpolation)
+- [Attention interpolation aware of contextual prompt editing](https://github.com/ljleb/prompt-fusion-extension/wiki/Attention-Interpolation)
+- [Alternation interpolation](https://github.com/ljleb/prompt-fusion-extension/wiki/Alternation-interpolation)
+- [Prompt weighted sum](https://github.com/ljleb/prompt-fusion-extension/wiki/Prompt-Average)
+- [Prompt variables and functions](https://github.com/ljleb/prompt-fusion-extension/wiki/Prompt-Variables)
+- Complete backwards compatibility with the builtin prompt syntax of the webui
+
+The prompt interpolation feature is similar to [Prompt Travel](https://github.com/Kahsolt/stable-diffusion-webui-prompt-travel), which allows to create videos of images generated by navigating the latent space iteratively. Unlike Prompt Travel however, instead of generating multiple images, Prompt Fusion allows you to travel during the sampling process of *a single image*. Also, instead of interpolating the latent space, it uses the embedding space to determine intermediate embeddings. 
+
+Prompt interpolation is also similar to [Prompt Blending](https://github.com/amotile/stable-diffusion-backend/tree/master/src/process/implementations/automatic1111_scripts). The main difference is that this extension calculates a new embedding for every step, as opposed to calculating it once and using that same one embedding for all the steps. 
+
+The attention interpolation feature is similar to [Shift Attention](https://github.com/yownas/shift-attention), which allows to generate multiple images with slight variations in the attention given to certain parts of the prompt. Unlike Shift Attention, instead of generating multiple images, Prompt Fusion allows to shift the attention of certain parts of a prompt during the sampling process of *a single image*.
+
+## Usage
+- Check the [wiki pages](https://github.com/ljleb/fusion/wiki) for the extension documentation.
+
+## Examples
+
+### 1. Influencing the beginning of the sampling process
+
+Interpolate linearly (by default) from `lion` (step 0) to `bird` (step 8) to `girl` (step 11), and stay at `girl` for the rest of the sampling steps:
+
+```
+[lion:bird:girl: , 7, 10]
+```
+
+![curve1](https://user-images.githubusercontent.com/32277961/214725976-b72bafc6-0c5d-4491-9c95-b73da41da082.gif)
+
+### 2. Influencing the middle of the sampling process
+
+Interpolate using a bezier curve from `fireball monster` (step 0) to `dragon monster` (step 12, because 30 steps * 0.4 = step 12), while using `seawater monster` as an intermediate control point to steer the curve away during interpolation and to get creative results:
+
+```
+[fireball:seawater:dragon: , .1, .4:bezier] monster
+```
+
+![curve2](https://user-images.githubusercontent.com/32277961/214941229-2dccad78-f856-42bb-ae6b-16b65b273cda.gif)
+
+## Webui supported releases
+
+The following webui releases are officially supported:
+- `v1.0.0-pre`
+- `master` (there may be a slight lag for issues arising during quick a1111 webui updates)
+
+## Installation
+1. Visit the **Extensions** tab of Automatic's WebUI.
+2. Visit the **Available** subtab.
+3. Look for **Prompt Fusion**.
+4. Press the **Install** button.
+5. Wait for the webui to finish downloading the extension.
+6. Visit the **Installed** subtab.
+7. click on **Apply and restart UI**.
+
+Alternatively, instead of steps 6 and 7, you can restart the webui completely.
+
+## Related Projects
+
+- Prompt Travel: https://github.com/Kahsolt/stable-diffusion-webui-prompt-travel
+- Shift Attention: https://github.com/yownas/shift-attention
+- Prompt Blending: https://github.com/amotile/stable-diffusion-backend/tree/master/src/process/implementations/automatic1111_scripts
+

z-patch_prompt-fusion-extension/requirements.txt

@@ -0,0 +1 @@
+

z-patch_prompt-fusion-extension/scripts/promptlang.py

@@ -0,0 +1,273 @@
+import gradio as gr
+from lib_prompt_fusion import hijacker, empty_cond, global_state, interpolation_tensor, prompt_parser as prompt_fusion_parser
+from modules import scripts, script_callbacks, prompt_parser, shared
+
+
+fusion_hijacker_attribute = '__fusion_hijacker'
+prompt_parser_hijacker = hijacker.ModuleHijacker.install_or_get(
+    module=prompt_parser,
+    hijacker_attribute=fusion_hijacker_attribute,
+    register_uninstall=script_callbacks.on_script_unloaded)
+
+
+def on_ui_settings():
+    section = ('prompt-fusion', 'Prompt Fusion')
+    shared.opts.add_option('prompt_fusion_enabled', shared.OptionInfo(True, 'Enable prompt-fusion extension', section=section))
+    shared.opts.add_option('prompt_fusion_slerp_scale', shared.OptionInfo(0, 'Slerp scale (0 = linear geometry, 1 = slerp geometry)', component=gr.Number, section=section))
+    shared.opts.add_option('prompt_fusion_slerp_negative_origin', shared.OptionInfo(True, 'use negative prompt as slerp origin', section=section))
+    shared.opts.add_option('prompt_fusion_slerp_epsilon', shared.OptionInfo(0.0001, 'Slerp epsilon (fallback on linear geometry when conds are too similar. 0 = parallel, 1 = perpendicular)', component=gr.Number, section=section))
+
+
+script_callbacks.on_ui_settings(on_ui_settings)
+
+
+@prompt_parser_hijacker.hijack('get_learned_conditioning')
+def _hijacked_get_learned_conditioning(model, prompts, total_steps, *args, original_function, **kwargs):
+    if not shared.opts.prompt_fusion_enabled:
+        return original_function(model, prompts, total_steps, *args, **kwargs)
+
+    hires_steps, use_old_scheduling, *_ = args if args else (None, True)
+    is_hires = hires_steps is not None
+    if is_hires:
+        real_total_steps = hires_steps
+    else:
+        real_total_steps = total_steps
+
+    if hasattr(prompts, 'is_negative_prompt'):
+        is_negative_prompt = prompts.is_negative_prompt
+    else:
+        is_negative_prompt = global_state.old_webui_is_negative
+
+    empty_cond.init(model)
+
+    tensor_builders = _parse_tensor_builders(prompts, real_total_steps, is_hires, use_old_scheduling)
+    if hasattr(prompt_parser, 'SdConditioning'):
+        empty_conditioning = prompt_parser.SdConditioning(prompts)
+        empty_conditioning.clear()
+    else:
+        empty_conditioning = []
+
+    flattened_prompts, consecutive_ranges = _get_flattened_prompts(tensor_builders, empty_conditioning)
+    flattened_schedules = original_function(model, flattened_prompts, total_steps, *args, **kwargs)
+
+    if isinstance(flattened_schedules[0][0].cond, dict): # sdxl
+        CondWrapper = interpolation_tensor.DictCondWrapper
+    else:
+        CondWrapper = interpolation_tensor.TensorCondWrapper
+
+    flattened_schedules = [
+        [
+            prompt_parser.ScheduledPromptConditioning(cond=CondWrapper(schedule.cond), end_at_step=schedule.end_at_step)
+            for schedule in subschedules
+        ]
+        for subschedules in flattened_schedules
+    ]
+
+    cond_tensors = (tensor_builder.build(flattened_schedules[begin:end], empty_cond.get())
+                    for begin, end, tensor_builder
+                    in zip(consecutive_ranges[:-1], consecutive_ranges[1:], tensor_builders))
+
+    schedules = [_sample_tensor_schedules(cond_tensor, real_total_steps, is_hires)
+                 for cond_tensor in cond_tensors]
+
+    if is_negative_prompt:
+        if hires_steps is not None:
+            global_state.negative_schedules_hires = schedules[0]
+        else:
+            global_state.negative_schedules = schedules[0]
+
+    schedules = [
+        [
+            prompt_parser.ScheduledPromptConditioning(cond=schedule.cond.original_cond, end_at_step=schedule.end_at_step)
+            for schedule in subschedules
+        ]
+        for subschedules in schedules
+    ]
+
+    return schedules
+
+
+def _pf_norm_lastdim(x: torch.Tensor, eps: float = 1e-8) -> torch.Tensor:
+    return x / (x.norm(dim=-1, keepdim=True) + eps)
+
+def _pf_stack_weight(tensors, weights):
+    base = tensors[0]
+    w = torch.as_tensor(weights, device=base.device, dtype=base.dtype)
+    # -> [N, 1, 1, ...] для широковещательного умножения
+    w = w.view(-1, *([1] * base.dim()))
+    return torch.stack(tensors, dim=0), w
+
+def _agg_tensors_linear(tensors, weights):
+    T, W = _pf_stack_weight(tensors, weights)
+    return (T * W).sum(dim=0) / (float(sum(weights)) + 1e-8)
+
+def _agg_tensors_spherical(tensors, weights):
+    normed = [_pf_norm_lastdim(t) for t in tensors]
+    T, W = _pf_stack_weight(normed, weights)
+    s = (T * W).sum(dim=0)
+    return _pf_norm_lastdim(s)
+
+def _agg_cond_list(conds, weights, mode: str):
+    # cond: либо тензор, либо dict SDXL (ключи: 'crossattn', 'vector/pooled' и т.п.)
+    if isinstance(conds[0], dict):
+        out = {}
+        for k in conds[0].keys():
+            vals = [c[k] for c in conds]
+            if isinstance(vals[0], list):  # SDXL может хранить списки тензоров
+                out[k] = [
+                    (_agg_tensors_linear if mode=="linear" else _agg_tensors_spherical)(
+                        [vv[i] for vv in vals], weights
+                    )
+                    for i in range(len(vals[0]))
+                ]
+            else:
+                out[k] = (_agg_tensors_linear if mode=="linear" else _agg_tensors_spherical)(vals, weights)
+        return out
+    else:
+        return (_agg_tensors_linear if mode=="linear" else _agg_tensors_spherical)(conds, weights)
+
+def _cond_at_step(schedule, step: int):
+    # schedule: List[ScheduledPromptConditioning]; step — граница сегмента
+    for sp in schedule:
+        if step <= sp.end_at_step:
+            return sp.cond
+    return schedule[-1].cond
+
+def _aggregate_neg_schedule_for_prompt(conds_parts, schedules_flat, mode: str, total_steps: int, pp):
+    """
+    conds_parts: List[(flat_index, weight)] для ОДНОГО исходного промпта (neg)
+    schedules_flat: List[List[ScheduledPromptConditioning]] индексируется flat_index
+    mode: 'linear' | 'spherical'
+    Возвращает агрегированное расписание (List[ScheduledPromptConditioning]).
+    """
+    boundaries = {total_steps}
+    for idx, _w in conds_parts:
+        for sp in schedules_flat[idx]:
+            boundaries.add(int(sp.end_at_step))
+    bps = sorted(boundaries)
+    out = []
+    for end in bps:
+        conds = []
+        ws = []
+        for idx, w in conds_parts:
+            conds.append(_cond_at_step(schedules_flat[idx], end))
+            ws.append(float(w))
+        agg = _agg_cond_list(conds, ws, mode)
+        out.append(pp.ScheduledPromptConditioning(end_at_step=int(end), cond=agg))
+    return out
+
+@prompt_parser_hijacker.hijack('get_multicond_learned_conditioning')
+def _hijacked_get_multicond_learned_conditioning(model, prompts, steps, *args, original_function, **kwargs):
+    # Если расширение отключено — пробрасываем как раньше
+    if not shared.opts.prompt_fusion_enabled:
+        res = original_function(model, prompts, steps, *args, **kwargs)
+        global_state.old_webui_is_negative = False
+        return res
+
+    # hires/шаги (совместимо с веб-UI)
+    hires_steps, use_old_scheduling, *_ = args if args else (None, True)
+    is_hires = hires_steps is not None
+    real_total_steps = hires_steps if is_hires else steps
+
+    # Разбиваем мультиконд на части по версии webui
+    try:
+        conds_list, all_prompts, _ = prompt_parser.get_multicond_prompt_list(prompts)
+    except Exception:
+        conds_list = [[(i, 1.0)] for i in range(len(prompts))]
+        all_prompts = list(prompts)
+
+    # Строим расписания для каждой части через одиночный перехват (Fusion уже внутри)
+    schedules_flat = prompt_parser.get_learned_conditioning(model, all_prompts, steps, *args, **kwargs)
+
+    # --- Агрегация негативного origin по выбранному режиму ---
+    if global_state.old_webui_is_negative and conds_list:
+        mode = global_state.get_neg_origin_mode()
+        if mode in ("linear","spherical"):
+            neg_parts = conds_list[0]  # [(flat_index, weight), ...]
+            agg_sched = _aggregate_neg_schedule_for_prompt(
+                conds_parts=neg_parts,
+                schedules_flat=schedules_flat,
+                mode=mode,
+                total_steps=real_total_steps,
+                pp=prompt_parser
+            )
+            if hires_steps is not None:
+                global_state.negative_schedules_hires = agg_sched
+            else:
+                global_state.negative_schedules = agg_sched
+
+    # Упаковка в MulticondLearnedConditioning
+    batch = []
+    for parts in conds_list:
+        comp = []
+        for flat_index, weight in parts:
+            comp.append(prompt_parser.ComposableScheduledPromptConditioning(schedules_flat[flat_index], weight))
+        batch.append(comp)
+
+    # Правильный shape для SDXL/SD1.x
+    if schedules_flat and schedules_flat[0]:
+        example_cond = schedules_flat[0][0].cond
+        if isinstance(example_cond, dict):  # SDXL
+            ca = example_cond.get('crossattn')
+            if isinstance(ca, list) and ca:
+                shape = getattr(ca[0], 'shape', None) or (0,)
+            else:
+                shape = getattr(ca, 'shape', None) or (0,)
+        else:  # SD 1.x
+            shape = getattr(example_cond, 'shape', None) or (0,)
+    else:
+        shape = (0,)
+
+    global_state.old_webui_is_negative = False
+    return prompt_parser.MulticondLearnedConditioning(shape, batch)
+
+
+def _parse_tensor_builders(prompts, total_steps, is_hires, use_old_scheduling):
+    tensor_builders = []
+
+    for prompt in prompts:
+        expr = prompt_fusion_parser.parse_prompt(prompt)
+        tensor_builder = interpolation_tensor.InterpolationTensorBuilder()
+        expr.extend_tensor(tensor_builder, (0, total_steps), total_steps, dict(), is_hires, use_old_scheduling)
+        tensor_builders.append(tensor_builder)
+
+    return tensor_builders
+
+
+def _get_flattened_prompts(tensor_builders, flattened_prompts=None):
+    if flattened_prompts is None:
+        flattened_prompts = []
+    consecutive_ranges = [0]
+
+    for tensor_builder in tensor_builders:
+        flattened_prompts.extend(tensor_builder.get_prompt_database())
+        consecutive_ranges.append(len(flattened_prompts))
+
+    return flattened_prompts, consecutive_ranges
+
+
+def _sample_tensor_schedules(tensor, steps, is_hires):
+    schedules = []
+
+    for step in range(steps):
+        origin_cond = global_state.get_origin_cond_at(step, is_hires)
+        params = interpolation_tensor.InterpolationParams(step / steps, step, steps, global_state.get_slerp_scale(), global_state.get_slerp_epsilon())
+        schedule_cond = tensor.interpolate(params, origin_cond, empty_cond.get())
+        if schedules and schedules[-1].cond == schedule_cond:
+            schedules[-1] = prompt_parser.ScheduledPromptConditioning(end_at_step=step, cond=schedules[-1].cond)
+        else:
+            schedules.append(prompt_parser.ScheduledPromptConditioning(end_at_step=step, cond=schedule_cond))
+
+    return schedules
+
+
+class PromptFusionScript(scripts.Script):
+    def title(self):
+        return 'Prompt Fusion'
+
+    def show(self, is_img2img):
+        return scripts.AlwaysVisible
+
+    def process(self, p, *args):
+        global_state.negative_schedules = None
+        global_state.old_webui_is_negative = True

z-patch_prompt-fusion-extension/test/parser_tests.py

@@ -0,0 +1,104 @@
+from lib_prompt_fusion.prompt_parser import parse_prompt
+from lib_prompt_fusion.interpolation_tensor import InterpolationTensorBuilder
+
+
+def run_functional_tests(total_steps=100):
+    for i, (given, expected) in enumerate(functional_parse_test_cases):
+        expr = parse_prompt(given)
+        tensor_builder = InterpolationTensorBuilder()
+        expr.extend_tensor(tensor_builder, (0, total_steps), total_steps, dict(), is_hires=False, use_old_scheduling=False)
+
+        actual = tensor_builder.get_prompt_database()
+
+        if type(expected) is set:
+            assert set(actual) == expected, f"{actual} != {expected}"
+        else:
+            assert len(actual) == 1 and actual[0] == expected, f"'{actual[0]}' != '{expected}'"
+
+
+functional_parse_test_cases = [
+    ('single',)*2,
+    ('some space separated text',)*2,
+    ('(legacy weighted prompt:-2.1)',)*2,
+    ('mixed (legacy weight:3.6) and text',)*2,
+    ('legacy [range begin:0] thingy',)*2,
+    ('legacy [range end::3] thingy',)*2,
+    ('legacy [[nested range::3]:2] thingy',)*2,
+    ('legacy [[nested range:2]::3] thingy',)*2,
+    ('sugar [range:,abc:3] thingy',)*2,
+    ('sugar [[(weight interpolation:0,12):0]::1] thingy', 'sugar [[(weight interpolation:0.0):0]::1] thingy'),
+    ('sugar [[(weight interpolation:0,12):0]::2] thingy', 'sugar [[[(weight interpolation:0.0)::1][(weight interpolation:12.0):1]:0]::2] thingy'),
+    ('sugar [[(weight interpolation:0,12):0]::3] thingy', 'sugar [[[(weight interpolation:0.0)::1][[(weight interpolation:6.0):1]::2][(weight interpolation:12.0):2]:0]::3] thingy'),
+    ('legacy [from:to:2] thingy',)*2,
+    ('legacy [negative weight]',)*2,
+    ('legacy (positive weight)',)*2,
+    ('[abc:1girl:2]',)*2,
+    ('[::]',)*2,
+    ('[a:b:]',)*2,
+    ('[[a:b:1,2]:b:]', {'[a:b:]', '[b:b:]'}),
+    ('1girl',)*2,
+    ('dashes-in-text',)*2,
+    ('text, separated with, comas',)*2,
+    ('{prompt}',)*2,
+    ('[abc|def ghi|jkl]',)*2,
+    ('merging this AND with this',)*2,
+    (':',)*2,
+    (r'portrait \(object\)',)*2,
+    (r'\[escaped square\]',)*2,
+    (r'\$var = abc',)*2,
+    (r'\\$ arst',)*2,
+    (r'$$ arst',)*2,
+    ('$var = abc', ''),
+    ('$a = prompt value\n$a', 'prompt value'),
+    ('$a = prompt value\n$b = $a\n$b', 'prompt value'),
+    ('$a = (multiline\nprompt\nvalue:1.0)\n$a', '(multiline prompt value:1.0)'),
+    ('$a = ($aa = nested variable\nmultiline\n$aa:1.0)\n$a', '(multiline nested variable:1.0)'),
+    ('a [b:c:-1, 10] d', {'a b d', 'a c d'}),
+    ('a [b:c:5, 6] d', {'a b d', 'a c d'}),
+    ('a [b:c:0.25, 0.5] d', {'a b d', 'a c d'}),
+    ('a [b:c:.25, .5] d', {'a b d', 'a c d'}),
+    ('a [b:c:,] d', {'a b d', 'a c d'}),
+    ('0[1.0:1.1:,]2[3.0:3.1:,]4', {
+        '0 1.0 2 3.0 4', '0 1.1 2 3.0 4',
+        '0 1.0 2 3.1 4', '0 1.1 2 3.1 4',
+    }),
+    ('0[1.0:1.1:1.2:,.5,]2[3.0:3.1:,]4', {
+        '0 1.0 2 3.0 4', '0 1.0 2 3.1 4',
+        '0 1.1 2 3.0 4', '0 1.1 2 3.1 4',
+        '0 1.2 2 3.0 4', '0 1.2 2 3.1 4',
+    }),
+    ('[0.0:0.1:,][1.0:1.1:,][2.0:2.1:,]', {
+        '0.0 1.0 2.0', '0.0 1.0 2.1',
+        '0.1 1.0 2.0', '0.1 1.0 2.1',
+        '0.0 1.1 2.0', '0.0 1.1 2.1',
+        '0.1 1.1 2.0', '0.1 1.1 2.1',
+    }),
+    ('[top level:interpolatin:lik a pro:1,3,5:linear]', {'top level', 'interpolatin', 'lik a pro'}),
+    ('[[nested:expr:,]:abc:,]', {'nested', 'expr', 'abc'}),
+    ('[(nested attention:2.0):abc:,]', {'(nested attention:2.0)', 'abc'}),
+    ('[[nested editing:15]:abc:,]', {'[nested editing:15]', 'abc'}),
+    ('[[nested interpolation:abc:,]:12]', {'[nested interpolation:12]', '[abc:12]'}),
+    ('[[nested interpolation:abc:,]::7]', {'[nested interpolation::7]', '[abc::7]'}),
+    ('$attention = 1.5\n(prompt:$attention)', '(prompt:1.5)'),
+    ('$a = 0\n$b = 12\n[[(prompt:$a,$b):0]::2]', '[[[(prompt:0.0)::1][(prompt:12.0):1]:0]::2]'),
+    ('$step = 5\n[legacy:editing:$step]', '[legacy:editing:5]'),
+    ('$begin = 2\n$end = 7\n[prompt:interpolation:$begin, $end]', {'prompt', 'interpolation'}),
+    ('$a($b, $c) = prompt with $b, prompt with $c\n$a(cat, dog)', 'prompt with cat , prompt with dog'),
+    ('$a($b) = prompt with $b\n$c($d) = yeay $a($d)\n$c(dog)', 'yeay prompt with dog'),
+    ('$a = a lot of animals\n$b($c) = I love $c\n$b($a)', 'I love a lot of animals'),
+    ('$a($b) = prompt with $b\n$c($d) = yeay $d\n$a($c(dog))', 'prompt with yeay dog'),
+    ('[a|b|c]', '[a|b|c]'),
+    ('[a|b|c:]', '[a|b|c]'),
+    ('[a|b|c:1]', {'a', 'b', 'c'}),
+    ('[a|b|c:2]', {'a', 'b', 'c'}),
+    ('[a|b|c:0.5]', {'a', 'b', 'c'}),
+    ('[a|b|c:1.1]', {'a', 'b', 'c'}),
+    ('[[[Imperial Yellow|Amber]:[Ruby|Plum|Bronze]:9]::39]',)*2,
+    ('[a:b:c::mean]', {'a', 'b', 'c'}),
+    ('[a:b:c:,,:mean]', {'a', 'b', 'c'}),
+    ('[a:b:c: 1, 2, 3:mean]', {'a', 'b', 'c'}),
+]
+
+
+def run_tests():
+    run_functional_tests()

z-patch_prompt-fusion-extension/test/run_all.py

@@ -0,0 +1,7 @@
+import sys
+sys.path.append('..')
+import parser_tests
+
+
+if __name__ == '__main__':
+    parser_tests.run_tests()