nodes/pupba-Comfy_ForEach
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

v0.1.0

Browse Source
This commit is contained in:
Pupbani
2025-04-22 11:01:58 +09:00
commit 12365cd792
11 changed files with 529 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+2
View File
@@ -0,0 +1,2 @@
__pycache__/
.DS_Store
README.md
+240
View File
@@ -0,0 +1,240 @@
# 📦 ComfyForEach: Custom ComfyUI Nodes for Batch Image Processing and Used in AWS
![version](https://img.shields.io/badge/version-0.1.0-blue.svg)
![license](https://img.shields.io/badge/license-MIT-green.svg)
A collection of ComfyUI custom nodes designed for image batch processing, per-index image operations, and AWS integration using EventBridge.
_🔥 `AWS Event Bridge` code updated after testing._
## 📁 Directory Structure
```
custom_nodes/
└── Comfy_ForEach/
├── __init__.py
├── context.py
├── task_manager.py
├── loader_nodes.py
├── index_selector_nodes.py
├── logic_nodes.py
├── save_nodes.py
├── aws_event_node.py
└── requirements.txt
```
## ✅ What needs to be modified in the ComfyUI
**`execution.py`**
```python
######### Error Logging #########
try:
results.append(getattr(obj, func)(**inputs))
except Exception as e:
import logging
from datetime import datetime
import traceback
logger = logging.getLogger("comfy_node_error")
node_name = obj.__class__.__name__
error_msg = f"[ERROR {datetime.now().isoformat()} | TaskID:{context.get_task_id()} | Node: {node_name} | Index: {index} | {type(e).__name__}: {e}"
logger.error(error_msg)
logger.error(traceback.format_exc())
# Here EventBridge Code
raise
#################################
```
**`main.py`**
```python
setup_logger(log_level=args.verbose, use_stdout=args.log_stdout,log_path="./logs")
```
**`app/logger`**
```python
...
from logging.handlers import TimedRotatingFileHandler
...
def setup_logger(log_level: str = 'INFO', capacity: int = 300, use_stdout: bool = False,log_path:str | None=None):
global logs
if logs:
return
# Override output streams and log to buffer
logs = deque(maxlen=capacity)
global stdout_interceptor
global stderr_interceptor
stdout_interceptor = sys.stdout = LogInterceptor(sys.stdout)
stderr_interceptor = sys.stderr = LogInterceptor(sys.stderr)
# Setup default global logger
logger = logging.getLogger()
logger.setLevel(log_level)
stream_handler = logging.StreamHandler()
# stream_handler.setFormatter(logging.Formatter("%(message)s"))
stream_handler.setFormatter(logging.Formatter("[%(asctime)s] [%(levelname)s] %(message)s"))
if use_stdout:
# Only errors and critical to stderr
stream_handler.addFilter(lambda record: not record.levelno < logging.ERROR)
# Lesser to stdout
stdout_handler = logging.StreamHandler(sys.stdout)
stdout_handler.setFormatter(logging.Formatter("%(message)s"))
stdout_handler.addFilter(lambda record: record.levelno < logging.ERROR)
logger.addHandler(stdout_handler)
logger.addHandler(stream_handler)
# Error Log
if log_path:
os.makedirs(log_path, exist_ok=True)
error_log_file = os.path.join(log_path, "errors.log")
error_file_handler = TimedRotatingFileHandler(
error_log_file,
when="midnight",
interval=1,
backupCount=7,
encoding="utf-8"
)
error_file_handler.suffix = "%Y-%m-%d"
error_file_handler.setLevel(logging.ERROR)
error_file_handler.setFormatter(logging.Formatter("[%(asctime)s] [%(levelname)s] %(message)s"))
logger.addHandler(error_file_handler)
```
## 🧩 Node Overview
### 🔹 TaskIDStorageNode
- Stores a task ID in global memory (context.py) for use across the workflow.
- **Category** : Workflow Utils
- **Output** : task_id (STRING)
### 🔹 FolderImageLoaderNode
- Loads all .png images from a specified folder.
- Outputs a list of image tensors and their filenames.
- **Category** : ComfyForEach/Load
- **Outputs** : image_list (IMAGE), image_name_list (STRING)
### 🔹 IndexedImageSelectorNode
- Selects a specific image from a list using an index.
- **Category** : ComfyForEach/Select
- **Output** : image (IMAGE)
### 🔹 IndexedNameWithPrefixNode
- Retrieves a specific image filename using an index and prepends a folder path.
- **Category** : ComfyForEach/Select
- **Outputs** : file_name, folder_path (STRING)
### 🔹 IsLastIndexNode
- Checks whether the current index is the last in a sequence.
- Useful for triggering events only once at the end of a loop.
- **Category** : ComfyForEach/Logic
- **Output** : is_last (BOOLEAN)
### 🔹 SaveExactNameImageNode
- Saves an image tensor with an exact filename and folder path.
- **Category** : ComfyForEach/Save
- **Output** : None (Terminal node)
### 🔹 EventBridgeTriggerNode
- Simulates AWS EventBridge notification.
- Writes a SUCCESS or FAILED event as a .json log based on is_last.
- **Category** : ComfyForEach/AWS
- **Output** : None (Terminal node)
## 🧪 Requirements
Install dependencies:
```bash
pip install -r requirements.txt
```
Contents of `requirements.txt`
```txt
pillow
boto3
opencv-python-headless
numpy
# torch (usually installed with ComfyUI)
```
## 🔧 How to Use
1. Clone or copy this into your ComfyUI/custom_nodes/Comfy_ForEach/ directory.
2. Launch ComfyUI. The nodes will appear under categories like:
- ComfyForEach/Load
- ComfyForEach/Select
- ComfyForEach/Save
- ComfyForEach/AWS
- ComfyForEach/TaskID
```bash
python3 main.py --use-split-cross-attention --fast --input-directory ./test --output-directory ./test --verbose ERROR
```
3. Build workflows that iterate over image folders and process each image index-by-index.
## 📌 Example Use Case
A loop-style batch processor that:
- Loads all images from a task-specific folder.
- Selects images by index.
- Saves outputs with structured filenames.
- Triggers an event only when the last image is processed.
Used in distributed processing systems where each image may be handled independently but result aggregation is done based on a `task_id` .
## 🛠 Maintainer Notes
- Uses `context.py` to globally store and retrieve task_id across node executions.
- EventBridge simulation can be replaced with real AWS API (boto3.client("events")) as needed.
__init__.py
+26
View File
@@ -0,0 +1,26 @@
from .loader_nodes import FolderImageLoaderNode
from .index_selector_nodes import IndexedImageSelectorNode, IndexedNameWithPrefixNode
from .save_nodes import SaveExactNameImageNode
from .logic_nodes import IsLastIndexNode
from .aws_event_node import EventBridgeTriggerNode
from .task_manager import TaskIDStorageNode
NODE_CLASS_MAPPINGS = {
"FolderImageLoaderNode": FolderImageLoaderNode,
"IndexedImageSelectorNode": IndexedImageSelectorNode,
"IndexedNameWithPrefixNode": IndexedNameWithPrefixNode,
"SaveExactNameImageNode": SaveExactNameImageNode,
"IsLastIndexNode": IsLastIndexNode,
"EventBridgeTriggerNode": EventBridgeTriggerNode,
"TaskIDStorageNode":TaskIDStorageNode,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"FolderImageLoaderNode": "Load Images from Folder",
"IndexedImageSelectorNode": "Select Image by Index",
"IndexedNameWithPrefixNode": "Get Filename by Index + Prefix",
"SaveExactNameImageNode": "Save Image with Exact Filename",
"IsLastIndexNode": "Check If Last Index",
"EventBridgeTriggerNode": "Trigger EventBridge (Simulated)",
"TaskIDStorageNode":"Get TaskID"
}
aws_event_node.py
+65
View File
@@ -0,0 +1,65 @@
import os
import json
from PIL import Image, ImageOps
import numpy as np
import torch
import logging
class EventBridgeTriggerNode:
"""
Triggers an EventBridge event when 'is_last' is True.
Writes event JSON to a file instead of sending via boto3.
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"is_last": ("BOOLEAN",),
"task_id": ("STRING", {"default": "unknown-task"}),
"log_folder": ("STRING", {"default": "./event_logs"})
}
}
RETURN_TYPES = ()
FUNCTION = "trigger"
OUTPUT_NODE = True
CATEGORY = "ComfyForEach/AWS"
def trigger(self, is_last, task_id, log_folder):
# ignore if not last
if not is_last:
print("[EventBridgeTriggerNode] Not the last index, skipping.")
return {}
# ensure log folder exists
os.makedirs(log_folder, exist_ok=True)
is_error = False
# create event payload
if is_error:
event = {
"task_id": task_id,
"status": "FAILED",
"reason": "Workflow execution error"
}
filename = f"{task_id}_failed.json"
else:
event = {
"task_id": task_id,
"status": "SUCCESS"
}
filename = f"{task_id}_success.json"
# Save to file
print(os.getcwd())
save_path = os.path.join(log_folder, filename)
with open(save_path, "w", encoding="utf-8") as f:
json.dump(event, f, indent=2)
print(f"[EventBridgeTriggerNode] Logged event: {save_path}")
# If using boto3:
# import boto3
# client = boto3.client("events")
# client.put_events(...)
return {}
context.py
+10
View File
@@ -0,0 +1,10 @@
# comfy_foreach/context.py
CURRENT_TASK_ID = "unknown"
def set_task_id(tid: str):
global CURRENT_TASK_ID
CURRENT_TASK_ID = tid
def get_task_id() -> str:
return CURRENT_TASK_ID
index_selector_nodes.py
+48
View File
@@ -0,0 +1,48 @@
class IndexedImageSelectorNode:
"""
Selects one image from image_list using the given index.
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"image_list": ("IMAGE",),
"index": ("INT", {"default": 0, "min": 0})
}
}
RETURN_TYPES = ("IMAGE",)
RETURN_NAMES = ("image",)
FUNCTION = "execute"
CATEGORY = "ComfyForEach/Select"
def execute(self, image_list, index):
if index < 0 or index >= len(image_list):
raise IndexError(f"Index {index} is out of bounds for image list of size {len(image_list)}")
return (image_list[index],)
class IndexedNameWithPrefixNode:
"""
Selects one image name using the index and adds a prefix to create a filename.
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"image_name_list": ("STRING",),
"index": ("INT", {"default": 0, "min": 0}),
"prefix": ("STRING", {"default": ""})
}
}
RETURN_TYPES = ("STRING","STRING")
RETURN_NAMES = ("file_name","folder_path")
FUNCTION = "execute"
CATEGORY = "ComfyForEach/Select"
def execute(self, image_name_list, index, prefix):
if index < 0 or index >= len(image_name_list):
raise IndexError(f"Index {index} is out of bounds for name list of size {len(image_name_list)}")
filename = str(image_name_list[index])
return (filename,prefix)
loader_nodes.py
+53
View File
@@ -0,0 +1,53 @@
from PIL import Image, ImageOps
import os
import numpy as np
import torch
class FolderImageLoaderNode:
"""
Loads all .png images from a folder and returns a list of images and their filenames.
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"taskid": ("STRING", {
"multiline": False,
"default": "abc123" # default path (can be changed in UI)
}),
"folder_path": ("STRING", {
"multiline": False,
"default": "./images" # default path (can be changed in UI)
})
}
}
RETURN_TYPES = ("IMAGE", "STRING")
RETURN_NAMES = ("image_list", "image_name_list")
FUNCTION = "execute"
CATEGORY = "ComfyForEach/Load"
def execute(self, taskid,folder_path):
# Get TaskID 노드를 가장 먼저 실행하기 위함 Dummy 호출
# Get list of .png files in the folder
if not os.path.isdir(folder_path):
raise ValueError(f"Folder not found: {folder_path}")
files = sorted(f for f in os.listdir(folder_path) if f.lower().endswith(".png"))
image_list = []
name_list = []
for fname in files:
file_path = os.path.join(folder_path, fname)
# Load image using PIL
img = Image.open(file_path)
img = ImageOps.exif_transpose(img) # handle orientation via EXIF&#8203;:contentReference[oaicite:4]{index=4}
# If image is 16-bit or has mode 'I', convert to 8-bit range
if img.mode == 'I':
img = img.point(lambda i: i * (1/255))
img = img.convert("RGB")
# Convert to numpy array (H, W, C) and normalize to [0,1]
np_img = np.array(img).astype(np.float32) / 255.0
# Convert to torch tensor and add batch dimension
tensor_img = torch.from_numpy(np_img) # shape (H, W, 3)
tensor_img = tensor_img.unsqueeze(0) # shape (1, H, W, 3)
image_list.append(tensor_img)
name_list.append(fname)
return (image_list, name_list)
logic_nodes.py
+20
View File
@@ -0,0 +1,20 @@
class IsLastIndexNode:
"""
Returns True if current index is the last index in total.
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"index": ("INT", {"default": 0, "min": 0}),
"total": ("INT", {"default": 1, "min": 1})
}
}
RETURN_TYPES = ("BOOLEAN",)
RETURN_NAMES = ("is_last",)
FUNCTION = "check"
CATEGORY = "ComfyForEach/Logic"
def check(self, index, total):
return (index == (total - 1),)
requirements.txt
+5
View File
@@ -0,0 +1,5 @@
pillow
boto3
opencv-python-headless
numpy
# torch
save_nodes.py
+38
View File
@@ -0,0 +1,38 @@
import os
from PIL import Image
import numpy as np
class SaveExactNameImageNode:
"""
Saves image to disk using exact user-provided filename.
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"image": ("IMAGE",),
"filename": ("STRING", {"default": "output.png"}),
"folder_path": ("STRING", {"default": "./output"}),
}
}
RETURN_TYPES = ()
FUNCTION = "save"
OUTPUT_NODE = True
CATEGORY = "ComfyForEach/Save"
def save(self, image, filename, folder_path):
# ensure folder exists
os.makedirs(folder_path, exist_ok=True)
# get image tensor (B,H,W,C)
img = image[0].cpu().numpy() # [H, W, C]
img = (img * 255).clip(0, 255).astype(np.uint8)
img = Image.fromarray(img)
# save
full_path = os.path.join(folder_path, filename)
img.save(full_path)
print(f"[Saved] → {full_path}")
return {}
task_manager.py
+22
View File
@@ -0,0 +1,22 @@
import custom_nodes.Comfy_ForEach.context as context
class TaskIDStorageNode:
"""
Stores and outputs a task_id string, to be used throughout the workflow.
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"task_id": ("STRING", {"default": "task-1234"})
}
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("task_id",)
FUNCTION = "output"
CATEGORY = "ComfyForEach/TaskID"
def output(self, task_id):
context.set_task_id(task_id)
return (task_id,)