"""BFS Shot Loop: plan a long video into model-sized shots, run any workflow once per shot, join. A video model only follows a guide reliably inside the clip length it was trained on (an H3 LoRA trained on 107 frames loses the scene at 243). This splits the source into shots that fit, at camera cuts when there are any, and hands them to the rest of the graph as a ComfyUI *list*. ComfyUI then runs every node that receives a list once per item, pairing the lists by index, so shot ``i`` always gets guide ``i``, reference ``i`` and prompt ``i``. ``BFS Shot Join`` takes the decoded list back, trims each shot to its true length and concatenates them in order. Every shot is generated at a length the model accepts (for example 17n+5 frames for MiniMax H3). When a shot is shorter than that, the extra guide frames are taken from the video that follows it, so the guide never repeats or stretches; the join then cuts the result back to the shot's own frames, which keeps the timing identical to the source. Those extra frames also give the join a real overlap to cross-fade across boundaries that are not camera cuts. """ from __future__ import annotations import base64 import hashlib import json import re import math import os import subprocess import time from typing import Any import numpy as np import torch import folder_paths try: from . import bfs_video_io as VIO except ImportError: # loaded as a plain module (the tests): the file next to this one import importlib.util as _ilu _vio = _ilu.spec_from_file_location("bfs_video_io", os.path.join(os.path.dirname(os.path.abspath(__file__)), "bfs_video_io.py")) VIO = _ilu.module_from_spec(_vio) _vio.loader.exec_module(VIO) # ---------------------------------------------------------------------------- frame grids GRIDS: dict[str, tuple[int, int]] = { "H3 (17n+5)": (17, 5), "LTX / Wan (8n+1)": (8, 1), "Wan (4n+1)": (4, 1), "any": (1, 0), } DEFAULT_GRID = "H3 (17n+5)" VIDEO_EXTS = (".mp4", ".mov", ".webm", ".mkv", ".avi", ".m4v", ".gif") IMAGE_EXTS = (".png", ".jpg", ".jpeg", ".webp", ".bmp") def snap_up(n: int, grid: str) -> int: """Smallest valid clip length >= n.""" step, off = GRIDS.get(grid, GRIDS[DEFAULT_GRID]) n = max(1, int(n)) if step == 1: return n if n <= off: return off return off + step * math.ceil((n - off) / step) def snap_down(n: int, grid: str) -> int: """Largest valid clip length <= n (never below the grid's smallest length).""" step, off = GRIDS.get(grid, GRIDS[DEFAULT_GRID]) n = max(1, int(n)) if step == 1: return n if n <= off: return off return off + step * ((n - off) // step) # ---------------------------------------------------------------------------- video io def _input_path(name: str) -> str: path = folder_paths.get_annotated_filepath(name) if name else "" if not path or not os.path.isfile(path): raise FileNotFoundError(f"BFS Shot Loop: file not found in the input folder: {name!r}") return path def _probe(path: str) -> dict: import cv2 cap = cv2.VideoCapture(path) if not cap.isOpened(): raise ValueError(f"BFS Shot Loop: cannot open video {path}") fps = cap.get(cv2.CAP_PROP_FPS) or 0.0 n = int(cap.get(cv2.CAP_PROP_FRAME_COUNT) or 0) w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH) or 0) h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT) or 0) cap.release() if fps <= 0 or n <= 0: raise ValueError(f"BFS Shot Loop: could not read fps/frame count from {path}") return {"fps_src": float(fps), "n_src": n, "width": w, "height": h, "duration": n / fps} def _timeline(n_src: int, fps_src: float, fps: float) -> np.ndarray: """Source frame index for each frame of the resampled timeline (nearest frame, real time).""" n = max(1, int(math.floor(n_src / fps_src * fps + 1e-6))) idx = np.round(np.arange(n) * fps_src / fps).astype(np.int64) return np.clip(idx, 0, n_src - 1) _STATUS_T = [0.0] def _status(stage: str, done: int = 0, total: int = 0, force: bool = False) -> None: """Progress for the planner panel's loading bar (throttled).""" import time now = time.time() if not force and now - _STATUS_T[0] < 0.25 and done < total: return _STATUS_T[0] = now _notify("bfs-shotloop-status", {"stage": stage, "done": int(done), "total": int(total)}) def _read_frames(path: str, src_indices: np.ndarray, size: tuple[int, int] | None, stage: str | None = None) -> list[np.ndarray]: """Decode the requested source frames (sorted, may repeat) as RGB uint8, sequentially.""" import cv2 want = sorted(set(int(i) for i in src_indices)) top = want[-1] + 1 if want else 0 got: dict[int, np.ndarray] = {} cap = cv2.VideoCapture(path) i, k = 0, 0 last = None while k < len(want): ok = cap.grab() if not ok: break if stage and i % 24 == 0: _status(stage, i, top) if i == want[k]: ok, fr = cap.retrieve() if ok: fr = cv2.cvtColor(fr, cv2.COLOR_BGR2RGB) if size is not None: fr = _fit(fr, size) last = fr got[i] = fr while k < len(want) and want[k] == i: k += 1 i += 1 cap.release() if not got: raise ValueError(f"BFS Shot Loop: no frames decoded from {path}") out = [] for s in src_indices: s = int(s) out.append(got.get(s, last if s > max(got) else got[min(got, key=lambda x: abs(x - s))])) return out def _fit(img: np.ndarray, size: tuple[int, int]) -> np.ndarray: """Cover-resize then center-crop to (w, h).""" import cv2 w, h = size ih, iw = img.shape[:2] s = max(w / iw, h / ih) r = cv2.resize(img, (max(w, round(iw * s)), max(h, round(ih * s))), interpolation=cv2.INTER_AREA if s < 1 else cv2.INTER_CUBIC) y, x = (r.shape[0] - h) // 2, (r.shape[1] - w) // 2 return r[y:y + h, x:x + w] def generation_size(src_w: int, src_h: int, megapixels: float, multiple: int) -> tuple[int, int]: """Aspect-preserving size with the given pixel area, both sides on the multiple grid.""" multiple = max(8, int(multiple)) area = max(0.05, float(megapixels)) * 1e6 ar = src_w / max(1, src_h) h = math.sqrt(area / ar) w = h * ar w = max(multiple, int(round(w / multiple)) * multiple) h = max(multiple, int(round(h / multiple)) * multiple) return w, h def plan_generation_size(src_w: int, src_h: int, plan: dict, multiple: int) -> tuple[int, int]: """Resolve the planner's megapixel, source-resolution, or custom generation size.""" mode = str(plan.get("generation_size_mode") or "megapixels") if mode == "original": mp = max(0.05, int(src_w) * int(src_h) / 1e6) return generation_size(src_w, src_h, mp, multiple) if mode == "custom": multiple = max(8, int(multiple)) w = max(multiple, int(round(max(1, int(plan.get("custom_width") or 1024)) / multiple)) * multiple) h = max(multiple, int(round(max(1, int(plan.get("custom_height") or 1024)) / multiple)) * multiple) return w, h return generation_size(src_w, src_h, float(plan.get("megapixels", 0.5)), multiple) def audio_info(path: str) -> dict: """Whether the video has a usable audio track (PyAV, no ffmpeg binary needed).""" try: import av with av.open(path) as c: if not c.streams.audio: return {"has_audio": False, "reason": "no audio track"} st = c.streams.audio[0] cc = st.codec_context ch = getattr(cc, "channels", 0) or (len(cc.layout.channels) if getattr(cc, "layout", None) else 0) return {"has_audio": True, "sample_rate": int(cc.sample_rate or 0), "channels": int(ch or 0), "codec": cc.name} except Exception as exc: # noqa: BLE001 return {"has_audio": False, "reason": f"audio unreadable ({type(exc).__name__})"} def silence(seconds: float, sample_rate: int = 44100, channels: int = 2) -> dict: """A silent AUDIO of the given length: Create Video and friends always get a valid track.""" n = max(1, int(round(max(0.0, seconds) * sample_rate))) return {"waveform": torch.zeros(1, channels, n), "sample_rate": sample_rate} def _usable(audio: dict | None) -> bool: try: return audio is not None and audio["waveform"].ndim == 3 and audio["waveform"].shape[-1] > 0 except Exception: # noqa: BLE001 return False def _read_audio_av(path: str, sample_rate: int) -> dict | None: """PyAV fallback when the ffmpeg binary is missing.""" import av with av.open(path) as c: if not c.streams.audio: return None st = c.streams.audio[0] layout = "stereo" if (getattr(st.codec_context, "channels", 2) or 2) >= 2 else "mono" rs = av.AudioResampler(format="fltp", layout=layout, rate=sample_rate) chunks = [] for fr in c.decode(st): for r in rs.resample(fr): chunks.append(r.to_ndarray()) for r in rs.resample(None): chunks.append(r.to_ndarray()) if not chunks: return None a = np.concatenate(chunks, axis=1).astype(np.float32) return {"waveform": torch.from_numpy(a)[None], "sample_rate": sample_rate} def _read_audio(path: str, sample_rate: int = 44100) -> dict | None: """Decode the whole soundtrack as float32 [1, C, T] (ffmpeg, else PyAV); None when there is none.""" a = _read_audio_ffmpeg(path, sample_rate) if not _usable(a): try: a = _read_audio_av(path, sample_rate) except Exception: # noqa: BLE001 - audio is optional a = None return a if _usable(a) else None def _read_audio_ffmpeg(path: str, sample_rate: int = 44100) -> dict | None: try: probe = subprocess.run([VIO.ffprobe_bin() or "ffprobe", "-v", "error", "-select_streams", "a:0", "-show_entries", "stream=channels", "-of", "csv=p=0", path], capture_output=True, text=True, timeout=60) ch = int((probe.stdout.strip() or "0").split(",")[0] or 0) if ch <= 0: return None ch = min(ch, 2) raw = subprocess.run([VIO.ffmpeg_bin() or "ffmpeg", "-v", "error", "-i", path, "-vn", "-ac", str(ch), "-ar", str(sample_rate), "-f", "f32le", "-"], capture_output=True, timeout=600).stdout if not raw: return None a = np.frombuffer(raw, dtype=np.float32).reshape(-1, ch).T.copy() return {"waveform": torch.from_numpy(a)[None], "sample_rate": sample_rate} except Exception: # noqa: BLE001 - audio is optional return None def _slice_audio(audio: dict | None, start_s: float, dur_s: float) -> dict | None: if audio is None: return None sr = audio["sample_rate"] wf = audio["waveform"] a, n = int(round(start_s * sr)), int(round(dur_s * sr)) seg = wf[..., a:a + n] if seg.shape[-1] < n: seg = torch.nn.functional.pad(seg, (0, n - seg.shape[-1])) return {"waveform": seg.contiguous(), "sample_rate": sr} # ---------------------------------------------------------------------------- analysis _ANALYSIS_CACHE: dict[tuple, dict] = {} def analyze(path: str, fps: float, thumbs: int = 120, thumb_h: int = 72) -> dict: """Probe, per-frame cut score on the resampled timeline, and a strip of thumbnails.""" import cv2 key = (path, os.path.getmtime(path), round(float(fps), 4), thumbs, thumb_h) if key in _ANALYSIS_CACHE: return _ANALYSIS_CACHE[key] info = _probe(path) src = _timeline(info["n_src"], info["fps_src"], fps) n = len(src) _status("Reading the video", 0, info["n_src"], force=True) # small frames straight from the decoder: the full-size frames of a long video do not fit in memory sw = 64 sh = max(8, int(round(sw * info["height"] / max(1, info["width"])))) # keep the aspect: nothing cropped small = _read_frames(path, src, (sw, sh), stage="Reading the video") hsv = [cv2.calcHist([cv2.cvtColor(s, cv2.COLOR_RGB2HSV)], [0, 1], None, [16, 8], [0, 180, 0, 256]) for s in small] hsv = [cv2.normalize(h, h).flatten() for h in hsv] raw = np.zeros(n, np.float32) for i in range(1, n): pix = np.abs(small[i].astype(np.float32) - small[i - 1].astype(np.float32)).mean() / 255.0 hist = cv2.compareHist(hsv[i - 1], hsv[i], cv2.HISTCMP_BHATTACHARYYA) raw[i] = 0.5 * pix + 0.5 * float(hist) # local normalisation: a cut stands out from the motion around it score = np.zeros(n, np.float32) w = max(4, int(round(fps))) for i in range(1, n): lo, hi = max(1, i - w), min(n, i + w + 1) neigh = np.concatenate([raw[lo:i], raw[i + 1:hi]]) base = float(np.median(neigh)) if len(neigh) else 0.0 score[i] = raw[i] / (base + 0.02) k = max(1, n // max(1, thumbs)) th = [] tw = max(16, int(round(thumb_h * info["width"] / max(1, info["height"])))) frames_for_thumbs = _read_frames(path, src[::k], (tw, thumb_h), stage="Making thumbnails") for j, fr in enumerate(frames_for_thumbs): ok, buf = cv2.imencode(".jpg", cv2.cvtColor(fr, cv2.COLOR_RGB2BGR), [cv2.IMWRITE_JPEG_QUALITY, 70]) th.append({"f": int(j * k), "src": "data:image/jpeg;base64," + base64.b64encode(buf.tobytes()).decode()}) out = dict(info, fps=float(fps), n=n, thumbs=th, thumb_w=tw, thumb_h=thumb_h, audio=audio_info(path), raw=[round(float(x), 4) for x in raw], score=[round(float(x), 3) for x in score]) _ANALYSIS_CACHE[key] = out return out _PSD_CACHE: dict[tuple, list[float]] = {} def scenedetect_cut_times(path: str, detector: str, sensitivity: float) -> list[float] | None: """Cut times in seconds from PySceneDetect, or None when the package is missing.""" try: from scenedetect import SceneManager, open_video from scenedetect.detectors import AdaptiveDetector, ContentDetector except ImportError: return None sensitivity = float(min(1.0, max(0.0, sensitivity))) key = (path, os.path.getmtime(path), detector, round(sensitivity, 3)) if key in _PSD_CACHE: return _PSD_CACHE[key] _status("Detecting camera cuts (PySceneDetect)", force=True) video = open_video(path) min_len = max(3, int(round(video.frame_rate * 0.25))) sm = SceneManager() if detector == "content": sm.add_detector(ContentDetector(threshold=45.0 - 33.0 * sensitivity, min_scene_len=min_len)) else: sm.add_detector(AdaptiveDetector(adaptive_threshold=5.0 - 3.5 * sensitivity, min_scene_len=min_len)) sm.detect_scenes(video, show_progress=False) times = [float(a.get_seconds()) for a, _ in sm.get_scene_list()[1:]] _PSD_CACHE[key] = times return times def find_cuts(path: str, analysis: dict, detector: str, sensitivity: float) -> tuple[list[int], str]: """Cut frames on the resampled timeline and which detector produced them.""" fps = float(analysis["fps"]) if detector in ("adaptive", "content"): times = scenedetect_cut_times(path, detector, sensitivity) if times is not None: n = analysis["n"] return sorted({int(round(t * fps)) for t in times if 0 < round(t * fps) < n}), f"PySceneDetect {detector}" return detect_cuts(analysis["score"], analysis["raw"], sensitivity, fps), "builtin" def detect_cuts(score: list[float], raw: list[float], sensitivity: float, fps: float) -> list[int]: """Frame indices that start a new shot (built-in detector). Higher sensitivity finds more cuts.""" sensitivity = float(min(1.0, max(0.0, sensitivity))) thr = 12.0 - 10.0 * sensitivity # local ratio threshold: 12 (strict) .. 2 (loose) abs_min = 0.12 - 0.09 * sensitivity # ignore tiny global changes gap = max(3, int(round(fps * 0.25))) # no two cuts closer than a quarter second cand = [i for i in range(1, len(score)) if score[i] >= thr and raw[i] >= abs_min] cuts: list[int] = [] for i in sorted(cand, key=lambda j: -score[j]): if all(abs(i - c) >= gap for c in cuts): cuts.append(i) return sorted(cuts) # ---------------------------------------------------------------------------- planning def plan_range(n: int, fps: float, start_s: float = 0.0, end_s: float = 0.0, max_total_s: float = 0.0, start_f: int | None = None, end_f: int | None = None) -> tuple[int, int]: """The part of the video the plan works on, in timeline frames [lo, hi): start / end frame (end 0 = the end of the video; older plans give seconds), then 'max total seconds' counted from the start.""" a = int(start_f) if start_f is not None else int(round(float(start_s or 0) * fps)) b = int(end_f) if end_f is not None else (int(round(float(end_s) * fps)) if end_s and float(end_s) > 0 else 0) lo = max(0, min(n - 1, a)) hi = n if b <= 0 else max(lo + 1, min(n, b)) if max_total_s and float(max_total_s) > 0: hi = min(hi, lo + max(1, int(round(float(max_total_s) * fps)))) return lo, hi def plan_segments(n: int, cuts: list[int], mode: str, max_len: int, min_len: int, max_parts: int = 0, max_total: int = 0, manual: list[int] | None = None, start: int = 0) -> list[dict]: """Split [start, n) into shots (n = the end frame). Returns [{start, end, cut_before}]. Lengths are in frames.""" lo = max(0, int(start or 0)) if max_total and max_total > 0: n = min(n, lo + int(max_total)) lo = min(lo, n - 1) max_len = max(1, int(max_len)) min_len = max(1, min(int(min_len), max_len)) cuts = sorted(c for c in set(cuts or []) if lo < c < n) cutset = set(cuts) if mode == "manual" and manual: bounds = sorted(set([lo, n] + [b for b in manual if lo < b < n])) elif mode == "single": # the whole range as ONE shot, whatever its length (the maximum is not applied) bounds = [lo, n] elif mode == "fixed": k = max(1, math.ceil((n - lo) / max_len)) bounds = sorted(set(lo + round((n - lo) * j / k) for j in range(k + 1))) else: # shots bounds = [lo] + cuts + [n] # merge shots that are too short into a neighbour, as long as the merge still fits changed = True while changed: changed = False for j in range(len(bounds) - 1): if bounds[j + 1] - bounds[j] >= min_len or len(bounds) <= 2: continue left = bounds[j + 1] - bounds[j - 1] if j > 0 else None right = bounds[j + 2] - bounds[j] if j + 2 < len(bounds) else None opts = [(v, side) for v, side in ((left, "l"), (right, "r")) if v is not None and v <= max_len] if not opts: continue side = min(opts)[1] bounds.pop(j if side == "l" else j + 1) changed = True break # split shots that are too long into equal parts out = [lo] for a, b in zip(bounds[:-1], bounds[1:]): k = max(1, math.ceil((b - a) / max_len)) out += [a + round((b - a) * j / k) for j in range(1, k + 1)] bounds = sorted(set(out)) segs = [{"start": a, "end": b, "cut_before": a in cutset} for a, b in zip(bounds[:-1], bounds[1:]) if b > a] if max_parts and max_parts > 0: segs = segs[:int(max_parts)] return segs DEFAULT_PLAN = { "video": "", "fps": 24.0, "grid": DEFAULT_GRID, "mode": "shots", "max_s": 4.5, "min_s": 1.0, "sensitivity": 0.5, "max_parts": 0, "max_total_s": 0.0, "start_s": 0.0, "end_s": 0.0, "bounds": [], "segs": [], "global_refs": [], "global_ref": "", "global_ref2": "", "global_prompt": "", "global_color_guide_ref1": "#00ff00", "global_color_guide_ref2": "#9b30ff", "picture_colors": [], "megapixels": 0.5, "multiple": 32, "generation_size_mode": "megapixels", "custom_width": 1024, "custom_height": 1024, "detector": "adaptive", "run": "recursive", "filters": {}, "skip_fill": "original", "cast": {}, "cast_split": False, "cast_only": False, "mask_cfg": {}, "vlm_cfg": {}, "audio_mode": "auto", "ref_details": {}, "picture_details": {}, "picture_cfg": {}, "auto_prompt": {}, "auto_prompts": {}, "between_shots": "keep", "shot_cache": "output", "shot_storage": "lossless", "unload_before_join": True, } # runs that render one shot at a time and store it on disk: "queue" (the panel re-queues the workflow after each # shot) and "recursive" (BFS Shot Join expands the graph for the next shot inside the same prompt) STORED_RUNS = ("queue", "recursive") # Second pass (upscale): the planner only works out the size; the graph does the upscale and the second sampling # (e.g. Minimax H3 Latent Upscaler (3D) in "target dimensions" mode, then a sampler on the upscaled latent). SECOND_PASS_DEFAULTS = {"enabled": False, "mode": "scale", "scale": 2.0, "width": 1280, "height": 704, "megapixels": 1.0, "align": 32} VAE_DOWNSAMPLE = 16 def second_pass_size(w: int, h: int, sp: dict | None) -> tuple[int, int]: """The upscaled size of a w x h generation, computed like Minimax H3 Latent Upscaler (3D): target pixels by multiplier, target dimensions or megapixels (keeping the aspect), rounded to `align`, then to whole latent cells.""" c = {**SECOND_PASS_DEFAULTS, **(sp or {})} mode = c["mode"] if mode == "dims": tw, th = float(c["width"]), float(c["height"]) elif mode == "megapixels": px = float(c["megapixels"]) * 1024 * 1024 th = (px / (w / h)) ** 0.5 tw = th * (w / h) else: tw, th = w * float(c["scale"]), h * float(c["scale"]) al = max(1, int(c["align"])) tw, th = round(tw / al) * al, round(th / al) * al d = VAE_DOWNSAMPLE return max(d, int(round(tw / d) * d)), max(d, int(round(th / d) * d)) def _load_plan(plan_json: str) -> dict: try: p = json.loads(plan_json or "{}") except json.JSONDecodeError as exc: raise ValueError(f"BFS Shot Planner: the plan is not valid JSON ({exc})") from exc out = dict(DEFAULT_PLAN) out.update({k: v for k, v in p.items() if v is not None}) out["global_refs"] = ref_list(p.get("global_refs"), out.get("global_ref"), out.get("global_ref2")) out["global_ref"], out["global_ref2"] = (out["global_refs"] + ["", ""])[:2] return out MAX_REFS = 100 # matches Comfy autogrow input capacity; continuity/setting pictures follow the shot refs def ref_list(refs, ref="", ref2="") -> list[str]: """A reference list of the plan: `refs` (plans from 1.113 on), else the old ref / ref2 pair. '' = not set (a shot's empty slot uses the global reference at the same position); empty slots at the end are dropped.""" out = [str(x or "") for x in refs] if isinstance(refs, list) else [str(ref or ""), str(ref2 or "")] out = out[:MAX_REFS] while out and not out[-1]: out.pop() return out def shot_ref_names(own: list, global_refs: list) -> list[str]: """The pictures a shot uses, in order: its own slot, else the global one at the same position. A slot with neither is left out, so the pictures after it move up one number.""" n = max(len(own), len(global_refs)) names = [(own[k] if k < len(own) else "") or (global_refs[k] if k < len(global_refs) else "") for k in range(n)] return [x for x in names if x][:MAX_REFS] def shot_refs(shot: dict) -> list: """The reference images of a shot handed out by the planner, in order.""" refs = shot.get("refs") if refs is None: refs = [shot.get("ref"), shot.get("ref2")] return [r for r in refs if r is not None] def resolve_plan(plan: dict, analysis: dict, path: str | None = None) -> list[dict]: """Shots for this plan: the boundaries edited in the panel, or an automatic split.""" fps = float(analysis["fps"]) grid = plan["grid"] max_len = snap_down(int(round(float(plan["max_s"]) * fps)), grid) min_len = max(1, int(round(float(plan["min_s"]) * fps))) lo, hi = plan_range(analysis["n"], fps, plan.get("start_s", 0), plan.get("end_s", 0), plan.get("max_total_s", 0), plan.get("start_f"), plan.get("end_f")) if path: cuts, _ = find_cuts(path, analysis, plan.get("detector", "adaptive"), float(plan["sensitivity"])) else: cuts = detect_cuts(analysis["score"], analysis["raw"], float(plan["sensitivity"]), fps) manual = plan.get("bounds") or None mode = "manual" if manual else plan["mode"] cast = cached_cast(path, analysis) if path else None # only once the people were analysed if cast is not None and plan.get("cast_split") and not manual: extra = [] bounds = [0] + cuts + [analysis["n"]] for a, b in zip(bounds[:-1], bounds[1:]): extra += person_change_points(cast, a, b, max(min_len, int(round(fps)))) cuts = sorted(set(cuts) | set(extra)) segs = plan_segments(hi, cuts, mode, max_len, min_len, int(plan.get("max_parts") or 0), 0, manual, start=lo) meta = plan.get("segs") or [] for i, s in enumerate(segs): m = meta[i] if i < len(meta) and isinstance(meta[i], dict) else {} s["enabled"] = bool(m.get("enabled", True)) s["refs"] = ref_list(m.get("refs"), m.get("ref"), m.get("ref2")) s["ref"], s["ref2"] = (s["refs"] + ["", ""])[:2] s["prompt"] = m.get("prompt") or "" s["chain"] = m.get("chain") or "head" # a continuing take starts on the previous end (cuts skip it) s["chain_frame"] = m.get("chain_frame") or "first" s["chain_history"] = int(m.get("chain_history") or 22) s["crop"] = bool(m.get("crop")) s["inpaint"] = bool(m.get("inpaint")) s["paste"] = bool(m.get("paste")) and not s["crop"] and not s["inpaint"] s["paste_text"] = str(m.get("paste_text") or "") s["color_guide"] = bool(m.get("color_guide")) s["color_guide_color"] = str(m.get("color_guide_color") or "") s["color_guide_text2"] = str(m.get("color_guide_text2") or "") s["color_guide_color2"] = str(m.get("color_guide_color2") or "") s["color_guide_points2"] = list(m.get("color_guide_points2") or []) s["color_guide_key2"] = int(m.get("color_guide_key2") or 0) s["color_guide_opacity"] = min(0.8, max(0.05, float(m.get("color_guide_opacity") or 0.35))) s["strength"] = min(1.0, max(0.05, float(m.get("strength") or 1.0))) s["mask"] = m.get("mask") or {} s["target"] = m.get("target") or "" s["cut_before"] = bool(s.get("cut_before")) or (s["start"] in cuts) s["gen_len"] = snap_up(s["end"] - s["start"], grid) s["people"], s["main"] = [], -1 s["color_guide_color"] = s["color_guide_color"] or str(plan.get("global_color_guide_ref1") or "#00ff00") s["color_guide_color2"] = s["color_guide_color2"] or str(plan.get("global_color_guide_ref2") or "#9b30ff") s["color_guide_marks"] = color_marks(plan, m) s["color_guide_color"] = s["color_guide_marks"][0]["color"] if cast is not None: sp = shot_people(cast, s["start"], s["end"]) s["people"], s["main"] = sp["people"], sp["main"] included = [p for p in sp["people"] if not (plan.get("cast") or {}).get(str(p), {}).get("ignore")] s["cast_skip"] = bool(plan.get("cast_only")) and not included return segs # ---------------------------------------------------------------------------- content filters DEFAULT_FILTERS = { "person": False, "min_person_area": 0.0, "max_persons": 0, "face": False, "skip_dark": False, "dark_level": 0.06, "skip_static": False, "static_level": 0.004, "min_frames": 0, "samples": 6, } _DET: dict[str, Any] = {} _STATS_CACHE: dict[tuple, dict] = {} def _yolo(kind: str): """YOLO model for 'person' or 'face' from models/ultralytics, or None (OpenCV fallback).""" if _DET.get(kind) is not None: return _DET[kind] model = None try: from ultralytics import YOLO root = os.path.join(folder_paths.models_dir, "ultralytics") cands = [] for sub in ("bbox", "segm", "pose", ""): d = os.path.join(root, sub) if os.path.isdir(d): cands += [os.path.join(d, f) for f in sorted(os.listdir(d)) if f.endswith(".pt") and kind in f.lower()] if cands: model = YOLO(cands[0]) except Exception: # noqa: BLE001 - fall back to OpenCV model = None _DET[kind] = model return model def _detect(frames: list[np.ndarray]) -> list[dict]: """Per frame: person boxes (area fraction) and face count.""" import cv2 out = [{"persons": [], "faces": 0} for _ in frames] pm, fm = _yolo("person"), _yolo("face") if pm is not None: for i, r in enumerate(pm(frames, verbose=False, conf=0.35, classes=[0])): h, w = frames[i].shape[:2] for b in r.boxes.xyxy.cpu().numpy() if r.boxes is not None else []: out[i]["persons"].append(float((b[2] - b[0]) * (b[3] - b[1]) / (w * h))) else: hog = cv2.HOGDescriptor() hog.setSVMDetector(cv2.HOGDescriptor_getDefaultPeopleDetector()) for i, fr in enumerate(frames): h, w = fr.shape[:2] rects, _ = hog.detectMultiScale(cv2.cvtColor(fr, cv2.COLOR_RGB2GRAY), winStride=(8, 8)) out[i]["persons"] = [float(rw * rh / (w * h)) for (_, _, rw, rh) in rects] if fm is not None: for i, r in enumerate(fm(frames, verbose=False, conf=0.4)): out[i]["faces"] = int(len(r.boxes)) if r.boxes is not None else 0 else: casc = cv2.CascadeClassifier(os.path.join(cv2.data.haarcascades, "haarcascade_frontalface_default.xml")) for i, fr in enumerate(frames): out[i]["faces"] = int(len(casc.detectMultiScale(cv2.cvtColor(fr, cv2.COLOR_RGB2GRAY), 1.1, 5))) return out def shot_stats(path: str, analysis: dict, seg: dict, samples: int = 6) -> dict: """Sampled content statistics for one shot (cached).""" key = (path, os.path.getmtime(path), float(analysis["fps"]), seg["start"], seg["end"], samples) if key in _STATS_CACHE: return _STATS_CACHE[key] src = _timeline(analysis["n_src"], analysis["fps_src"], float(analysis["fps"])) k = max(1, min(samples, seg["end"] - seg["start"])) pick = np.linspace(seg["start"], seg["end"] - 1, k).round().astype(int) w = 640 h = max(32, int(round(w * analysis["height"] / max(1, analysis["width"])))) frames = _read_frames(path, src[pick], (w, h)) det = _detect(frames) raw = analysis.get("raw") or [] motion = float(np.mean(raw[seg["start"] + 1:seg["end"]])) if seg["end"] - seg["start"] > 1 and raw else 0.0 st = { "persons": int(max(len(d["persons"]) for d in det)), "person_area": round(float(max([max(d["persons"]) for d in det if d["persons"]] or [0.0])), 4), "person_frames": int(sum(1 for d in det if d["persons"])), "faces": int(max(d["faces"] for d in det)), "brightness": round(float(np.mean([f.mean() / 255.0 for f in frames])), 4), "motion": round(motion, 4), "sampled": int(k), } _STATS_CACHE[key] = st return st def skip_reason(stats: dict, length: int, f: dict) -> str: """Why a shot should be skipped under these filters ('' = keep).""" if f.get("min_frames") and length < int(f["min_frames"]): return f"shorter than {int(f['min_frames'])} frames" if f.get("skip_dark") and stats["brightness"] < float(f.get("dark_level", 0.06)): return "dark / fade" if f.get("skip_static") and stats["motion"] < float(f.get("static_level", 0.004)): return "static" if f.get("person") and stats["persons"] == 0: return "no person" if f.get("person") and float(f.get("min_person_area") or 0) > 0 and stats["person_area"] < float(f["min_person_area"]): return f"person smaller than {float(f['min_person_area']) * 100:.0f}% of the frame" if int(f.get("max_persons") or 0) > 0 and stats["persons"] > int(f["max_persons"]): return f"more than {int(f['max_persons'])} people" if f.get("face") and stats["faces"] == 0: return "no face" return "" def filters_active(f: dict) -> bool: return any(f.get(k) for k in ("person", "face", "skip_dark", "skip_static")) or \ int(f.get("max_persons") or 0) > 0 or int(f.get("min_frames") or 0) > 0 def apply_filters(plan: dict, analysis: dict, path: str, segs: list[dict]) -> list[dict]: """Mark each shot with stats and an automatic skip reason; per-shot 'force' overrides it.""" f = dict(DEFAULT_FILTERS); f.update(plan.get("filters") or {}) meta = plan.get("segs") or [] need = filters_active(f) for i, s in enumerate(segs): force = (meta[i] or {}).get("force", "auto") if i < len(meta) and isinstance(meta[i], dict) else "auto" s["force"] = force s["stats"] = shot_stats(path, analysis, s, int(f.get("samples") or 6)) if need else None s["skip_reason"] = skip_reason(s["stats"], s["end"] - s["start"], f) if need else "" if not s["skip_reason"] and s.get("cast_skip"): s["skip_reason"] = "no detected person" s["run"] = s["enabled"] and (force == "run" or (force != "skip" and not s["skip_reason"])) return segs # ---------------------------------------------------------------------------- cast (people by face) _FACE_APP: Any = None _CAST_CACHE: dict[tuple, dict] = {} def _face_app(): """InsightFace detector + ArcFace recogniser (buffalo_l), or None when unavailable.""" global _FACE_APP if _FACE_APP is not None: return _FACE_APP or None try: import insightface root = os.path.join(folder_paths.models_dir, "insightface") # CPU only: light, no VRAM, and no CUDA/cuDNN loading (a missing cuDNN aborts the whole process) app = insightface.app.FaceAnalysis(name="buffalo_l", root=root, allowed_modules=["detection", "recognition"], providers=["CPUExecutionProvider"]) app.prepare(ctx_id=-1, det_size=(320, 320)) # small models: a few threads are faster than onnxruntime's default pool (one per core) and stay light import onnxruntime as ort so = ort.SessionOptions() so.intra_op_num_threads = min(4, os.cpu_count() or 4) so.inter_op_num_threads = 1 for model in app.models.values(): model.session = ort.InferenceSession(model.model_file, so, providers=["CPUExecutionProvider"]) _FACE_APP = app except Exception as exc: # noqa: BLE001 print(f"[BFS Shot Planner] face analysis unavailable: {exc!r}") _FACE_APP = False return _FACE_APP or None def _cast_key(path: str, analysis: dict, step_s: float = 0.5, threshold: float = 0.42, min_share: float = 0.01) -> tuple: return (path, os.path.getmtime(path), float(analysis["fps"]), round(step_s, 3), round(threshold, 3), round(min_share, 4), "insightface" if _face_app() is not None else "opencv" if _cv_face() is not None else "sam3") APPEARANCE_THRESHOLD = 0.86 # cosine on the colour signature (only when DINOv2 cannot load) SFACE_THRESHOLD = 0.36 # cosine on SFace features (OpenCV's same-identity threshold is 0.363) _CV_FACE: dict = {} CV_FACE_MODELS = { "face_detection_yunet_2023mar.onnx": "https://github.com/opencv/opencv_zoo/raw/main/models/face_detection_yunet/face_detection_yunet_2023mar.onnx", "face_recognition_sface_2021dec.onnx": "https://github.com/opencv/opencv_zoo/raw/main/models/face_recognition_sface/face_recognition_sface_2021dec.onnx", } def _cv_face(): """OpenCV Zoo's YuNet (detection) + SFace (recognition): Apache-2.0, run by OpenCV itself (no extra package), downloaded once into models/opencv_face (0.2 + 37 MB). None when it cannot load.""" if "m" not in _CV_FACE: try: import cv2 import urllib.request d = os.path.join(folder_paths.models_dir, "opencv_face") os.makedirs(d, exist_ok=True) for name, url in CV_FACE_MODELS.items(): f = os.path.join(d, name) if not os.path.exists(f) or os.path.getsize(f) < 100000: print(f"[BFS Shot Planner] downloading {name} (OpenCV Zoo, Apache-2.0)", flush=True) urllib.request.urlretrieve(url, f + ".part") os.replace(f + ".part", f) det = cv2.FaceDetectorYN.create(os.path.join(d, "face_detection_yunet_2023mar.onnx"), "", (320, 320), 0.6, 0.3, 50) rec = cv2.FaceRecognizerSF.create(os.path.join(d, "face_recognition_sface_2021dec.onnx"), "") _CV_FACE["m"] = (det, rec) except Exception as exc: # noqa: BLE001 print(f"[BFS Shot Planner] OpenCV face models unavailable ({exc!r})", flush=True) _CV_FACE["m"] = None return _CV_FACE["m"] def appearance_signature(frame: np.ndarray, box) -> np.ndarray: """A license-free identity signature for grouping people inside ONE video: colour histograms of the head (hair, skin) and of the torso under it (clothes), plus a tiny grey thumbnail of the face. Weaker than ArcFace across videos, enough to tell the people of a clip apart.""" import cv2 H, W = frame.shape[:2] x0, y0, x1, y1 = [int(v) for v in box] bw, bh = max(2, x1 - x0), max(2, y1 - y0) head = frame[max(0, y0 - bh // 4):min(H, y1), max(0, x0 - bw // 6):min(W, x1 + bw // 6)] torso = frame[min(H - 1, y1):min(H, y1 + 2 * bh), max(0, x0 - bw // 3):min(W, x1 + bw // 3)] def hist(img): if img.size == 0: return np.zeros(12 * 4 * 4, np.float32) hsv = cv2.cvtColor(img, cv2.COLOR_RGB2HSV) h = cv2.calcHist([hsv], [0, 1, 2], None, [12, 4, 4], [0, 180, 0, 256, 0, 256]).ravel().astype(np.float32) return np.sqrt(h / max(1.0, h.sum())) # Hellinger: cosine of sqrt-histograms face = frame[max(0, y0):min(H, y1), max(0, x0):min(W, x1)] thumb = cv2.resize(cv2.cvtColor(face, cv2.COLOR_RGB2GRAY), (8, 8), interpolation=cv2.INTER_AREA).astype(np.float32).ravel() \ if face.size else np.zeros(64, np.float32) thumb = (thumb - thumb.mean()) / (thumb.std() + 1e-6) * 0.05 v = np.concatenate([hist(head), 1.3 * hist(torso), thumb]) return v / (np.linalg.norm(v) + 1e-8) def _sam3_faces(frames: list, threshold: float = 0.4) -> list: """Faces per frame with SAM 3.1 (ComfyUI core): [[(x0, y0, x1, y1, score), ...], ...] in pixels.""" from comfy_extras.nodes_sam3 import SAM3_Detect model, clip = _sam3() cond = clip.encode_from_tokens_scheduled(clip.tokenize("face")) out = [] for k in range(0, len(frames), 8): _status("Finding faces (SAM 3.1)", k, len(frames)) batch = torch.from_numpy(np.stack(frames[k:k + 8]).astype(np.float32) / 255.0) boxes = SAM3_Detect.execute(model=model, image=batch, conditioning=cond, threshold=threshold, refine_iterations=0).args[1] for fb in boxes: out.append([(b["x"], b["y"], b["x"] + b["width"], b["y"] + b["height"], b.get("score", 1.0)) for b in (fb or [])]) return out def cached_cast(path: str, analysis: dict) -> dict | None: """The cast of a video when it was already analysed with the default settings, else None (never computes).""" try: return _CAST_CACHE.get(_cast_key(path, analysis)) except OSError: return None def analyze_cast(path: str, analysis: dict, step_s: float = 0.5, threshold: float = 0.42, min_share: float = 0.01, max_faces: int = 3) -> dict: """Detect faces every `step_s` seconds (CPU), keep the `max_faces` most probable ones per frame, group them into people by ArcFace similarity. Returns {"people": [{id, thumb, count, share}], "samples": [{f, faces: [{pid, area}]}]} where `f` is a timeline frame. Person ids are stable for the same video and settings. """ import cv2 key = _cast_key(path, analysis, step_s, threshold, min_share) if key in _CAST_CACHE: return _CAST_CACHE[key] # InsightFace (ArcFace) when installed; else OpenCV's YuNet + SFace (Apache-2.0, no extra package); else SAM 3.1 # faces grouped by colours (hair, skin, clothes) app = _face_app() cvf = _cv_face() if app is None else None if app is None: threshold = SFACE_THRESHOLD if cvf is not None else APPEARANCE_THRESHOLD print("[BFS Shot Planner] Find people: insightface is not installed, using " + ("OpenCV YuNet + SFace faces" if cvf is not None else "SAM 3.1 faces + colours (hair, skin, clothes)"), flush=True) fps = float(analysis["fps"]) step = max(1, int(round(step_s * fps))) frames_idx = np.arange(0, analysis["n"], step) src = _timeline(analysis["n_src"], analysis["fps_src"], fps) w = 960 h = max(32, int(round(w * analysis["height"] / max(1, analysis["width"])))) frames = _read_frames(path, src[frames_idx], (w, h), stage="Reading frames for faces") dets = [] # (sample index, embedding, area fraction, crop) if app is None and cvf is not None: det, rec = cvf det.setInputSize((w, h)) for si, fr in enumerate(frames): _status("Finding faces", si, len(frames)) bgr = cv2.cvtColor(fr, cv2.COLOR_RGB2BGR) _, faces = det.detect(bgr) cand = [] for row in (faces if faces is not None else []): x, y, bw, bh, score = float(row[0]), float(row[1]), float(row[2]), float(row[3]), float(row[-1]) area = max(0.0, bw) * max(0.0, bh) / float(w * h) if area >= 0.0015 and score >= 0.6: cand.append((score, row, area)) for score, row, area in sorted(cand, key=lambda c: -c[0])[:max_faces]: feat = rec.feature(rec.alignCrop(bgr, row)).ravel().astype(np.float32) x0, y0, x1, y1 = int(row[0]), int(row[1]), int(row[0] + row[2]), int(row[1] + row[3]) pad = int(0.25 * max(x1 - x0, y1 - y0)) crop = fr[max(0, y0 - pad):min(h, y1 + pad), max(0, x0 - pad):min(w, x1 + pad)] dets.append((si, feat / (np.linalg.norm(feat) + 1e-8), area, crop)) elif app is None: for si, (fr, faces) in enumerate(zip(frames, _sam3_faces(frames, threshold=0.55))): cand = [] for x0, y0, x1, y1, score in faces: bw, bh = max(1.0, x1 - x0), max(1.0, y1 - y0) area = bw * bh / float(w * h) if area >= 0.0015 and 0.55 <= bw / bh <= 1.4: # a face-shaped box (logos and text are skipped) cand.append((float(score), (x0, y0, x1, y1), area)) for score, box, area in sorted(cand, reverse=True)[:max_faces]: x0, y0, x1, y1 = [int(v) for v in box] pad = int(0.25 * max(x1 - x0, y1 - y0)) crop = fr[max(0, y0 - pad):min(h, y1 + pad), max(0, x0 - pad):min(w, x1 + pad)] dets.append((si, appearance_signature(fr, box), area, crop)) frames_for_faces = frames if app is not None else [] from_insight = app is not None if from_insight: from insightface.app.common import Face rec = app.models["recognition"] for si, fr in enumerate(frames_for_faces): _status("Finding faces", si, len(frames)) bgr = cv2.cvtColor(fr, cv2.COLOR_RGB2BGR) boxes, kpss = app.det_model.detect(bgr, max_num=0, metric="default") # only the most probable faces of the frame get recognised, so a crowd stays cheap and out of the cast cand = [] for k in range(boxes.shape[0]): x0, y0, x1, y1 = [int(v) for v in boxes[k, :4]] area = max(0, x1 - x0) * max(0, y1 - y0) / float(w * h) if area >= 0.0015 and boxes[k, 4] >= 0.5: cand.append((float(boxes[k, 4]), k, area)) for score, k, area in sorted(cand, reverse=True)[:max_faces]: f = Face(bbox=boxes[k, :4], kps=kpss[k] if kpss is not None else None, det_score=score) rec.get(bgr, f) x0, y0, x1, y1 = [int(v) for v in f.bbox] pad = int(0.25 * max(x1 - x0, y1 - y0)) crop = fr[max(0, y0 - pad):min(h, y1 + pad), max(0, x0 - pad):min(w, x1 + pad)] dets.append((si, f.normed_embedding.astype(np.float32), area, crop)) # greedy online clustering on cosine similarity, then merge close clusters cents, members = [], [] for k, (_, e, _, _) in enumerate(dets): if cents: sims = np.array([c @ e / (np.linalg.norm(c) + 1e-8) for c in cents]) j = int(sims.argmax()) if sims[j] >= threshold: members[j].append(k); cents[j] = cents[j] + e continue cents.append(e.copy()); members.append([k]) merged = True while merged: merged = False for a in range(len(cents)): for b in range(a + 1, len(cents)): ca, cb = cents[a] / np.linalg.norm(cents[a]), cents[b] / np.linalg.norm(cents[b]) if ca @ cb >= threshold: cents[a] = cents[a] + cents[b]; members[a] += members[b] del cents[b]; del members[b]; merged = True break if merged: break total = max(1, len(frames)) groups = [m for m in members if len({dets[k][0] for k in m}) / total >= min_share] groups.sort(key=lambda m: -len({dets[k][0] for k in m})) people, owner = [], {} for pid, m in enumerate(groups): best = max(m, key=lambda k: dets[k][2]) crop = cv2.resize(dets[best][3], (96, 96), interpolation=cv2.INTER_AREA) ok, buf = cv2.imencode(".jpg", cv2.cvtColor(crop, cv2.COLOR_RGB2BGR), [cv2.IMWRITE_JPEG_QUALITY, 80]) seen = sorted({dets[k][0] for k in m}) people.append({"id": pid, "thumb": "data:image/jpeg;base64," + base64.b64encode(buf.tobytes()).decode(), "count": len(seen), "share": round(len(seen) / total, 4), "first": int(frames_idx[seen[0]]), "last": int(frames_idx[seen[-1]])}) for k in m: owner[k] = pid samples = [{"f": int(f), "faces": []} for f in frames_idx] for k, (si, _, area, _) in enumerate(dets): if k in owner: samples[si]["faces"].append({"pid": owner[k], "area": round(float(area), 5)}) out = {"people": people, "samples": samples, "step": step} _CAST_CACHE[key] = out return out def shot_people(cast: dict, start: int, end: int) -> dict: """Who appears in [start, end): per person, sampled frames seen and summed face area; plus the main one.""" score: dict[int, list[float]] = {} for smp in cast["samples"]: if start <= smp["f"] < end: for fc in smp["faces"]: v = score.setdefault(fc["pid"], [0, 0.0]); v[0] += 1; v[1] += fc["area"] main = max(score, key=lambda p: (score[p][0], score[p][1])) if score else -1 return {"people": sorted(score), "main": main} def person_change_points(cast: dict, start: int, end: int, min_run: int) -> list[int]: """Frames inside [start, end) where the main (largest) face switches to another person for at least min_run.""" runs = [] for smp in cast["samples"]: if start <= smp["f"] < end: main = max(smp["faces"], key=lambda fc: fc["area"])["pid"] if smp["faces"] else -1 if main >= 0 and (not runs or runs[-1][1] != main): runs.append([smp["f"], main]) pts = [] for (f0, p0), (f1, p1) in zip(runs, runs[1:]): if f1 - (pts[-1] if pts else start) >= min_run and end - f1 >= min_run: pts.append(f1) return pts # ---------------------------------------------------------------------------- shot masks (SAM 3) and crop / uncrop SAM3_FILE = "sam3.1_multiplex_fp16.safetensors" SAM3_URL = "https://huggingface.co/Comfy-Org/sam3.1/resolve/main/checkpoints/sam3.1_multiplex_fp16.safetensors" _SAM3: dict[str, Any] = {} _MASK_CACHE: dict[tuple, dict] = {} # global mask settings (the planner's Mask card); a shot only says what to segment (text / points / key frame) DEFAULT_MASK = {"text": "", "points": [], "key": 0, "threshold": 0.5, "max_objects": 4, "invert": False, "fill_holes": True, "temporal_expand": 2, "blockify": 0, "padding": 0.15, "expand": 16, "feather": 12, "paste": "box"} SHOT_MASK_KEYS = ("text", "points", "key") def _sam3_path() -> str: """The official SAM 3.1 checkpoint in models/checkpoints, downloaded once when missing.""" p = folder_paths.get_full_path("checkpoints", SAM3_FILE) if p: return p import urllib.request dst = os.path.join(folder_paths.get_folder_paths("checkpoints")[0], SAM3_FILE) tmp = dst + ".part" print(f"[BFS Shot Planner] downloading {SAM3_FILE} (1.7 GB) from Comfy-Org/sam3.1 ...") with urllib.request.urlopen(SAM3_URL) as r, open(tmp, "wb") as f: total, done, step = int(r.headers.get("Content-Length") or 0), 0, 0 while True: chunk = r.read(1 << 22) if not chunk: break f.write(chunk); done += len(chunk) if total and done * 10 // total > step: step = done * 10 // total print(f"[BFS Shot Planner] {SAM3_FILE}: {step * 10}%") os.replace(tmp, dst) return dst def _sam3(): if "model" not in _SAM3: import comfy.sd out = comfy.sd.load_checkpoint_guess_config(_sam3_path(), output_vae=False, output_clip=True, embedding_directory=folder_paths.get_folder_paths("embeddings")) _SAM3["model"], _SAM3["clip"] = out[0], out[1] return _SAM3["model"], _SAM3["clip"] def mask_spec(m: dict | None, cfg: dict | None = None) -> dict: """Global settings (cfg) with the shot's own target (text / points / key frame) on top.""" spec = dict(DEFAULT_MASK) spec.update({k: v for k, v in (cfg or {}).items() if v is not None and k not in SHOT_MASK_KEYS}) spec.update({k: v for k, v in (m or {}).items() if v is not None and k in SHOT_MASK_KEYS}) return spec # external masks (e.g. rotoscoped in another tool) replace SAM 3. Most specific first: the shot's own mask video # (covers only that shot), the shot's own SAM 3 text / points, the planner's `mask` input (the whole video), the # panel's mask video (the whole video). _EXT_MASK: dict[str, torch.Tensor] = {} # video path -> the planner's `mask` input from its last run def plan_spec(p: dict, m: dict | None, path: str = "") -> dict: """mask_spec plus the external mask this shot uses, if any (spec['ext']).""" spec = mask_spec(m, p.get("mask_cfg")) if (m or {}).get("video"): spec["ext"] = {"file": m["video"], "scope": "shot"} elif (m or {}).get("points") or str((m or {}).get("text") or "").strip(): pass # the shot's own SAM 3 selection elif path and (p.get("track") or {}).get("points"): # one selection tracked through the whole video fps = float(p.get("fps") or 24) spec["ext"] = {"track": {**p["track"], "start_s": p.get("start_s", 0), "end_s": p.get("end_s", 0), "start_f": p.get("start_f"), "end_f": p.get("end_f"), "max_total_s": p.get("max_total_s", 0), "fps": fps}, "path": path} elif path and path in _EXT_MASK: spec["ext"] = {"tensor": path} elif p.get("mask_video"): spec["ext"] = {"file": p["mask_video"], "scope": "video"} return spec def spec_has_mask(spec: dict) -> bool: return bool(spec.get("ext") or spec.get("points") or str(spec.get("text") or "").strip()) def ext_masks(ext: dict, analysis: dict, start: int, length: int, w: int, h: int) -> torch.Tensor: """[length,h,w] uint8 from an external mask (white = the subject) for timeline frames start..start+length.""" idx = np.clip(np.arange(start, start + length), 0, analysis["n"] - 1) if "track" in ext: tm, lo = track_video(ext["path"], analysis, ext["track"]) sel = torch.from_numpy(np.clip(idx - lo, 0, tm.shape[0] - 1).astype(np.int64)) m = torch.nn.functional.interpolate(tm[sel][:, None].float(), size=(h, w), mode="nearest")[:, 0] return (m > 0.5).to(torch.uint8) if "tensor" in ext: t = _EXT_MASK[ext["tensor"]].float() if t.ndim == 4: # an IMAGE batch (e.g. Load Video of a black/white video) t = t[..., :3].mean(-1) if t.ndim == 2: t = t[None] src = _timeline(analysis["n_src"], analysis["fps_src"], float(analysis["fps"]))[idx] n = t.shape[0] # one mask per source frame; otherwise spread over the video ii = src if n == analysis["n_src"] else np.round(src * (n - 1) / max(1, analysis["n_src"] - 1)) m = t[torch.from_numpy(np.clip(ii, 0, n - 1).astype(np.int64))] m = torch.nn.functional.interpolate(m[:, None], size=(h, w), mode="bilinear", align_corners=False)[:, 0] else: fpath = _input_path(ext["file"]) pr = _probe(fpath) t = (idx - start) / float(analysis["fps"]) if ext.get("scope") == "shot" else idx / float(analysis["fps"]) mi = np.clip(np.round(t * pr["fps_src"]), 0, pr["n_src"] - 1).astype(np.int64) fr = _read_frames(fpath, mi, (w, h), stage="Reading the mask video") m = torch.from_numpy(np.stack(fr).astype(np.float32).mean(-1) / 255.0) return (m > 0.5).to(torch.uint8) def _ext_name(ext: dict) -> str: if "track" in ext: t = ext["track"] return f"the whole-video track ({len(t.get('points') or [])} points at frame {t.get('frame')})" if "tensor" in ext: t = _EXT_MASK.get(ext["tensor"]) return f"the planner's mask input ({tuple(t.shape) if t is not None else 'gone'})" return f"mask video '{ext['file']}'" + (" (this shot)" if ext.get("scope") == "shot" else " (whole video)") def shape_mask(masks: torch.Tensor, spec: dict) -> torch.Tensor: """invert -> fill holes -> temporal expand -> blockify, on [N,H,W] 0/1 masks.""" x = masks.float() if spec.get("invert"): x = 1 - x if spec.get("fill_holes"): try: from scipy.ndimage import binary_fill_holes x = torch.from_numpy(np.stack([binary_fill_holes(f > 0.5) for f in x.numpy()]).astype(np.float32)) except ImportError: pass t = int(spec.get("temporal_expand") or 0) if t > 0 and x.shape[0] > 1: N, H, W = x.shape x = torch.nn.functional.max_pool1d(x.permute(1, 2, 0).reshape(-1, 1, N), 2 * t + 1, 1, t) \ .reshape(H, W, N).permute(2, 0, 1) b = int(spec.get("blockify") or 0) if b > 1: H, W = x.shape[1:] cov = torch.nn.functional.avg_pool2d(x[:, None], b, b, ceil_mode=True)[:, 0] x = (cov >= 0.5).float().repeat_interleave(b, 1).repeat_interleave(b, 2)[:, :H, :W] return (x > 0.5).to(torch.uint8) _TRACK_CACHE: dict[tuple, tuple] = {} TRACK_CHUNK = 150 # frames tracked at once; each chunk starts from the previous one's last mask def track_video(path: str, analysis: dict, t: dict) -> tuple[torch.Tensor, int]: """One selection (points on one frame, optional text) tracked by SAM 3 through the plan's whole range, forward and backward from that frame, in chunks that hand their last mask to the next (the text, when given, finds the subject again after a camera cut). Returns (masks [hi-lo, h, w] uint8 at a 640 px working size, lo).""" fps = float(analysis["fps"]) lo, hi = plan_range(analysis["n"], fps, t.get("start_s", 0), t.get("end_s", 0), t.get("max_total_s", 0), t.get("start_f"), t.get("end_f")) key = (path, os.path.getmtime(path), fps, lo, hi, json.dumps({k: t.get(k) for k in ("points", "frame", "text", "threshold")}, sort_keys=True)) if key in _TRACK_CACHE: return _TRACK_CACHE[key] from comfy_extras.nodes_sam3 import SAM3_Detect model, clip = _sam3() sw, sh = analysis["width"], analysis["height"] sc = 640 / max(sw, sh) w, h = max(32, int(sw * sc) // 2 * 2), max(32, int(sh * sc) // 2 * 2) src = _timeline(analysis["n_src"], analysis["fps_src"], fps) k = max(lo, min(hi - 1, int(t.get("frame") or lo))) text = str(t.get("text") or "").strip() cond = clip.encode_from_tokens_scheduled(clip.tokenize(text)) if text else None spec = dict(DEFAULT_MASK, threshold=float(t.get("threshold") or DEFAULT_MASK["threshold"]), max_objects=1) out = np.zeros((hi - lo, h, w), np.uint8) t0 = time.time() print(f"[BFS Shot] SAM 3: tracking {len(t.get('points') or [])} points from frame {k} through frames {lo}-{hi - 1}" + (f" (re-found by '{text}')" if text else "") + "…", flush=True) allf = _read_frames(path, src[np.arange(lo, hi)], (w, h), stage="Reading the video for the track") # read once key_img = torch.from_numpy(allf[k - lo][None].astype(np.float32) / 255.0) pts = t.get("points") or [] pos = [{"x": q["x"] * w, "y": q["y"] * h} for q in pts if q.get("label", 1)] neg = [{"x": q["x"] * w, "y": q["y"] * h} for q in pts if not q.get("label", 1)] first = SAM3_Detect.execute(model=model, image=key_img, positive_coords=json.dumps(pos), negative_coords=json.dumps(neg), threshold=float(spec["threshold"]), refine_iterations=2).args[0][:1].float() for direction in (1, -1): # forward from the key frame, then backward frames = list(range(k, hi)) if direction == 1 else list(range(k, lo - 1, -1)) init, done = first, 0 while done < len(frames): part = frames[max(0, done - 1): done + TRACK_CHUNK] # one frame of overlap carries the mask on seed = init if init is not None and bool(init.any()) else None if seed is None and cond is None: break # the subject is lost and there is no text to find it again imgs = torch.from_numpy(np.stack([allf[f - lo] for f in part]).astype(np.float32) / 255.0) mk = (_track(model, imgs, seed, cond, spec) > 0.5).numpy().astype(np.uint8) for j, f in enumerate(part): out[f - lo] |= mk[j] init = torch.from_numpy(mk[-1:]).float() done += TRACK_CHUNK _status(f"Tracking with SAM 3: {min(done, len(frames))}/{len(frames)} frames", min(done, len(frames)), len(frames)) _node_boundary() res = (torch.from_numpy(out), lo) _TRACK_CACHE[key] = res print(f"[BFS Shot] SAM 3: track done in {time.time() - t0:.0f}s, {mask_report(res[0])}", flush=True) return res def _track(model, imgs: torch.Tensor, init: torch.Tensor | None, cond, spec: dict) -> torch.Tensor: from comfy_extras.nodes_sam3 import SAM3_TrackToMask, SAM3_VideoTrack data = SAM3_VideoTrack.execute(images=imgs, model=model, initial_mask=init, conditioning=cond, detection_threshold=float(spec["threshold"]), max_objects=int(spec["max_objects"]), detect_interval=1).args[0] return SAM3_TrackToMask.execute(track_data=data, object_indices="").args[0].float().cpu() def segment_frames(imgs: torch.Tensor, spec: dict) -> torch.Tensor: """[N,H,W,3] -> [N,H,W] masks with SAM 3: tracked from the points on the key frame, or from the text.""" from comfy_extras.nodes_sam3 import SAM3_Detect model, clip = _sam3() text = str(spec.get("text") or "").strip() cond = clip.encode_from_tokens_scheduled(clip.tokenize(text)) if text else None pts = spec.get("points") or [] N, H, W = imgs.shape[:3] if not pts: if cond is None: raise ValueError("the shot's mask needs a text prompt or points") return _track(model, imgs, None, cond, spec) key = max(0, min(N - 1, int(spec.get("key") or 0))) pos = [{"x": p["x"] * W, "y": p["y"] * H} for p in pts if p.get("label", 1)] neg = [{"x": p["x"] * W, "y": p["y"] * H} for p in pts if not p.get("label", 1)] first = SAM3_Detect.execute(model=model, image=imgs[key:key + 1], positive_coords=json.dumps(pos), negative_coords=json.dumps(neg), threshold=float(spec["threshold"]), refine_iterations=2).args[0][:1].float() fwd = _track(model, imgs[key:], first, None, spec) # key frame -> end if key == 0: return fwd back = _track(model, imgs[:key + 1].flip(0), first, None, spec).flip(0) # key frame -> start return torch.cat([back[:-1], fwd], 0) def _mask_key(path: str, analysis: dict, start: int, length: int, spec: dict) -> tuple: ext = spec.get("ext") if ext and "track" in ext: e = ("track", json.dumps(ext["track"], sort_keys=True)) elif ext and "tensor" in ext: t = _EXT_MASK.get(ext["tensor"]) e = ("tensor", id(t), tuple(t.shape) if t is not None else None) elif ext: f = folder_paths.get_annotated_filepath(ext["file"]) e = ("file", ext["file"], ext.get("scope"), os.path.getmtime(f) if f and os.path.isfile(f) else 0) else: e = None return (path, os.path.getmtime(path), float(analysis["fps"]), int(start), int(length), json.dumps({k: spec[k] for k in ("text", "points", "key", "threshold", "max_objects")}, sort_keys=True), e) def shot_mask(path: str, analysis: dict, start: int, length: int, spec: dict) -> dict: """Masks of one shot (timeline frames start..start+length) at a working size, cached; plus the crop box.""" key = _mask_key(path, analysis, start, length, spec) if key not in _MASK_CACHE: src = _timeline(analysis["n_src"], analysis["fps_src"], float(analysis["fps"])) idx = np.clip(np.arange(start, start + length), 0, analysis["n"] - 1) sw, sh = analysis["width"], analysis["height"] s = 640 / max(sw, sh) w, h = max(32, int(sw * s) // 2 * 2), max(32, int(sh * s) // 2 * 2) if spec.get("ext"): t0 = time.time() _MASK_CACHE[key] = {"masks": ext_masks(spec["ext"], analysis, start, length, w, h), "size": (w, h)} print(f"[BFS Shot] {_ext_name(spec['ext'])} instead of SAM 3, timeline {start}-{start + length - 1}: " f"{mask_report(_MASK_CACHE[key]['masks'])} ({time.time() - t0:.1f}s)", flush=True) out = dict(_MASK_CACHE[key]) out["masks"] = shape_mask(out["masks"], spec) out["box"] = crop_box(out["masks"], float(spec["padding"])) return out frames = _read_frames(path, src[idx], (w, h), stage="Reading the shot") imgs = torch.from_numpy(np.stack(frames).astype(np.float32) / 255.0) _status(f"Segmenting {len(frames)} frames with SAM 3", force=True) what = f"{len(spec.get('points') or [])} points on frame {int(spec.get('key') or 0)}" if spec.get("points") \ else f"'{spec.get('text', '')}'" print(f"[BFS Shot] SAM 3: segmenting {what} on {len(frames)} frames ({w}x{h}, timeline {start}-{start + length - 1})…", flush=True) t0 = time.time() _MASK_CACHE[key] = {"masks": (segment_frames(imgs, spec) > 0.5).to(torch.uint8), "size": (w, h)} print(f"[BFS Shot] SAM 3: done in {time.time() - t0:.1f}s, {mask_report(_MASK_CACHE[key]['masks'])}", flush=True) _node_boundary() out = dict(_MASK_CACHE[key]) out["masks"] = shape_mask(out["masks"], spec) out["box"] = crop_box(out["masks"], float(spec["padding"])) return out def crop_box(masks: torch.Tensor, padding: float) -> list[float] | None: """One box for the whole shot (union of every frame's mask) plus padding, normalised [x0, y0, x1, y1].""" if masks.numel() == 0 or not bool(masks.any()): return None union = masks.amax(0) ys, xs = torch.nonzero(union, as_tuple=True) H, W = union.shape x0, x1, y0, y1 = xs.min().item() / W, (xs.max().item() + 1) / W, ys.min().item() / H, (ys.max().item() + 1) / H px, py = (x1 - x0) * padding, (y1 - y0) * padding return [max(0.0, x0 - px), max(0.0, y0 - py), min(1.0, x1 + px), min(1.0, y1 + py)] def grow_blur(m: torch.Tensor, grow: int, blur: int) -> torch.Tensor: """[N,H,W] in 0-1: dilate by `grow` px, then soften the edge by `blur` px.""" x = m.float()[:, None] if grow > 0: # separable: a square max = max over rows then over columns (much faster for big kernels) k = 2 * grow + 1 x = torch.nn.functional.max_pool2d(x, (1, k), 1, (0, grow)) x = torch.nn.functional.max_pool2d(x, (k, 1), 1, (grow, 0)) for _ in range(2 if blur > 0 else 0): # box blur, separable too k = 2 * blur + 1 x = torch.nn.functional.avg_pool2d(torch.nn.functional.pad(x, (blur, blur, 0, 0), mode="replicate"), (1, k), 1) x = torch.nn.functional.avg_pool2d(torch.nn.functional.pad(x, (0, 0, blur, blur), mode="replicate"), (k, 1), 1) return x[:, 0].clamp(0, 1) def box_px(box: list[float], W: int, H: int, multiple: int = 2) -> tuple[int, int, int, int]: x0, y0 = int(box[0] * W), int(box[1] * H) x1, y1 = max(x0 + multiple, int(round(box[2] * W))), max(y0 + multiple, int(round(box[3] * H))) return x0, y0, min(W, x1), min(H, y1) def crop_panel(img: torch.Tensor, shot: dict) -> torch.Tensor: """Cut a duet panel off a decoded canvas (the shot was conditioned with a duet panel).""" info = shot.get("panel") if not info or img.shape[1:3] == (shot.get("height"), shot.get("width")): return img h, w = info["h"] * 16, info["w"] * 16 y0 = info["strip_h"] * 16 if info["position"] == "top" else 0 x0 = info["strip_w"] * 16 if info["position"] == "left" else 0 if img.shape[1] < y0 + h or img.shape[2] < x0 + w: return img return img[:, y0:y0 + h, x0:x0 + w] # --- stitch finishing: edge hardness, Reinhard colour match (LAB) and Poisson seamless clone. The three operations are # adapted from Neko (Nekodificador)'s ComfyUI-NKD-Basic-Tools (NKD Inpaint Stitch, MIT licence), built on AbleJones's # workflow and nodes. Differences: the colour # statistics are taken over the whole shot (no flicker), and by default from a ring AROUND the subject, so a swap's new # person keeps its own colours while the model's light / colour drift is corrected. def _alpha_hardness(alpha: torch.Tensor, hardness: float) -> torch.Tensor: """Black / white point remap of the paste alpha: 0 = unchanged, 1 = hard cut at 0.5 (removes the halo of the original subject in the soft edge).""" if hardness <= 0.0: return alpha bp = min(hardness * 0.5, 0.499) return ((alpha - bp) / (1.0 - 2 * bp)).clamp(0.0, 1.0) def _rgb_to_lab(rgb): lin = np.where(rgb <= 0.04045, rgb / 12.92, ((rgb + 0.055) / 1.055) ** 2.4) M = np.array([[0.4124564, 0.3575761, 0.1804375], [0.2126729, 0.7151522, 0.0721750], [0.0193339, 0.1191920, 0.9503041]], np.float32) xyz = lin @ M.T / np.array([0.95047, 1.0, 1.08883], np.float32) d = 6.0 / 29.0 f = lambda t: np.where(t > d ** 3, np.cbrt(t), t / (3 * d * d) + 4.0 / 29.0) fx, fy, fz = f(xyz[..., 0]), f(xyz[..., 1]), f(xyz[..., 2]) return np.stack([116.0 * fy - 16.0, 500.0 * (fx - fy), 200.0 * (fy - fz)], -1).astype(np.float32) def _lab_to_rgb(lab): fy = (lab[..., 0] + 16.0) / 116.0 fx, fz = lab[..., 1] / 500.0 + fy, fy - lab[..., 2] / 200.0 d = 6.0 / 29.0 fi = lambda t: np.where(t > d, t ** 3, 3 * d * d * (t - 4.0 / 29.0)) xyz = np.stack([fi(fx) * 0.95047, fi(fy), fi(fz) * 1.08883], -1) Mi = np.array([[3.2404542, -1.5371385, -0.4985314], [-0.9692660, 1.8760108, 0.0415560], [0.0556434, -0.2040259, 1.0572252]], np.float32) lin = np.clip(xyz @ Mi.T, 0.0, None) return np.clip(np.where(lin <= 0.0031308, lin * 12.92, 1.055 * lin ** (1 / 2.4) - 0.055), 0, 1).astype(np.float32) def _shot_color_transfer(orig: torch.Tensor, gen: torch.Tensor, region: torch.Tensor, strength: float) -> torch.Tensor: """Reinhard transfer in LAB with statistics over the whole shot inside `region` [N,H,W] (original vs generated), applied to every generated frame. Returns gen unchanged when the region is too small.""" if strength <= 0.0: return gen sel = region > 0.5 if int(sel.sum()) < 400: return gen step = max(1, int(sel.sum()) // 200000) o = _rgb_to_lab(orig[sel].numpy()[::step].astype(np.float32)) g = _rgb_to_lab(gen[sel].numpy()[::step].astype(np.float32)) om, os_, gm, gs = o.mean(0), o.std(0) + 1e-5, g.mean(0), g.std(0) + 1e-5 ratio = np.clip(os_ / gs, 0.5, 2.0) # a flat region has ~0 spread: never blow the contrast up lab = _rgb_to_lab(gen.numpy().astype(np.float32)) out = _lab_to_rgb((lab - gm) * ratio + om) return torch.from_numpy(out) * strength + gen * (1.0 - strength) def _seamless(orig: np.ndarray, gen: np.ndarray, alpha: np.ndarray) -> np.ndarray: """Poisson clone of gen onto orig inside alpha (one frame, float RGB 0-1).""" import cv2 binary = (alpha > 0.1).astype(np.uint8) * 255 binary[0, :] = binary[-1, :] = 0 binary[:, 0] = binary[:, -1] = 0 x, y, w, h = cv2.boundingRect(binary) a3 = alpha[..., None] if w == 0 or h == 0: return orig * (1 - a3) + gen * a3 try: c = cv2.seamlessClone((np.clip(gen, 0, 1) * 255).astype(np.uint8), (np.clip(orig, 0, 1) * 255).astype(np.uint8), binary, (x + w // 2, y + h // 2), cv2.NORMAL_CLONE).astype(np.float32) / 255.0 return orig * (1 - a3) + c * a3 except Exception: # noqa: BLE001 return orig * (1 - a3) + gen * a3 MATCH_REGIONS = ["around the subject (swap)", "inside the subject (same content)"] def paste_back(result: torch.Tensor, shot: dict, new_outline: bool = True, **finish) -> torch.Tensor: """'Frame + paste' shots: the whole frame was generated (so the model followed the guide freely); paste only the subject onto the original frame. The pasted area is the shot's mask, plus (new_outline) the new subject found by SAM 3 on the result near the original one (the shot's 'new subject' text, default 'person'), so a bigger silhouette is not cut.""" base = shot["frames"] H, W = base.shape[1:3] ms = shot["mask_src"] tag = f"[BFS Shot Join] shot {shot.get('index', 0) + 1}" sm = shot_mask(ms["path"], ms["analysis"], ms["start"], ms["length"], ms["spec"])["masks"] n = min(result.shape[0], base.shape[0], sm.shape[0]) m = torch.nn.functional.interpolate(sm[:n, None].float(), size=(H, W), mode="nearest")[:, 0] for sm2, _ in extra_mark_masks(shot, tag) if shot.get("color_guide") else []: # every marked person is pasted sm2 = torch.nn.functional.interpolate(sm2[:n, None].float(), size=(H, W), mode="nearest")[:, 0] m = torch.maximum(m, sm2) if new_outline and bool(m.any()): x0, y0, x1, y1 = box_px(crop_box(m, 0.5), W, H) # where the new person may be s = 640 / max(H, W) small = torch.nn.functional.interpolate(result[:n, ..., :3].movedim(-1, 1).float(), size=(max(32, int(H * s) // 2 * 2), max(32, int(W * s) // 2 * 2)), mode="bilinear", align_corners=False).movedim(1, -1).clamp(0, 1) t0 = time.time() what = str(shot.get("paste_text") or "").strip() or "person" new = segment_frames(small, dict(DEFAULT_MASK, text=what, max_objects=8)) _node_boundary() new = torch.nn.functional.interpolate(new[:, None].float(), size=(H, W), mode="nearest")[:, 0] near = torch.zeros(H, W); near[y0:y1, x0:x1] = 1 m = torch.maximum(m, new * near) print(f"{tag}: Frame + paste, new outline ('{what}') by SAM 3 in {time.time() - t0:.1f}s", flush=True) sp = ms["spec"] crop = {"box": [0.0, 0.0, 1.0, 1.0], "mask": m, "paste": "mask", "expand": int(sp["expand"] if sp.get("expand") is not None else 16), "feather": int(sp["feather"] if sp.get("feather") is not None else 12)} print(f"{tag}: pasting the person onto the original frame: {mask_report(m)}", flush=True) return uncrop(result, {"crop": crop, "full_frames": base[:n]}, **finish) def uncrop(result: torch.Tensor, shot: dict, edge_hardness: float = 0.0, match_colors: float = 0.0, match_region: str = MATCH_REGIONS[0], seamless: bool = False) -> torch.Tensor: """Paste a cropped shot's result back into its full frames (feathered by the mask or the box), with optional edge hardness, colour match and seamless clone.""" c = shot["crop"] full = shot["full_frames"] W, H = full.shape[2], full.shape[1] x0, y0, x1, y1 = box_px(c["box"], W, H) n = min(result.shape[0], full.shape[0]) res = torch.nn.functional.interpolate(result[:n, ..., :3].movedim(-1, 1).float(), size=(y1 - y0, x1 - x0), mode="bilinear", align_corners=False).movedim(1, -1).clamp(0, 1) out = full[:n].clone() if c.get("paste") == "box": a = torch.zeros(n, H, W) a[:, y0:y1, x0:x1] = 1 alpha = grow_blur(a, 0, int(c.get("feather", 12))) else: m = c["mask"][:n].float() alpha = grow_blur(m, int(c.get("expand", 8)), int(c.get("feather", 12))) alpha = _alpha_hardness(alpha, float(edge_hardness)) if match_colors > 0 and c.get("mask") is not None: m = c["mask"][:n].float()[:, y0:y1, x0:x1] if match_region == MATCH_REGIONS[0]: # ring around the subject: background the model also regenerated k = max(3, int(0.06 * max(y1 - y0, x1 - x0))) region = grow_blur(m, k, 0) - grow_blur(m, max(2, k // 3), 0) else: region = m res = _shot_color_transfer(full[:n, y0:y1, x0:x1].float(), res, region, float(match_colors)) a = alpha[:, y0:y1, x0:x1, None] if seamless: for i in range(n): out[i, y0:y1, x0:x1] = torch.from_numpy(_seamless(out[i, y0:y1, x0:x1].numpy().astype(np.float32), res[i].numpy().astype(np.float32), a[i, ..., 0].numpy())) else: out[:, y0:y1, x0:x1] = res * a + out[:, y0:y1, x0:x1] * (1 - a) if result.shape[0] > n: # frames past the planned length stay as generated, pasted the same way out = torch.cat([out, out[-1:].expand(result.shape[0] - n, -1, -1, -1)], 0) return out def mask_preview(path: str, analysis: dict, start: int, length: int, spec: dict, count: int = 6, color: str = "#ff283c", opacity: float = 0.55) -> dict: """A few frames of the shot with the selected region tinted and the crop box, as data URLs.""" import cv2 r = shot_mask(path, analysis, start, length, spec) masks, (w, h), box = r["masks"], r["size"], r["box"] src = _timeline(analysis["n_src"], analysis["fps_src"], float(analysis["fps"])) picks = sorted(set(int(round(i)) for i in np.linspace(0, length - 1, min(count, length)))) frames = _read_frames(path, src[np.clip(np.array(picks) + start, 0, analysis["n"] - 1)], (w, h)) try: tint = np.array([int(color[i:i + 2], 16) for i in (1, 3, 5)], dtype=np.float32) alpha = min(0.8, max(0.05, float(opacity))) except (TypeError, ValueError, IndexError): tint, alpha = np.array([255, 40, 60], dtype=np.float32), 0.55 out = [] for i, fr in zip(picks, frames): img = fr.copy() m = masks[min(i, masks.shape[0] - 1)].numpy().astype(bool) img[m] = (img[m].astype(np.float32) * (1 - alpha) + tint * alpha).astype(np.uint8) if box: x0, y0, x1, y1 = box_px(box, w, h) cv2.rectangle(img, (x0, y0), (x1 - 1, y1 - 1), (255, 210, 90), 2) ok, buf = cv2.imencode(".jpg", cv2.cvtColor(img, cv2.COLOR_RGB2BGR), [cv2.IMWRITE_JPEG_QUALITY, 80]) out.append({"f": start + i, "src": "data:image/jpeg;base64," + base64.b64encode(buf.tobytes()).decode()}) cover = float(masks.float().mean()) if masks.numel() else 0.0 return {"box": box, "frames": out, "coverage": cover, "empty": box is None} # ---------------------------------------------------------------------------- {target}: who is replaced TARGET_Q = ("Describe the main subject in this picture (a person, an animal, a creature, a robot or a cartoon character) as " "one short English noun phrase for a video prompt, at most 14 words: what it is first, then for a person the " "apparent gender and age group, hair and main clothing with colours, for anything else its kind and most " "visible colours or features. Examples: 'the young woman with long black hair in a white T-shirt and denim " "shorts', 'the brown dog with a red collar', 'the green goblin in a torn leather vest'. Start with 'the'. " "Reply with the phrase only.") def target_crop(path: str, analysis: dict, start: int, length: int, spec: dict, pad: float = 0.12): """The selected subject (the shot's SAM 3 points / text, on its key frame) cropped from that frame, RGB uint8.""" src = _timeline(analysis["n_src"], analysis["fps_src"], float(analysis["fps"])) key = max(0, min(length - 1, int(spec.get("key") or length // 2))) fr = _read_frames(path, src[np.clip(np.array([start + key]), 0, analysis["n"] - 1)], None)[0] if spec.get("ext"): sm = shot_mask(path, analysis, start, length, spec)["masks"] m = torch.nn.functional.interpolate(sm[min(key, sm.shape[0] - 1)][None, None].float(), size=fr.shape[:2], mode="nearest")[0, 0] > 0.5 else: img = torch.from_numpy(fr.astype(np.float32) / 255.0)[None] m = segment_frames(img, dict(spec, key=0))[0] > 0.5 _node_boundary() if not bool(m.any()): return None ys, xs = np.where(m.numpy()) H, W = fr.shape[:2] py, px = int((ys.max() - ys.min()) * pad) + 4, int((xs.max() - xs.min()) * pad) + 4 return fr[max(0, ys.min() - py):min(H, ys.max() + py), max(0, xs.min() - px):min(W, xs.max() + px)] def prompt_slot(prompt: str, marker: str, limit: int = 360) -> str: """The sentence of `prompt` that holds `marker`, with the marker as a blank ('___') and the other placeholders made readable: what the VLM fills in, so the description fits the prompt (and not the other way round).""" text = str(prompt or "") if marker not in text: return "" i = text.index(marker) starts = [text.rfind(sep, 0, i) for sep in (". ", "; ", ": ", "\n")] a = max(starts) a = 0 if a < 0 else a + 2 ends = [x for x in (text.find(sep, i + len(marker)) for sep in (". ", "; ", "\n")) if x >= 0] b = min(ends) + 1 if ends else len(text) snippet = text[a:b].strip() snippet = snippet.replace(marker, "___", 1) for other, plain in (("{target}", "the guide subject"), ("{details2}", "the second character"), ("{details}", "the character"), ("{shot}", ""), ("{setting}", "")): snippet = snippet.replace(other, plain) snippet = PICTURE_DETAILS.sub(lambda m: f"the subject of ", snippet) return snippet[:limit] # {details@Picture 3} (or {details@Picture3}): the description of the shot's third picture alone PICTURE_DETAILS = re.compile(r"\{details@Picture\s*(\d+)\}") def picture_marker(prompt: str, k: int | None = None) -> str: """The prompt's {details@Picture k} marker as written (the first one of any picture with k=None), or ''.""" for m in PICTURE_DETAILS.finditer(str(prompt or "")): if k is None or int(m.group(1)) == k: return m.group(0) return "" def expand_two_ref_prompt(prompt: str) -> str: """Add a second subject slot when a shot has a second reference and its template has only one (not when the prompt already says what is, or names its pictures with {details@Picture n}).""" text = str(prompt or "") if "{details2}" in text or "" in text or "" in text or "{details@" in text: return text # Only expand an explicit legacy H3 template. A complete short edit request # must not acquire a phantom second performer/setting or a partial H3 header. if "{details}" not in text or not re.search(r"(?im)^\s*(?:[\w-]+:\s*)?subject_definitions\s*:", text): return text second = " is the {details2} shown in ." if re.search(r"(?im)^\s*subject_definitions\s*:", text): if re.search(r"(?im)^\s*summary\s*:", text): return re.sub(r"(?im)(^\s*summary\s*:)", lambda m: second + "\n" + m.group(1), text, count=1) return text.rstrip() + "\n" + second return "subject_definitions: " + second + "\n\n" + text def fit_note(slot: str) -> str: return (f" Your answer goes into the blank (___) of this part of the video prompt: \"{slot}\". Reply with ONLY the " "words that go in the blank, so that the sentence reads naturally with them in place: do not repeat the " "sentence, do not write '___', no full sentence.") if slot else "" _ANSWER_LABEL = re.compile(r"(?:\*\*|__|#+\s*)?\s*(?:final\s+|short\s+)?(?:answer|description|phrase|noun\s+phrase|output|result|response|caption)" r"\s*(?:\*\*|__)?\s*[:\-\u2013]\s*(?:\*\*|__)?\s*", re.I) def strip_answer_label(text: str) -> str: """'**Answer:** elderly woman with ...' -> 'elderly woman with ...'. The VLM sometimes labels its reply (or reasons first and then writes 'Answer: ...'): keep what follows the LAST label, without markdown marks around it.""" t = str(text or "").strip() last = None for m in _ANSWER_LABEL.finditer(t): if m.start() == 0 or t[m.start() - 1] in " \n\t.*_#>": last = m if last is not None and t[last.end():].strip(): t = t[last.end():] return t.strip().strip("*_`#").strip().strip('"\'').strip() def tidy_slot(text: str, slot: str, max_words: int = 25) -> str: """Only the words for the blank: cut an echo of the sentence around it (the VLM sometimes rewrites the whole sentence), drop a leading article when the prompt already has one before the blank, keep it short.""" t = strip_answer_label(" ".join(str(text or "").strip().split("\n")[0].split()).strip().strip('"\'')) if "___" in slot: head, tail = slot.split("___", 1) tail_key = " ".join(tail.split()[:2]) # e.g. "shown in" if "___" in t: # "humanoid robot is the ___ shown in ..." -> before the echo t = t.split("___", 1)[0] if tail_key and tail_key.lower() in t.lower(): t = t[:t.lower().index(tail_key.lower())] hw = head.strip().split() for n in range(min(4, len(hw)), 0, -1): # trailing "is the" copied from before the blank end = " " + " ".join(hw[-n:]) if t.lower().rstrip().endswith(end.lower()): t = t.rstrip()[: -len(end)] break lead = " ".join(hw[-2:]).lower() # a leading " is the" copied from the prompt k = t.lower().find(lead + " ") if lead else -1 if 0 <= k <= 20: t = t[k + len(lead) + 1:] t = " ".join(t.split()[:max_words]).strip(" ,;:.") before = slot.split("___")[0].rstrip().lower().split()[-1:] if "___" in slot else [] if before and before[0] in ("the", "a", "an"): for art in ("the ", "a ", "an "): if t.lower().startswith(art): t = t[len(art):] break return t def describe_target(clip, crop: np.ndarray, max_tokens: int = 64, slot: str = "", temperature: float = 0.0, seed: int = 0, image_max_side: int = 448) -> str: img = torch.from_numpy(crop.astype(np.float32) / 255.0)[None] side = max(224, min(4096, int(image_max_side or 1024))) scale = min(1.0, side / max(crop.shape[:2])) if scale < 1.0: h, w = max(2, round(crop.shape[0] * scale)), max(2, round(crop.shape[1] * scale)) img = torch.nn.functional.interpolate(img.movedim(-1, 1), size=(h, w), mode="bilinear", align_corners=False).movedim(1, -1) t = vlm_generate(clip, TARGET_Q + fit_note(slot), img, max_tokens, temperature, seed) t = strip_answer_label(t).split("\n")[0].strip().strip('"\'').rstrip(".") if slot: t = tidy_slot(t, slot) elif t and not t.lower().startswith("the "): t = "the " + t[0].lower() + t[1:] return " ".join(t.split()[:16]) or "the person" def fill_target(text: str, desc: str) -> str: """{target} = who is replaced in this shot (picked + described in the panel), else 'the person'.""" return text.replace("{target}", desc.strip() or "the person") if "{target}" in text else text # ---------------------------------------------------------------------------- VLM suggestions (optional) DEFAULT_VLM = {"enabled": False, "frames": 3, "max_tokens": 1024, "auto_segment": True, "auto_shot": True, "instruction": "", "describe_preset": "short", "describe_custom": "", "fit_prompt": True, "temperature": 0.0, "seed": 0, "image_max_side": 1024} _VLM: dict[str, Any] = {} # the VLM connected to a planner (kept for the panel's Analyse button) _VLM_CACHE: dict[tuple, dict] = {} VLM_QUESTION = """You see {n} frames (in time order) of ONE camera shot from a video that will be edited with an AI video model. Answer with one JSON object only, no code fence: {{"segment": "the main person or object a user would edit, as a short English noun phrase a segmentation model understands, describing what it looks like (e.g. 'woman in a black top', 'man in a grey suit', 'red car'); never on-screen text", "shot": "one or two sentences for a video prompt: camera distance and angle, camera movement, where the subject is in the frame and what they do, in time order; describe actions and framing, not the person's face or clothes", "people": , "recommend": "run" or "skip" (skip when the shot has nobody to edit, is a title card, a black frame or text only), "reason": "a few words"}}{extra}""" async def _off_loop(fn): """Run slow work in a thread so the server keeps answering (and the panel gets progress events).""" import asyncio _outside_prompt() return await asyncio.get_running_loop().run_in_executor(None, fn) def _outside_prompt() -> None: """Models run from a panel button (no prompt running): ComfyUI's progress hook reads the server's last prompt and node ids, which do not exist until a first prompt has run.""" try: from server import PromptServer srv = PromptServer.instance for attr in ("last_prompt_id", "last_node_id"): if not hasattr(srv, attr): setattr(srv, attr, None) except Exception: # noqa: BLE001 pass def _node_boundary() -> None: """What ComfyUI does between two nodes: drop the per-thread CUDA malloc graph and prefetch queues. Needed when one node runs a model several times (a second text generation in the same node otherwise hits a device assert).""" try: import comfy.model_prefetch comfy.model_prefetch.cleanup_prefetch_queues() except Exception: # noqa: BLE001 - older ComfyUI without it pass def _textgenerate_execute(TextGenerate, **kwargs): """Keep MTP disabled when TextGenerate supports it, and tolerate older signatures without it.""" import inspect try: params = inspect.signature(TextGenerate.execute).parameters supports_mtp = "mtp" in params or any(p.kind is inspect.Parameter.VAR_KEYWORD for p in params.values()) except (TypeError, ValueError): supports_mtp = False call_kwargs = dict(kwargs) if supports_mtp: # MTP is speculative decoding. It is optional and off preserves the planner's prior behavior. call_kwargs["mtp"] = "off" try: return TextGenerate.execute(**call_kwargs) except TypeError as e: # Some wrappers advertise **kwargs but forward to an older execute() implementation. if "mtp" not in call_kwargs or "unexpected keyword argument 'mtp'" not in str(e): raise call_kwargs.pop("mtp") return TextGenerate.execute(**call_kwargs) def vlm_generate(clip, prompt: str, images: torch.Tensor, max_tokens: int, temperature: float = 0.0, seed: int = 0, sampling: dict | None = None) -> str: """Text from a vision-language CLIP. Some models end the answer before writing anything for some wordings: retry with a reworded request, then with sampling, and clean leftovers of the chat template. temperature > 0 samples (with `seed`) from the first try, so another seed gives another answer.""" from comfy_extras.nodes_textgen import TextGenerate first = ({"sampling_mode": "on", "temperature": float(temperature), "top_k": 40, "top_p": 0.9, "min_p": 0.0, "repetition_penalty": 1.05, "presence_penalty": 0.0, "seed": int(seed)} if temperature and temperature > 0 else {"sampling_mode": "off"}) if sampling: # e.g. long answers: greedy decoding falls into loops there first = {"sampling_mode": "on", "top_k": 40, "top_p": 0.9, "min_p": 0.0, "seed": int(seed), **sampling} tries = [(prompt, first), ("Look at the picture(s) carefully. " + prompt + " Write the answer now.", {"sampling_mode": "off"}), (prompt, {"sampling_mode": "on", "temperature": 0.7, "top_k": 40, "top_p": 0.9, "min_p": 0.0, "repetition_penalty": 1.05, "presence_penalty": 0.0, "seed": 1})] text = "" for q, mode in tries: text = _textgenerate_execute(TextGenerate, clip=clip, prompt=q, max_length=int(max_tokens), sampling_mode=mode, image=images).args[0] _node_boundary() text = str(text or "").strip() for junk in ("assistant\n", "assistant:", "assistant"): if text.lower().startswith(junk): text = text[len(junk):].strip() if text: break return text def vlm_cfg(cfg: dict | None) -> dict: out = dict(DEFAULT_VLM) out.update({k: v for k, v in (cfg or {}).items() if v is not None}) return out def _parse_json(text: str) -> dict: t = text.strip() if "```" in t: t = t.split("```")[1] t = t[4:] if t.lower().startswith("json") else t a, b = t.find("{"), t.rfind("}") try: return json.loads(t[a:b + 1]) if a >= 0 and b > a else {} except json.JSONDecodeError: return {} def vlm_shot(clip, path: str, analysis: dict, start: int, end: int, cfg: dict) -> dict: """Ask the VLM about one shot (a few frames); cached per shot and settings.""" key = (path, os.path.getmtime(path), int(start), int(end), id(clip), json.dumps({k: cfg[k] for k in ("frames", "max_tokens", "instruction", "image_max_side")}, sort_keys=True)) if key in _VLM_CACHE: return _VLM_CACHE[key] from comfy_extras.nodes_textgen import TextGenerate _status("Asking the VLM about the shot", force=True) n = max(1, min(8, int(cfg["frames"]))) src = _timeline(analysis["n_src"], analysis["fps_src"], float(analysis["fps"])) picks = np.clip(np.linspace(start, end - 1, n).round().astype(int), 0, analysis["n"] - 1) side = max(224, min(4096, int(cfg.get("image_max_side") or 1024))) s = side / max(analysis["width"], analysis["height"]) size = (max(32, int(analysis["width"] * s) // 2 * 2), max(32, int(analysis["height"] * s) // 2 * 2)) imgs = torch.from_numpy(np.stack(_read_frames(path, src[picks], size)).astype(np.float32) / 255.0) extra = ("\n" + cfg["instruction"].strip()) if str(cfg.get("instruction") or "").strip() else "" q = VLM_QUESTION.format(n=n, extra=extra) text = vlm_generate(clip, q, imgs, int(cfg["max_tokens"])) out = _parse_json(text) out = {"segment": str(out.get("segment") or "").strip(), "shot": str(out.get("shot") or "").strip(), "people": out.get("people"), "recommend": str(out.get("recommend") or "run").lower(), "reason": str(out.get("reason") or "").strip(), "raw": "" if out else text} _VLM_CACHE[key] = out return out # ---------------------------------------------------------------------------- reference descriptions (VLM) # the subject can be a person, a creature, a robot, an animal or a cartoon character: say what it is first, then the # traits that apply to it (a goblin is not described as "a person with greenish skin") _WHAT = ("First say what the subject is in a few words (e.g. 'A young woman', 'A goblin', 'A humanoid robot', 'A golden " "retriever', 'An anime girl'), then describe it with the traits that apply to that kind of subject: ") _DESCRIBE_ONLY = ("Treat all pictures as views of the same subject and use only clearly visible details. Return only " "the requested English description, with no heading, label, preamble, bullet list or Markdown.") DESCRIBE_PRESETS = { "scene / setting": ("Describe the environment in this reference image for a video-generation prompt, in two or " "three concise English sentences: location, spatial layout, architecture, scenery, furniture, " "materials, colors and visible lighting. This image is a setting reference, not a replacement " "character. Describe people only if needed to distinguish the space; do not turn them into " "subject identities. Use only visible facts. Return only the description."), # default: a noun phrase that fits inside the prompt, " is the {details} shown in " "short": ("Describe the subject of these reference pictures for a video-generation prompt. Write one English noun " "phrase of at most 20 words: first identify what the subject is (such as a young woman, goblin, humanoid " "robot, dog or anime character), then give only its most distinctive visible features and clothing with " "colours. The phrase must fit after ' is the' and before 'shown in ', so do not start " "with an article. Never describe pose, action, expression, camera or background. " + _DESCRIBE_ONLY), "full body": ("Describe the subject of these reference pictures for a video-generation prompt: one paragraph of 3 to 5 " "sentences in English. " + _WHAT + "for people and creatures the apparent gender and age, skin or fur " "colour and texture, face, head shape, ears, horns or other distinctive features, hair, body build and " "proportions, then the clothing piece by piece with colours and materials, shoes and accessories; for " "robots the design, materials, colours, head or face plate, panels, joints and lights; for animals the " "breed or species, fur colour, length and markings, ears, tail, build and any collar or accessory. Only " "what is visible. Describe only appearance: never the pose, gesture, action, expression, camera, framing " "or background. Concrete words only: no negations, no names, no opinions. " + _DESCRIBE_ONLY), "head / face": ("Describe the head and face of the subject of these reference pictures for a video-generation prompt: one " "paragraph of 3 to 4 sentences in English. " + _WHAT + "skin, fur or surface (colour and texture), " "head and face shape, eyes, eyebrows, nose or snout, mouth, jawline, ears, horns, facial hair, " "wrinkles, marks, glasses or face plate, and the hair or fur: colour, length, texture, style. Only " "what is visible. Describe only appearance: never the head angle, gaze, expression, action, camera or " "background. Concrete words only: no negations, no names, no opinions. " + _DESCRIBE_ONLY), "face attributes": ("Describe the face and head in these reference pictures for a video-generation prompt. Write one " "short comma-separated English attribute list. First identify the apparent gender and age group " "for a person, or identify the kind of creature, robot, animal or cartoon character. Then cover, " "in order: head and face shape and proportions (such as oval, round, square, heart-shaped, " "narrow/wide, jaw structure, whether the head reads as small/average/large relative to the " "shoulders), skin, fur or surface colour, eye color, hair or fur color and style, and any distinctive " "features (ears, horns, facial hair, jewelry, makeup, glasses, etc.). Physical traits only, never the " "pose, expression or background. " + _DESCRIBE_ONLY), # an object (prop, product, vehicle, piece of furniture...): a noun phrase that fits inside the prompt like "short" "object": ("Describe the object in these reference pictures for a video-generation prompt. Write one English noun " "phrase of at most 25 words: first say what the object is (such as a leather handbag, red sports car, " "wooden chair, ceramic mug or sword), then its most distinctive visible features: shape, materials, colors, " "finish, patterns, printed text or logos, and notable parts. The phrase must fit after 'the' and before " "'shown in ', so do not start with an article. Never describe the background, a person holding " "it, the camera or the lighting. Treat all pictures as views of the same object and use only clearly " "visible details. Return only the requested English description, with no heading, label, preamble, " "bullet list or Markdown."), "outfit": ("Describe only what the subject of these reference pictures wears, piece by piece, for a video-generation " "prompt: garments, armour or accessories, colours, materials, fit and shoes, in 2 or 3 sentences in " "English. Start with what the subject is in two or three words (e.g. 'The goblin wears...'). Never the " "pose, action or background. No negations, no names. " + _DESCRIBE_ONLY), } _DESCRIBE_WORD_LIMITS = {"short": 20, "object": 25, "full body": 150, "head / face": 110, "face attributes": 65, "outfit": 90, "scene / setting": 90} # presets that write a noun phrase for a blank inside a prompt sentence (the others write paragraphs) PHRASE_PRESETS = ("short", "object") _DESCRIBE_CACHE: dict[tuple, str] = {} def _image_key(img: torch.Tensor | None) -> str: if img is None: return "" x = torch.nn.functional.interpolate(img[:1, ..., :3].movedim(-1, 1).float(), size=(32, 32), mode="area") return hashlib.sha1((x * 255).round().to(torch.uint8).numpy().tobytes()).hexdigest()[:16] def clean_slot(d: str) -> str: """A description going into ' is the {details} shown in ...': without the prompt words a VLM (or an old saved description) repeats around it, which would come out twice.""" t = str(d or "").strip() for _ in range(3): t = re.sub(r"^(\s*)?(is\s+)?(the|a|an)\s+", "", t, flags=re.I).strip() t = re.sub(r"\s*\bshown in(\s+)?[\s.]*$", "", t, flags=re.I).strip() t = re.sub(r"\s+is(\s+(the|a|an))?[\s.]*$", "", t, flags=re.I).strip() return t def tidy_details(text: str, cfg: dict) -> str: """Normalise every description preset. In particular, remove labels the VLM adds despite the prompt and keep automatic descriptions inside the size promised by the selected preset. The short phrase also has to fit inside ' is the {details} shown in ...'.""" preset = cfg.get("describe_preset") or "short" t = str(text or "").replace("```", " ").strip() t = strip_answer_label(t) if preset in PHRASE_PRESETS: t = t.splitlines()[0] if t.splitlines() else "" t = " ".join(t.split()).strip().strip('"\'').strip() if preset in PHRASE_PRESETS: # the VLM sometimes repeats the prompt sentence around its answer ("is the ... shown in ", "... is # the shown in"): those words are already in the prompt, so they would come out twice for _ in range(3): t = re.sub(r"^(\s*)?(is\s+)?(the|a|an)\s+", "", t, flags=re.I).strip() t = re.sub(r"\s*\bshown in(\s+)?[\s.]*$", "", t, flags=re.I).strip() t = re.sub(r"\s+is(\s+(the|a|an))?[\s.]*$", "", t, flags=re.I).strip() t = t.rstrip(" .").strip('"\'').rstrip(" .") limit = _DESCRIBE_WORD_LIMITS.get(preset, 180) words = t.split() if len(words) > limit: t = " ".join(words[:limit]).rstrip(" ,;:") if preset not in PHRASE_PRESETS: t = t.rstrip(".") + "." return t def describe_instruction(cfg: dict, slot: str = "") -> str: """The describe instruction; with fit_prompt and the prompt sentence that holds {details}, the VLM writes for it.""" preset = cfg.get("describe_preset", "short") if preset == "custom": base = str(cfg.get("describe_custom") or DESCRIBE_PRESETS["short"]).strip() else: base = DESCRIBE_PRESETS.get(preset, DESCRIBE_PRESETS["short"]) # only a short phrase fits a blank inside a sentence; the paragraph presets are written on their own fits = cfg.get("fit_prompt", True) and preset in PHRASE_PRESETS return base + (fit_note(slot) if fits else "") def vlm_describe(clip, images: list, instruction: str, max_tokens: int = 320, temperature: float = 0.0, seed: int = 0, image_max_side: int = 448) -> str: """One description for a set of reference pictures (e.g. a close-up and a full-body photo of one person).""" images = [im for im in images if im is not None] if not images: return "" side = max(224, min(4096, int(image_max_side or 1024))) key = (tuple(_image_key(im) for im in images), instruction, id(clip), int(max_tokens), float(temperature), int(seed), side) if key in _DESCRIBE_CACHE: return _DESCRIBE_CACHE[key] from comfy_extras.nodes_textgen import TextGenerate _status("Describing the references with the VLM", force=True) # Keep each image's aspect ratio and letterbox to a shared canvas. The setting controls # the longest side, before any additional resizing done by the model's own processor. batch = [] for im in images: x = im[:1, ..., :3].movedim(-1, 1).float() s = min(1.0, side / max(x.shape[-2:])) nh, nw = max(2, int(x.shape[-2] * s)), max(2, int(x.shape[-1] * s)) x = torch.nn.functional.interpolate(x, size=(nh, nw), mode="bilinear", align_corners=False) canvas = torch.full((1, 3, side, side), 0.5) canvas[:, :, (side - nh) // 2:(side - nh) // 2 + nh, (side - nw) // 2:(side - nw) // 2 + nw] = x batch.append(canvas) imgs = torch.cat(batch, 0).movedim(1, -1).clamp(0, 1) text = vlm_generate(clip, instruction, imgs, int(max_tokens), temperature, seed) text = " ".join(str(text).replace("```", " ").split()) _DESCRIBE_CACHE[key] = text return text def describe_picture(clip, image, prompt: str, marker: str, cfg: dict) -> str: """One picture described alone ({details@Picture k}), written for the prompt sentence that holds `marker`.""" slot = prompt_slot(prompt, marker) if marker and cfg.get("fit_prompt", True) else "" d = tidy_details(vlm_describe(clip, [image], describe_instruction(cfg, slot), int(cfg["max_tokens"]), float(cfg.get("temperature") or 0), int(cfg.get("seed") or 0), int(cfg.get("image_max_side") or 1024)), cfg) return tidy_slot(d, slot) if slot and (cfg.get("describe_preset") or "short") in PHRASE_PRESETS else d def picture_markers(text: str) -> str: """{details} and {details2} are the descriptions of pictures 1 and 2: {details@Picture 1} / {details@Picture 2}.""" return str(text or "").replace("{details2}", "{details@Picture 2}").replace("{details}", "{details@Picture 1}") def saved_picture_details(plan: dict) -> dict: """One description per picture: picture_details, plus the ones plans before 1.113 saved per reference set in ref_details ('a|b' -> {details, details2}, or one text for a).""" out = {} for key, value in (plan.get("ref_details") or {}).items(): a, _, b = str(key).partition("|") pairs = ((a, value.get("details") or value.get("legacy")), (b, value.get("details2"))) \ if isinstance(value, dict) else ((a, value),) for name, text in pairs: if name and text and name not in out: out[name] = str(text) out.update({k: v for k, v in (plan.get("picture_details") or {}).items() if v}) return out def picture_cfg(cfg: dict, plan: dict, name: str) -> dict: """The describe settings of one picture: the VLM card's, with the picture's own preset (and custom instruction) when it has one (plan['picture_cfg'][name] = {preset, custom}).""" own = (plan.get("picture_cfg") or {}).get(name) or {} if not own.get("preset"): return cfg return dict(cfg, describe_preset=own["preset"], describe_custom=own.get("custom") or cfg.get("describe_custom", "")) def fill_picture_details(text: str, names: list, images: list, saved: dict, clip=None, cfg: dict | None = None, tag: str = "", keep_missing: bool = False) -> str: """{details@Picture k} = the description of the shot's k-th picture: the one saved for that image (📝 Describe each picture, or typed in the panel), else the VLM describes it now, else 'character' (keep_missing: the marker stays, for the conditioning node's VLM).""" if "{details@" not in (text or ""): return text def sub(m): k = int(m.group(1)) - 1 name = names[k] if 0 <= k < len(names) else "" image = images[k] if 0 <= k < len(images) else None d = (saved or {}).get(name, "") if name else "" if not d and clip is not None and image is not None: d = describe_picture(clip, image, text, m.group(0), cfg or vlm_cfg({})) if d: print(f"{tag}: {m.group(0)} = {d!r} (described automatically)", flush=True) if d: return clean_slot(d) if image is None: print(f"{tag}: {m.group(0)}: the shot has no picture {k + 1}", flush=True) elif keep_missing: return m.group(0) return "character" return PICTURE_DETAILS.sub(sub, text) # ---------------------------------------------------------------------------- prompt writer (VLM) WRITER_TASKS = ["character swap", "style", "setting", "appearance", "lighting / weather", "custom"] WRITER_INSTRUCTION = """You write the prompt for MiniMax H3, a video model, for a split-screen "duet": the KEPT FOOTAGE (the first {nf} images, one camera shot in time order) stays exactly as it is in one half of the frame, and the other half is generated moving in sync with it. {refs_txt} Task: {task}. {change} Rules (render-tested, follow them exactly): 1. Never describe the kept footage's performer, face, hair, clothes, props or room, not even to contrast them: whatever you name gets drawn, whatever you leave out is copied from the kept footage. 2. The kept footage has no tag: call it "the kept footage". Never write