From 12365cd7925fc1a950bb94bb87f15bfc60afc064 Mon Sep 17 00:00:00 2001 From: Pupbani Date: Tue, 22 Apr 2025 11:01:58 +0900 Subject: [PATCH] v0.1.0 --- .gitignore | 2 + README.md | 240 ++++++++++++++++++++++++++++++++++++++++ __init__.py | 26 +++++ aws_event_node.py | 65 +++++++++++ context.py | 10 ++ index_selector_nodes.py | 48 ++++++++ loader_nodes.py | 53 +++++++++ logic_nodes.py | 20 ++++ requirements.txt | 5 + save_nodes.py | 38 +++++++ task_manager.py | 22 ++++ 11 files changed, 529 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 __init__.py create mode 100644 aws_event_node.py create mode 100644 context.py create mode 100644 index_selector_nodes.py create mode 100644 loader_nodes.py create mode 100644 logic_nodes.py create mode 100644 requirements.txt create mode 100644 save_nodes.py create mode 100644 task_manager.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..46c9745 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +__pycache__/ +.DS_Store \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..bbaa504 --- /dev/null +++ b/README.md @@ -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. diff --git a/__init__.py b/__init__.py new file mode 100644 index 0000000..a387582 --- /dev/null +++ b/__init__.py @@ -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" +} diff --git a/aws_event_node.py b/aws_event_node.py new file mode 100644 index 0000000..1a14db6 --- /dev/null +++ b/aws_event_node.py @@ -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 {} \ No newline at end of file diff --git a/context.py b/context.py new file mode 100644 index 0000000..ffa4cd0 --- /dev/null +++ b/context.py @@ -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 diff --git a/index_selector_nodes.py b/index_selector_nodes.py new file mode 100644 index 0000000..85f5a45 --- /dev/null +++ b/index_selector_nodes.py @@ -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) diff --git a/loader_nodes.py b/loader_nodes.py new file mode 100644 index 0000000..374ea02 --- /dev/null +++ b/loader_nodes.py @@ -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​: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) \ No newline at end of file diff --git a/logic_nodes.py b/logic_nodes.py new file mode 100644 index 0000000..87ee689 --- /dev/null +++ b/logic_nodes.py @@ -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),) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..1030207 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +pillow +boto3 +opencv-python-headless +numpy +# torch \ No newline at end of file diff --git a/save_nodes.py b/save_nodes.py new file mode 100644 index 0000000..ed981b0 --- /dev/null +++ b/save_nodes.py @@ -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 {} + \ No newline at end of file diff --git a/task_manager.py b/task_manager.py new file mode 100644 index 0000000..fa48ec7 --- /dev/null +++ b/task_manager.py @@ -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,) \ No newline at end of file