init
This commit is contained in:
Yash Gohel
@@ -0,0 +1,92 @@
|
||||
# ComfyUI Roop Custom Nodes
|
||||
|
||||
Custom nodes for ComfyUI to enable face swapping using the Roop library.
|
||||
|
||||
Features
|
||||
|
||||
- Face swap between two images
|
||||
- Optional face enhancement
|
||||
- Batch image folder processing
|
||||
- Face swap for videos
|
||||
- Webhook sender for images (tensor) or file paths
|
||||
- Auto-increment filenames to prevent overwriting
|
||||
|
||||
## Node Descriptions
|
||||
|
||||
RoopFaceSwap
|
||||
- Basic face swap using source and target image tensors
|
||||
- Returns a single swapped image tensor
|
||||
|
||||
RoopFaceSwapWithEnhancer
|
||||
- Like RoopFaceSwap but also runs a face enhancer after swapping
|
||||
- Useful for improving visual quality
|
||||
|
||||
RoopBatchFaceSwap
|
||||
- Takes a source image and a folder of input images
|
||||
- Applies face swap to each and saves results to an output folder
|
||||
- Supports enhancement and multi-face mode
|
||||
|
||||
RoopFaceSwapVideo
|
||||
- Swaps a face from an image into a video file
|
||||
- Keeps original FPS and frames
|
||||
- Saves a new video file
|
||||
|
||||
RoopSendWebhookImage
|
||||
- Sends an image tensor to a webhook URL
|
||||
- Accepts optional secret for HMAC verification
|
||||
- Ideal for live image results
|
||||
|
||||
RoopSendWebhookFile
|
||||
- Sends a file path (e.g. video or saved image) to a webhook
|
||||
- Supports HMAC-secured sending
|
||||
- Useful for post-processing delivery
|
||||
|
||||
## Requirements
|
||||
|
||||
- Python 3.8+
|
||||
- torch
|
||||
- torchvision
|
||||
- Pillow (PIL)
|
||||
- requests
|
||||
- roop (clone from GitHub)
|
||||
|
||||
## Installation
|
||||
|
||||
1. Clone this repo into your ComfyUI custom nodes folder:
|
||||
|
||||
git clone https://github.com/glitchinthemetrix16/ComfyUI-Roop
|
||||
|
||||
2. Install required Python packages:
|
||||
|
||||
pip install torch torchvision pillow requests
|
||||
|
||||
3. Clone the roop repo for Google Colab Environment:
|
||||
|
||||
git clone https://github.com/FurkanGozukara/roop /content/roop
|
||||
|
||||
4. Ensure that run.py exists in /content/roop
|
||||
|
||||
## Installation For Google Colab With T4 GPU (Tested and working)
|
||||
|
||||
```
|
||||
%cd /content/
|
||||
|
||||
!git clone https://github.com/FurkanGozukara/roop
|
||||
%cd roop
|
||||
|
||||
#Tested and updated 23 August 2023 commit
|
||||
#!git checkout da1ef285f1d43bd0cc8b9cdb9a0f80f7ae793a97
|
||||
!pip install onnxruntime-gpu && pip install -r requirements.txt
|
||||
!pip install onnxruntime-gpu --upgrade
|
||||
!apt-get update --yes
|
||||
!apt install nvidia-cuda-toolkit --yes
|
||||
!pip install opennsfw2 keras --upgrade
|
||||
!pip install numpy==1.26.4
|
||||
```
|
||||
|
||||
Usage Notes
|
||||
|
||||
- All temporary files are saved inside roop_dir/temp_io
|
||||
- File overwrites are avoided via auto-renaming
|
||||
- Video swapping works best with shorter clips due to performance
|
||||
- Webhook nodes allow sending output to remote servers or APIs
|
||||
@@ -0,0 +1,3 @@
|
||||
from .roop_custom_nodes import NODE_CLASS_MAPPINGS, NODE_DISPLAY_NAME_MAPPINGS
|
||||
|
||||
__all__ = ['NODE_CLASS_MAPPINGS', 'NODE_DISPLAY_NAME_MAPPINGS']
|
||||
@@ -0,0 +1,401 @@
|
||||
# roop_custom_nodes.py
|
||||
import os
|
||||
import subprocess
|
||||
import glob
|
||||
from PIL import Image
|
||||
import numpy as np
|
||||
import torch
|
||||
import hmac
|
||||
import hashlib
|
||||
import requests
|
||||
|
||||
# Helper: save tensor image
|
||||
def save_tensor_as_image(tensor, path):
|
||||
np_img = (tensor[0].cpu().numpy() * 255).astype(np.uint8)
|
||||
if np_img.shape[0] == 3: # If format is CHW
|
||||
np_img = np.transpose(np_img, (1, 2, 0)) # Convert to HWC for PIL
|
||||
img = Image.fromarray(np_img)
|
||||
img.save(path)
|
||||
|
||||
# Helper: load image as tensor
|
||||
def load_image_as_tensor(path):
|
||||
img = Image.open(path).convert("RGB")
|
||||
np_img = np.array(img).astype(np.float32) / 255.0
|
||||
return torch.from_numpy(np_img).unsqueeze(0)
|
||||
|
||||
def get_unique_filename(path):
|
||||
base, ext = os.path.splitext(path)
|
||||
counter = 1
|
||||
new_path = path
|
||||
while os.path.exists(new_path):
|
||||
new_path = f"{base}_{counter}{ext}"
|
||||
counter += 1
|
||||
return new_path
|
||||
|
||||
def send_webhook_image(webhook_url, webhook_secret, output_path, extra_data=None):
|
||||
if not webhook_url:
|
||||
return
|
||||
|
||||
with open(output_path, 'rb') as f:
|
||||
image_data = f.read()
|
||||
|
||||
# GitHub-style HMAC signature
|
||||
signature = 'sha256=' + hmac.new(
|
||||
webhook_secret.encode('utf-8'),
|
||||
image_data,
|
||||
hashlib.sha256
|
||||
).hexdigest()
|
||||
|
||||
headers = {
|
||||
'X-Hub-Signature-256': signature
|
||||
}
|
||||
|
||||
files = {
|
||||
'file': ('swapped.png', image_data, 'image/png')
|
||||
}
|
||||
|
||||
data = extra_data or {}
|
||||
|
||||
try:
|
||||
resp = requests.post(webhook_url, headers=headers, files=files, data=data)
|
||||
resp.raise_for_status()
|
||||
print(f"[Webhook] Sent successfully to {webhook_url}")
|
||||
except Exception as e:
|
||||
print(f"[Webhook Error] {e}")
|
||||
|
||||
|
||||
class RoopFaceSwap:
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return {
|
||||
"required": {
|
||||
"source_image": ("IMAGE",),
|
||||
"target_image": ("IMAGE",),
|
||||
"roop_dir": ("STRING", {"default": "/content/roop"}),
|
||||
"output_name": ("STRING", {"default": "roop_output.png"}),
|
||||
"many_faces": ("BOOLEAN", {"default": False})
|
||||
}
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("IMAGE",)
|
||||
RETURN_NAMES = ("swapped_image",)
|
||||
FUNCTION = "run"
|
||||
CATEGORY = "Roop/Basic"
|
||||
|
||||
def run(self, source_image, target_image, roop_dir, output_name, many_faces):
|
||||
temp_dir = os.path.join(roop_dir, "temp_io")
|
||||
os.makedirs(temp_dir, exist_ok=True)
|
||||
|
||||
source_path = os.path.join(temp_dir, "source.png")
|
||||
target_path = os.path.join(temp_dir, "target.png")
|
||||
output_path = get_unique_filename(os.path.join(temp_dir, output_name))
|
||||
|
||||
save_tensor_as_image(source_image, source_path)
|
||||
save_tensor_as_image(target_image, target_path)
|
||||
|
||||
cmd = [
|
||||
"python", "run.py",
|
||||
"-s", source_path,
|
||||
"-t", target_path,
|
||||
"-o", output_path,
|
||||
"--execution-provider", "cuda",
|
||||
"--frame-processor", "face_swapper"
|
||||
]
|
||||
if many_faces:
|
||||
cmd.append("--many-faces")
|
||||
|
||||
subprocess.run(cmd, check=True, cwd=roop_dir)
|
||||
|
||||
if not os.path.exists(output_path):
|
||||
print(f"[Warning] Roop did not produce output: {output_path}")
|
||||
blank = torch.zeros_like(target_image)
|
||||
return (blank,)
|
||||
|
||||
return (load_image_as_tensor(output_path),)
|
||||
|
||||
|
||||
class RoopFaceSwapWithEnhancer:
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return {
|
||||
"required": {
|
||||
"source_image": ("IMAGE",),
|
||||
"target_image": ("IMAGE",),
|
||||
"roop_dir": ("STRING", {"default": "/content/roop"}),
|
||||
"output_name": ("STRING", {"default": "roop_output.png"}),
|
||||
"many_faces": ("BOOLEAN", {"default": False})
|
||||
}
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("IMAGE",)
|
||||
RETURN_NAMES = ("swapped_image",)
|
||||
FUNCTION = "run"
|
||||
CATEGORY = "Roop/Enhanced"
|
||||
|
||||
def run(self, source_image, target_image, roop_dir, output_name, many_faces):
|
||||
temp_dir = os.path.join(roop_dir, "temp_io")
|
||||
os.makedirs(temp_dir, exist_ok=True)
|
||||
|
||||
source_path = os.path.join(temp_dir, "source.png")
|
||||
target_path = os.path.join(temp_dir, "target.png")
|
||||
output_path = get_unique_filename(os.path.join(temp_dir, output_name))
|
||||
|
||||
save_tensor_as_image(source_image, source_path)
|
||||
save_tensor_as_image(target_image, target_path)
|
||||
|
||||
cmd = [
|
||||
"python", "run.py",
|
||||
"-s", source_path,
|
||||
"-t", target_path,
|
||||
"-o", output_path,
|
||||
"--execution-provider", "cuda",
|
||||
"--frame-processor", "face_swapper", "face_enhancer"
|
||||
]
|
||||
if many_faces:
|
||||
cmd.append("--many-faces")
|
||||
|
||||
subprocess.run(cmd, check=True, cwd=roop_dir)
|
||||
|
||||
if not os.path.exists(output_path):
|
||||
print(f"[Warning] Roop did not produce output: {output_path}")
|
||||
blank = torch.zeros_like(target_image)
|
||||
return (blank,)
|
||||
|
||||
return (load_image_as_tensor(output_path),)
|
||||
|
||||
|
||||
class RoopBatchFaceSwap:
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return {
|
||||
"required": {
|
||||
"source_image": ("IMAGE",),
|
||||
"input_dir": ("STRING", {"default": "/input/images"}),
|
||||
"output_dir": ("STRING", {"default": "/output/images"}),
|
||||
"roop_dir": ("STRING", {"default": "/content/roop"}),
|
||||
"use_enhancer": ("BOOLEAN", {"default": False}),
|
||||
"many_faces": ("BOOLEAN", {"default": False})
|
||||
}
|
||||
}
|
||||
|
||||
RETURN_TYPES = ()
|
||||
RETURN_NAMES = ()
|
||||
FUNCTION = "run"
|
||||
CATEGORY = "Roop/Batch"
|
||||
|
||||
def run(self, source_image, input_dir, output_dir, roop_dir, use_enhancer, many_faces):
|
||||
os.makedirs(output_dir, exist_ok=True)
|
||||
temp_dir = os.path.join(roop_dir, "temp_io")
|
||||
os.makedirs(temp_dir, exist_ok=True)
|
||||
|
||||
source_path = os.path.join(temp_dir, "source.png")
|
||||
save_tensor_as_image(source_image, source_path)
|
||||
|
||||
image_paths = glob.glob(os.path.join(input_dir, "*.jpg")) + \
|
||||
glob.glob(os.path.join(input_dir, "*.png"))
|
||||
|
||||
for img_path in image_paths:
|
||||
target_name = os.path.basename(img_path)
|
||||
target_path = os.path.join(temp_dir, "target.png")
|
||||
output_path = os.path.join(output_dir, f"out_{target_name}")
|
||||
|
||||
Image.open(img_path).save(target_path)
|
||||
|
||||
cmd = [
|
||||
"python", "run.py",
|
||||
"-s", source_path,
|
||||
"-t", target_path,
|
||||
"-o", output_path,
|
||||
"--execution-provider", "cuda",
|
||||
"--frame-processor", "face_swapper"
|
||||
]
|
||||
|
||||
if use_enhancer:
|
||||
cmd[-1] += " face_enhancer"
|
||||
if many_faces:
|
||||
cmd.append("--many-faces")
|
||||
|
||||
subprocess.run(cmd, check=True, cwd=roop_dir)
|
||||
|
||||
if not os.path.exists(output_path):
|
||||
print(f"[Skipped] NSFW or error: {img_path} -> No output generated.")
|
||||
continue
|
||||
|
||||
return ()
|
||||
|
||||
|
||||
class RoopFaceSwapVideo:
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return {
|
||||
"required": {
|
||||
"source_image": ("IMAGE",),
|
||||
"target_video_path": ("STRING", {"default": "/path/to/video.mp4"}),
|
||||
"roop_dir": ("STRING", {"default": "/content/roop"}),
|
||||
"output_name": ("STRING", {"default": "swapped_video.mp4"}),
|
||||
"many_faces": ("BOOLEAN", {"default": False})
|
||||
}
|
||||
}
|
||||
|
||||
RETURN_TYPES = ()
|
||||
RETURN_NAMES = ()
|
||||
FUNCTION = "run"
|
||||
CATEGORY = "Roop/Video"
|
||||
|
||||
def run(self, source_image, target_video_path, roop_dir, output_name, many_faces):
|
||||
temp_dir = os.path.join(roop_dir, "temp_io")
|
||||
os.makedirs(temp_dir, exist_ok=True)
|
||||
|
||||
source_path = os.path.join(temp_dir, "source.png")
|
||||
output_path = get_unique_filename(os.path.join(temp_dir, output_name))
|
||||
save_tensor_as_image(source_image, source_path)
|
||||
|
||||
cmd = [
|
||||
"python", "run.py",
|
||||
"-s", source_path,
|
||||
"-t", target_video_path,
|
||||
"-o", output_path,
|
||||
"--keep-fps",
|
||||
"--keep-frames",
|
||||
"--execution-provider", "cuda",
|
||||
"--frame-processor", "face_swapper"
|
||||
]
|
||||
if many_faces:
|
||||
cmd.append("--many-faces")
|
||||
|
||||
subprocess.run(cmd, check=True, cwd=roop_dir)
|
||||
|
||||
if not os.path.exists(output_path):
|
||||
print(f"[Warning] Roop did not produce video output: {output_path}")
|
||||
|
||||
return ()
|
||||
|
||||
|
||||
class RoopSendWebhookImage:
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return {
|
||||
"required": {
|
||||
"image_tensor": ("IMAGE",),
|
||||
"filename": ("STRING", {"default": "output.png"}),
|
||||
"webhook_url": ("STRING", {"default": ""}),
|
||||
"webhook_secret": ("STRING", {"default": ""}),
|
||||
"enable_webhook": ("BOOLEAN", {"default": True}),
|
||||
"roop_dir": ("STRING", {"default": "/content/roop"})
|
||||
}
|
||||
}
|
||||
|
||||
RETURN_TYPES = ()
|
||||
RETURN_NAMES = ()
|
||||
FUNCTION = "run"
|
||||
CATEGORY = "Roop/Webhook"
|
||||
|
||||
def run(self, image_tensor, filename, webhook_url, webhook_secret, enable_webhook, roop_dir):
|
||||
if not enable_webhook or not webhook_url:
|
||||
print("[WebhookImage] Disabled or URL not set — skipping.")
|
||||
return ()
|
||||
|
||||
temp_dir = os.path.join(roop_dir, "temp_io")
|
||||
os.makedirs(temp_dir, exist_ok=True)
|
||||
output_path = os.path.join(temp_dir, filename)
|
||||
|
||||
save_tensor_as_image(image_tensor, output_path)
|
||||
|
||||
try:
|
||||
with open(output_path, 'rb') as f:
|
||||
file_data = f.read()
|
||||
|
||||
headers = {}
|
||||
if webhook_secret:
|
||||
signature = 'sha256=' + hmac.new(
|
||||
webhook_secret.encode('utf-8'),
|
||||
file_data,
|
||||
hashlib.sha256
|
||||
).hexdigest()
|
||||
headers['X-Hub-Signature-256'] = signature
|
||||
|
||||
files = {
|
||||
'file': (filename, file_data, 'image/png')
|
||||
}
|
||||
|
||||
resp = requests.post(webhook_url, headers=headers, files=files)
|
||||
resp.raise_for_status()
|
||||
print(f"[WebhookImage] Sent image: {filename} → {webhook_url}")
|
||||
except Exception as e:
|
||||
print(f"[WebhookImage Error] {e}")
|
||||
|
||||
return ()
|
||||
|
||||
|
||||
class RoopSendWebhookFile:
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return {
|
||||
"required": {
|
||||
"file_path": ("STRING",),
|
||||
"filename": ("STRING", {"default": "output.mp4"}),
|
||||
"webhook_url": ("STRING", {"default": ""}),
|
||||
"webhook_secret": ("STRING", {"default": ""}),
|
||||
"enable_webhook": ("BOOLEAN", {"default": True})
|
||||
}
|
||||
}
|
||||
|
||||
RETURN_TYPES = ()
|
||||
RETURN_NAMES = ()
|
||||
FUNCTION = "run"
|
||||
CATEGORY = "Roop/Webhook"
|
||||
|
||||
def run(self, file_path, filename, webhook_url, webhook_secret, enable_webhook):
|
||||
if not enable_webhook or not webhook_url:
|
||||
print("[WebhookFile] Disabled or URL not set — skipping.")
|
||||
return ()
|
||||
|
||||
if not os.path.exists(file_path):
|
||||
print(f"[WebhookFile] File does not exist: {file_path}")
|
||||
return ()
|
||||
|
||||
try:
|
||||
with open(file_path, 'rb') as f:
|
||||
file_data = f.read()
|
||||
|
||||
headers = {}
|
||||
if webhook_secret:
|
||||
signature = 'sha256=' + hmac.new(
|
||||
webhook_secret.encode('utf-8'),
|
||||
file_data,
|
||||
hashlib.sha256
|
||||
).hexdigest()
|
||||
headers['X-Hub-Signature-256'] = signature
|
||||
|
||||
content_type = "video/mp4" if filename.endswith(".mp4") else "application/octet-stream"
|
||||
files = {
|
||||
'file': (filename, file_data, content_type)
|
||||
}
|
||||
|
||||
resp = requests.post(webhook_url, headers=headers, files=files)
|
||||
resp.raise_for_status()
|
||||
print(f"[WebhookFile] Sent file: {filename} → {webhook_url}")
|
||||
except Exception as e:
|
||||
print(f"[WebhookFile Error] {e}")
|
||||
|
||||
return ()
|
||||
|
||||
|
||||
# Register with ComfyUI
|
||||
NODE_CLASS_MAPPINGS = {
|
||||
"RoopFaceSwap": RoopFaceSwap,
|
||||
"RoopFaceSwapWithEnhancer": RoopFaceSwapWithEnhancer,
|
||||
"RoopBatchFaceSwap": RoopBatchFaceSwap,
|
||||
"RoopFaceSwapVideo": RoopFaceSwapVideo,
|
||||
"RoopSendWebhookImage": RoopSendWebhookImage,
|
||||
"RoopSendWebhookFile": RoopSendWebhookFile,
|
||||
}
|
||||
|
||||
NODE_DISPLAY_NAME_MAPPINGS = {
|
||||
"RoopFaceSwap": "Roop Face Swap (Image)",
|
||||
"RoopFaceSwapWithEnhancer": "Roop Face Swap + Enhancer",
|
||||
"RoopBatchFaceSwap": "Roop Batch Image Folder",
|
||||
"RoopFaceSwapVideo": "Roop Face Swap (Video)",
|
||||
"RoopSendWebhookImage": "Roop Webhook: Image Tensor",
|
||||
"RoopSendWebhookFile": "Roop Webhook: File Path",
|
||||
}
|
||||
Reference in New Issue
Block a user