CURVE node

* CURVE type

* fix: update typed wrapper unwrap keys to __type__ and __value__

* code improve

* code improve

comfy/supported_models.py

@@ -1127,7 +1127,7 @@ class ZImagePixelSpace(ZImage):
     latent_format = latent_formats.ZImagePixelSpace
 
     # Much lower memory than latent-space models (no VAE, small patches).
-    memory_usage_factor = 0.05 # TODO: figure out the optimal value for this.
+    memory_usage_factor = 0.03 # TODO: figure out the optimal value for this.
 
     def get_model(self, state_dict, prefix="", device=None):
         return model_base.ZImagePixelSpace(self, device=device)

comfy_api/latest/_io.py

@@ -1240,6 +1240,19 @@ class BoundingBox(ComfyTypeIO):
             return d
 
 
+@comfytype(io_type="CURVE")
+class Curve(ComfyTypeIO):
+    CurvePoint = tuple[float, float]
+    Type = list[CurvePoint]
+
+    class Input(WidgetInput):
+        def __init__(self, id: str, display_name: str=None, optional=False, tooltip: str=None,
+                     socketless: bool=True, default: list[tuple[float, float]]=None, advanced: bool=None):
+            super().__init__(id, display_name, optional, tooltip, None, default, socketless, None, None, None, None, advanced)
+            if default is None:
+                self.default = [(0.0, 0.0), (1.0, 1.0)]
+
+
 DYNAMIC_INPUT_LOOKUP: dict[str, Callable[[dict[str, Any], dict[str, Any], tuple[str, dict[str, Any]], str, list[str] | None], None]] = {}
 def register_dynamic_input_func(io_type: str, func: Callable[[dict[str, Any], dict[str, Any], tuple[str, dict[str, Any]], str, list[str] | None], None]):
     DYNAMIC_INPUT_LOOKUP[io_type] = func
@@ -2226,5 +2239,6 @@ __all__ = [
     "PriceBadgeDepends",
     "PriceBadge",
     "BoundingBox",
+    "Curve",
     "NodeReplace",
 ]

comfy_extras/nodes_custom_sampler.py

@@ -1049,6 +1049,48 @@ class ManualSigmas(io.ComfyNode):
         sigmas = torch.FloatTensor(sigmas)
         return io.NodeOutput(sigmas)
 
+class CurveToSigmas(io.ComfyNode):
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="CurveToSigmas",
+            display_name="Curve to Sigmas",
+            category="sampling/custom_sampling/schedulers",
+            inputs=[
+                io.Curve.Input("curve", default=[[0.0, 1.0], [1.0, 0.0]]),
+                io.Model.Input("model", optional=True),
+                io.Int.Input("steps", default=20, min=1, max=10000),
+                io.Float.Input("sigma_max", default=14.614642, min=0.0, max=5000.0, step=0.01, round=False, optional=True),
+            ],
+            outputs=[io.Sigmas.Output()]
+        )
+
+    @classmethod
+    def execute(cls, curve, steps, sigma_max=14.614642, model=None) -> io.NodeOutput:
+        points = sorted(curve, key=lambda p: p[0])
+        model_sampling = model.get_model_object("model_sampling") if model is not None else None
+
+        sigmas = []
+        for i in range(steps + 1):
+            t = i / steps
+            y = points[0][1] if t < points[0][0] else points[-1][1]
+            for j in range(len(points) - 1):
+                if points[j][0] <= t <= points[j + 1][0]:
+                    x0, y0 = points[j]
+                    x1, y1 = points[j + 1]
+                    y = y0 if x1 == x0 else y0 + (y1 - y0) * (t - x0) / (x1 - x0)
+                    break
+            if model_sampling is not None:
+                sigmas.append(float(model_sampling.percent_to_sigma(1.0 - y)))
+            else:
+                sigmas.append(y * sigma_max)
+
+        sigmas[-1] = 0.0
+        return io.NodeOutput(torch.FloatTensor(sigmas))
+
+    get_sigmas = execute
+
+
 class CustomSamplersExtension(ComfyExtension):
     @override
     async def get_node_list(self) -> list[type[io.ComfyNode]]:
@@ -1088,6 +1130,7 @@ class CustomSamplersExtension(ComfyExtension):
             AddNoise,
             SamplerCustomAdvanced,
             ManualSigmas,
+            CurveToSigmas,
         ]
 
 

comfy_extras/nodes_painter.py

@@ -1,127 +0,0 @@
-from __future__ import annotations
-
-import hashlib
-import os
-
-import numpy as np
-import torch
-from PIL import Image
-
-import folder_paths
-import node_helpers
-from comfy_api.latest import ComfyExtension, io, UI
-from typing_extensions import override
-
-
-def hex_to_rgb(hex_color: str) -> tuple[float, float, float]:
-    hex_color = hex_color.lstrip("#")
-    if len(hex_color) != 6:
-        return (0.0, 0.0, 0.0)
-    r = int(hex_color[0:2], 16) / 255.0
-    g = int(hex_color[2:4], 16) / 255.0
-    b = int(hex_color[4:6], 16) / 255.0
-    return (r, g, b)
-
-
-class PainterNode(io.ComfyNode):
-    @classmethod
-    def define_schema(cls):
-        return io.Schema(
-            node_id="Painter",
-            display_name="Painter",
-            category="image",
-            inputs=[
-                io.Image.Input(
-                    "image",
-                    optional=True,
-                    tooltip="Optional base image to paint over",
-                ),
-                io.String.Input(
-                    "mask",
-                    default="",
-                    socketless=True,
-                    extra_dict={"widgetType": "PAINTER", "image_upload": True},
-                ),
-                io.Int.Input(
-                    "width",
-                    default=512,
-                    min=64,
-                    max=4096,
-                    step=64,
-                    socketless=True,
-                    extra_dict={"hidden": True},
-                ),
-                io.Int.Input(
-                    "height",
-                    default=512,
-                    min=64,
-                    max=4096,
-                    step=64,
-                    socketless=True,
-                    extra_dict={"hidden": True},
-                ),
-                io.Color.Input("bg_color", default="#000000"),
-            ],
-            outputs=[
-                io.Image.Output("IMAGE"),
-                io.Mask.Output("MASK"),
-            ],
-        )
-
-    @classmethod
-    def execute(cls, mask, width, height, bg_color="#000000", image=None) -> io.NodeOutput:
-        if image is not None:
-            base_image = image[:1]
-            h, w = base_image.shape[1], base_image.shape[2]
-        else:
-            h, w = height, width
-            r, g, b = hex_to_rgb(bg_color)
-            base_image = torch.zeros((1, h, w, 3), dtype=torch.float32)
-            base_image[0, :, :, 0] = r
-            base_image[0, :, :, 1] = g
-            base_image[0, :, :, 2] = b
-
-        if mask and mask.strip():
-            mask_path = folder_paths.get_annotated_filepath(mask)
-            painter_img = node_helpers.pillow(Image.open, mask_path)
-            painter_img = painter_img.convert("RGBA")
-
-            if painter_img.size != (w, h):
-                painter_img = painter_img.resize((w, h), Image.LANCZOS)
-
-            painter_np = np.array(painter_img).astype(np.float32) / 255.0
-            painter_rgb = painter_np[:, :, :3]
-            painter_alpha = painter_np[:, :, 3:4]
-
-            mask_tensor = torch.from_numpy(painter_np[:, :, 3]).unsqueeze(0)
-
-            base_np = base_image[0].cpu().numpy()
-            composited = painter_rgb * painter_alpha + base_np * (1.0 - painter_alpha)
-            out_image = torch.from_numpy(composited).unsqueeze(0)
-        else:
-            mask_tensor = torch.zeros((1, h, w), dtype=torch.float32)
-            out_image = base_image
-
-        return io.NodeOutput(out_image, mask_tensor, ui=UI.PreviewImage(out_image))
-
-    @classmethod
-    def fingerprint_inputs(cls, mask, width, height, bg_color="#000000", image=None):
-        if mask and mask.strip():
-            mask_path = folder_paths.get_annotated_filepath(mask)
-            if os.path.exists(mask_path):
-                m = hashlib.sha256()
-                with open(mask_path, "rb") as f:
-                    m.update(f.read())
-                return m.digest().hex()
-        return ""
-
-
-
-class PainterExtension(ComfyExtension):
-    @override
-    async def get_node_list(self):
-        return [PainterNode]
-
-
-async def comfy_entrypoint():
-    return PainterExtension()

execution.py

@@ -876,12 +876,14 @@ async def validate_inputs(prompt_id, prompt, item, validated):
                 continue
         else:
             try:
-                # Unwraps values wrapped in __value__ key. This is used to pass
-                # list widget value to execution, as by default list value is
-                # reserved to represent the connection between nodes.
-                if isinstance(val, dict) and "__value__" in val:
-                    val = val["__value__"]
-                    inputs[x] = val
+                # Unwraps values wrapped in __value__ key or typed wrapper.
+                # This is used to pass list widget values to execution,
+                # as by default list value is reserved to represent the
+                # connection between nodes.
+                if isinstance(val, dict):
+                    if "__value__" in val:
+                        val = val["__value__"]
+                        inputs[x] = val
 
                 if input_type == "INT":
                     val = int(val)

nodes.py

@@ -2034,6 +2034,24 @@ class ImagePadForOutpaint:
         return (new_image, mask.unsqueeze(0))
 
 
+class CurveEditor:
+    @classmethod
+    def INPUT_TYPES(s):
+        return {
+            "required": {
+                "curve": ("CURVE", {"default": [[0, 0], [1, 1]]}),
+            }
+        }
+
+    RETURN_TYPES = ("CURVE",)
+    RETURN_NAMES = ("curve",)
+    FUNCTION = "execute"
+    CATEGORY = "utils"
+
+    def execute(self, curve):
+        return (curve,)
+
+
 NODE_CLASS_MAPPINGS = {
     "KSampler": KSampler,
     "CheckpointLoaderSimple": CheckpointLoaderSimple,
@@ -2102,6 +2120,7 @@ NODE_CLASS_MAPPINGS = {
     "ConditioningZeroOut": ConditioningZeroOut,
     "ConditioningSetTimestepRange": ConditioningSetTimestepRange,
     "LoraLoaderModelOnly": LoraLoaderModelOnly,
+    "CurveEditor": CurveEditor,
 }
 
 NODE_DISPLAY_NAME_MAPPINGS = {
@@ -2170,6 +2189,7 @@ NODE_DISPLAY_NAME_MAPPINGS = {
     # _for_testing
     "VAEDecodeTiled": "VAE Decode (Tiled)",
     "VAEEncodeTiled": "VAE Encode (Tiled)",
+    "CurveEditor": "Curve Editor",
 }
 
 EXTENSION_WEB_DIRS = {}
@@ -2449,7 +2469,6 @@ async def init_builtin_extra_nodes():
         "nodes_replacements.py",
         "nodes_nag.py",
         "nodes_sdpose.py",
-        "nodes_painter.py",
     ]
 
     import_failed = []