This commit is contained in:
Ubuntu
2025-12-03 10:23:07 +00:00
parent a5a88e2dbd
commit 6a94e8b91b
8 changed files with 617 additions and 1 deletions
+19 -1
View File
@@ -26,7 +26,13 @@ from .nodes import (
ShotByTextManualPlacementNode,
ShotByTextManualPaddingNode,
ShotByTextCustomCoordinatesNode,
AttributionByImageNode
AttributionByImageNode,
RemoveVideoBackgroundNode,
VideoSolidColorBackgroundNode,
VideoMaskByPromptNode,
VideoMaskByKeyPointsNode,
VideoIncreaseResolutionNode,
VideoEraseElementsNode,
)
# Map the node class to a name used internally by ComfyUI
@@ -59,6 +65,12 @@ NODE_CLASS_MAPPINGS = {
"AttributionByImageNode": AttributionByImageNode,
"GenerateImageNodeV2": GenerateImageNodeV2,
"RefineImageNodeV2": RefineImageNodeV2,
"RemoveVideoBackgroundNode":RemoveVideoBackgroundNode,
"VideoSolidColorBackgroundNode":VideoSolidColorBackgroundNode,
"VideoMaskByPromptNode":VideoMaskByPromptNode,
"VideoMaskByKeyPointsNode":VideoMaskByKeyPointsNode,
"VideoIncreaseResolutionNode":VideoIncreaseResolutionNode,
"VideoEraseElementsNode":VideoEraseElementsNode,
}
# Map the node display name to the one shown in the ComfyUI node interface
NODE_DISPLAY_NAME_MAPPINGS = {
@@ -90,4 +102,10 @@ NODE_DISPLAY_NAME_MAPPINGS = {
"AttributionByImageNode": "Attribution By Image Node",
"GenerateImageNodeV2": "Generate Image",
"RefineImageNodeV2": "Refine and Regenerate Image",
"RemoveVideoBackgroundNode": "Bria Remove Video Background",
"VideoSolidColorBackgroundNode":"Bria SolidColor Background Video",
"VideoMaskByPromptNode":"Bria Video Mask By Prompt",
"VideoMaskByKeyPointsNode":"Bria Video Mask By Key Points",
"VideoIncreaseResolutionNode":"Bria Video Increase Resolution",
"VideoEraseElementsNode":"Bria Video Erase Elements",
}
+6
View File
@@ -28,3 +28,9 @@ from .shot_by_image_node import ShotByImageOriginalNode
from .shot_by_image_manual_placement_node import ShotByImageManualPlacementNode
from .shot_by_image_manual_padding_node import ShotByImageManualPaddingNode
from .attribution_by_image_node import AttributionByImageNode
from .video_nodes.remove_video_background_node import RemoveVideoBackgroundNode
from .video_nodes.video_increase_resolution_node import VideoIncreaseResolutionNode
from .video_nodes.video_solid_color_background_node import VideoSolidColorBackgroundNode
from .video_nodes.video_erase_elements_node import VideoEraseElementsNode
from .video_nodes.video_mask_by_prompt_node import VideoMaskByPromptNode
from .video_nodes.video_mask_by_key_points_node import VideoMaskByKeyPointsNode
@@ -0,0 +1,97 @@
import requests
from ..common import deserialize_and_get_comfy_key, poll_status_until_completed
class RemoveVideoBackgroundNode():
"""
Bria Remove Video Background Node
This node removes the background from videos using the Bria API.
It accepts a publicly accessible video URL and returns a processed video URL
with the background removed.
Supported input resolution: up to 16000x16000 (16K)
Parameters:
- video_url: Publicly accessible URL of the input video
- api_key: Your Bria API token
- output_container_and_codec: Output video format and codec (default: webm_vp9)
"""
@classmethod
def INPUT_TYPES(self):
return {
"required": {
"video_url": ("STRING", {"default": ""}),
"api_key": ("STRING", {"default": "BRIA_API_TOKEN"}),
},
"optional": {
"preserve_audio": ("BOOLEAN", {"default":True}),
"output_container_and_codec": ([
"mp4_h264",
"mp4_h265",
"webm_vp9",
"mov_h265",
"mov_proresks",
"mkv_h264",
"mkv_h265",
"mkv_vp9",
"gif"
], {"default": "webm_vp9"}),
}
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("video_url_response",)
CATEGORY = "API Nodes"
FUNCTION = "execute"
def __init__(self):
self.api_url = "https://engine.prod.bria-api.com/v2/video/edit/remove_background" # Video RMBG API URL
# Define the execute method as expected by ComfyUI
def execute(self, video_url, api_key, preserve_audio, output_container_and_codec="webm_vp9"):
if api_key.strip() == "" or api_key.strip() == "BRIA_API_TOKEN":
raise Exception("Please insert a valid API key.")
api_key = deserialize_and_get_comfy_key(api_key)
# Prepare the API request payload
payload = {
"video": video_url,
"preserve_audio": preserve_audio,
"output_container_and_codec": output_container_and_codec
}
headers = {
"Content-Type": "application/json",
"api_token": f"{api_key}"
}
try:
response = requests.post(self.api_url, json=payload, headers=headers)
if response.status_code == 200 or response.status_code == 202:
print('Initial Video RMBG request successful, polling for completion...')
response_dict = response.json()
status_url = response_dict.get('status_url')
request_id = response_dict.get('request_id')
if not status_url:
raise Exception("No status_url returned from API")
print(f"Request ID: {request_id}, Status URL: {status_url}")
final_response = poll_status_until_completed(status_url, api_key, timeout=3600, check_interval=5)
result_video_url = final_response['result']['video_url']
print(f"Video processing completed. Result URL: {result_video_url}")
return (result_video_url,)
else:
raise Exception(f"Error: API request failed with status code {response.status_code} {response.text}")
except Exception as e:
raise Exception(f"{e}")
@@ -0,0 +1,97 @@
import requests
from ..common import deserialize_and_get_comfy_key, poll_status_until_completed
class VideoEraseElementsNode():
"""
Bria Video Erase Elements Node
This node erases specific elements from videos based on a text prompt using the Bria API.
It accepts a publicly accessible video URL and a text instruction describing the object
to be masked and removed.
Parameters:
- video_url: Publicly accessible URL of the input video
- api_key: Your Bria API token
- prompt: Text instruction describing the object to be masked
- output_container_and_codec: Output video format and codec (default: mp4_h264)
- preserve_audio: Audio preservation (default: True)
"""
@classmethod
def INPUT_TYPES(self):
return {
"required": {
"video_url": ("STRING", {"default": ""}),
"mask_url": ("STRING", {"default": ""}),
"prompt": ("STRING", {"default": ""}),
"api_key": ("STRING", {"default": "BRIA_API_TOKEN"}),
},
"optional": {
"output_container_and_codec": ([
"mp4_h264",
"mp4_h265",
"webm_vp9",
"mov_h265",
"mov_proresks",
"mkv_h264",
"mkv_h265",
"mkv_vp9",
"gif"
], {"default": "mp4_h264"}),
"preserve_audio": ("BOOLEAN", {"default": True}),
}
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("video_url_response",)
CATEGORY = "API Nodes"
FUNCTION = "execute"
def __init__(self):
self.api_url = "https://engine.prod.bria-api.com/v2/video/edit/erase"
def execute(self, video_url, mask_url, prompt, api_key, output_container_and_codec="mp4_h264", preserve_audio=True):
if api_key.strip() == "" or api_key.strip() == "BRIA_API_TOKEN":
raise Exception("Please insert a valid API key.")
api_key = deserialize_and_get_comfy_key(api_key)
# Prepare the API request payload
payload = {
"video": video_url,
"mask":mask_url,
"prompt": prompt,
"output_container_and_codec": output_container_and_codec,
"preserve_audio": preserve_audio
}
headers = {
"Content-Type": "application/json",
"api_token": f"{api_key}"
}
try:
response = requests.post(self.api_url, json=payload, headers=headers)
if response.status_code == 200 or response.status_code == 202:
print('Initial Video Erase Elements request successful, polling for completion...')
response_dict = response.json()
status_url = response_dict.get('status_url')
request_id = response_dict.get('request_id')
if not status_url:
raise Exception("No status_url returned from API")
print(f"Request ID: {request_id}, Status URL: {status_url}")
final_response = poll_status_until_completed(status_url, api_key, timeout=3600, check_interval=5)
result_video_url = final_response['result']['video_url']
print(f"Video processing completed. Result URL: {result_video_url}")
return (result_video_url,)
else:
raise Exception(f"Error: API request failed with status code {response.status_code} {response.text}")
except Exception as e:
raise Exception(f"{e}")
@@ -0,0 +1,95 @@
import requests
from ..common import deserialize_and_get_comfy_key, poll_status_until_completed
class VideoIncreaseResolutionNode():
"""
Bria Video Increase Resolution Node
This node increases the resolution of videos using the Bria API.
It accepts a publicly accessible video URL and returns a processed video URL
with increased resolution.
Parameters:
- video_url: Publicly accessible URL of the input video
- api_key: Your Bria API token
- desired_increase: Integer scale factor for upscaling (2 or 4)
- output_container_and_codec: Output video format and codec (default: mp4_h264)
- preserve_audio: Audio preservation (default: True)
"""
@classmethod
def INPUT_TYPES(self):
return {
"required": {
"video_url": ("STRING", {"default": ""}),
"api_key": ("STRING", {"default": "BRIA_API_TOKEN"}),
},
"optional": {
"desired_increase": (['2', '4'], {"default": '2'}),
"output_container_and_codec": ([
"mp4_h264",
"mp4_h265",
"webm_vp9",
"mov_h265",
"mov_proresks",
"mkv_h264",
"mkv_h265",
"mkv_vp9",
"gif"
], {"default": "mp4_h264"}),
"preserve_audio": ("BOOLEAN", {"default": True}),
}
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("video_url_response",)
CATEGORY = "API Nodes"
FUNCTION = "execute"
def __init__(self):
self.api_url = "https://engine.prod.bria-api.com/v2/video/edit/increase_resolution"
def execute(self, video_url, api_key, desired_increase, output_container_and_codec="mp4_h264", preserve_audio=True):
if api_key.strip() == "" or api_key.strip() == "BRIA_API_TOKEN":
raise Exception("Please insert a valid API key.")
api_key = deserialize_and_get_comfy_key(api_key)
# Prepare the API request payload
payload = {
"video": video_url,
"desired_increase": desired_increase,
"output_container_and_codec": output_container_and_codec,
"preserve_audio": preserve_audio
}
headers = {
"Content-Type": "application/json",
"api_token": f"{api_key}"
}
try:
response = requests.post(self.api_url, json=payload, headers=headers)
if response.status_code == 200 or response.status_code == 202:
print('Initial Video Increase Resolution request successful, polling for completion...')
response_dict = response.json()
status_url = response_dict.get('status_url')
request_id = response_dict.get('request_id')
if not status_url:
raise Exception("No status_url returned from API")
print(f"Request ID: {request_id}, Status URL: {status_url}")
final_response = poll_status_until_completed(status_url, api_key, timeout=3600, check_interval=5)
result_video_url = final_response['result']['video_url']
print(f"Video processing completed. Result URL: {result_video_url}")
return (result_video_url,)
else:
raise Exception(f"Error: API request failed with status code {response.status_code} {response.text}")
except Exception as e:
raise Exception(f"{e}")
@@ -0,0 +1,101 @@
import requests
from ..common import deserialize_and_get_comfy_key, poll_status_until_completed
import json
class VideoMaskByKeyPointsNode():
"""
Bria Video Mask by Key Points Node
This node generates a mask for specific elements in videos based on key point coordinates
using the Bria API. It accepts a publicly accessible video URL and an array of coordinate
objects defining the mask hints.
Parameters:
- video_url: Publicly accessible URL of the input video
- api_key: Your Bria API token
- key_points: Array of coordinate objects defining the mask hints (JSON format)
- output_container_and_codec: Output video format and codec (default: mp4_h264)
- preserve_audio: Audio preservation (default: True)
"""
@classmethod
def INPUT_TYPES(self):
return {
"required": {
"video_url": ("STRING", {"default": ""}),
"key_points": ("STRING", {"default": "[]", "multiline": True}),
"api_key": ("STRING", {"default": "BRIA_API_TOKEN"}),
},
"optional": {
"output_container_and_codec": ([
"mp4_h264",
"mp4_h265",
"webm_vp9",
"mov_h265",
"mov_proresks",
"mkv_h264",
"mkv_h265",
"mkv_vp9",
"gif"
], {"default": "mp4_h264"}),
"preserve_audio": ("BOOLEAN", {"default": True}),
}
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("mask_url_response",)
CATEGORY = "API Nodes"
FUNCTION = "execute"
def __init__(self):
self.api_url = "https://engine.prod.bria-api.com/v2/video/segment/mask_by_key_points"
def execute(self, video_url, key_points, api_key, output_container_and_codec="mp4_h264", preserve_audio=True):
if api_key.strip() == "" or api_key.strip() == "BRIA_API_TOKEN":
raise Exception("Please insert a valid API key.")
api_key = deserialize_and_get_comfy_key(api_key)
try:
key_points_array = json.loads(key_points)
except json.JSONDecodeError as e:
raise Exception(f"Invalid JSON format for key_points: {e}")
# Prepare the API request payload
payload = {
"video": video_url,
"key_points": key_points_array,
"output_container_and_codec": output_container_and_codec,
"preserve_audio": preserve_audio
}
headers = {
"Content-Type": "application/json",
"api_token": f"{api_key}"
}
try:
response = requests.post(self.api_url, json=payload, headers=headers)
if response.status_code == 200 or response.status_code == 202:
print('Initial Video Mask by Key Points request successful, polling for completion...')
response_dict = response.json()
status_url = response_dict.get('status_url')
request_id = response_dict.get('request_id')
if not status_url:
raise Exception("No status_url returned from API")
print(f"Request ID: {request_id}, Status URL: {status_url}")
final_response = poll_status_until_completed(status_url, api_key, timeout=3600, check_interval=5)
result_mask_url = final_response['result']['mask_url']
print(f"Video mask processing completed. Result URL: {result_mask_url}")
return (result_mask_url,)
else:
raise Exception(f"Error: API request failed with status code {response.status_code} {response.text}")
except Exception as e:
raise Exception(f"{e}")
@@ -0,0 +1,95 @@
import requests
from ..common import deserialize_and_get_comfy_key, poll_status_until_completed
class VideoMaskByPromptNode():
"""
Bria Video Mask by Prompt Node
This node generates a mask for specific elements in videos based on a text prompt
using the Bria API. It accepts a publicly accessible video URL and a text instruction
describing the object to be masked.
Parameters:
- video_url: Publicly accessible URL of the input video
- api_key: Your Bria API token
- prompt: Text instruction describing the object to be masked
- output_container_and_codec: Output video format and codec (default: mp4_h264)
- preserve_audio: Audio preservation (default: True)
"""
@classmethod
def INPUT_TYPES(self):
return {
"required": {
"video_url": ("STRING", {"default": ""}),
"prompt": ("STRING", {"default": ""}),
"api_key": ("STRING", {"default": "BRIA_API_TOKEN"}),
},
"optional": {
"output_container_and_codec": ([
"mp4_h264",
"mp4_h265",
"webm_vp9",
"mov_h265",
"mov_proresks",
"mkv_h264",
"mkv_h265",
"mkv_vp9",
"gif"
], {"default": "mp4_h264"}),
"preserve_audio": ("BOOLEAN", {"default": True}),
}
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("mask_url_response",)
CATEGORY = "API Nodes"
FUNCTION = "execute"
def __init__(self):
self.api_url = "https://engine.prod.bria-api.com/v2/video/segment/mask_by_prompt"
def execute(self, video_url, prompt, api_key, output_container_and_codec="mp4_h264", preserve_audio=True):
if api_key.strip() == "" or api_key.strip() == "BRIA_API_TOKEN":
raise Exception("Please insert a valid API key.")
api_key = deserialize_and_get_comfy_key(api_key)
# Prepare the API request payload
payload = {
"video": video_url,
"prompt": prompt,
"output_container_and_codec": output_container_and_codec,
"preserve_audio": preserve_audio
}
headers = {
"Content-Type": "application/json",
"api_token": f"{api_key}"
}
try:
response = requests.post(self.api_url, json=payload, headers=headers)
if response.status_code == 200 or response.status_code == 202:
print('Initial Video Mask by Prompt request successful, polling for completion...')
response_dict = response.json()
status_url = response_dict.get('status_url')
request_id = response_dict.get('request_id')
if not status_url:
raise Exception("No status_url returned from API")
print(f"Request ID: {request_id}, Status URL: {status_url}")
final_response = poll_status_until_completed(status_url, api_key, timeout=3600, check_interval=5)
result_mask_url = final_response['result']['mask_url']
print(f"Video mask processing completed. Result URL: {result_mask_url}")
return (result_mask_url,)
else:
raise Exception(f"Error: API request failed with status code {response.status_code} {response.text}")
except Exception as e:
raise Exception(f"{e}")
@@ -0,0 +1,107 @@
import requests
from ..common import deserialize_and_get_comfy_key, poll_status_until_completed
class VideoSolidColorBackgroundNode():
"""
Bria Video Solid Color Background Node
This node removes the background from videos and replaces it with a solid color
using the Bria API. It accepts a publicly accessible video URL and returns a
processed video URL with the specified background color.
Parameters:
- video_url: Publicly accessible URL of the input video
- api_key: Your Bria API token
- background_color: Predefined color string (default: Transparent)
- output_container_and_codec: Output video format and codec (default: mp4_h264)
- preserve_audio: Audio preservation (default: True)
"""
@classmethod
def INPUT_TYPES(self):
return {
"required": {
"video_url": ("STRING", {"default": ""}),
"api_key": ("STRING", {"default": "BRIA_API_TOKEN"}),
},
"optional": {
"background_color": ([
"Transparent",
"Black",
"White",
"Gray",
"Red",
"Green",
"Blue",
"Yellow",
"Cyan",
"Magenta",
"Orange"
], {"default": "Transparent"}),
"output_container_and_codec": ([
"mp4_h264",
"mp4_h265",
"webm_vp9",
"mov_h265",
"mov_proresks",
"mkv_h264",
"mkv_h265",
"mkv_vp9",
"gif"
], {"default": "mp4_h264"}),
"preserve_audio": ("BOOLEAN", {"default": True}),
}
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("video_url_response",)
CATEGORY = "API Nodes"
FUNCTION = "execute"
def __init__(self):
self.api_url = "https://engine.prod.bria-api.com/v2/video/edit/remove_background"
def execute(self, video_url, api_key, background_color="Transparent", output_container_and_codec="mp4_h264", preserve_audio=True):
if api_key.strip() == "" or api_key.strip() == "BRIA_API_TOKEN":
raise Exception("Please insert a valid API key.")
api_key = deserialize_and_get_comfy_key(api_key)
# Prepare the API request payload
payload = {
"video": video_url,
"background_color": background_color,
"output_container_and_codec": output_container_and_codec,
"preserve_audio": preserve_audio
}
headers = {
"Content-Type": "application/json",
"api_token": f"{api_key}"
}
try:
response = requests.post(self.api_url, json=payload, headers=headers)
if response.status_code == 200 or response.status_code == 202:
print('Initial Video Solid Color Background request successful, polling for completion...')
response_dict = response.json()
status_url = response_dict.get('status_url')
request_id = response_dict.get('request_id')
if not status_url:
raise Exception("No status_url returned from API")
print(f"Request ID: {request_id}, Status URL: {status_url}")
final_response = poll_status_until_completed(status_url, api_key, timeout=3600, check_interval=5)
result_video_url = final_response['result']['video_url']
print(f"Video processing completed. Result URL: {result_video_url}")
return (result_video_url,)
else:
raise Exception(f"Error: API request failed with status code {response.status_code} {response.text}")
except Exception as e:
raise Exception(f"{e}")