From 9af2c7e0a25a07ea0f4b44280928fee20beb88e4 Mon Sep 17 00:00:00 2001 From: Yash Gohel <18393798+Ygohel18@users.noreply.github.com> Date: Thu, 10 Jul 2025 20:20:23 +0530 Subject: [PATCH] init --- README.md | 92 ++++++++++ __init__.py | 3 + roop_custom_nodes.py | 401 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 496 insertions(+) create mode 100644 README.md create mode 100644 __init__.py create mode 100644 roop_custom_nodes.py diff --git a/README.md b/README.md new file mode 100644 index 0000000..89a3d0a --- /dev/null +++ b/README.md @@ -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 \ No newline at end of file diff --git a/__init__.py b/__init__.py new file mode 100644 index 0000000..174adfb --- /dev/null +++ b/__init__.py @@ -0,0 +1,3 @@ +from .roop_custom_nodes import NODE_CLASS_MAPPINGS, NODE_DISPLAY_NAME_MAPPINGS + +__all__ = ['NODE_CLASS_MAPPINGS', 'NODE_DISPLAY_NAME_MAPPINGS'] \ No newline at end of file diff --git a/roop_custom_nodes.py b/roop_custom_nodes.py new file mode 100644 index 0000000..56b66e5 --- /dev/null +++ b/roop_custom_nodes.py @@ -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", +}