| |
| """Queue a folder of clip prompts through a ComfyUI image-to-video graph. |
| |
| Reads clip-*.txt in order. A header of `handoff-from: clip-NN last frame` |
| extracts the last frame of that clip's video and loads it as the graph's |
| first frame. `handoff-from: none` does not reuse a frame. |
| |
| A fresh clip uses --first-image, or a still beside the prompt |
| (clip-01.png / .jpg / .webp). With neither, first_frame is left unconnected |
| and the clip is text-to-video. |
| |
| Default graph: quickies/comfyui/minimax-fl2v.json |
| Default host: http://127.0.0.1:9000 |
| |
| When the queue finishes, the clips are joined in order into |
| renders/<film>.mp4. The picture and the sound are also written apart, as |
| renders/<film>_video.mp4 and renders/<film>_audio.m4a. |
| |
| Several project folders run one after another. Pass them on the command |
| line, or list them in a --queue file. The next project starts when the |
| one before it has finished. The guide is quickies/comfyui/README.md, |
| beside this program. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import random |
| import re |
| import shutil |
| import subprocess |
| import sys |
| import time |
| import urllib.error |
| import urllib.parse |
| import urllib.request |
| from pathlib import Path |
|
|
| DEFAULT_WORKFLOW = Path(__file__).resolve().parent / "minimax-fl2v.json" |
| DEFAULT_HOST = "http://127.0.0.1:9000" |
| DEFAULT_INPUT = Path("/home/wizdroid/Applications/comfyui/input") |
| CLIP_RE = re.compile(r"^clip-(\d+)", re.IGNORECASE) |
| HANDOFF_RE = re.compile(r"^handoff-from:\s*(.+?)\s*$", re.IGNORECASE | re.MULTILINE) |
| DURATION_RE = re.compile(r"clip-\d+-(\d+)s", re.IGNORECASE) |
| STILL_EXTS = (".png", ".jpg", ".jpeg", ".webp") |
|
|
|
|
| class BatchError(RuntimeError): |
| pass |
|
|
|
|
| def prompt_files(folder: Path) -> list[Path]: |
| files = [p for p in folder.glob("clip-*.txt") if p.is_file()] |
|
|
| def key(path: Path) -> tuple: |
| match = CLIP_RE.match(path.name) |
| number = int(match.group(1)) if match else 10**9 |
| return (number, path.name) |
|
|
| return sorted(files, key=key) |
|
|
|
|
| def handoff_source(text: str) -> str | None: |
| """Return the clip id that must supply the first frame, or None.""" |
| match = HANDOFF_RE.search(text) |
| if not match: |
| return None |
| value = match.group(1).strip().lower() |
| if value == "none" or value.startswith("new ") or "do not" in value: |
| return None |
| clip = re.search(r"clip-\d+", value, re.IGNORECASE) |
| if not clip: |
| raise BatchError(f"cannot read handoff-from: {match.group(1).strip()}") |
| return clip.group(0).lower() |
|
|
|
|
| def duration_seconds(path: Path) -> float | None: |
| match = DURATION_RE.search(path.name) |
| if not match: |
| return None |
| return float(match.group(1)) |
|
|
|
|
| def find_nodes(graph: dict, class_type: str) -> list[tuple[str, dict]]: |
| return [ |
| (node_id, node) |
| for node_id, node in graph.items() |
| if isinstance(node, dict) and node.get("class_type") == class_type |
| ] |
|
|
|
|
| def bind_graph(graph: dict) -> dict: |
| i2v = find_nodes(graph, "MiniMaxH3ImageToVideo") |
| if len(i2v) != 1: |
| raise BatchError( |
| f"expected one MiniMaxH3ImageToVideo node, found {len(i2v)}" |
| ) |
| prompt_id, prompt_node = i2v[0] |
| if "prompt" not in prompt_node.get("inputs", {}): |
| raise BatchError(f"node {prompt_id} has no prompt input") |
| first = prompt_node["inputs"].get("first_frame") |
| image_id = None |
| if isinstance(first, list) and first: |
| image_id = str(first[0]) |
| image_node = graph.get(image_id, {}) |
| if image_node.get("class_type") != "LoadImage": |
| raise BatchError( |
| f"first_frame comes from {image_id} ({image_node.get('class_type')}), not LoadImage" |
| ) |
| savers = find_nodes(graph, "SaveVideo") |
| if len(savers) != 1: |
| raise BatchError(f"expected one SaveVideo node, found {len(savers)}") |
| duration_id = None |
| for node_id, node in graph.items(): |
| title = (node.get("_meta") or {}).get("title", "") |
| if node.get("class_type") == "PrimitiveFloat" and "duration" in title.lower(): |
| duration_id = node_id |
| break |
| noise = find_nodes(graph, "RandomNoise") |
| return { |
| "prompt_id": prompt_id, |
| "image_id": image_id, |
| "save_id": savers[0][0], |
| "duration_id": duration_id, |
| "noise_id": noise[0][0] if noise else None, |
| } |
|
|
|
|
| def apply_job(graph: dict, binding: dict, prompt: str, prefix: str, seconds: float | None, image_name: str | None) -> dict: |
| job = json.loads(json.dumps(graph)) |
| job[binding["prompt_id"]]["inputs"]["prompt"] = prompt |
| job[binding["save_id"]]["inputs"]["filename_prefix"] = prefix |
| if binding["duration_id"] and seconds is not None: |
| job[binding["duration_id"]]["inputs"]["value"] = seconds |
| if binding["noise_id"]: |
| job[binding["noise_id"]]["inputs"]["noise_seed"] = random.randint(1, 2**31 - 1) |
| if image_name and binding["image_id"]: |
| job[binding["image_id"]]["inputs"]["image"] = image_name |
| elif binding["image_id"]: |
| job[binding["prompt_id"]]["inputs"].pop("first_frame", None) |
| image_id = binding["image_id"] |
| used = False |
| for node_id, node in job.items(): |
| if node_id == image_id: |
| continue |
| for value in node.get("inputs", {}).values(): |
| if isinstance(value, list) and value and str(value[0]) == image_id: |
| used = True |
| if not used: |
| job.pop(image_id, None) |
| return job |
|
|
|
|
|
|
|
|
|
|
| def still_beside(prompt_path: Path) -> Path | None: |
| for ext in STILL_EXTS: |
| candidate = prompt_path.with_suffix(ext) |
| if candidate.is_file(): |
| return candidate |
| return None |
|
|
|
|
| def frame_count(video: Path) -> int: |
| probe = [ |
| "ffprobe", "-v", "error", "-select_streams", "v:0", |
| "-show_entries", "stream=nb_frames", |
| "-of", "default=nokey=1:noprint_wrappers=1", str(video), |
| ] |
| number = subprocess.run(probe, check=False, capture_output=True, text=True).stdout.strip() |
| if number.isdigit() and int(number) > 0: |
| return int(number) |
| probe = [ |
| "ffprobe", "-v", "error", "-count_frames", "-select_streams", "v:0", |
| "-show_entries", "stream=nb_read_frames", |
| "-of", "default=nokey=1:noprint_wrappers=1", str(video), |
| ] |
| number = subprocess.run(probe, check=False, capture_output=True, text=True).stdout.strip() |
| if number.isdigit() and int(number) > 0: |
| return int(number) |
| raise BatchError(f"could not count frames in {video}") |
|
|
|
|
| def extract_last_frame(video: Path, dest: Path) -> Path: |
| dest.parent.mkdir(parents=True, exist_ok=True) |
| count = frame_count(video) |
| index = count - 1 |
| cmd = [ |
| "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", |
| "-i", str(video), |
| "-map", "0:v:0", |
| "-vf", f"select=eq(n\\,{index})", |
| "-frames:v", "1", |
| str(dest), |
| ] |
| subprocess.run(cmd, check=True) |
| if not dest.is_file(): |
| raise BatchError(f"ffmpeg wrote no frame: {dest}") |
| return dest |
|
|
|
|
| def stage_image(src: Path, comfy_input: Path, name: str) -> str: |
| comfy_input.mkdir(parents=True, exist_ok=True) |
| dest = comfy_input / name |
| shutil.copyfile(src, dest) |
| return name |
|
|
|
|
| def request_json(url: str, payload: dict | None = None, timeout: float = 60) -> dict: |
| data = None |
| headers = {} |
| if payload is not None: |
| data = json.dumps(payload).encode() |
| headers["Content-Type"] = "application/json" |
| req = urllib.request.Request(url, data=data, headers=headers) |
| try: |
| with urllib.request.urlopen(req, timeout=timeout) as response: |
| return json.loads(response.read().decode()) |
| except urllib.error.HTTPError as exc: |
| body = exc.read().decode(errors="replace") |
| raise BatchError(f"{url} returned {exc.code}: {body}") from exc |
| except urllib.error.URLError as exc: |
| raise BatchError(f"cannot reach {url}: {exc.reason}") from exc |
|
|
|
|
| def queue_prompt(host: str, graph: dict) -> str: |
| result = request_json(f"{host}/prompt", {"prompt": graph}) |
| if result.get("node_errors"): |
| raise BatchError(f"ComfyUI rejected the graph: {result['node_errors']}") |
| prompt_id = result.get("prompt_id") |
| if not prompt_id: |
| raise BatchError(f"ComfyUI returned no prompt_id: {result}") |
| return prompt_id |
|
|
|
|
| def wait_history(host: str, prompt_id: str, timeout: float) -> dict: |
| deadline = time.time() + timeout |
| url = f"{host}/history/{prompt_id}" |
| while time.time() < deadline: |
| try: |
| history = request_json(url, timeout=30) |
| except BatchError: |
| history = {} |
| entry = history.get(prompt_id) |
| if entry: |
| status = entry.get("status") or {} |
| if status.get("status_str") == "error" or status.get("completed") is False and status.get("status_str"): |
| if status.get("status_str") == "error": |
| raise BatchError(f"ComfyUI failed {prompt_id}: {status}") |
| if status.get("completed") or entry.get("outputs"): |
| return entry |
| time.sleep(3) |
| raise BatchError(f"timed out after {int(timeout)}s waiting for {prompt_id}") |
|
|
|
|
| def output_videos(entry: dict) -> list[dict]: |
| found = [] |
| for node_out in (entry.get("outputs") or {}).values(): |
| for key in ("videos", "gifs", "images"): |
| for item in node_out.get(key) or []: |
| name = item.get("filename", "") |
| if name.lower().endswith((".mp4", ".webm", ".mkv", ".mov")): |
| found.append(item) |
| return found |
|
|
|
|
| def download_output(host: str, item: dict, dest: Path) -> Path: |
| query = urllib.parse.urlencode({ |
| "filename": item["filename"], |
| "subfolder": item.get("subfolder", ""), |
| "type": item.get("type", "output"), |
| }) |
| dest.parent.mkdir(parents=True, exist_ok=True) |
| url = f"{host}/view?{query}" |
| try: |
| with urllib.request.urlopen(url, timeout=120) as response: |
| dest.write_bytes(response.read()) |
| except urllib.error.URLError as exc: |
| raise BatchError(f"could not download {url}: {exc.reason}") from exc |
| return dest |
|
|
|
|
| def clip_key(path: Path) -> str: |
| match = CLIP_RE.match(path.name) |
| if not match: |
| return path.stem |
| return f"clip-{match.group(1)}" |
|
|
|
|
| def lookup_source(source: str, rendered: dict[str, Path]) -> Path: |
| if source in rendered: |
| return rendered[source] |
| matches = [(key, path) for key, path in rendered.items() if key.startswith(source)] |
| if len(matches) == 1: |
| return matches[0][1] |
| known = ", ".join(sorted(rendered)) or "none yet" |
| raise BatchError( |
| f"{source} needs a finished video for its last frame; have: {known}" |
| ) |
|
|
|
|
| def fresh_still(prompt_path: Path, first_image: Path | None, used_first: bool) -> tuple[str, Path, bool]: |
| beside = still_beside(prompt_path) |
| if beside: |
| return (f"still {beside.name}", beside, used_first) |
| if first_image and not used_first: |
| return (f"starting still {first_image.name}", first_image, True) |
| return ("no still, first frame bypassed", None, used_first) |
|
|
|
|
| def stream_kinds(path: Path) -> set[str]: |
| probe = subprocess.run( |
| [ |
| "ffprobe", "-v", "error", "-show_entries", "stream=codec_type", |
| "-of", "csv=p=0", str(path), |
| ], |
| check=False, capture_output=True, text=True, |
| ) |
| return {line.strip() for line in probe.stdout.splitlines() if line.strip()} |
|
|
|
|
| def concat_list(clips: list[Path], dest: Path) -> None: |
| script = dest.with_suffix(".concat.txt") |
| lines = [] |
| for clip in clips: |
| escaped = str(clip.resolve()).replace("'", "'\\''") |
| lines.append(f"file '{escaped}'") |
| script.write_text("\n".join(lines) + "\n") |
| cmd = [ |
| "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", |
| "-f", "concat", "-safe", "0", "-i", str(script), |
| "-c", "copy", str(dest), |
| ] |
| try: |
| subprocess.run(cmd, check=True) |
| finally: |
| script.unlink(missing_ok=True) |
|
|
|
|
| def split_av(sequence: Path, video_path: Path, audio_path: Path) -> Path | None: |
| subprocess.run( |
| [ |
| "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", |
| "-i", str(sequence), "-map", "0:v:0", "-c", "copy", "-an", |
| str(video_path), |
| ], |
| check=True, |
| ) |
| if "audio" not in stream_kinds(sequence): |
| return None |
| subprocess.run( |
| [ |
| "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", |
| "-i", str(sequence), "-map", "0:a:0", "-c", "copy", "-vn", |
| str(audio_path), |
| ], |
| check=True, |
| ) |
| return audio_path |
|
|
|
|
| def assemble(folder: Path, files: list[Path], dry_run: bool = False) -> None: |
| renders = folder / "renders" |
| clips = [] |
| missing = [] |
| for path in files: |
| video = renders / f"{path.stem}.mp4" |
| if video.is_file(): |
| clips.append(video) |
| else: |
| missing.append(video.name) |
| sequence = renders / f"{folder.name}.mp4" |
| picture = renders / f"{folder.name}_video.mp4" |
| sound = renders / f"{folder.name}_audio.m4a" |
| if dry_run: |
| print(f"assemble: {' '.join(p.name for p in clips) or '(no renders yet)'}") |
| print(f" {sequence}") |
| print(f" {picture}") |
| print(f" {sound}") |
| return |
| if missing: |
| raise BatchError(f"cannot stitch; missing {', '.join(missing)}") |
| if not clips: |
| raise BatchError("no clips to stitch") |
| renders.mkdir(parents=True, exist_ok=True) |
| concat_list(clips, sequence) |
| audio = split_av(sequence, picture, sound) |
| print(f"stitched {sequence}") |
| print(f"video {picture}") |
| if audio: |
| print(f"audio {audio}") |
| else: |
| print("audio: none in the clips, so no audio file was written") |
|
|
|
|
| def run(args: argparse.Namespace) -> int: |
| folder = Path(args.directory).expanduser().resolve() |
| if not folder.is_dir(): |
| raise BatchError(f"not a directory: {folder}") |
| files = prompt_files(folder) |
| if not files: |
| raise BatchError(f"no clip-*.txt in {folder}") |
| if args.assemble_only and args.no_assemble: |
| raise BatchError("--assemble-only and --no-assemble cannot be used together") |
| if args.assemble_only: |
| assemble(folder, files, dry_run=args.dry_run) |
| return 0 |
| workflow = Path(args.workflow).expanduser().resolve() |
| graph = json.loads(workflow.read_text()) |
| binding = bind_graph(graph) |
| if binding["image_id"] is None: |
| raise BatchError("graph has no first_frame LoadImage; this runner only feeds that input") |
| renders = folder / "renders" |
| rendered: dict[str, Path] = {} |
| for path in files: |
| existing = renders / f"{path.stem}.mp4" |
| if existing.is_file(): |
| rendered[clip_key(path)] = existing |
| comfy_input = Path(args.input_dir).expanduser() |
| film = folder.name |
| used_first = False |
| first_image = Path(args.first_image).expanduser().resolve() if args.first_image else None |
| if first_image and not first_image.is_file(): |
| raise BatchError(f"starting still not found: {first_image}") |
|
|
| for path in files: |
| key = clip_key(path) |
| text = path.read_text() |
| source = handoff_source(text) |
| dest = renders / f"{path.stem}.mp4" |
| if args.skip_existing and dest.is_file(): |
| rendered[key] = dest |
| print(f"skip {path.name} (render exists)") |
| continue |
| seconds = duration_seconds(path) |
| prefix = f"videos/quickies/{film}/{path.stem}" |
| local_image = None |
| video = None |
| if source: |
| video = lookup_source(source, rendered) |
| reason = f"last frame of {source}" |
| else: |
| reason, local_image, used_first = fresh_still(path, first_image, used_first) |
| print(f"{path.name}: {reason}") |
| print(f" prefix {prefix}" + (f" duration {seconds:g}s" if seconds else "")) |
| if args.dry_run: |
| if source and video is not None and not video.is_file(): |
| print(f" would extract the last frame of {video.name} after it renders") |
| elif source and video is not None: |
| print(f" would extract the last frame of {video}") |
| rendered[key] = dest |
| continue |
| if source: |
| if video is None or not video.is_file(): |
| raise BatchError(f"{source} has no video to take a last frame from") |
| frame = renders / f"{video.stem}_last.png" |
| extract_last_frame(video, frame) |
| image_name = stage_image(frame, comfy_input, f"{film}_{video.stem}_last.png") |
| elif local_image is not None: |
| image_name = stage_image(local_image, comfy_input, f"{film}_{path.stem}{local_image.suffix.lower()}") |
| else: |
| image_name = None |
| job = apply_job(graph, binding, text, prefix, seconds, image_name) |
| prompt_id = queue_prompt(args.host.rstrip("/"), job) |
| print(f" queued {prompt_id}") |
| entry = wait_history(args.host.rstrip("/"), prompt_id, args.timeout) |
| videos = output_videos(entry) |
| if not videos: |
| raise BatchError(f"{path.name} finished with no video in the history") |
| download_output(args.host.rstrip("/"), videos[-1], dest) |
| rendered[key] = dest |
| print(f" wrote {dest}") |
| if args.dry_run and not args.no_assemble: |
| assemble(folder, files, dry_run=True) |
| if not args.dry_run and not args.no_assemble: |
| assemble(folder, files) |
| return 0 |
|
|
|
|
| def self_test() -> int: |
| text_none = "handoff-from: none\nend-frame: hold\n\nintegrated_multimodal_description: hello\n" |
| text_link = "handoff-from: clip-03 last frame\nnext: clip-05\n\nbody\n" |
| assert handoff_source(text_none) is None |
| assert handoff_source(text_link) == "clip-03" |
| assert handoff_source("no header\n") is None |
| graph = json.loads(DEFAULT_WORKFLOW.read_text()) |
| binding = bind_graph(graph) |
| assert binding["image_id"] == "144" |
| assert binding["prompt_id"] == "105:104" |
| job = apply_job(graph, binding, "PROMPT", "videos/quickies/demo/clip-01-15s", 15, "demo.png") |
| assert job["105:104"]["inputs"]["prompt"] == "PROMPT" |
| assert job["144"]["inputs"]["image"] == "demo.png" |
| bare = apply_job(graph, binding, "PROMPT", "videos/quickies/demo/clip-01-15s", 15, None) |
| assert "first_frame" not in bare["105:104"]["inputs"] |
| assert "144" not in bare |
| assert job["92"]["inputs"]["filename_prefix"] == "videos/quickies/demo/clip-01-15s" |
| assert job["105:111"]["inputs"]["value"] == 15 |
| folder = Path("/tmp/batch-clips-self") |
| if folder.exists(): |
| shutil.rmtree(folder) |
| folder.mkdir() |
| (folder / "clip-02-15s.txt").write_text(text_link) |
| (folder / "clip-01-15s.txt").write_text(text_none) |
| (folder / "arc.md").write_text("skip") |
| ordered = [p.name for p in prompt_files(folder)] |
| assert ordered == ["clip-01-15s.txt", "clip-02-15s.txt"], ordered |
| video = folder / "clip.mp4" |
| subprocess.run([ |
| "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", |
| "-f", "lavfi", "-i", "color=c=black:s=32x32:r=10:d=0.2", |
| "-f", "lavfi", "-i", "color=c=white:s=32x32:r=10:d=0.1", |
| "-filter_complex", "[0][1]concat=n=2:v=1:a=0", |
| "-pix_fmt", "yuv420p", str(video), |
| ], check=True) |
| frame = extract_last_frame(video, folder / "last.png") |
| raw = subprocess.check_output([ |
| "ffmpeg", "-v", "error", "-i", str(frame), |
| "-vf", "scale=1:1,format=rgb24", "-f", "rawvideo", "-", |
| ]) |
| assert list(raw[:3]) == [255, 255, 255], list(raw[:3]) |
| try: |
| lookup_source("clip-09", {}) |
| except BatchError: |
| pass |
| else: |
| raise SystemExit("missing source video should fail") |
| still = folder / "start.png" |
| subprocess.run([ |
| "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", |
| "-f", "lavfi", "-i", "color=c=red:s=16x16:d=0.1", |
| "-frames:v", "1", "-update", "1", str(still), |
| ], check=True) |
| (folder / "clip-02-15s.txt").write_text( |
| "handoff-from: clip-01 last frame\n\nbody\n" |
| ) |
| code = main([ |
| "--dry-run", "--first-image", str(still), str(folder), |
| "--workflow", str(DEFAULT_WORKFLOW), |
| ]) |
| if code != 0: |
| raise SystemExit(f"dry-run failed: {code}") |
| code = main([ |
| "--dry-run", "--no-assemble", str(folder), |
| "--workflow", str(DEFAULT_WORKFLOW), |
| ]) |
| if code != 0: |
| raise SystemExit(f"text-only dry-run failed: {code}") |
| renders = folder / "renders" |
| renders.mkdir() |
| for name, color in (("clip-01-15s.mp4", "black"), ("clip-02-15s.mp4", "white")): |
| subprocess.run([ |
| "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", |
| "-f", "lavfi", "-i", f"color=c={color}:s=32x32:r=10:d=0.3", |
| "-f", "lavfi", "-i", "sine=frequency=440:duration=0.3", |
| "-shortest", "-pix_fmt", "yuv420p", str(renders / name), |
| ], check=True) |
| assemble(folder, prompt_files(folder)) |
| sequence = renders / "batch-clips-self.mp4" |
| picture = renders / "batch-clips-self_video.mp4" |
| sound = renders / "batch-clips-self_audio.m4a" |
| assert sequence.is_file() and picture.is_file() and sound.is_file() |
| assert "audio" not in stream_kinds(picture) |
| assert "video" not in stream_kinds(sound) |
| assert "video" in stream_kinds(sequence) and "audio" in stream_kinds(sequence) |
| queue_file = folder / "queue.txt" |
| first = folder / "proj-a" |
| second = folder / "proj-b" |
| for project in (first, second): |
| project.mkdir() |
| (project / "clip-01-15s.txt").write_text(text_none) |
| queue_file.write_text( |
| f"# overnight\nproj-a {still.name}\n\nproj-b {still.name}\n" |
| ) |
| code = main([ |
| "--dry-run", "--no-assemble", "--queue", str(queue_file), |
| "--workflow", str(DEFAULT_WORKFLOW), |
| ]) |
| if code != 0: |
| raise SystemExit(f"queue dry-run failed: {code}") |
| code = main([ |
| "--dry-run", "--no-assemble", "--first-image", str(still), |
| str(first), str(second), "--workflow", str(DEFAULT_WORKFLOW), |
| ]) |
| if code != 0: |
| raise SystemExit(f"multi-dir dry-run failed: {code}") |
| shutil.rmtree(folder) |
| print("self-test ok") |
| return 0 |
|
|
|
|
| def load_queue(path: Path) -> list[tuple[str, str | None]]: |
| """Each line is FOLDER [FIRST_IMAGE]. Paths are relative to the queue file.""" |
| if not path.is_file(): |
| raise BatchError(f"queue file not found: {path}") |
| base = path.parent |
| jobs = [] |
| for lineno, raw in enumerate(path.read_text().splitlines(), 1): |
| line = raw.split("#", 1)[0].strip() |
| if not line: |
| continue |
| parts = line.split() |
| if len(parts) > 2: |
| raise BatchError(f"{path}:{lineno}: expected FOLDER [FIRST_IMAGE]") |
| folder = Path(parts[0]) |
| if not folder.is_absolute(): |
| folder = base / folder |
| image = None |
| if len(parts) == 2: |
| image = Path(parts[1]) |
| if not image.is_absolute(): |
| image = base / image |
| image = str(image) |
| jobs.append((str(folder), image)) |
| if not jobs: |
| raise BatchError(f"no projects in {path}") |
| return jobs |
|
|
|
|
| def project_list(args: argparse.Namespace) -> list[tuple[str, str | None]]: |
| jobs = [] |
| if args.queue: |
| jobs.extend(load_queue(Path(args.queue).expanduser())) |
| for directory in args.directories: |
| jobs.append((directory, None)) |
| return jobs |
|
|
|
|
| def parse_args(argv: list[str]) -> argparse.Namespace: |
| parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) |
| parser.add_argument("directories", nargs="*", help="prompt folders, run in the order given") |
| parser.add_argument("--queue", help="text file of projects: FOLDER [FIRST_IMAGE] per line") |
| parser.add_argument("--workflow", default=str(DEFAULT_WORKFLOW), help="ComfyUI API graph") |
| parser.add_argument("--host", default=DEFAULT_HOST, help="ComfyUI base URL") |
| parser.add_argument("--input-dir", default=str(DEFAULT_INPUT), help="ComfyUI input folder") |
| parser.add_argument("--first-image", help="still for the first clip that has no handoff") |
| parser.add_argument("--timeout", type=float, default=2700, help="seconds to wait for one clip") |
| parser.add_argument("--skip-existing", action="store_true", help="keep renders/<clip>.mp4") |
| parser.add_argument("--no-assemble", action="store_true", help="leave the clips separate") |
| parser.add_argument("--assemble-only", action="store_true", help="stitch renders already on disk") |
| parser.add_argument("--dry-run", action="store_true", help="print the queue and do not generate") |
| parser.add_argument("--self-test", action="store_true", help="check parsing, the graph, and last-frame extract") |
| return parser.parse_args(argv) |
|
|
|
|
| def main(argv: list[str]) -> int: |
| args = parse_args(argv) |
| if args.self_test: |
| return self_test() |
| try: |
| jobs = project_list(args) |
| except BatchError as exc: |
| print(f"batch-clips: {exc}", file=sys.stderr) |
| return 1 |
| if not jobs: |
| parse_args(["--help"]) |
| return 2 |
| failures = [] |
| for index, (directory, image) in enumerate(jobs, 1): |
| print(f"\n=== project {index}/{len(jobs)}: {directory} ===") |
| one = argparse.Namespace(**vars(args)) |
| one.directory = directory |
| if image is not None: |
| one.first_image = image |
| try: |
| run(one) |
| except BatchError as exc: |
| print(f"batch-clips: {directory}: {exc}", file=sys.stderr) |
| failures.append(str(directory)) |
| if failures: |
| print( |
| f"batch-clips: {len(failures)} of {len(jobs)} project(s) failed", |
| file=sys.stderr, |
| ) |
| return 1 |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main(sys.argv[1:])) |
|
|
