Add Grok video Comfy API node
This commit is contained in:
pandy
@@ -35,6 +35,13 @@ pip install -r comfyui-PD_comfy-api-node/requirements.txt
|
||||
|
||||
## 📝 更新日志
|
||||
|
||||
### 2026-10-11 - Grok Video
|
||||
- 新增 `PD- grok video comfyapi`（`PDGrokVideoAPIKey`），位于 `PD_Tools/Video_Generation`，手动填写 Comfy.org API Key。
|
||||
- 支持 `grok-imagine-video` / `grok-imagine-video-1.5` / `grok-imagine-video-1.5-lite`，文生视频或单张起始图生视频，1–15 秒；参数与当前官方 Grok Video 节点一致。普通版不支持 1080p，1.5 系列有起始图时可省略提示词，seed 不保证输出复现。
|
||||
- 输出 `VIDEO` 可直接连接 Save Video，`info` 包含任务 ID、模型、服务返回的 usage，以及响应提供的实际 credits。费用角标明确标为估算。
|
||||
- 一次执行只提交一次生成请求；轮询、下载期间每秒显示耗时，每 0.1 秒检查取消，整次执行限时 600 秒，不自动重试付费请求。取消不保证退还已受理任务费用。
|
||||
- 验证：`python custom_nodes/comfyui-PD_comfy-api-node/tests/test_grok_video.py`。离线覆盖三个模型文/图生视频、MP4 输出可解码、任务/费用信息、提交/轮询/下载取消、超时不重试及请求前校验。本次没有真实付费生成。
|
||||
|
||||
### 2026-10-11 - Grok 文生图
|
||||
- 新增 `PDGrokImageAPIKey` 文生图节点，复用现有 Grok 模型列表；API Key 由节点输入提供。同步更新 Grok 编辑节点的取消检查与失败状态。
|
||||
|
||||
|
||||
@@ -0,0 +1,79 @@
|
||||
import io
|
||||
import json
|
||||
|
||||
from comfy_api.latest import IO, VideoFromFile
|
||||
from comfy_api_nodes.nodes_grok import GrokVideoNode, tensor_to_base64_string
|
||||
|
||||
from .OPENAI_latest_apikey import BASE_URL, _Request
|
||||
|
||||
|
||||
class PDGrokVideoAPIKey(IO.ComfyNode):
|
||||
@classmethod
|
||||
def define_schema(cls):
|
||||
schema = GrokVideoNode.define_schema()
|
||||
schema.node_id = "PDGrokVideoAPIKey"
|
||||
schema.display_name = "PD- grok video comfyapi"
|
||||
schema.category = "PD_Tools/Video_Generation"
|
||||
schema.inputs.insert(0, IO.String.Input("comfy_api_key", default="", tooltip="Comfy.org API Key"))
|
||||
schema.outputs.append(IO.String.Output(display_name="info"))
|
||||
schema.hidden = [IO.Hidden.unique_id]
|
||||
schema.is_api_node = False
|
||||
schema.price_badge.expr = schema.price_badge.expr.replace(
|
||||
'"usd": $is15 ? $total * 1.43 : $total}',
|
||||
'"usd": $is15 ? $total * 1.43 : $total, "format": {"approximate": true}}',
|
||||
)
|
||||
return schema
|
||||
|
||||
@classmethod
|
||||
def execute(cls, comfy_api_key, model, prompt, resolution, aspect_ratio, duration, seed, image=None):
|
||||
if model not in ("grok-imagine-video", "grok-imagine-video-1.5", "grok-imagine-video-1.5-lite"):
|
||||
raise ValueError(f"Unknown Grok video model: {model}")
|
||||
if resolution == "1080p" and model == "grok-imagine-video":
|
||||
raise ValueError("grok-imagine-video 不支持 1080p。")
|
||||
if not 1 <= duration <= 15:
|
||||
raise ValueError("视频时长须为 1–15 秒。")
|
||||
if image is not None and image.shape[0] != 1:
|
||||
raise ValueError("Grok Video 只支持一张起始图。")
|
||||
if not prompt.strip() and (image is None or model == "grok-imagine-video"):
|
||||
raise ValueError("请填写提示词。")
|
||||
payload = {"model": model, "prompt": prompt, "resolution": resolution,
|
||||
"duration": duration, "seed": seed}
|
||||
if aspect_ratio != "auto":
|
||||
payload["aspect_ratio"] = aspect_ratio
|
||||
with _Request(cls.hidden.unique_id, comfy_api_key) as api:
|
||||
api.check()
|
||||
if image is not None:
|
||||
payload["image"] = {"url": "data:image/png;base64," + tensor_to_base64_string(image)}
|
||||
created = api.request("POST", BASE_URL + "/proxy/xai/v1/videos/generations", json=payload)
|
||||
api.response_id = created.json().get("request_id", "")
|
||||
if not api.response_id:
|
||||
raise ValueError("Grok 响应缺少 request_id。")
|
||||
while True:
|
||||
response = api.request("GET", BASE_URL + "/proxy/xai/v1/videos/" + api.response_id)
|
||||
result = response.json()
|
||||
status = result.get("status")
|
||||
if status in ("failed", "error", "cancelled", "canceled", "expired") or result.get("error"):
|
||||
raise ValueError(f"Grok video {status}: {result.get('error', '')}")
|
||||
video = result.get("video") or {}
|
||||
if video.get("url") and status in (None, "done", "complete", "completed", "success", "succeeded"):
|
||||
break
|
||||
api.sleep(2)
|
||||
api.usage = result.get("usage") or {}
|
||||
info = {"request_id": api.response_id, "model": model, "usage": api.usage}
|
||||
credits = response.headers.get("X-Comfy-Credits-Used") or created.headers.get("X-Comfy-Credits-Used")
|
||||
if credits is not None:
|
||||
info["credits_used"] = credits
|
||||
url = video["url"]
|
||||
if url.startswith("/proxy/"):
|
||||
url = BASE_URL + url
|
||||
downloaded = api.request("GET", url)
|
||||
api.check()
|
||||
output = VideoFromFile(io.BytesIO(downloaded.content))
|
||||
# Reject invalid downloads before reporting success.
|
||||
output.get_dimensions()
|
||||
api.check()
|
||||
return IO.NodeOutput(output, json.dumps(info, ensure_ascii=False))
|
||||
|
||||
|
||||
NODE_CLASS_MAPPINGS = {"PDGrokVideoAPIKey": PDGrokVideoAPIKey}
|
||||
NODE_DISPLAY_NAME_MAPPINGS = {"PDGrokVideoAPIKey": "PD- grok video comfyapi"}
|
||||
@@ -0,0 +1,168 @@
|
||||
import importlib
|
||||
import io
|
||||
import json
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
import types
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import av
|
||||
import numpy as np
|
||||
import requests
|
||||
import torch
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[3]))
|
||||
sys.argv = [sys.argv[0], "--cpu"]
|
||||
package = types.ModuleType("pd_grok_video_test")
|
||||
package.__path__ = [str(Path(__file__).resolve().parents[1])]
|
||||
sys.modules[package.__name__] = package
|
||||
module = importlib.import_module("pd_grok_video_test.py.VIDEO_Grok_apikey")
|
||||
common = importlib.import_module("pd_grok_video_test.py.OPENAI_latest_apikey")
|
||||
Node = module.PDGrokVideoAPIKey
|
||||
Node.hidden = types.SimpleNamespace(unique_id="test-node")
|
||||
|
||||
|
||||
def make_video():
|
||||
buffer = io.BytesIO()
|
||||
with av.open(buffer, "w", format="mp4") as container:
|
||||
stream = container.add_stream("mpeg4", rate=24)
|
||||
stream.width = stream.height = 16
|
||||
stream.pix_fmt = "yuv420p"
|
||||
frame = av.VideoFrame.from_ndarray(np.zeros((16, 16, 3), dtype=np.uint8), format="rgb24")
|
||||
for packet in stream.encode(frame):
|
||||
container.mux(packet)
|
||||
for packet in stream.encode():
|
||||
container.mux(packet)
|
||||
return buffer.getvalue()
|
||||
|
||||
|
||||
class Response:
|
||||
def __init__(self, result=None):
|
||||
self.result = result or {}
|
||||
self.headers = {"X-Comfy-Credits-Used": "12.3"}
|
||||
self.content = make_video()
|
||||
|
||||
def json(self):
|
||||
return self.result
|
||||
|
||||
def raise_for_status(self):
|
||||
pass
|
||||
|
||||
|
||||
class Session:
|
||||
def __init__(self, slow=None, timeout=False, failed=False):
|
||||
self.calls = []
|
||||
self.closed = threading.Event()
|
||||
self.slow, self.timeout, self.failed = slow, timeout, failed
|
||||
|
||||
def request(self, method, url, **kwargs):
|
||||
self.calls.append((method, url, kwargs))
|
||||
stage = "submit" if method == "POST" else ("download" if "example.test" in url else "poll")
|
||||
if self.timeout:
|
||||
raise requests.Timeout("simulated")
|
||||
if stage == self.slow:
|
||||
self.closed.wait(5)
|
||||
if stage == "submit":
|
||||
return Response({"request_id": "test-task"})
|
||||
if stage == "poll":
|
||||
if self.failed:
|
||||
return Response({"status": "failed", "error": "simulated failure"})
|
||||
return Response({"status": "done", "video": {"url": "https://example.test/video.mp4"},
|
||||
"usage": {"cost_in_usd_ticks": 1000}})
|
||||
return Response()
|
||||
|
||||
def close(self):
|
||||
self.closed.set()
|
||||
|
||||
|
||||
class GrokVideoTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.cancel = threading.Event()
|
||||
self.events = []
|
||||
self.patches = [
|
||||
patch.object(common.PromptServer, "instance", types.SimpleNamespace(send_sync=lambda event, data: self.events.append(data)), create=True),
|
||||
patch.object(common.comfy.model_management, "throw_exception_if_processing_interrupted", self.check),
|
||||
]
|
||||
for item in self.patches:
|
||||
item.start()
|
||||
|
||||
def tearDown(self):
|
||||
for item in reversed(self.patches):
|
||||
item.stop()
|
||||
|
||||
def check(self):
|
||||
if self.cancel.is_set():
|
||||
raise common.comfy.model_management.InterruptProcessingException()
|
||||
|
||||
def run_node(self, **kwargs):
|
||||
values = dict(comfy_api_key="test-key", model="grok-imagine-video-1.5-lite", prompt="test prompt",
|
||||
resolution="720p", aspect_ratio="auto", duration=6, seed=42)
|
||||
values.update(kwargs)
|
||||
return Node.execute(**values)
|
||||
|
||||
def test_models_text_and_image_video_output(self):
|
||||
for model in ("grok-imagine-video", "grok-imagine-video-1.5", "grok-imagine-video-1.5-lite"):
|
||||
for image in (None, torch.zeros((1, 8, 8, 3))):
|
||||
session = Session()
|
||||
with patch.object(common.requests, "Session", return_value=session):
|
||||
result = self.run_node(model=model, image=image)
|
||||
self.assertEqual(result.result[0].get_dimensions(), (16, 16))
|
||||
info = json.loads(result.result[1])
|
||||
self.assertEqual(info["request_id"], "test-task")
|
||||
self.assertEqual(info["credits_used"], "12.3")
|
||||
self.assertEqual([item[0] for item in session.calls], ["POST", "GET", "GET"])
|
||||
payload = session.calls[0][2]["json"]
|
||||
self.assertEqual(payload["model"], model)
|
||||
self.assertEqual(payload["seed"], 42)
|
||||
self.assertEqual("image" in payload, image is not None)
|
||||
self.assertNotIn("aspect_ratio", payload)
|
||||
self.assertEqual(session.calls[-1][2]["headers"], {})
|
||||
|
||||
def test_cancel_submit_poll_download(self):
|
||||
for stage in ("submit", "poll", "download"):
|
||||
self.cancel.clear()
|
||||
self.events.clear()
|
||||
errors = []
|
||||
session = Session(slow=stage)
|
||||
def run():
|
||||
try:
|
||||
self.run_node()
|
||||
except common.comfy.model_management.InterruptProcessingException as exc:
|
||||
errors.append(exc)
|
||||
with patch.object(common.requests, "Session", return_value=session):
|
||||
worker = threading.Thread(target=run)
|
||||
worker.start()
|
||||
time.sleep(1.25)
|
||||
started = time.monotonic()
|
||||
self.cancel.set()
|
||||
worker.join(0.5)
|
||||
self.assertFalse(worker.is_alive())
|
||||
self.assertLess(time.monotonic() - started, 0.5)
|
||||
self.assertEqual(len(errors), 1)
|
||||
self.assertTrue(session.closed.is_set())
|
||||
self.assertEqual(sum(item[0] == "POST" for item in session.calls), 1)
|
||||
self.assertTrue(self.events[-1]["text"].startswith("Cancelled"))
|
||||
self.assertGreaterEqual(sum(item["text"].startswith("Running") for item in self.events), 2)
|
||||
|
||||
def test_timeout_failure_and_validation(self):
|
||||
for session, error in ((Session(timeout=True), requests.Timeout), (Session(failed=True), ValueError)):
|
||||
with patch.object(common.requests, "Session", return_value=session), self.assertRaises(error):
|
||||
self.run_node()
|
||||
self.assertEqual(sum(item[0] == "POST" for item in session.calls), 1)
|
||||
session = Session()
|
||||
with patch.object(common.requests, "Session", return_value=session):
|
||||
for kwargs in ({"model": "grok-imagine-video", "resolution": "1080p"},
|
||||
{"image": torch.zeros((2, 8, 8, 3))}, {"prompt": ""}, {"duration": 16}):
|
||||
with self.assertRaises(ValueError):
|
||||
self.run_node(**kwargs)
|
||||
self.cancel.set()
|
||||
with self.assertRaises(common.comfy.model_management.InterruptProcessingException):
|
||||
self.run_node()
|
||||
self.assertEqual(session.calls, [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(argv=[sys.argv[0]])
|
||||
Reference in New Issue
Block a user