feat: add video metadata support for VHS integration
This commit is contained in:
LuqP2
+5
-1
@@ -1,22 +1,26 @@
|
||||
"""
|
||||
MetaHub Save Image - ComfyUI Custom Node
|
||||
MetaHub Save Image/Video - ComfyUI Custom Node
|
||||
Entry point for node registration
|
||||
"""
|
||||
|
||||
try:
|
||||
from .metahub_save_node import MetaHubSaveNode
|
||||
from .metahub_save_video_node import MetaHubSaveVideoNode
|
||||
from .timer_node import MetaHubTimerNode
|
||||
except ImportError:
|
||||
from metahub_save_node import MetaHubSaveNode
|
||||
from metahub_save_video_node import MetaHubSaveVideoNode
|
||||
from timer_node import MetaHubTimerNode
|
||||
|
||||
NODE_CLASS_MAPPINGS = {
|
||||
"MetaHubSaveNode": MetaHubSaveNode,
|
||||
"MetaHubSaveVideoNode": MetaHubSaveVideoNode,
|
||||
"MetaHubTimerNode": MetaHubTimerNode
|
||||
}
|
||||
|
||||
NODE_DISPLAY_NAME_MAPPINGS = {
|
||||
"MetaHubSaveNode": "MetaHub Save Image",
|
||||
"MetaHubSaveVideoNode": "MetaHub Save Video",
|
||||
"MetaHubTimerNode": "MetaHub Timer"
|
||||
}
|
||||
|
||||
|
||||
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,410 @@
|
||||
"""
|
||||
MetaHub Save Video Node for ComfyUI
|
||||
Adds metadata to video files generated by VHS (Video Helper Suite)
|
||||
"""
|
||||
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
try:
|
||||
from . import metadata_utils as utils
|
||||
from . import video_metadata_utils as video_utils
|
||||
from .workflow_extractor import WorkflowExtractor
|
||||
except ImportError:
|
||||
import metadata_utils as utils
|
||||
import video_metadata_utils as video_utils
|
||||
from workflow_extractor import WorkflowExtractor
|
||||
|
||||
|
||||
class MetaHubSaveVideoNode:
|
||||
"""
|
||||
ComfyUI custom node for adding MetaHub metadata to video files.
|
||||
|
||||
Accepts VHS_FILENAMES from VHS Video Combine node and injects metadata
|
||||
into the video container (MP4, WebM, MKV) without re-encoding.
|
||||
|
||||
Features:
|
||||
- A1111/Civitai compatible metadata (description field)
|
||||
- Video MetaHub metadata (comment field as JSON)
|
||||
- Auto-detection of workflow parameters
|
||||
- Frame rate and frame count tracking
|
||||
- Pass-through design for workflow chaining
|
||||
"""
|
||||
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return {
|
||||
"required": {
|
||||
"filenames": ("VHS_FILENAMES",),
|
||||
},
|
||||
"optional": {
|
||||
# Generation parameter overrides
|
||||
"seed": ("INT", {
|
||||
"default": None,
|
||||
"forceInput": True,
|
||||
"tooltip": "Override seed"
|
||||
}),
|
||||
"steps": ("INT", {
|
||||
"default": None,
|
||||
"forceInput": True,
|
||||
"tooltip": "Override steps"
|
||||
}),
|
||||
"cfg": ("FLOAT", {
|
||||
"default": None,
|
||||
"forceInput": True,
|
||||
"tooltip": "Override CFG scale"
|
||||
}),
|
||||
"sampler_name": ("STRING", {
|
||||
"default": None,
|
||||
"forceInput": True,
|
||||
"tooltip": "Override sampler name"
|
||||
}),
|
||||
"scheduler": ("STRING", {
|
||||
"default": None,
|
||||
"forceInput": True,
|
||||
"tooltip": "Override scheduler"
|
||||
}),
|
||||
"model_name": ("STRING", {
|
||||
"default": None,
|
||||
"forceInput": True,
|
||||
"tooltip": "Override model filename"
|
||||
}),
|
||||
"positive": ("STRING", {
|
||||
"multiline": True,
|
||||
"default": None,
|
||||
"forceInput": True,
|
||||
"tooltip": "Override positive prompt"
|
||||
}),
|
||||
"negative": ("STRING", {
|
||||
"multiline": True,
|
||||
"default": None,
|
||||
"forceInput": True,
|
||||
"tooltip": "Override negative prompt"
|
||||
}),
|
||||
"denoise": ("FLOAT", {
|
||||
"default": None,
|
||||
"forceInput": True,
|
||||
"tooltip": "Override denoise strength"
|
||||
}),
|
||||
"vae_name": ("STRING", {
|
||||
"default": None,
|
||||
"forceInput": True,
|
||||
"tooltip": "Override VAE filename"
|
||||
}),
|
||||
# Video-specific fields
|
||||
"frame_rate": ("FLOAT", {
|
||||
"default": 0.0,
|
||||
"min": 0.0,
|
||||
"max": 120.0,
|
||||
"tooltip": "Video frame rate (FPS). 0 = auto-detect from file"
|
||||
}),
|
||||
"frame_count": ("INT", {
|
||||
"default": 0,
|
||||
"min": 0,
|
||||
"tooltip": "Number of frames. 0 = auto-detect from file"
|
||||
}),
|
||||
"motion_model_name": ("STRING", {
|
||||
"default": "",
|
||||
"tooltip": "Motion model name (AnimateDiff, etc.)"
|
||||
}),
|
||||
# IMH Pro fields
|
||||
"user_tags": ("STRING", {
|
||||
"default": "",
|
||||
"tooltip": "User tags (comma-separated)"
|
||||
}),
|
||||
"notes": ("STRING", {
|
||||
"multiline": True,
|
||||
"default": "",
|
||||
"tooltip": "Notes"
|
||||
}),
|
||||
"project_name": ("STRING", {
|
||||
"default": "",
|
||||
"tooltip": "Project name"
|
||||
}),
|
||||
# Timer integration
|
||||
"generation_time_override": ("FLOAT", {
|
||||
"default": None,
|
||||
"forceInput": True,
|
||||
"tooltip": "Timestamp from MetaHub Timer Node"
|
||||
}),
|
||||
},
|
||||
"hidden": {
|
||||
"prompt": "PROMPT",
|
||||
"extra_pnginfo": "EXTRA_PNGINFO",
|
||||
"unique_id": "UNIQUE_ID"
|
||||
},
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("VHS_FILENAMES",)
|
||||
RETURN_NAMES = ("filenames",)
|
||||
FUNCTION = "add_metadata"
|
||||
OUTPUT_NODE = True
|
||||
CATEGORY = "video/save"
|
||||
DESCRIPTION = "Add A1111/Civitai and Video MetaHub metadata to video files from VHS"
|
||||
|
||||
def add_metadata(
|
||||
self,
|
||||
filenames,
|
||||
seed=None,
|
||||
steps=None,
|
||||
cfg=None,
|
||||
sampler_name=None,
|
||||
scheduler=None,
|
||||
model_name=None,
|
||||
positive=None,
|
||||
negative=None,
|
||||
denoise=None,
|
||||
vae_name=None,
|
||||
frame_rate=0.0,
|
||||
frame_count=0,
|
||||
motion_model_name="",
|
||||
user_tags="",
|
||||
notes="",
|
||||
project_name="",
|
||||
generation_time_override=None,
|
||||
prompt=None,
|
||||
extra_pnginfo=None,
|
||||
unique_id=None,
|
||||
):
|
||||
"""
|
||||
Adds metadata to video files from VHS_FILENAMES.
|
||||
|
||||
1. Extracts video paths from VHS_FILENAMES tuple
|
||||
2. Uses WorkflowExtractor to get generation parameters
|
||||
3. Builds A1111 and MetaHub metadata structures
|
||||
4. Injects metadata into video using ffmpeg
|
||||
5. Returns original filenames for chaining
|
||||
"""
|
||||
# Check ffmpeg availability first
|
||||
ffmpeg = video_utils.find_ffmpeg_binary()
|
||||
if not ffmpeg:
|
||||
print("[MetaHub Video] Warning: FFmpeg not found. Metadata injection skipped.")
|
||||
print("[MetaHub Video] Install FFmpeg and ensure it's in PATH, or set FFMPEG_PATH.")
|
||||
return (filenames,)
|
||||
|
||||
# Extract video files from VHS_FILENAMES
|
||||
video_paths = video_utils.extract_video_files(filenames)
|
||||
|
||||
if not video_paths:
|
||||
print("[MetaHub Video] No video files found in VHS_FILENAMES")
|
||||
return (filenames,)
|
||||
|
||||
# Extract workflow data
|
||||
try:
|
||||
workflow_json = utils.get_workflow_json(extra_pnginfo)
|
||||
prompt_data = prompt if isinstance(prompt, dict) else workflow_json.get("prompt", {})
|
||||
if not isinstance(prompt_data, dict):
|
||||
prompt_data = {}
|
||||
|
||||
workflow_json = utils.ensure_prompt_in_workflow(workflow_json, prompt_data)
|
||||
|
||||
# Use WorkflowExtractor to get generation params
|
||||
extractor = WorkflowExtractor(prompt_data)
|
||||
extracted, missing_fields = extractor.extract(save_node_id=None)
|
||||
lora_list = extracted.get("lora_list") or utils.extract_loras_from_workflow(workflow_json)
|
||||
|
||||
except Exception as e:
|
||||
print(f"[MetaHub Video] Warning: Could not extract workflow data: {e}")
|
||||
extracted = {}
|
||||
lora_list = []
|
||||
workflow_json = {}
|
||||
|
||||
# Resolve values (manual override > extracted > default)
|
||||
def resolve_value(manual_value, extracted_value, default_value):
|
||||
if manual_value is not None:
|
||||
return manual_value
|
||||
if extracted_value is not None:
|
||||
return extracted_value
|
||||
return default_value
|
||||
|
||||
def normalize_int(value, default_value):
|
||||
try:
|
||||
return int(value)
|
||||
except (TypeError, ValueError):
|
||||
return default_value
|
||||
|
||||
def normalize_float(value, default_value):
|
||||
try:
|
||||
return float(value)
|
||||
except (TypeError, ValueError):
|
||||
return default_value
|
||||
|
||||
seed_value = normalize_int(resolve_value(seed, extracted.get("seed"), 0), 0)
|
||||
steps_value = normalize_int(resolve_value(steps, extracted.get("steps"), 20), 20)
|
||||
cfg_value = normalize_float(resolve_value(cfg, extracted.get("cfg"), 7.0), 7.0)
|
||||
sampler_value = resolve_value(sampler_name, extracted.get("sampler_name"), "euler")
|
||||
scheduler_value = resolve_value(scheduler, extracted.get("scheduler"), "normal")
|
||||
model_name_value = resolve_value(model_name, extracted.get("model_name"), "")
|
||||
positive_value = resolve_value(positive, extracted.get("positive"), "")
|
||||
negative_value = resolve_value(negative, extracted.get("negative"), "")
|
||||
denoise_value = normalize_float(resolve_value(denoise, extracted.get("denoise"), 1.0), 1.0)
|
||||
vae_name_value = resolve_value(vae_name, extracted.get("vae_name"), "")
|
||||
|
||||
# Calculate model hash
|
||||
if model_name_value:
|
||||
model_hash = utils.calculate_model_hash(model_name_value, model_type="checkpoint")
|
||||
else:
|
||||
model_hash = "0000000000"
|
||||
lora_hashes = utils.calculate_lora_hashes(lora_list) if lora_list else {}
|
||||
|
||||
# Calculate generation time
|
||||
if generation_time_override is not None and generation_time_override > 0:
|
||||
final_time = time.time() - generation_time_override
|
||||
else:
|
||||
final_time = 0.0
|
||||
|
||||
generation_time_ms = int(final_time * 1000) if final_time > 0 else None
|
||||
steps_per_second = None
|
||||
if generation_time_ms and generation_time_ms > 0 and steps_value > 0:
|
||||
steps_per_second = round((steps_value / (generation_time_ms / 1000)), 2)
|
||||
|
||||
# Collect GPU metrics
|
||||
gpu_metrics = utils.collect_gpu_metrics()
|
||||
version_info = utils.collect_version_info()
|
||||
|
||||
# Process each video file
|
||||
for video_path in video_paths:
|
||||
try:
|
||||
self._process_video(
|
||||
video_path=video_path,
|
||||
seed=seed_value,
|
||||
steps=steps_value,
|
||||
cfg=cfg_value,
|
||||
sampler=sampler_value,
|
||||
scheduler=scheduler_value,
|
||||
model_name=model_name_value,
|
||||
model_hash=model_hash,
|
||||
positive=positive_value,
|
||||
negative=negative_value,
|
||||
denoise=denoise_value,
|
||||
vae_name=vae_name_value,
|
||||
frame_rate=frame_rate,
|
||||
frame_count=frame_count,
|
||||
motion_model_name=motion_model_name,
|
||||
lora_list=lora_list,
|
||||
lora_hashes=lora_hashes,
|
||||
user_tags=user_tags,
|
||||
notes=notes,
|
||||
project_name=project_name,
|
||||
generation_time_ms=generation_time_ms,
|
||||
steps_per_second=steps_per_second,
|
||||
gpu_metrics=gpu_metrics,
|
||||
version_info=version_info,
|
||||
workflow_json=workflow_json,
|
||||
)
|
||||
except Exception as e:
|
||||
print(f"[MetaHub Video] Warning: Failed to add metadata to {video_path.name}: {e}")
|
||||
|
||||
return (filenames,)
|
||||
|
||||
def _process_video(
|
||||
self,
|
||||
video_path: Path,
|
||||
seed: int,
|
||||
steps: int,
|
||||
cfg: float,
|
||||
sampler: str,
|
||||
scheduler: str,
|
||||
model_name: str,
|
||||
model_hash: str,
|
||||
positive: str,
|
||||
negative: str,
|
||||
denoise: float,
|
||||
vae_name: str,
|
||||
frame_rate: float,
|
||||
frame_count: int,
|
||||
motion_model_name: str,
|
||||
lora_list: List[Dict],
|
||||
lora_hashes: Dict[str, str],
|
||||
user_tags: str,
|
||||
notes: str,
|
||||
project_name: str,
|
||||
generation_time_ms: Optional[int],
|
||||
steps_per_second: Optional[float],
|
||||
gpu_metrics: Dict[str, Any],
|
||||
version_info: Dict[str, str],
|
||||
workflow_json: Dict,
|
||||
) -> None:
|
||||
"""
|
||||
Processes a single video file and injects metadata.
|
||||
"""
|
||||
# Get video info from file if not provided
|
||||
video_info = video_utils.get_video_info(video_path)
|
||||
|
||||
if video_info:
|
||||
if frame_rate <= 0:
|
||||
frame_rate = video_info.get('frame_rate') or 0
|
||||
if frame_count <= 0:
|
||||
frame_count = video_info.get('frame_count') or 0
|
||||
width = video_info.get('width', 512)
|
||||
height = video_info.get('height', 512)
|
||||
video_codec = video_info.get('codec', 'h264')
|
||||
else:
|
||||
width = 512
|
||||
height = 512
|
||||
video_codec = 'h264'
|
||||
|
||||
video_format = video_utils.detect_video_format(video_path) or 'mp4'
|
||||
|
||||
# Build params dict
|
||||
params = {
|
||||
"positive": positive,
|
||||
"negative": negative,
|
||||
"steps": steps,
|
||||
"sampler": sampler,
|
||||
"scheduler": scheduler,
|
||||
"cfg": cfg,
|
||||
"seed": seed,
|
||||
"width": width,
|
||||
"height": height,
|
||||
"model_name": model_name,
|
||||
"model_hash": model_hash,
|
||||
"vae_name": vae_name,
|
||||
"denoise": denoise,
|
||||
# Video-specific
|
||||
"frame_rate": frame_rate,
|
||||
"frame_count": frame_count,
|
||||
"video_format": video_format,
|
||||
"video_codec": video_codec,
|
||||
"motion_model_name": motion_model_name,
|
||||
# LoRAs
|
||||
"lora_list": lora_list,
|
||||
"lora_hashes": lora_hashes,
|
||||
# IMH Pro
|
||||
"user_tags": user_tags,
|
||||
"notes": notes,
|
||||
"project_name": project_name,
|
||||
# Performance metrics
|
||||
"vram_peak_mb": gpu_metrics.get("vram_peak_mb"),
|
||||
"gpu_device": gpu_metrics.get("gpu_device"),
|
||||
"generation_time_ms": generation_time_ms,
|
||||
"steps_per_second": steps_per_second,
|
||||
"comfyui_version": version_info.get("comfyui_version"),
|
||||
"torch_version": version_info.get("torch_version"),
|
||||
"python_version": version_info.get("python_version"),
|
||||
}
|
||||
|
||||
# Build metadata
|
||||
a1111_metadata = video_utils.build_video_a1111_metadata(params)
|
||||
metahub_metadata = video_utils.build_video_metahub_metadata(params, workflow_json)
|
||||
|
||||
# Inject metadata into video
|
||||
video_utils.inject_video_metadata(
|
||||
video_path=video_path,
|
||||
a1111_metadata=a1111_metadata,
|
||||
metahub_metadata=metahub_metadata,
|
||||
)
|
||||
|
||||
print(f"[MetaHub Video] Added metadata to: {video_path.name}")
|
||||
|
||||
# Verify metadata was injected
|
||||
verification = video_utils.verify_video_metadata(video_path)
|
||||
if verification and verification.get('has_metahub_data'):
|
||||
print(f"[MetaHub Video] Metadata verified successfully")
|
||||
else:
|
||||
print(f"[MetaHub Video] Warning: Metadata verification failed")
|
||||
|
||||
|
||||
NODE_CLASS_MAPPINGS = {"MetaHubSaveVideoNode": MetaHubSaveVideoNode}
|
||||
NODE_DISPLAY_NAME_MAPPINGS = {"MetaHubSaveVideoNode": "MetaHub Save Video"}
|
||||
@@ -0,0 +1,566 @@
|
||||
"""
|
||||
Video Metadata Utils for MetaHub Save Node
|
||||
Handles video metadata injection using ffmpeg.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
# Video extensions supported
|
||||
VIDEO_EXTENSIONS = {'.mp4', '.webm', '.mkv', '.mov', '.avi'}
|
||||
|
||||
|
||||
def find_ffmpeg_binary() -> Optional[Path]:
|
||||
"""
|
||||
Locates ffmpeg executable.
|
||||
|
||||
Search order:
|
||||
1. FFMPEG_PATH environment variable
|
||||
2. System PATH (shutil.which)
|
||||
3. Common installation locations
|
||||
|
||||
Returns:
|
||||
Path to ffmpeg executable or None if not found
|
||||
"""
|
||||
# Check environment variable first
|
||||
env_path = os.environ.get('FFMPEG_PATH')
|
||||
if env_path:
|
||||
ffmpeg_path = Path(env_path)
|
||||
if ffmpeg_path.exists():
|
||||
return ffmpeg_path
|
||||
|
||||
# Check system PATH
|
||||
which_result = shutil.which('ffmpeg')
|
||||
if which_result:
|
||||
return Path(which_result)
|
||||
|
||||
# Common locations on Windows
|
||||
common_paths = [
|
||||
Path('C:/ffmpeg/bin/ffmpeg.exe'),
|
||||
Path('C:/Program Files/ffmpeg/bin/ffmpeg.exe'),
|
||||
Path('C:/Program Files (x86)/ffmpeg/bin/ffmpeg.exe'),
|
||||
]
|
||||
|
||||
for path in common_paths:
|
||||
if path.exists():
|
||||
return path
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def find_ffprobe_binary() -> Optional[Path]:
|
||||
"""
|
||||
Locates ffprobe executable (for verification).
|
||||
|
||||
Returns:
|
||||
Path to ffprobe executable or None if not found
|
||||
"""
|
||||
# Check environment variable
|
||||
env_path = os.environ.get('FFPROBE_PATH')
|
||||
if env_path:
|
||||
ffprobe_path = Path(env_path)
|
||||
if ffprobe_path.exists():
|
||||
return ffprobe_path
|
||||
|
||||
# Check system PATH
|
||||
which_result = shutil.which('ffprobe')
|
||||
if which_result:
|
||||
return Path(which_result)
|
||||
|
||||
# Try to find next to ffmpeg
|
||||
ffmpeg = find_ffmpeg_binary()
|
||||
if ffmpeg:
|
||||
ffprobe = ffmpeg.parent / ('ffprobe' + ffmpeg.suffix)
|
||||
if ffprobe.exists():
|
||||
return ffprobe
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def detect_video_format(file_path: Path) -> Optional[str]:
|
||||
"""
|
||||
Detects video format from file extension.
|
||||
|
||||
Args:
|
||||
file_path: Path to video file
|
||||
|
||||
Returns:
|
||||
Format string ('mp4', 'webm', 'mkv') or None if not a video
|
||||
"""
|
||||
ext = file_path.suffix.lower()
|
||||
if ext in VIDEO_EXTENSIONS:
|
||||
return ext[1:] # Remove the dot
|
||||
return None
|
||||
|
||||
|
||||
def is_video_file(file_path: Path) -> bool:
|
||||
"""
|
||||
Checks if file is a supported video format.
|
||||
|
||||
Args:
|
||||
file_path: Path to check
|
||||
|
||||
Returns:
|
||||
True if file is a supported video format
|
||||
"""
|
||||
return file_path.suffix.lower() in VIDEO_EXTENSIONS
|
||||
|
||||
|
||||
def build_video_a1111_metadata(params: dict) -> str:
|
||||
"""
|
||||
Builds A1111/Civitai compatible metadata string for video.
|
||||
|
||||
Similar to image metadata but includes video-specific fields.
|
||||
|
||||
Format:
|
||||
{positive_prompt}
|
||||
Negative prompt: {negative_prompt}
|
||||
Steps: {steps}, Sampler: {sampler}, CFG scale: {cfg}, Seed: {seed},
|
||||
Size: {width}x{height}, Model: {model_name}, Model hash: {hash},
|
||||
Frames: {frame_count}, FPS: {frame_rate}
|
||||
|
||||
Args:
|
||||
params: Dict containing all generation parameters
|
||||
|
||||
Returns:
|
||||
Formatted A1111 metadata string
|
||||
"""
|
||||
# Build LoRA hashes string
|
||||
lora_hashes_str = ", ".join(
|
||||
f"{name}: {hash_val}"
|
||||
for name, hash_val in params.get('lora_hashes', {}).items()
|
||||
)
|
||||
|
||||
# Build config parts
|
||||
config_parts = [
|
||||
f"Steps: {params.get('steps', 20)}",
|
||||
f"Sampler: {params.get('sampler', 'euler')}",
|
||||
f"CFG scale: {params.get('cfg', 7.0)}",
|
||||
f"Seed: {params.get('seed', 0)}",
|
||||
f"Size: {params.get('width', 512)}x{params.get('height', 512)}",
|
||||
f"Model: {params.get('model_name', '')}",
|
||||
f"Model hash: {params.get('model_hash', '0000000000')}",
|
||||
]
|
||||
|
||||
# Add video-specific fields
|
||||
if params.get('frame_count'):
|
||||
config_parts.append(f"Frames: {params['frame_count']}")
|
||||
if params.get('frame_rate'):
|
||||
config_parts.append(f"FPS: {params['frame_rate']}")
|
||||
|
||||
# Add motion model if present
|
||||
motion_model = params.get('motion_model_name')
|
||||
if motion_model:
|
||||
config_parts.append(f"Motion Model: {motion_model}")
|
||||
|
||||
# Add LoRA hashes if present
|
||||
if lora_hashes_str:
|
||||
config_parts.append(f'Lora hashes: "{lora_hashes_str}"')
|
||||
|
||||
config_line = ", ".join(config_parts)
|
||||
|
||||
# Build final metadata string
|
||||
positive = params.get('positive', '')
|
||||
negative = params.get('negative', '')
|
||||
|
||||
metadata = f"""{positive}
|
||||
Negative prompt: {negative}
|
||||
{config_line}"""
|
||||
|
||||
return metadata
|
||||
|
||||
|
||||
def build_video_metahub_metadata(params: dict, workflow_json: dict) -> dict:
|
||||
"""
|
||||
Builds Video MetaHub metadata as JSON object.
|
||||
|
||||
Similar to image metadata but with video-specific fields.
|
||||
|
||||
Args:
|
||||
params: Dict containing generation parameters
|
||||
workflow_json: Complete ComfyUI workflow from extra_pnginfo
|
||||
|
||||
Returns:
|
||||
JSON-serializable dict for video metadata
|
||||
"""
|
||||
import numpy as np
|
||||
|
||||
def _sanitize(value: Any) -> Any:
|
||||
if isinstance(value, float) and np.isnan(value):
|
||||
return None
|
||||
if isinstance(value, dict):
|
||||
return {k: _sanitize(v) for k, v in value.items()}
|
||||
if isinstance(value, list):
|
||||
return [_sanitize(v) for v in value]
|
||||
return value
|
||||
|
||||
safe_workflow = _sanitize(workflow_json.get('workflow', {}))
|
||||
safe_prompt = _sanitize(workflow_json.get('prompt', {}))
|
||||
|
||||
# Calculate duration if we have frame count and rate
|
||||
duration_seconds = None
|
||||
frame_count = params.get('frame_count')
|
||||
frame_rate = params.get('frame_rate')
|
||||
if frame_count and frame_rate and frame_rate > 0:
|
||||
duration_seconds = round(frame_count / frame_rate, 2)
|
||||
|
||||
return {
|
||||
# CRITICAL: Required field for detection
|
||||
"generator": "ComfyUI",
|
||||
"media_type": "video",
|
||||
|
||||
# Main generation fields
|
||||
"prompt": params.get('positive', ''),
|
||||
"negativePrompt": params.get('negative', ''),
|
||||
"seed": params.get('seed', 0),
|
||||
"steps": params.get('steps', 20),
|
||||
"cfg": params.get('cfg', 7.0),
|
||||
"sampler_name": params.get('sampler', 'euler'),
|
||||
"scheduler": params.get('scheduler', 'normal'),
|
||||
"model": params.get('model_name', ''),
|
||||
"model_hash": params.get('model_hash', ''),
|
||||
"vae": params.get('vae_name', ''),
|
||||
"denoise": params.get('denoise', 1.0),
|
||||
"width": params.get('width', 512),
|
||||
"height": params.get('height', 512),
|
||||
|
||||
# Video-specific fields
|
||||
"video": {
|
||||
"frame_rate": frame_rate,
|
||||
"frame_count": frame_count,
|
||||
"duration_seconds": duration_seconds,
|
||||
"width": params.get('width', 512),
|
||||
"height": params.get('height', 512),
|
||||
"format": params.get('video_format', 'mp4'),
|
||||
"codec": params.get('video_codec', 'h264'),
|
||||
},
|
||||
|
||||
# Motion model (AnimateDiff, etc.)
|
||||
"motion_model": {
|
||||
"name": params.get('motion_model_name'),
|
||||
"hash": params.get('motion_model_hash'),
|
||||
} if params.get('motion_model_name') else None,
|
||||
|
||||
# LoRAs in array format
|
||||
"loras": [
|
||||
{"name": lora['name'], "weight": lora['weight']}
|
||||
for lora in params.get('lora_list', [])
|
||||
],
|
||||
|
||||
# IMH Pro fields
|
||||
"imh_pro": {
|
||||
"user_tags": params.get('user_tags', ''),
|
||||
"notes": params.get('notes', ''),
|
||||
"project_name": params.get('project_name', ''),
|
||||
},
|
||||
|
||||
# Analytics/Performance metrics
|
||||
"analytics": {
|
||||
"vram_peak_mb": params.get('vram_peak_mb'),
|
||||
"gpu_device": params.get('gpu_device'),
|
||||
"generation_time_ms": params.get('generation_time_ms'),
|
||||
"steps_per_second": params.get('steps_per_second'),
|
||||
"comfyui_version": params.get('comfyui_version'),
|
||||
"torch_version": params.get('torch_version'),
|
||||
"python_version": params.get('python_version'),
|
||||
},
|
||||
|
||||
# Complete workflow data
|
||||
"workflow": safe_workflow,
|
||||
"prompt_api": safe_prompt,
|
||||
}
|
||||
|
||||
|
||||
def inject_video_metadata(
|
||||
video_path: Path,
|
||||
a1111_metadata: str,
|
||||
metahub_metadata: dict,
|
||||
output_path: Optional[Path] = None,
|
||||
keep_original: bool = False,
|
||||
) -> Path:
|
||||
"""
|
||||
Injects metadata into video container without re-encoding.
|
||||
|
||||
Uses ffmpeg with -c copy to add metadata to video file.
|
||||
Metadata is stored in:
|
||||
- description: A1111/Civitai format
|
||||
- comment: JSON videometahub_data
|
||||
|
||||
Args:
|
||||
video_path: Path to input video file
|
||||
a1111_metadata: A1111 formatted metadata string
|
||||
metahub_metadata: MetaHub metadata dict (will be JSON serialized)
|
||||
output_path: Optional output path (default: replace original)
|
||||
keep_original: If True, keep original file (only works if output_path differs)
|
||||
|
||||
Returns:
|
||||
Path to video with injected metadata
|
||||
|
||||
Raises:
|
||||
RuntimeError: If ffmpeg is not found or injection fails
|
||||
"""
|
||||
ffmpeg = find_ffmpeg_binary()
|
||||
if not ffmpeg:
|
||||
raise RuntimeError(
|
||||
"FFmpeg not found. Please install FFmpeg and ensure it's in PATH, "
|
||||
"or set FFMPEG_PATH environment variable."
|
||||
)
|
||||
|
||||
video_path = Path(video_path)
|
||||
if not video_path.exists():
|
||||
raise FileNotFoundError(f"Video file not found: {video_path}")
|
||||
|
||||
# Determine output path
|
||||
if output_path is None:
|
||||
# Replace original - use temp file then rename
|
||||
replace_original = True
|
||||
temp_fd, temp_path = tempfile.mkstemp(suffix=video_path.suffix)
|
||||
os.close(temp_fd)
|
||||
final_output = Path(temp_path)
|
||||
else:
|
||||
replace_original = False
|
||||
final_output = Path(output_path)
|
||||
|
||||
# Serialize metadata JSON
|
||||
metahub_json = json.dumps(metahub_metadata, ensure_ascii=False, separators=(',', ':'))
|
||||
|
||||
# Detect format for format-specific flags
|
||||
video_format = detect_video_format(video_path)
|
||||
|
||||
# Build ffmpeg command
|
||||
cmd = [
|
||||
str(ffmpeg),
|
||||
'-y', # Overwrite output
|
||||
'-i', str(video_path),
|
||||
'-c', 'copy', # No re-encoding
|
||||
'-map_metadata', '0', # Preserve existing metadata
|
||||
]
|
||||
|
||||
# Add format-specific flags
|
||||
if video_format == 'mp4':
|
||||
cmd.extend(['-movflags', 'use_metadata_tags'])
|
||||
|
||||
# Add metadata fields
|
||||
# Use description for A1111 format (human-readable)
|
||||
cmd.extend(['-metadata', f'description={a1111_metadata}'])
|
||||
# Use comment for full JSON data
|
||||
cmd.extend(['-metadata', f'comment={metahub_json}'])
|
||||
# Also add a title with basic info
|
||||
title = f"Generated with ComfyUI - Seed: {metahub_metadata.get('seed', 'unknown')}"
|
||||
cmd.extend(['-metadata', f'title={title}'])
|
||||
|
||||
# Output file
|
||||
cmd.append(str(final_output))
|
||||
|
||||
# Execute ffmpeg
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=60, # 1 minute timeout
|
||||
)
|
||||
|
||||
if result.returncode != 0:
|
||||
raise RuntimeError(f"FFmpeg failed: {result.stderr}")
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
raise RuntimeError("FFmpeg timed out while injecting metadata")
|
||||
except Exception as e:
|
||||
# Clean up temp file on error
|
||||
if replace_original and final_output.exists():
|
||||
final_output.unlink()
|
||||
raise RuntimeError(f"Failed to inject metadata: {e}")
|
||||
|
||||
# If replacing original, move temp file to original location
|
||||
if replace_original:
|
||||
try:
|
||||
# Remove original
|
||||
video_path.unlink()
|
||||
# Move temp to original location
|
||||
shutil.move(str(final_output), str(video_path))
|
||||
return video_path
|
||||
except Exception as e:
|
||||
# Try to clean up
|
||||
if final_output.exists():
|
||||
final_output.unlink()
|
||||
raise RuntimeError(f"Failed to replace original file: {e}")
|
||||
|
||||
return final_output
|
||||
|
||||
|
||||
def verify_video_metadata(video_path: Path) -> Optional[dict]:
|
||||
"""
|
||||
Verifies that metadata was successfully injected into video.
|
||||
|
||||
Uses ffprobe to read metadata from video file.
|
||||
|
||||
Args:
|
||||
video_path: Path to video file
|
||||
|
||||
Returns:
|
||||
Dict with metadata fields or None if verification failed
|
||||
"""
|
||||
ffprobe = find_ffprobe_binary()
|
||||
if not ffprobe:
|
||||
print("[MetaHub Video] Warning: ffprobe not found, cannot verify metadata")
|
||||
return None
|
||||
|
||||
try:
|
||||
cmd = [
|
||||
str(ffprobe),
|
||||
'-v', 'quiet',
|
||||
'-print_format', 'json',
|
||||
'-show_format',
|
||||
str(video_path),
|
||||
]
|
||||
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=10,
|
||||
)
|
||||
|
||||
if result.returncode != 0:
|
||||
return None
|
||||
|
||||
data = json.loads(result.stdout)
|
||||
tags = data.get('format', {}).get('tags', {})
|
||||
|
||||
return {
|
||||
'title': tags.get('title'),
|
||||
'description': tags.get('description'),
|
||||
'comment': tags.get('comment'),
|
||||
'has_metahub_data': 'videometahub_data' in tags.get('comment', '') or
|
||||
'"generator":"ComfyUI"' in tags.get('comment', ''),
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
print(f"[MetaHub Video] Warning: Could not verify metadata: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def get_video_info(video_path: Path) -> Optional[dict]:
|
||||
"""
|
||||
Gets video information using ffprobe.
|
||||
|
||||
Args:
|
||||
video_path: Path to video file
|
||||
|
||||
Returns:
|
||||
Dict with video info (frame_count, frame_rate, width, height, duration) or None
|
||||
"""
|
||||
ffprobe = find_ffprobe_binary()
|
||||
if not ffprobe:
|
||||
return None
|
||||
|
||||
try:
|
||||
cmd = [
|
||||
str(ffprobe),
|
||||
'-v', 'quiet',
|
||||
'-print_format', 'json',
|
||||
'-show_streams',
|
||||
'-show_format',
|
||||
str(video_path),
|
||||
]
|
||||
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=10,
|
||||
)
|
||||
|
||||
if result.returncode != 0:
|
||||
return None
|
||||
|
||||
data = json.loads(result.stdout)
|
||||
|
||||
# Find video stream
|
||||
video_stream = None
|
||||
for stream in data.get('streams', []):
|
||||
if stream.get('codec_type') == 'video':
|
||||
video_stream = stream
|
||||
break
|
||||
|
||||
if not video_stream:
|
||||
return None
|
||||
|
||||
# Parse frame rate (can be "30/1" or "29.97")
|
||||
frame_rate = None
|
||||
r_frame_rate = video_stream.get('r_frame_rate', '')
|
||||
if '/' in r_frame_rate:
|
||||
num, den = r_frame_rate.split('/')
|
||||
if int(den) != 0:
|
||||
frame_rate = round(int(num) / int(den), 2)
|
||||
else:
|
||||
try:
|
||||
frame_rate = float(r_frame_rate)
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# Get frame count
|
||||
frame_count = video_stream.get('nb_frames')
|
||||
if frame_count:
|
||||
frame_count = int(frame_count)
|
||||
|
||||
# Get duration
|
||||
duration = data.get('format', {}).get('duration')
|
||||
if duration:
|
||||
duration = float(duration)
|
||||
|
||||
return {
|
||||
'width': int(video_stream.get('width', 0)),
|
||||
'height': int(video_stream.get('height', 0)),
|
||||
'frame_rate': frame_rate,
|
||||
'frame_count': frame_count,
|
||||
'duration': duration,
|
||||
'codec': video_stream.get('codec_name'),
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
print(f"[MetaHub Video] Warning: Could not get video info: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def extract_video_files(filenames_tuple: tuple) -> List[Path]:
|
||||
"""
|
||||
Extracts video file paths from VHS_FILENAMES tuple.
|
||||
|
||||
VHS_FILENAMES format: (save_output: bool, [file_paths: List[str]])
|
||||
|
||||
Args:
|
||||
filenames_tuple: VHS_FILENAMES tuple from Video Combine node
|
||||
|
||||
Returns:
|
||||
List of Path objects for video files
|
||||
"""
|
||||
video_paths = []
|
||||
|
||||
try:
|
||||
if not isinstance(filenames_tuple, (tuple, list)):
|
||||
return video_paths
|
||||
|
||||
# VHS_FILENAMES is (bool, [paths])
|
||||
if len(filenames_tuple) < 2:
|
||||
return video_paths
|
||||
|
||||
file_list = filenames_tuple[1]
|
||||
if not isinstance(file_list, (list, tuple)):
|
||||
return video_paths
|
||||
|
||||
for path_str in file_list:
|
||||
path = Path(path_str)
|
||||
if path.exists() and is_video_file(path):
|
||||
video_paths.append(path)
|
||||
|
||||
except Exception as e:
|
||||
print(f"[MetaHub Video] Warning: Could not parse VHS_FILENAMES: {e}")
|
||||
|
||||
return video_paths
|
||||
Reference in New Issue
Block a user