263 lines
6.6 KiB
Markdown
263 lines
6.6 KiB
Markdown
# 📦 ComfyForEach: Custom ComfyUI Nodes for Batch Image Processing and Used in AWS
|
|
|
|
|
|
|
|
|
|
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
|
|
import boto3
|
|
|
|
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())
|
|
|
|
event = boto3.client("events",region_name="us-east-1")
|
|
|
|
# Changed Your Message
|
|
resp = event.put_events(
|
|
Entries=[
|
|
{
|
|
"Source":"comfyui.ec2",
|
|
"DetailType":"ComfyUI Task State",
|
|
"Detail":json.dumps({
|
|
"task_id":task_id,
|
|
"status":"FAILED",
|
|
"timestamp":datetime.utcnow().isoformat(),
|
|
"error_msg":error_msg
|
|
}),
|
|
"EventBusName":"default"
|
|
}
|
|
]
|
|
)
|
|
if resp.get("FailedEntryCount", 0) > 0:
|
|
raise RuntimeError("❌ EventBridge Send Failed")
|
|
else:
|
|
print("✅ EventBridge Send Success")
|
|
|
|
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)
|
|
|
|
### 🔹 IndexedNameSelectorNode
|
|
|
|
- Selects a specific image from a list using an index.
|
|
|
|
- **Category** : ComfyForEach/Select
|
|
|
|
- **Outputs** : file_name (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.
|