Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Jedrzej KosinskiandAmp
|
69af6f818d | ||
|
Jedrzej Kosinski
|
8e91b27453 | ||
|
Jedrzej KosinskiandAmp
|
11a84be06d | ||
|
Jedrzej KosinskiandAmp
|
6b66f6c5a4 | ||
|
Jedrzej KosinskiandAmp
|
9843d63569 | ||
|
Jedrzej KosinskiandAmp
|
d7e2683e9d |
@@ -28,69 +28,24 @@ from .protocol import (
|
||||
log = logging.getLogger("comfy_remote_nodes.client")
|
||||
|
||||
|
||||
# Capability tokens advertised on every outbound request. Currently
|
||||
# covers the input/output encodings the registered providers (LTXV,
|
||||
# Gemini, Ideogram, ...) need, plus the ``schema:v3`` marker so
|
||||
# descriptors come back in the V3 ``get_v1_info()`` shape. Servers use
|
||||
# this list to negotiate per-request: e.g. a provider only emits the
|
||||
# multi-frame ``image:png_base64_batch`` envelope when this list
|
||||
# contains :data:`Capability.IMAGE_PNG_BASE64_BATCH`, otherwise it
|
||||
# falls back to the single-frame ``image:png_base64`` shape (with a
|
||||
# server-side truncation warning logged when extras are dropped).
|
||||
# Capability tokens advertised on every outbound request. Covers every
|
||||
# input/output encoding ``serialization.py`` knows how to round-trip,
|
||||
# plus the ``schema:v3`` marker so descriptors come back in the V3
|
||||
# ``get_v1_info()`` shape. Audio rides as ``mp3_base64`` outbound (the
|
||||
# encoder always re-encodes a ComfyUI AUDIO dict via
|
||||
# ``audio_input_to_mp3``) and decodes any of mp3_base64/mp3_inline/
|
||||
# wav_base64/wav_inline inbound — Stability returns wav_base64 by
|
||||
# default.
|
||||
CLIENT_CAPABILITIES = [
|
||||
Capability.SCHEMA_V3,
|
||||
Capability.VIDEO_MP4_INLINE,
|
||||
Capability.VIDEO_MP4_BASE64,
|
||||
Capability.IMAGE_PNG_BASE64,
|
||||
Capability.IMAGE_PNG_BASE64_BATCH,
|
||||
# Advertises that this client can emit a multi-frame IMAGE
|
||||
# envelope with per-frame presigned-GET URIs instead of inline
|
||||
# base64 frames when the summed batch payload would overflow the
|
||||
# 8 MiB inline cap (see ``serialization.maybe_externalize``).
|
||||
# Opportunistic: the client falls back to it whenever a batch
|
||||
# envelope it produced exceeds the cap, so descriptors don't
|
||||
# need to publish it in their ``input_serialization`` to
|
||||
# benefit — any input that already accepts ``png_base64_batch``
|
||||
# transparently accepts the externalised variant on the server
|
||||
# via the shared ``resolve_image_envelope_pngs`` dispatch.
|
||||
Capability.IMAGE_PNG_BASE64_BATCH_URI,
|
||||
Capability.MASK_PNG_BASE64,
|
||||
Capability.AUDIO_MP3_INLINE,
|
||||
Capability.AUDIO_MP3_BASE64,
|
||||
Capability.AUDIO_WAV_BASE64,
|
||||
Capability.AUDIO_MP3_INLINE,
|
||||
Capability.AUDIO_WAV_INLINE,
|
||||
# OGG/Opus decode path. PyAV / FFmpeg handle OGG/Opus natively in
|
||||
# ``audio_bytes_to_audio_input`` so the proxy_node side is the
|
||||
# same as MP3/WAV — advertising the token unlocks per-provider
|
||||
# ``output_format`` choices that emit Opus (e.g. ElevenLabs'
|
||||
# ``opus_48000_192``) without requiring servers to label-launder
|
||||
# the upstream container as ``mp3_base64``.
|
||||
Capability.AUDIO_OPUS_BASE64,
|
||||
# Advertises that this client can drive the
|
||||
# ``execute_async`` + poll-loop lifecycle (submit, poll, cancel,
|
||||
# TASK_LOST replay). Servers gate ``async_polling``-mode
|
||||
# descriptors on the presence of this token so an older client
|
||||
# that only knows ``/execute`` doesn't silently submit a
|
||||
# long-poll node into the sync endpoint.
|
||||
Capability.ASYNC_EXECUTE,
|
||||
# Generic opaque pass-through bucket for partner helper-config
|
||||
# custom IO types (OPENAI_INPUT_FILES, OPENAI_CHAT_CONFIG,
|
||||
# GEMINI_INPUT_FILES, RECRAFT_*, future partner helper types).
|
||||
# Servers may gate descriptors that publish ``input_serialization``
|
||||
# / ``output_serialization`` entries with the ``"opaque"`` marker
|
||||
# on the presence of this token; legacy clients that don't know
|
||||
# about IO.Custom fallthrough get filtered out before they try to
|
||||
# parse a descriptor whose io_type strings they wouldn't recognise.
|
||||
Capability.IO_OPAQUE,
|
||||
# 3D MODEL envelope (binary glTF 2.0 — single-mesh GLB). The
|
||||
# decoder in ``serialization.decode_model3d_envelope`` hands raw
|
||||
# GLB bytes to ``comfy_api.latest._util.File3D`` so downstream
|
||||
# nodes (preview3d / SaveGLB / Load3D) see the same value shape
|
||||
# they would from a local Tripo / Rodin / Meshy node. Servers
|
||||
# gate ``MODEL_3D`` provider descriptors on this token so a
|
||||
# legacy client without the decoder surfaces NEGOTIATION_FAILED
|
||||
# at descriptor-load time rather than crashing at execute.
|
||||
Capability.MODEL_3D_GLB_INLINE,
|
||||
Capability.SVG_XML_BASE64,
|
||||
]
|
||||
|
||||
# Reported in the X-RNP-Client-Version header.
|
||||
|
||||
@@ -1,352 +0,0 @@
|
||||
"""Smoke test for the three batch-URI followups (PR-2/PR-3 cleanup):
|
||||
|
||||
1. ``serialization.maybe_externalize`` keys batch dispatch off the
|
||||
``encoding`` string (``png_base64_batch``) — NOT field presence
|
||||
(``frames``).
|
||||
2. ``serialization.maybe_externalize`` rejects an empty/missing
|
||||
``frames`` list on a ``png_base64_batch`` envelope as a hard
|
||||
``RnpProtocolError`` rather than silently no-op-ing.
|
||||
3. ``proxy_node._deserialize_output`` accepts the externalised
|
||||
``png_base64_batch_uri`` shape on the response path: fetches every
|
||||
``uri`` (in parallel via ``asyncio.gather``), rewrites to inline
|
||||
``frames``, and decodes into a ``[B,H,W,C]`` tensor with frame
|
||||
ordering preserved.
|
||||
|
||||
Run with any python that has ``torch`` and ``Pillow`` available:
|
||||
|
||||
python notes/run_batch_uri_followups.py
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import base64
|
||||
import http.server
|
||||
import importlib.util
|
||||
import io
|
||||
import os
|
||||
import socketserver
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
import types
|
||||
|
||||
import numpy as np
|
||||
import torch
|
||||
from PIL import Image
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Locate the worktree under test.
|
||||
# ---------------------------------------------------------------------------
|
||||
_HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
_CLIENT_DIR = os.path.dirname(_HERE)
|
||||
|
||||
|
||||
def _load_module(name: str, path: str) -> types.ModuleType:
|
||||
spec = importlib.util.spec_from_file_location(name, path)
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
sys.modules[name] = mod
|
||||
spec.loader.exec_module(mod)
|
||||
return mod
|
||||
|
||||
|
||||
# Stub out client.upload_asset so the dispatch tests don't need a real
|
||||
# RNP server — the externalize batch path will call this once per frame.
|
||||
_uploads: list[bytes] = []
|
||||
|
||||
|
||||
async def _fake_upload_asset(server_url, raw, *, content_type, auth_headers=None): # noqa: ANN001
|
||||
_uploads.append(bytes(raw))
|
||||
return f"https://fake-storage/{len(_uploads) - 1}"
|
||||
|
||||
|
||||
def _install_fake_client() -> None:
|
||||
fake_client = types.ModuleType("comfy_remote_nodes.client")
|
||||
fake_client.upload_asset = _fake_upload_asset
|
||||
sys.modules["comfy_remote_nodes.client"] = fake_client
|
||||
|
||||
|
||||
def _png_b64_for_color(color: tuple[int, int, int]) -> str:
|
||||
img = Image.new("RGB", (4, 4), color=color)
|
||||
buf = io.BytesIO()
|
||||
img.save(buf, format="PNG")
|
||||
return base64.b64encode(buf.getvalue()).decode("ascii")
|
||||
|
||||
|
||||
def _png_bytes_for_color(color: tuple[int, int, int]) -> bytes:
|
||||
img = Image.new("RGB", (4, 4), color=color)
|
||||
buf = io.BytesIO()
|
||||
img.save(buf, format="PNG")
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test 1+2: maybe_externalize dispatch + empty-frames
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
async def _test_maybe_externalize() -> list[str]:
|
||||
failures: list[str] = []
|
||||
|
||||
# Re-import serialization fresh because it caches a `client` ref.
|
||||
sys.modules.pop("comfy_remote_nodes", None)
|
||||
sys.modules.pop("comfy_remote_nodes.serialization", None)
|
||||
sys.modules.pop("comfy_remote_nodes.protocol", None)
|
||||
|
||||
# Build a minimal package shim so `from . import client` works.
|
||||
pkg = types.ModuleType("comfy_remote_nodes")
|
||||
pkg.__path__ = [_CLIENT_DIR]
|
||||
sys.modules["comfy_remote_nodes"] = pkg
|
||||
_install_fake_client()
|
||||
protocol = _load_module(
|
||||
"comfy_remote_nodes.protocol",
|
||||
os.path.join(_CLIENT_DIR, "protocol.py"),
|
||||
)
|
||||
serialization = _load_module(
|
||||
"comfy_remote_nodes.serialization",
|
||||
os.path.join(_CLIENT_DIR, "serialization.py"),
|
||||
)
|
||||
|
||||
# --- Case A: batch envelope with payload bigger than cap → externalises
|
||||
_uploads.clear()
|
||||
big_frames = [_png_b64_for_color((i * 30, 0, 0)) for i in range(4)]
|
||||
env = {
|
||||
"type": "image",
|
||||
"encoding": "png_base64_batch",
|
||||
"frames": big_frames,
|
||||
"shape": [4, 4, 4, 3],
|
||||
"dtype": "uint8",
|
||||
"color_space": "srgb",
|
||||
}
|
||||
out = await serialization.maybe_externalize(
|
||||
env, server_url="http://fake", max_inline_bytes=1, auth_headers={},
|
||||
)
|
||||
if out.get("encoding") != "png_base64_batch_uri":
|
||||
failures.append(
|
||||
f"A: expected encoding=png_base64_batch_uri, got {out.get('encoding')!r}"
|
||||
)
|
||||
if "frames" in out:
|
||||
failures.append("A: 'frames' field still present after externalise")
|
||||
if not isinstance(out.get("uris"), list) or len(out["uris"]) != 4:
|
||||
failures.append(f"A: expected 4 uris, got {out.get('uris')!r}")
|
||||
if len(_uploads) != 4:
|
||||
failures.append(f"A: expected 4 uploads, got {len(_uploads)}")
|
||||
|
||||
# --- Case B: dispatch keys off encoding, NOT field presence.
|
||||
# An envelope with a stray `frames` field but a non-batch encoding
|
||||
# must NOT be treated as a batch (would have under the old
|
||||
# `isinstance(frames, list) and frames` dispatch).
|
||||
_uploads.clear()
|
||||
rogue = {
|
||||
"type": "image",
|
||||
"encoding": "png_base64", # singleton encoding
|
||||
"data": _png_b64_for_color((10, 20, 30)) * 200, # > cap
|
||||
"frames": ["bogus-extra-field"],
|
||||
}
|
||||
out_b = await serialization.maybe_externalize(
|
||||
rogue, server_url="http://fake", max_inline_bytes=1, auth_headers={},
|
||||
)
|
||||
if out_b.get("encoding") == "png_base64_batch_uri":
|
||||
failures.append(
|
||||
"B: dispatch took batch branch on a singleton encoding (rogue frames field)"
|
||||
)
|
||||
if "uri" not in out_b:
|
||||
failures.append("B: singleton encoding should have externalised to 'uri'")
|
||||
if len(_uploads) != 1:
|
||||
failures.append(f"B: expected 1 upload (singleton path), got {len(_uploads)}")
|
||||
|
||||
# --- Case C: empty frames on a batch encoding is a hard error.
|
||||
bad = {
|
||||
"type": "image",
|
||||
"encoding": "png_base64_batch",
|
||||
"frames": [],
|
||||
"shape": [0, 4, 4, 3],
|
||||
}
|
||||
try:
|
||||
await serialization.maybe_externalize(
|
||||
bad, server_url="http://fake", max_inline_bytes=1, auth_headers={},
|
||||
)
|
||||
except protocol.RnpProtocolError:
|
||||
pass
|
||||
else:
|
||||
failures.append("C: empty 'frames' on batch encoding did not raise RnpProtocolError")
|
||||
|
||||
# --- Case D: under-cap batch: no upload, envelope unchanged.
|
||||
_uploads.clear()
|
||||
small = {
|
||||
"type": "image",
|
||||
"encoding": "png_base64_batch",
|
||||
"frames": big_frames,
|
||||
"shape": [4, 4, 4, 3],
|
||||
}
|
||||
out_d = await serialization.maybe_externalize(
|
||||
small,
|
||||
server_url="http://fake",
|
||||
max_inline_bytes=10 * 1024 * 1024,
|
||||
auth_headers={},
|
||||
)
|
||||
if out_d is not small and out_d != small:
|
||||
failures.append("D: under-cap batch should be a no-op")
|
||||
if _uploads:
|
||||
failures.append(f"D: under-cap batch should not upload, got {len(_uploads)}")
|
||||
|
||||
return failures
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test 3: _deserialize_output handles png_base64_batch_uri on response
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
DELAY_S = 0.4
|
||||
N_FRAMES = 4
|
||||
|
||||
|
||||
class _SlowPngHandler(http.server.BaseHTTPRequestHandler):
|
||||
def do_GET(self) -> None: # noqa: N802
|
||||
time.sleep(DELAY_S)
|
||||
try:
|
||||
idx = int(self.path.rsplit("/", 1)[-1])
|
||||
except ValueError:
|
||||
self.send_response(404); self.end_headers(); return
|
||||
# Distinct color per frame so we can verify ordering after decode.
|
||||
body = _png_bytes_for_color((idx * 60 % 256, 80, 120))
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "image/png")
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def log_message(self, *_a, **_kw) -> None:
|
||||
return
|
||||
|
||||
|
||||
async def _test_deserialize_output() -> list[str]:
|
||||
failures: list[str] = []
|
||||
|
||||
server = socketserver.ThreadingTCPServer(("127.0.0.1", 0), _SlowPngHandler)
|
||||
host, port = server.server_address
|
||||
th = threading.Thread(target=server.serve_forever, daemon=True)
|
||||
th.start()
|
||||
try:
|
||||
# Stub the proxy_node's _fetch_url_envelope so we don't have to
|
||||
# bring up the whole comfy_api_nodes.util.client stack.
|
||||
import urllib.request as _urlreq
|
||||
|
||||
async def fake_fetch(envelope, policy, *, auth_headers, node_cls): # noqa: ANN001, ARG001
|
||||
# Real ``_fetch_url_envelope`` is properly async (aiohttp);
|
||||
# the smoke stub uses sync urllib so we hop to a thread to
|
||||
# preserve the parallel-gather semantics under test.
|
||||
uri = envelope["uri"]
|
||||
def _do_fetch() -> bytes:
|
||||
with _urlreq.urlopen(uri, timeout=10.0) as resp:
|
||||
return resp.read()
|
||||
return await asyncio.to_thread(_do_fetch)
|
||||
|
||||
# Re-import proxy_node helpers via AST to avoid loading ComfyUI.
|
||||
import ast
|
||||
|
||||
with open(os.path.join(_CLIENT_DIR, "proxy_node.py"), "r", encoding="utf-8") as fh:
|
||||
src = fh.read()
|
||||
tree = ast.parse(src)
|
||||
wanted = {"_deserialize_output"}
|
||||
nodes = [n for n in tree.body if isinstance(n, ast.AsyncFunctionDef) and n.name in wanted]
|
||||
if len(nodes) != len(wanted):
|
||||
failures.append(f"missing AST nodes; found {[n.name for n in nodes]}")
|
||||
return failures
|
||||
|
||||
# Re-import the protocol + serialization modules with the package
|
||||
# shim so `from . import` works inside serialization.
|
||||
sys.modules.pop("comfy_remote_nodes.serialization", None)
|
||||
sys.modules.pop("comfy_remote_nodes.protocol", None)
|
||||
pkg = types.ModuleType("comfy_remote_nodes")
|
||||
pkg.__path__ = [_CLIENT_DIR]
|
||||
sys.modules["comfy_remote_nodes"] = pkg
|
||||
_install_fake_client()
|
||||
protocol = _load_module(
|
||||
"comfy_remote_nodes.protocol",
|
||||
os.path.join(_CLIENT_DIR, "protocol.py"),
|
||||
)
|
||||
serialization = _load_module(
|
||||
"comfy_remote_nodes.serialization",
|
||||
os.path.join(_CLIENT_DIR, "serialization.py"),
|
||||
)
|
||||
|
||||
ns: dict = {
|
||||
"Any": object,
|
||||
"RnpProtocolError": protocol.RnpProtocolError,
|
||||
"ErrorCode": protocol.ErrorCode,
|
||||
"is_envelope": protocol.is_envelope,
|
||||
"serialization": serialization,
|
||||
"_fetch_url_envelope": fake_fetch,
|
||||
"log": types.SimpleNamespace(warning=lambda *a, **kw: None),
|
||||
}
|
||||
mod_src = ast.Module(body=nodes, type_ignores=[])
|
||||
exec(compile(mod_src, "<batch_uri_smoke>", "exec"), ns)
|
||||
deserialize = ns["_deserialize_output"]
|
||||
|
||||
# Build the externalised batch envelope on the response path.
|
||||
uris = [f"http://{host}:{port}/frame/{i}" for i in range(N_FRAMES)]
|
||||
envelope = {
|
||||
"type": "image",
|
||||
"encoding": "png_base64_batch_uri",
|
||||
"uris": uris,
|
||||
"shape": [N_FRAMES, 4, 4, 3],
|
||||
"dtype": "uint8",
|
||||
"color_space": "srgb",
|
||||
}
|
||||
url_fetch_policy = {"image": {"url_kind": "presigned", "timeout_s": 10}}
|
||||
|
||||
t0 = time.monotonic()
|
||||
tensor = await deserialize(
|
||||
envelope, meta=None,
|
||||
url_fetch_policy=url_fetch_policy,
|
||||
server_url="http://fake",
|
||||
auth_headers={},
|
||||
node_cls=None,
|
||||
)
|
||||
elapsed = time.monotonic() - t0
|
||||
|
||||
if not isinstance(tensor, torch.Tensor):
|
||||
failures.append(f"expected torch.Tensor, got {type(tensor).__name__}")
|
||||
return failures
|
||||
if tuple(tensor.shape) != (N_FRAMES, 4, 4, 3):
|
||||
failures.append(f"shape {tuple(tensor.shape)} != ({N_FRAMES}, 4, 4, 3)")
|
||||
|
||||
# Each frame's R-channel should match the expected color.
|
||||
for i in range(N_FRAMES):
|
||||
expected_r = (i * 60 % 256) / 255.0
|
||||
actual_r = float(tensor[i, 0, 0, 0])
|
||||
if abs(actual_r - expected_r) > 1e-3:
|
||||
failures.append(
|
||||
f"frame {i}: R channel {actual_r:.3f} != expected {expected_r:.3f} "
|
||||
f"(ordering broken or wrong fetch)"
|
||||
)
|
||||
|
||||
# Wall-clock: parallel should be ~DELAY_S, sequential would be ~N*DELAY_S.
|
||||
sequential_lower = N_FRAMES * DELAY_S * 0.7
|
||||
if elapsed >= sequential_lower:
|
||||
failures.append(
|
||||
f"wall clock {elapsed:.2f}s >= sequential lower bound "
|
||||
f"{sequential_lower:.2f}s — gather not running in parallel"
|
||||
)
|
||||
|
||||
return failures
|
||||
finally:
|
||||
server.shutdown(); server.server_close()
|
||||
|
||||
|
||||
def main() -> int:
|
||||
failures: list[str] = []
|
||||
failures += asyncio.run(_test_maybe_externalize())
|
||||
failures += asyncio.run(_test_deserialize_output())
|
||||
if failures:
|
||||
print()
|
||||
print("FAIL:")
|
||||
for f in failures:
|
||||
print(" -", f)
|
||||
return 1
|
||||
print("PASS: maybe_externalize dispatches on encoding + rejects empty frames; "
|
||||
"_deserialize_output handles png_base64_batch_uri in parallel")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -1,441 +0,0 @@
|
||||
"""Smoke test for the descriptor-driven ``local_validate.image_max_batch``
|
||||
client-side fail-fast.
|
||||
|
||||
Exercises the proxy_node's ``_collect_local_validate`` /
|
||||
``_enforce_local_validate`` pair against a fake descriptor that mirrors
|
||||
the Flux2 wire shape. Asserts that:
|
||||
|
||||
1. ``_collect_local_validate`` extracts ``{"images": {"image_max_batch": 2}}``
|
||||
from the descriptor's IMAGE-input options dict.
|
||||
2. Encoding a 3-frame tensor for that input raises ``RnpProtocolError``
|
||||
with ``code=INPUT_INVALID`` and ``user_facing=True`` *before* any
|
||||
serialization / HTTP attempt.
|
||||
3. A 1-frame tensor passes through cleanly.
|
||||
4. A descriptor without ``local_validate`` is a no-op.
|
||||
5. ``cap=1`` surfaces the single-frame copy.
|
||||
6. A ``local_validate`` block missing the ``image_max_batch`` key is a no-op.
|
||||
|
||||
Run with any python that has ``torch`` available:
|
||||
|
||||
python notes/run_image_max_batch.py
|
||||
|
||||
The test loads the helper functions out of ``proxy_node.py`` via AST
|
||||
extraction so it doesn't pay the cost of importing ComfyUI's
|
||||
``comfy_api`` / ``comfy_api_nodes`` packages (which the helper
|
||||
functions don't touch). Standalone — no test runner / fixtures needed.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import asyncio
|
||||
import importlib.util
|
||||
import os
|
||||
import sys
|
||||
import types
|
||||
|
||||
import torch
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Locate the worktree under test (this file's parent's parent).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
_CLIENT_DIR = os.path.dirname(_HERE)
|
||||
|
||||
|
||||
def _load_protocol() -> types.ModuleType:
|
||||
"""Load ``protocol.py`` standalone (no relative imports)."""
|
||||
path = os.path.join(_CLIENT_DIR, "protocol.py")
|
||||
spec = importlib.util.spec_from_file_location("rnp_smoke_protocol", path)
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
sys.modules[spec.name] = mod
|
||||
spec.loader.exec_module(mod)
|
||||
return mod
|
||||
|
||||
|
||||
def _build_serialization_stub(protocol: types.ModuleType) -> types.ModuleType:
|
||||
"""Minimal ``serialization`` shim covering exactly the surface
|
||||
``_encode_one`` / ``_enforce_local_validate`` / ``_encode_inputs`` use:
|
||||
|
||||
* ``_is_torch_tensor(value)`` — duck-type check for torch tensors.
|
||||
* ``is_audio_input(value)`` — always False (no audio in this test).
|
||||
* ``encode_image_tensor(tensor, *, accepts_batch)`` — fake envelope.
|
||||
* ``encode_mask_tensor(tensor)`` — fake envelope.
|
||||
* ``is_envelope(value)`` — False so ``maybe_externalize`` is skipped.
|
||||
"""
|
||||
mod = types.ModuleType("rnp_smoke_serialization")
|
||||
|
||||
def _is_torch_tensor(value): # noqa: ANN001
|
||||
try:
|
||||
import torch as _torch
|
||||
except ImportError:
|
||||
return False
|
||||
return isinstance(value, _torch.Tensor)
|
||||
|
||||
def is_audio_input(value): # noqa: ANN001
|
||||
return isinstance(value, dict) and {"waveform", "sample_rate"} <= set(value.keys())
|
||||
|
||||
def encode_image_tensor(tensor, *, accepts_batch=False): # noqa: ANN001, ARG001
|
||||
return {"type": "image", "encoding": "png_base64", "frames": int(tensor.shape[0])}
|
||||
|
||||
def encode_mask_tensor(tensor): # noqa: ANN001, ARG001
|
||||
return {"type": "mask", "encoding": "png_base64"}
|
||||
|
||||
def encode_audio_input(value): # noqa: ANN001, ARG001
|
||||
return {"type": "audio", "encoding": "mp3_base64"}
|
||||
|
||||
def is_envelope(value): # noqa: ANN001
|
||||
return False # smoke test never externalizes
|
||||
|
||||
mod._is_torch_tensor = _is_torch_tensor
|
||||
mod.is_audio_input = is_audio_input
|
||||
mod.encode_image_tensor = encode_image_tensor
|
||||
mod.encode_mask_tensor = encode_mask_tensor
|
||||
mod.encode_audio_input = encode_audio_input
|
||||
mod.is_envelope = is_envelope
|
||||
|
||||
async def maybe_externalize(*args, **kwargs): # pragma: no cover — unused
|
||||
return args[0] if args else None
|
||||
mod.maybe_externalize = maybe_externalize
|
||||
return mod
|
||||
|
||||
|
||||
def _build_proxy_node_stub(
|
||||
protocol: types.ModuleType, serialization: types.ModuleType,
|
||||
) -> types.ModuleType:
|
||||
"""Extract the four helper functions under test from ``proxy_node.py``
|
||||
and exec them in a minimal namespace so we don't drag the rest of
|
||||
the module's heavy ComfyUI imports into the test process.
|
||||
"""
|
||||
mod = types.ModuleType("rnp_smoke_proxy_node")
|
||||
mod.__dict__.update({
|
||||
"Any": object,
|
||||
"log": types.SimpleNamespace(warning=lambda *a, **kw: None),
|
||||
"serialization": serialization,
|
||||
"RnpProtocolError": protocol.RnpProtocolError,
|
||||
"ErrorCode": protocol.ErrorCode,
|
||||
})
|
||||
src_path = os.path.join(_CLIENT_DIR, "proxy_node.py")
|
||||
with open(src_path, "r", encoding="utf-8") as fh:
|
||||
src = fh.read()
|
||||
tree = ast.parse(src)
|
||||
wanted = {
|
||||
"_collect_local_validate",
|
||||
"_rules_from_input_spec",
|
||||
"_enforce_local_validate",
|
||||
"_check_image_max_batch",
|
||||
"_encode_one",
|
||||
"_encode_inputs",
|
||||
}
|
||||
nodes = [
|
||||
n for n in tree.body
|
||||
if isinstance(n, (ast.AsyncFunctionDef, ast.FunctionDef)) and n.name in wanted
|
||||
]
|
||||
snippet = ast.Module(body=nodes, type_ignores=[])
|
||||
code = compile(snippet, src_path, "exec")
|
||||
exec(code, mod.__dict__)
|
||||
missing = wanted - set(mod.__dict__)
|
||||
if missing:
|
||||
raise RuntimeError(f"proxy_node helpers missing from extract: {missing}")
|
||||
return mod
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _flux2_like_descriptor(cap: int) -> dict:
|
||||
"""Mirrors the Flux2 ``images`` input descriptor surface."""
|
||||
return {
|
||||
"name": "FakeFlux2_RNP",
|
||||
"input": {
|
||||
"required": {
|
||||
"prompt": ["STRING", {"multiline": True, "default": ""}],
|
||||
},
|
||||
"optional": {
|
||||
"images": ["IMAGE", {
|
||||
"tooltip": "Up to N images to be used as references.",
|
||||
"local_validate": {"image_max_batch": cap},
|
||||
}],
|
||||
},
|
||||
"hidden": {},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _no_validate_descriptor() -> dict:
|
||||
return {
|
||||
"name": "FakeNoValidate_RNP",
|
||||
"input": {
|
||||
"required": {},
|
||||
"optional": {
|
||||
"images": ["IMAGE", {"tooltip": "no cap"}],
|
||||
},
|
||||
"hidden": {},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _autogrow_descriptor(cap: int, prefix: str = "reference") -> dict:
|
||||
"""Mirrors GrokVideoReferenceNode_RNP's top-level Autogrow input
|
||||
after server descriptor unwraps the nested DynamicCombo branch
|
||||
(kept simpler here for direct AUTOGROW coverage)."""
|
||||
return {
|
||||
"name": "FakeAutogrow_RNP",
|
||||
"input": {
|
||||
"required": {
|
||||
"reference_images": ["AUTOGROW", {
|
||||
"template": ["IMAGE", {
|
||||
"local_validate": {"image_max_batch": cap},
|
||||
}],
|
||||
"prefix": prefix,
|
||||
"min": 1,
|
||||
"max": cap,
|
||||
}],
|
||||
},
|
||||
"optional": {},
|
||||
"hidden": {},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _dyncombo_with_autogrow_descriptor() -> dict:
|
||||
"""Mirrors GrokImageEditNodeV2_RNP's ``model`` DynamicCombo with
|
||||
two branches each carrying an Autogrow ``images`` template — one
|
||||
capped at 3, the other at 1."""
|
||||
def branch(key: str, cap: int) -> dict:
|
||||
return {
|
||||
"key": key,
|
||||
"inputs": [
|
||||
["images", "AUTOGROW", {
|
||||
"template": ["IMAGE", {
|
||||
"local_validate": {"image_max_batch": cap},
|
||||
}],
|
||||
"prefix": "image",
|
||||
"min": 1,
|
||||
"max": cap,
|
||||
}],
|
||||
["resolution", "COMBO", {"options": ["1K", "2K"]}],
|
||||
],
|
||||
}
|
||||
return {
|
||||
"name": "FakeDynComboAutogrow_RNP",
|
||||
"input": {
|
||||
"required": {
|
||||
"model": ["DYNAMIC_COMBO", {
|
||||
"options": [
|
||||
branch("grok-imagine-image-quality", 3),
|
||||
branch("grok-imagine-image-pro", 1),
|
||||
],
|
||||
}],
|
||||
},
|
||||
"optional": {},
|
||||
"hidden": {},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
async def main() -> int:
|
||||
protocol = _load_protocol()
|
||||
serialization = _build_serialization_stub(protocol)
|
||||
proxy_node = _build_proxy_node_stub(protocol, serialization)
|
||||
|
||||
# ---- 1. _collect_local_validate extracts the rule
|
||||
desc = _flux2_like_descriptor(cap=2)
|
||||
rules = proxy_node._collect_local_validate(desc)
|
||||
assert rules == {"images": {"image_max_batch": 2}}, rules
|
||||
print("ok: _collect_local_validate extracted", rules)
|
||||
|
||||
# ---- 2. over-cap tensor raises INPUT_INVALID before encoding
|
||||
over_cap = torch.zeros((3, 64, 64, 3), dtype=torch.float32)
|
||||
raised = False
|
||||
try:
|
||||
await proxy_node._encode_inputs(
|
||||
{"images": over_cap},
|
||||
input_serialization={"images": ["png_base64", "png_base64_batch"]},
|
||||
local_validate=rules,
|
||||
node_id="FakeFlux2_RNP",
|
||||
)
|
||||
except protocol.RnpProtocolError as e:
|
||||
raised = True
|
||||
assert e.code == protocol.ErrorCode.INPUT_INVALID, e.code
|
||||
assert e.user_facing is True, e.user_facing
|
||||
msg = str(e)
|
||||
assert "FakeFlux2_RNP" in msg, msg
|
||||
assert "3" in msg and "2" in msg, msg
|
||||
print("ok: over-cap raised INPUT_INVALID:", msg)
|
||||
assert raised, "over-cap tensor did not raise"
|
||||
|
||||
# ---- 3. at-cap tensor passes through (no exception)
|
||||
at_cap = torch.zeros((2, 64, 64, 3), dtype=torch.float32)
|
||||
out = await proxy_node._encode_inputs(
|
||||
{"images": at_cap},
|
||||
input_serialization={"images": ["png_base64", "png_base64_batch"]},
|
||||
local_validate=rules,
|
||||
node_id="FakeFlux2_RNP",
|
||||
)
|
||||
assert "images" in out
|
||||
print("ok: at-cap (B=2) encoded without error")
|
||||
|
||||
# ---- 4. cap=1 / multi-frame raises with the single-frame copy
|
||||
desc1 = _flux2_like_descriptor(cap=1)
|
||||
rules1 = proxy_node._collect_local_validate(desc1)
|
||||
raised = False
|
||||
try:
|
||||
await proxy_node._encode_inputs(
|
||||
{"images": over_cap},
|
||||
local_validate=rules1,
|
||||
node_id="FakeI2V_RNP",
|
||||
)
|
||||
except protocol.RnpProtocolError as e:
|
||||
raised = True
|
||||
assert e.code == protocol.ErrorCode.INPUT_INVALID
|
||||
assert "single image" in str(e), str(e)
|
||||
print("ok: cap=1 single-frame copy fired:", str(e))
|
||||
assert raised
|
||||
|
||||
# ---- 5. descriptor without local_validate is a no-op
|
||||
desc_n = _no_validate_descriptor()
|
||||
rules_n = proxy_node._collect_local_validate(desc_n)
|
||||
assert rules_n == {}, rules_n
|
||||
out = await proxy_node._encode_inputs(
|
||||
{"images": over_cap},
|
||||
input_serialization={"images": ["png_base64", "png_base64_batch"]},
|
||||
local_validate=rules_n,
|
||||
node_id="FakeNoValidate_RNP",
|
||||
)
|
||||
assert "images" in out
|
||||
print("ok: missing local_validate is a no-op")
|
||||
|
||||
# ---- 6. local_validate block without image_max_batch sub-key is a no-op
|
||||
out = await proxy_node._encode_inputs(
|
||||
{"images": over_cap},
|
||||
input_serialization={"images": ["png_base64", "png_base64_batch"]},
|
||||
local_validate={"images": {"some_other_rule": 7}},
|
||||
node_id="FakeOther_RNP",
|
||||
)
|
||||
assert "images" in out
|
||||
print("ok: missing image_max_batch sub-key is a no-op")
|
||||
|
||||
# ---- 7. AUTOGROW: each slot value enforced against the template cap
|
||||
desc_ag = _autogrow_descriptor(cap=2)
|
||||
rules_ag = proxy_node._collect_local_validate(desc_ag)
|
||||
assert "reference_images" in rules_ag, rules_ag
|
||||
assert rules_ag["reference_images"].get("__template__") == {"image_max_batch": 2}, rules_ag
|
||||
print("ok: AUTOGROW collected __template__ rules:", rules_ag)
|
||||
|
||||
one_frame = torch.zeros((1, 64, 64, 3), dtype=torch.float32)
|
||||
raised = False
|
||||
try:
|
||||
# Slot 1 is over-cap (B=3 > cap=2) → should fire on that slot
|
||||
await proxy_node._encode_inputs(
|
||||
{"reference_images": {"reference0": one_frame, "reference1": over_cap}},
|
||||
local_validate=rules_ag,
|
||||
node_id="FakeAutogrow_RNP",
|
||||
)
|
||||
except protocol.RnpProtocolError as e:
|
||||
raised = True
|
||||
assert e.code == protocol.ErrorCode.INPUT_INVALID
|
||||
msg = str(e)
|
||||
# Slot path label should mention reference_images.<slot_key>
|
||||
assert "reference_images" in msg, msg
|
||||
assert "reference1" in msg, msg
|
||||
print("ok: AUTOGROW slot over-cap fired:", msg)
|
||||
assert raised, "AUTOGROW slot over-cap did not raise"
|
||||
|
||||
# All-under-cap autogrow passes (no exception)
|
||||
out = await proxy_node._encode_inputs(
|
||||
{"reference_images": {"reference0": one_frame, "reference1": one_frame, "reference2": None}},
|
||||
local_validate=rules_ag,
|
||||
node_id="FakeAutogrow_RNP",
|
||||
)
|
||||
assert "reference_images" in out
|
||||
print("ok: AUTOGROW all-under-cap (None slots tolerated) passed")
|
||||
|
||||
# ---- 8. DYNAMIC_COMBO with nested AUTOGROW: branch-aware enforcement
|
||||
desc_dc = _dyncombo_with_autogrow_descriptor()
|
||||
rules_dc = proxy_node._collect_local_validate(desc_dc)
|
||||
assert "model" in rules_dc, rules_dc
|
||||
branches = rules_dc["model"].get("__branches__")
|
||||
assert isinstance(branches, dict), rules_dc
|
||||
assert "grok-imagine-image-quality" in branches
|
||||
assert "grok-imagine-image-pro" in branches
|
||||
assert branches["grok-imagine-image-quality"]["images"]["__template__"] == {"image_max_batch": 3}
|
||||
assert branches["grok-imagine-image-pro"]["images"]["__template__"] == {"image_max_batch": 1}
|
||||
print("ok: DYNAMIC_COMBO collected branch-keyed templates")
|
||||
|
||||
# Branch A (cap=3): slot with B=4 should raise
|
||||
over_4 = torch.zeros((4, 64, 64, 3), dtype=torch.float32)
|
||||
raised = False
|
||||
try:
|
||||
await proxy_node._encode_inputs(
|
||||
{"model": {
|
||||
"model": "grok-imagine-image-quality",
|
||||
"resolution": "1K",
|
||||
"images": {"image0": over_4},
|
||||
}},
|
||||
local_validate=rules_dc,
|
||||
node_id="FakeDynComboAutogrow_RNP",
|
||||
)
|
||||
except protocol.RnpProtocolError as e:
|
||||
raised = True
|
||||
assert e.code == protocol.ErrorCode.INPUT_INVALID
|
||||
msg = str(e)
|
||||
assert "image0" in msg and "3" in msg and "4" in msg, msg
|
||||
print("ok: DYNAMIC_COMBO branch-A (cap=3) fired on B=4 slot:", msg)
|
||||
assert raised
|
||||
|
||||
# Branch A (cap=3): all slots within cap → passes
|
||||
out = await proxy_node._encode_inputs(
|
||||
{"model": {
|
||||
"model": "grok-imagine-image-quality",
|
||||
"resolution": "2K",
|
||||
"images": {"image0": one_frame, "image1": one_frame, "image2": one_frame},
|
||||
}},
|
||||
local_validate=rules_dc,
|
||||
node_id="FakeDynComboAutogrow_RNP",
|
||||
)
|
||||
assert "model" in out
|
||||
print("ok: DYNAMIC_COMBO branch-A all-under-cap passed")
|
||||
|
||||
# Branch B (cap=1): a B=3 slot fires the single-frame copy
|
||||
raised = False
|
||||
try:
|
||||
await proxy_node._encode_inputs(
|
||||
{"model": {
|
||||
"model": "grok-imagine-image-pro",
|
||||
"resolution": "2K",
|
||||
"images": {"image0": over_cap},
|
||||
}},
|
||||
local_validate=rules_dc,
|
||||
node_id="FakeDynComboAutogrow_RNP",
|
||||
)
|
||||
except protocol.RnpProtocolError as e:
|
||||
raised = True
|
||||
assert e.code == protocol.ErrorCode.INPUT_INVALID
|
||||
assert "single image" in str(e), str(e)
|
||||
print("ok: DYNAMIC_COMBO branch-B (cap=1) fired single-frame copy:", str(e))
|
||||
assert raised
|
||||
|
||||
# Unknown branch key → no enforcement (defensive)
|
||||
out = await proxy_node._encode_inputs(
|
||||
{"model": {
|
||||
"model": "unknown-branch-key",
|
||||
"images": {"image0": over_cap},
|
||||
}},
|
||||
local_validate=rules_dc,
|
||||
node_id="FakeDynComboAutogrow_RNP",
|
||||
)
|
||||
assert "model" in out
|
||||
print("ok: DYNAMIC_COMBO unknown branch key is a defensive no-op")
|
||||
|
||||
print("PASS: image_max_batch local_validate smoke test (incl. nested traversal)")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(asyncio.run(main()))
|
||||
@@ -1,316 +0,0 @@
|
||||
"""Smoke test for the client-side MODEL_3D decoder.
|
||||
|
||||
Verifies:
|
||||
|
||||
1. The vendored ``protocol.HEAVY_TYPES`` now recognises ``model_3d`` so
|
||||
``is_envelope`` returns True for an inbound MODEL_3D envelope.
|
||||
2. ``serialization.decode_model3d_envelope`` round-trips an inline
|
||||
``glb_inline`` envelope into a ``comfy_api.latest._util.File3D``
|
||||
whose ``BytesIO`` payload matches the original GLB bytes.
|
||||
3. ``decode_envelope`` dispatches ``type=model_3d`` to the new decoder.
|
||||
4. The proxy_node ``_OUTPUT_CLASSES`` now maps ``"MODEL_3D"`` to
|
||||
``IO.File3DGLB.Output`` so a descriptor with ``output=["MODEL_3D"]``
|
||||
parses without falling through to the opaque IO.Custom branch.
|
||||
5. ``client.CLIENT_CAPABILITIES`` advertises
|
||||
``model_3d:glb_inline`` on every outbound request so descriptors
|
||||
gated on the token negotiate cleanly.
|
||||
6. End-to-end against the server's ``make_model3d_envelope`` helper
|
||||
(imported from the comfy-rnp-server worktree) — the bytes the
|
||||
server packs are exactly the bytes the client decodes.
|
||||
|
||||
Run with the ComfyUI venv (needs torch / comfy_api on PYTHONPATH):
|
||||
|
||||
python notes/run_model3d_decoder.py
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import types
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Locate the worktree under test.
|
||||
# ---------------------------------------------------------------------------
|
||||
_HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
_CLIENT_DIR = os.path.dirname(_HERE)
|
||||
_COMFYUI_DIR = os.path.dirname(os.path.dirname(_CLIENT_DIR))
|
||||
_SERVER_DIR = os.path.normpath(
|
||||
os.path.join(_COMFYUI_DIR, "..", "comfy-rnp-server")
|
||||
)
|
||||
|
||||
|
||||
def _load_module(name: str, path: str) -> types.ModuleType:
|
||||
spec = importlib.util.spec_from_file_location(name, path)
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
sys.modules[name] = mod
|
||||
spec.loader.exec_module(mod)
|
||||
return mod
|
||||
|
||||
|
||||
# Make ComfyUI importable so ``comfy_api.latest._util.File3D`` resolves.
|
||||
if _COMFYUI_DIR not in sys.path:
|
||||
sys.path.insert(0, _COMFYUI_DIR)
|
||||
# And the server worktree so we can import the ENCODER side end-to-end.
|
||||
if _SERVER_DIR not in sys.path:
|
||||
sys.path.insert(0, _SERVER_DIR)
|
||||
|
||||
# ComfyUI's full ``comfy_api`` tree pulls torch / PIL / comfy /
|
||||
# comfy_execution at import time — way too heavy for a decoder
|
||||
# round-trip check. Stub just the leaf module our decoder imports
|
||||
# (``comfy_api.latest._util``) with a faithful pure-Python ``File3D``.
|
||||
class _StubFile3D:
|
||||
def __init__(self, source, file_format: str = "") -> None:
|
||||
self._source = source
|
||||
self._format = file_format
|
||||
|
||||
_comfy_api = types.ModuleType("comfy_api")
|
||||
_comfy_api_latest = types.ModuleType("comfy_api.latest")
|
||||
_comfy_api_latest_util = types.ModuleType("comfy_api.latest._util")
|
||||
_comfy_api_latest_util.File3D = _StubFile3D
|
||||
_comfy_api.latest = _comfy_api_latest
|
||||
_comfy_api_latest._util = _comfy_api_latest_util
|
||||
sys.modules.setdefault("comfy_api", _comfy_api)
|
||||
sys.modules.setdefault("comfy_api.latest", _comfy_api_latest)
|
||||
sys.modules.setdefault("comfy_api.latest._util", _comfy_api_latest_util)
|
||||
|
||||
# Load the client modules as if they were a package — the ``protocol``
|
||||
# module imports from ``.protocol`` relative to its package, so we
|
||||
# emulate the package layout under the name ``comfy_remote_nodes``.
|
||||
pkg = types.ModuleType("comfy_remote_nodes_test")
|
||||
pkg.__path__ = [_CLIENT_DIR]
|
||||
sys.modules["comfy_remote_nodes_test"] = pkg
|
||||
|
||||
protocol = _load_module(
|
||||
"comfy_remote_nodes_test.protocol",
|
||||
os.path.join(_CLIENT_DIR, "protocol.py"),
|
||||
)
|
||||
# serialization imports ``client`` for ``rnp_client`` but we don't need
|
||||
# the HTTP client for the decoder path; provide a tiny stub so the
|
||||
# import succeeds.
|
||||
client_stub = types.ModuleType("comfy_remote_nodes_test.client")
|
||||
sys.modules["comfy_remote_nodes_test.client"] = client_stub
|
||||
serialization = _load_module(
|
||||
"comfy_remote_nodes_test.serialization",
|
||||
os.path.join(_CLIENT_DIR, "serialization.py"),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Build a minimal valid binary glTF 2.0 (GLB) payload for the round-trip.
|
||||
# Format: 12-byte header + N chunks. Header: magic(4) + version(4) + length(4).
|
||||
# Magic must be b"glTF", version=2. We include exactly one JSON chunk that
|
||||
# describes an empty asset (enough to satisfy any GLB-magic check; downstream
|
||||
# nodes do their own validation).
|
||||
# ---------------------------------------------------------------------------
|
||||
def _make_minimal_glb() -> bytes:
|
||||
json_payload = b'{"asset":{"version":"2.0"}}'
|
||||
# Chunks must be 4-byte aligned; pad JSON with spaces (0x20).
|
||||
pad = (4 - (len(json_payload) % 4)) % 4
|
||||
json_payload = json_payload + (b" " * pad)
|
||||
json_chunk = (
|
||||
len(json_payload).to_bytes(4, "little") # chunk length
|
||||
+ b"JSON" # chunk type
|
||||
+ json_payload # chunk data
|
||||
)
|
||||
total = 12 + len(json_chunk)
|
||||
header = (
|
||||
b"glTF"
|
||||
+ (2).to_bytes(4, "little")
|
||||
+ total.to_bytes(4, "little")
|
||||
)
|
||||
return header + json_chunk
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 1. HEAVY_TYPES recognises model_3d.
|
||||
# ---------------------------------------------------------------------------
|
||||
assert "model_3d" in protocol.HEAVY_TYPES, (
|
||||
f"HEAVY_TYPES missing 'model_3d': {sorted(protocol.HEAVY_TYPES)}"
|
||||
)
|
||||
assert protocol.Capability.MODEL_3D_GLB_INLINE == "model_3d:glb_inline", (
|
||||
f"Capability.MODEL_3D_GLB_INLINE wrong value: "
|
||||
f"{protocol.Capability.MODEL_3D_GLB_INLINE!r}"
|
||||
)
|
||||
print("PASS: HEAVY_TYPES includes 'model_3d'; Capability constant intact.")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 2. is_envelope returns True for a MODEL_3D envelope.
|
||||
# ---------------------------------------------------------------------------
|
||||
glb_bytes = _make_minimal_glb()
|
||||
envelope = {
|
||||
"type": "model_3d",
|
||||
"encoding": "glb_inline",
|
||||
"data": base64.b64encode(glb_bytes).decode("ascii"),
|
||||
"format": "glb",
|
||||
"byte_size": len(glb_bytes),
|
||||
}
|
||||
assert protocol.is_envelope(envelope), (
|
||||
"is_envelope returned False for a well-formed MODEL_3D envelope; "
|
||||
"HEAVY_TYPES update did not take effect."
|
||||
)
|
||||
print("PASS: is_envelope recognises model_3d envelopes.")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 3. decode_model3d_envelope round-trips inline bytes.
|
||||
# ---------------------------------------------------------------------------
|
||||
file3d = serialization.decode_model3d_envelope(envelope)
|
||||
from comfy_api.latest._util import File3D as _File3D_for_check # noqa: E402
|
||||
assert isinstance(file3d, _File3D_for_check), (
|
||||
f"Expected File3D, got {type(file3d).__name__}"
|
||||
)
|
||||
# File3D exposes get_data() / get_source(); the public API on this
|
||||
# version returns a BytesIO when constructed with one. Drain bytes:
|
||||
src = file3d._source # internal, but stable across this code path
|
||||
assert hasattr(src, "read"), f"File3D source not BytesIO-like: {type(src)}"
|
||||
src.seek(0)
|
||||
round_trip = src.read()
|
||||
assert round_trip == glb_bytes, (
|
||||
f"GLB byte mismatch after decode: input={len(glb_bytes)} bytes, "
|
||||
f"output={len(round_trip)} bytes"
|
||||
)
|
||||
assert round_trip[:4] == b"glTF", "GLB magic bytes lost in round-trip"
|
||||
print(f"PASS: decode_model3d_envelope round-trips {len(glb_bytes)} GLB bytes.")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 4. decode_envelope dispatches model_3d to the new decoder.
|
||||
# ---------------------------------------------------------------------------
|
||||
file3d2 = serialization.decode_envelope(envelope)
|
||||
assert isinstance(file3d2, _File3D_for_check), (
|
||||
f"decode_envelope dispatch failed for type=model_3d: got "
|
||||
f"{type(file3d2).__name__}"
|
||||
)
|
||||
print("PASS: decode_envelope dispatches type='model_3d' correctly.")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 5. Unsupported encoding raises a hard error (defence-in-depth so a
|
||||
# future encoding never silently downgrades to garbage bytes).
|
||||
# ---------------------------------------------------------------------------
|
||||
bad = dict(envelope, encoding="obj_inline")
|
||||
try:
|
||||
serialization.decode_model3d_envelope(bad)
|
||||
except protocol.RnpProtocolError as e:
|
||||
assert "Unsupported model_3d encoding" in str(e), (
|
||||
f"Wrong error message for unsupported encoding: {e!r}"
|
||||
)
|
||||
assert e.code == protocol.ErrorCode.INTERNAL
|
||||
print("PASS: unsupported model_3d encoding raises RnpProtocolError.")
|
||||
else:
|
||||
raise AssertionError(
|
||||
"decode_model3d_envelope accepted an unsupported encoding"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 6. End-to-end: server's make_model3d_envelope -> client decode.
|
||||
# Skips when the server worktree isn't on disk (e.g. CI running
|
||||
# against just the client repo).
|
||||
# ---------------------------------------------------------------------------
|
||||
server_protocol_path = os.path.join(
|
||||
_SERVER_DIR, "comfy_rnp_protocol", "envelopes.py",
|
||||
)
|
||||
if os.path.exists(server_protocol_path):
|
||||
# Import the server-side envelopes module standalone (it has no
|
||||
# ComfyUI deps and uses its own sibling ``constants`` import).
|
||||
server_pkg = types.ModuleType("comfy_rnp_protocol_e2e")
|
||||
server_pkg.__path__ = [os.path.join(_SERVER_DIR, "comfy_rnp_protocol")]
|
||||
sys.modules["comfy_rnp_protocol_e2e"] = server_pkg
|
||||
server_constants = _load_module(
|
||||
"comfy_rnp_protocol_e2e.constants",
|
||||
os.path.join(_SERVER_DIR, "comfy_rnp_protocol", "constants.py"),
|
||||
)
|
||||
server_envelopes = _load_module(
|
||||
"comfy_rnp_protocol_e2e.envelopes",
|
||||
os.path.join(_SERVER_DIR, "comfy_rnp_protocol", "envelopes.py"),
|
||||
)
|
||||
# Build using the server helper exactly as a provider would.
|
||||
server_env = server_envelopes.make_model3d_envelope(
|
||||
encoding="glb_inline",
|
||||
data=base64.b64encode(glb_bytes).decode("ascii"),
|
||||
format="glb",
|
||||
byte_size=len(glb_bytes),
|
||||
)
|
||||
assert server_env["type"] == "model_3d", server_env
|
||||
assert server_env["encoding"] == "glb_inline", server_env
|
||||
assert "model_3d" in server_envelopes.HEAVY_TYPES, (
|
||||
"Server HEAVY_TYPES missing 'model_3d' — server PR #37 should "
|
||||
"have added it; reinstall comfy_rnp_protocol if this fires."
|
||||
)
|
||||
# Pipe through the client decoder.
|
||||
file3d3 = serialization.decode_envelope(server_env)
|
||||
src = file3d3._source
|
||||
src.seek(0)
|
||||
assert src.read() == glb_bytes, "End-to-end byte mismatch"
|
||||
print(
|
||||
"PASS: end-to-end server make_model3d_envelope -> client "
|
||||
"decode_envelope preserves bytes."
|
||||
)
|
||||
else:
|
||||
print(
|
||||
f"SKIP end-to-end: server worktree not found at {_SERVER_DIR!r}"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 7. proxy_node._OUTPUT_CLASSES maps "MODEL_3D" -> IO.File3DGLB.Output.
|
||||
# proxy_node pulls in the full comfy_api / server / comfy_api_nodes
|
||||
# tree at import time (too heavy to stub here), so we verify the
|
||||
# mapping with a source-level check instead. The pattern is exact:
|
||||
# a quoted ``"MODEL_3D":`` key whose value is the
|
||||
# ``IO.File3DGLB.Output`` class reference inside the
|
||||
# ``_OUTPUT_CLASSES`` dict literal. The same source check would
|
||||
# catch a typo (e.g. ``IO.File3DAny.Output``) that a smoke test
|
||||
# importing the live module would miss too — the dict literal is
|
||||
# the contract.
|
||||
# ---------------------------------------------------------------------------
|
||||
proxy_src = open(
|
||||
os.path.join(_CLIENT_DIR, "proxy_node.py"), "r", encoding="utf-8",
|
||||
).read()
|
||||
assert "_OUTPUT_CLASSES = {" in proxy_src, "proxy_node missing _OUTPUT_CLASSES dict"
|
||||
assert '"MODEL_3D":' in proxy_src and "IO.File3DGLB.Output" in proxy_src, (
|
||||
"proxy_node._OUTPUT_CLASSES does not appear to map "
|
||||
"'MODEL_3D' -> IO.File3DGLB.Output"
|
||||
)
|
||||
# Verify the mapping is in the same dict literal (not somewhere else).
|
||||
oc_start = proxy_src.index("_OUTPUT_CLASSES = {")
|
||||
oc_end = proxy_src.index("}", oc_start)
|
||||
oc_block = proxy_src[oc_start:oc_end]
|
||||
assert '"MODEL_3D":' in oc_block, (
|
||||
"'MODEL_3D' key not inside the _OUTPUT_CLASSES dict literal"
|
||||
)
|
||||
assert "IO.File3DGLB.Output" in oc_block, (
|
||||
"IO.File3DGLB.Output value not inside the _OUTPUT_CLASSES dict literal"
|
||||
)
|
||||
print(
|
||||
"PASS: proxy_node._OUTPUT_CLASSES dict literal contains "
|
||||
"'MODEL_3D' -> IO.File3DGLB.Output."
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 8. CLIENT_CAPABILITIES advertises model_3d:glb_inline.
|
||||
# client.py imports aiohttp / comfy_api_nodes too — same source-level
|
||||
# contract check.
|
||||
# ---------------------------------------------------------------------------
|
||||
client_src = open(
|
||||
os.path.join(_CLIENT_DIR, "client.py"), "r", encoding="utf-8",
|
||||
).read()
|
||||
assert "CLIENT_CAPABILITIES = [" in client_src, (
|
||||
"client.py missing CLIENT_CAPABILITIES list"
|
||||
)
|
||||
cc_start = client_src.index("CLIENT_CAPABILITIES = [")
|
||||
cc_end = client_src.index("]", cc_start)
|
||||
cc_block = client_src[cc_start:cc_end]
|
||||
assert "Capability.MODEL_3D_GLB_INLINE" in cc_block, (
|
||||
"Capability.MODEL_3D_GLB_INLINE not advertised in CLIENT_CAPABILITIES"
|
||||
)
|
||||
print("PASS: CLIENT_CAPABILITIES advertises Capability.MODEL_3D_GLB_INLINE.")
|
||||
|
||||
print("ALL CHECKS PASSED.")
|
||||
+1
-52
@@ -104,69 +104,18 @@ class Capability:
|
||||
IMAGE_PNG_BASE64 = "image:png_base64"
|
||||
IMAGE_JPEG_BASE64 = "image:jpeg_base64"
|
||||
IMAGE_RAW_CHW_UINT8 = "image:raw_b64_chw_uint8"
|
||||
# Multi-frame IMAGE envelope: a single value carries N PNG-base64
|
||||
# frames sharing the same H/W/C, decoded into a [B,H,W,C] tensor on
|
||||
# this client. Mirrors the AUDIO sweep precedent (parallel encodings
|
||||
# rather than extending ``png_base64``) so legacy server emits still
|
||||
# decode correctly when the new token isn't advertised.
|
||||
IMAGE_PNG_BASE64_BATCH = "image:png_base64_batch"
|
||||
# Externalised counterpart of ``image:png_base64_batch``: same
|
||||
# multi-frame IMAGE envelope shape, but every frame's PNG bytes
|
||||
# are uploaded out-of-band via the presigned-PUT path (see
|
||||
# ``serialization.maybe_externalize``) and the envelope carries
|
||||
# a parallel ``uris`` list instead of inline ``frames``. Required
|
||||
# because the original ``maybe_externalize`` only inspected the
|
||||
# singleton ``data`` field, so a batch of N PNGs whose summed
|
||||
# base64 payload crossed the 8 MiB inline cap silently rode the
|
||||
# wire as a 30+ MiB POST body. Per-frame URIs (NOT a tar/zip
|
||||
# blob) keep this composable with the existing per-frame fetch
|
||||
# helpers and let the server fetch in parallel. Decoder dispatch
|
||||
# keys off the ``encoding`` string — ``frames`` vs ``uris`` is a
|
||||
# payload field, not a shape selector.
|
||||
IMAGE_PNG_BASE64_BATCH_URI = "image:png_base64_batch_uri"
|
||||
VIDEO_MP4_INLINE = "video:mp4_inline"
|
||||
VIDEO_MP4_BASE64 = "video:mp4_base64"
|
||||
AUDIO_MP3_INLINE = "audio:mp3_inline"
|
||||
AUDIO_MP3_BASE64 = "audio:mp3_base64"
|
||||
AUDIO_WAV_BASE64 = "audio:wav_base64"
|
||||
AUDIO_WAV_INLINE = "audio:wav_inline"
|
||||
# OGG/Opus container as base64 in ``data``. Mirrors the parallel-
|
||||
# encoding negotiation pattern already used for the AUDIO sweep
|
||||
# (mp3 vs. wav) — servers only emit ``audio:opus_base64`` when
|
||||
# this client advertises the token, otherwise they fall back to
|
||||
# ``mp3_base64``. PyAV/FFmpeg decodes OGG/Opus natively so
|
||||
# ``decode_audio_envelope`` ships the bytes through
|
||||
# ``audio_bytes_to_audio_input`` the same way it ships MP3/WAV.
|
||||
AUDIO_OPUS_BASE64 = "audio:opus_base64"
|
||||
MASK_PNG_BASE64 = "mask:png_base64"
|
||||
SVG_XML_BASE64 = "svg:svg_xml_base64"
|
||||
SCHEMA_V3 = "schema:v3"
|
||||
ASYNC_EXECUTE = "execute:async"
|
||||
# Pass-through custom-IO bucket: when an input/output type string
|
||||
# is not in the standard primitive/media set, the proxy_node parser
|
||||
# falls back to ``IO.Custom(<type>)`` so partner helper-config nodes
|
||||
# (OpenAIInputFiles, OpenAIChatConfig, GeminiInputFiles, Recraft
|
||||
# style/color/controls helpers, future partner helper-config types)
|
||||
# round-trip without per-type encoder plumbing. The descriptor's
|
||||
# original io_type string is preserved on the V3 input so node-to-
|
||||
# node connections only chain when the strings match. Encoder
|
||||
# dispatch passes opaque dict/list/string values through
|
||||
# ``_encode_one`` unchanged — they're already JSON-serialisable.
|
||||
IO_OPAQUE = "io:opaque"
|
||||
# 3D MODEL envelope: a single-mesh GLB (binary glTF 2.0, magic
|
||||
# b"glTF") wrapped opaquely in ``data`` (base64). Mirrors the
|
||||
# parallel-encoding pattern already used for VIDEO_MP4_INLINE — the
|
||||
# server emits this token only when the inbound request advertised
|
||||
# it; legacy clients without the token will surface
|
||||
# NEGOTIATION_FAILED at descriptor-load time. The decoder on the
|
||||
# client side hands raw GLB bytes straight to ComfyUI's
|
||||
# ``Types.File3D`` / ``IO.File3DGLB`` machinery (no in-server GLB
|
||||
# parse — the magic-byte check at offset 0 is the only validity
|
||||
# guard the server applies).
|
||||
MODEL_3D_GLB_INLINE = "model_3d:glb_inline"
|
||||
|
||||
|
||||
HEAVY_TYPES = frozenset({"image", "video", "audio", "mask", "model_3d"})
|
||||
HEAVY_TYPES = frozenset({"image", "video", "audio", "mask", "svg"})
|
||||
|
||||
|
||||
def is_envelope(value: Any) -> bool:
|
||||
|
||||
+213
-383
@@ -3,13 +3,11 @@
|
||||
The descriptor wire format is V3 ``Schema.get_v1_info()`` verbatim, so
|
||||
per-input dicts are handed straight to V3's IO classes without
|
||||
reinterpretation. ``_parse_input_spec`` knows STRING / INT / FLOAT /
|
||||
BOOLEAN / COMBO / IMAGE / VIDEO / AUDIO / MASK; any other io_type
|
||||
string falls through to ``IO.Custom(io_type)`` (the §B opaque
|
||||
pass-through bucket — partner helper-config types like RECRAFT_*,
|
||||
OPENAI_INPUT_FILES, OPENAI_CHAT_CONFIG, GEMINI_INPUT_FILES round-trip
|
||||
as raw JSON blobs with the original io_type preserved on the V3
|
||||
socket, so node-to-node connections only chain when the strings
|
||||
match). Only a non-string / empty io_type causes a skip.
|
||||
BOOLEAN / COMBO / IMAGE / VIDEO / AUDIO / MASK natively; any other
|
||||
``io_type`` string is treated as an opaque pass-through socket (see
|
||||
``IO.Custom``) so partner-defined types like ``RECRAFT_STYLE_V3`` /
|
||||
``RECRAFT_COLOR`` round-trip between RNP nodes without the descriptor
|
||||
being rejected.
|
||||
|
||||
Hidden inputs (``auth_token_comfy_org`` / ``api_key_comfy_org`` /
|
||||
``unique_id`` / ``prompt`` / ``extra_pnginfo`` / ``dynprompt``) are
|
||||
@@ -47,6 +45,7 @@ from .protocol import (
|
||||
DEFAULT_POLL_INTERVAL_S,
|
||||
DEFAULT_RETRY_BACKOFF_PER_POLL,
|
||||
DEFAULT_RETRY_DELAY_PER_POLL_S,
|
||||
DEFAULT_SOFT_TIMEOUT_S,
|
||||
DEFAULT_TIMEOUT_PER_POLL_S,
|
||||
ErrorCode,
|
||||
ExecutionMode,
|
||||
@@ -309,32 +308,6 @@ def build_node_class(
|
||||
schema_hash = remote.get("schema_hash")
|
||||
execute_path = execute_endpoint["path"]
|
||||
execution = remote.get("execution") or {}
|
||||
# ``descriptor.remote.input_serialization`` advertises the wire
|
||||
# encodings each input accepts. The wire shape is
|
||||
# ``{input_name: <str | list[str]>}`` — single string for a
|
||||
# single-encoding input (e.g. ``"png_base64"``), list when the
|
||||
# input accepts multiple encodings (e.g.
|
||||
# ``["png_base64", "png_base64_batch"]`` for a multi-frame IMAGE).
|
||||
# Normalize to ``dict[str, list[str]]`` so the encoder can do a
|
||||
# single ``in`` check per input without re-parsing per call.
|
||||
raw_input_serialization = (
|
||||
remote.get("input_serialization") if isinstance(remote.get("input_serialization"), dict) else {}
|
||||
)
|
||||
input_serialization: dict[str, list[str]] = {}
|
||||
for name, val in raw_input_serialization.items():
|
||||
if isinstance(val, str):
|
||||
input_serialization[name] = [val]
|
||||
elif isinstance(val, list):
|
||||
input_serialization[name] = [v for v in val if isinstance(v, str)]
|
||||
|
||||
# ``descriptor.input.<section>.<name>[1].local_validate`` carries
|
||||
# declarative client-side validation rules (e.g.
|
||||
# ``{"image_max_batch": 9}`` for a Flux2 reference-image input).
|
||||
# The encoder consults this map and raises ``INPUT_INVALID``
|
||||
# locally before any HTTP round trip when an over-cap batch is
|
||||
# wired in. Lookup is defensive: missing ``local_validate`` →
|
||||
# no validation; missing per-rule key → no validation.
|
||||
local_validate = _collect_local_validate(descriptor)
|
||||
timeout_s = float(
|
||||
execution.get("hard_timeout_s")
|
||||
or execution.get("soft_timeout_s")
|
||||
@@ -351,6 +324,16 @@ def build_node_class(
|
||||
url_fetch_policy = (
|
||||
remote.get("url_fetch") if isinstance(remote.get("url_fetch"), dict) else {}
|
||||
)
|
||||
# ``descriptor.remote.hidden_forwarded`` opt-in for non-auth hidden
|
||||
# markers (``prompt`` / ``extra_pnginfo`` / ``dynprompt``). Auth
|
||||
# tokens and ``unique_id`` always forward when declared. Anything
|
||||
# else is declared on the schema (so the executor populates
|
||||
# ``cls.hidden``) but dropped before building the request body so
|
||||
# large EXTRA_PNGINFO blobs don't ship by default.
|
||||
hidden_forwarded_raw = remote.get("hidden_forwarded") or []
|
||||
hidden_forwarded: list[str] = [
|
||||
h for h in hidden_forwarded_raw if isinstance(h, str)
|
||||
] if isinstance(hidden_forwarded_raw, (list, tuple)) else []
|
||||
|
||||
display_name = descriptor.get("display_name") or node_id
|
||||
category = descriptor.get("category") or "remote"
|
||||
@@ -409,14 +392,13 @@ def build_node_class(
|
||||
timeout_s=timeout_s,
|
||||
outputs_meta=schema_outputs,
|
||||
hidden_names=hidden_names,
|
||||
hidden_forwarded=hidden_forwarded,
|
||||
inputs=kwargs,
|
||||
max_inline_bytes=max_inline_bytes,
|
||||
estimated_duration_s=estimated_duration_s,
|
||||
execution_mode=execution_mode,
|
||||
execution_policy=execution_policy,
|
||||
url_fetch_policy=url_fetch_policy,
|
||||
input_serialization=input_serialization,
|
||||
local_validate=local_validate,
|
||||
)
|
||||
|
||||
RemoteProxyNode.__name__ = node_id
|
||||
@@ -442,111 +424,6 @@ _HIDDEN_MARKERS = {
|
||||
}
|
||||
|
||||
|
||||
def _collect_local_validate(descriptor: dict[str, Any]) -> dict[str, dict[str, Any]]:
|
||||
"""Pull ``local_validate`` rule dicts off every required/optional input
|
||||
spec in the descriptor, recursing into AUTOGROW templates and
|
||||
DYNAMIC_COMBO branches.
|
||||
|
||||
Returns ``{input_name: rules_dict}`` for every input that carries a
|
||||
non-empty ``local_validate`` block (or whose nested template /
|
||||
branches do). The returned ``rules_dict`` may carry the literal
|
||||
``local_validate`` keys (e.g. ``{"image_max_batch": 9}``) plus two
|
||||
private composite keys consumed by ``_enforce_local_validate``:
|
||||
|
||||
* ``__template__`` — for AUTOGROW inputs, the rules harvested from
|
||||
the template's options dict. The enforcer applies these to every
|
||||
slot value in the runtime Autogrow dict
|
||||
(``{"image0": <env>, "image1": <env>, ...}``).
|
||||
* ``__branches__`` — for DYNAMIC_COMBO inputs, a
|
||||
``{branch_key: {sub_name: rules}}`` map. The enforcer reads the
|
||||
selected branch from ``value[<name>]`` and walks the matching
|
||||
branch's sub-inputs (used by GPT Image V2 / Grok ImageEditV2 /
|
||||
Grok VideoReference, where the cap lives on a nested Autogrow
|
||||
template inside a DynamicCombo branch).
|
||||
|
||||
Inputs whose entire subtree is empty are omitted so the encoder can
|
||||
use a single ``in`` check per call.
|
||||
"""
|
||||
out: dict[str, dict[str, Any]] = {}
|
||||
input_def = descriptor.get("input") or {}
|
||||
for kind in ("required", "optional"):
|
||||
defs = input_def.get(kind) or {}
|
||||
for name, spec in defs.items():
|
||||
rules = _rules_from_input_spec(spec)
|
||||
if rules:
|
||||
out[name] = rules
|
||||
return out
|
||||
|
||||
|
||||
def _rules_from_input_spec(spec: Any) -> dict[str, Any]:
|
||||
"""Extract ``local_validate`` rules from one V3 input spec.
|
||||
|
||||
Accepts both wire shapes:
|
||||
|
||||
* ``[io_type, options_dict]`` — top-level inputs (the descriptor's
|
||||
``input.required[name]`` value, or an AUTOGROW template).
|
||||
* ``[name, io_type, options_dict]`` — DynamicCombo branch sub-inputs
|
||||
(each entry of ``options.options[i].inputs``).
|
||||
|
||||
For ``AUTOGROW`` specs, recurses into ``options["template"]`` and
|
||||
stashes the result under ``__template__``. For ``DYNAMIC_COMBO``,
|
||||
walks each branch's sub-inputs and stashes the resulting
|
||||
``{branch_key: {sub_name: sub_rules}}`` map under ``__branches__``.
|
||||
"""
|
||||
if not isinstance(spec, (list, tuple)) or len(spec) < 2:
|
||||
return {}
|
||||
if (
|
||||
len(spec) >= 3
|
||||
and isinstance(spec[1], str)
|
||||
and isinstance(spec[2], dict)
|
||||
):
|
||||
# [name, io_type, options]
|
||||
io_type, options = spec[1], spec[2]
|
||||
elif isinstance(spec[1], dict):
|
||||
# [io_type, options]
|
||||
io_type, options = spec[0], spec[1]
|
||||
else:
|
||||
return {}
|
||||
rules: dict[str, Any] = {}
|
||||
direct = options.get("local_validate")
|
||||
if isinstance(direct, dict) and direct:
|
||||
rules.update(direct)
|
||||
if io_type == "AUTOGROW":
|
||||
tmpl_spec = options.get("template")
|
||||
tmpl_rules = (
|
||||
_rules_from_input_spec(tmpl_spec)
|
||||
if isinstance(tmpl_spec, (list, tuple))
|
||||
else {}
|
||||
)
|
||||
if tmpl_rules:
|
||||
rules["__template__"] = tmpl_rules
|
||||
elif io_type == "DYNAMIC_COMBO":
|
||||
branches: dict[str, dict[str, Any]] = {}
|
||||
for opt in (options.get("options") or []):
|
||||
if not isinstance(opt, dict):
|
||||
continue
|
||||
key = opt.get("key")
|
||||
if not isinstance(key, str):
|
||||
continue
|
||||
sub_rules: dict[str, dict[str, Any]] = {}
|
||||
for sub_spec in (opt.get("inputs") or []):
|
||||
if (
|
||||
not isinstance(sub_spec, (list, tuple))
|
||||
or len(sub_spec) < 1
|
||||
or not isinstance(sub_spec[0], str)
|
||||
):
|
||||
continue
|
||||
sub_name = sub_spec[0]
|
||||
r = _rules_from_input_spec(sub_spec)
|
||||
if r:
|
||||
sub_rules[sub_name] = r
|
||||
if sub_rules:
|
||||
branches[key] = sub_rules
|
||||
if branches:
|
||||
rules["__branches__"] = branches
|
||||
return rules
|
||||
|
||||
|
||||
def _parse_inputs(descriptor: dict[str, Any]) -> tuple[list[Any] | None, list[str]]:
|
||||
"""Return ``(input_objs, hidden_input_names)``.
|
||||
|
||||
@@ -591,13 +468,17 @@ def _parse_input_spec(name: str, spec: list[Any], optional: bool) -> Any | None:
|
||||
if isinstance(io_type, list):
|
||||
# Legacy V1-style combo: io_type *is* the options list.
|
||||
return IO.Combo.Input(name, options=list(io_type),
|
||||
default=options.get("default"), **common)
|
||||
default=options.get("default"),
|
||||
control_after_generate=options.get("control_after_generate"),
|
||||
**common)
|
||||
if io_type == "COMBO":
|
||||
opts = options.get("options")
|
||||
if not isinstance(opts, (list, tuple)):
|
||||
return None
|
||||
return IO.Combo.Input(name, options=list(opts),
|
||||
default=options.get("default"), **common)
|
||||
default=options.get("default"),
|
||||
control_after_generate=options.get("control_after_generate"),
|
||||
**common)
|
||||
if io_type == "STRING":
|
||||
return IO.String.Input(
|
||||
name,
|
||||
@@ -612,6 +493,7 @@ def _parse_input_spec(name: str, spec: list[Any], optional: bool) -> Any | None:
|
||||
min=options.get("min", 0),
|
||||
max=options.get("max", 2147483647),
|
||||
step=options.get("step", 1),
|
||||
control_after_generate=options.get("control_after_generate"),
|
||||
**common,
|
||||
)
|
||||
if io_type == "FLOAT":
|
||||
@@ -635,20 +517,130 @@ def _parse_input_spec(name: str, spec: list[Any], optional: bool) -> Any | None:
|
||||
return IO.Audio.Input(name, **common)
|
||||
if io_type == "MASK":
|
||||
return IO.Mask.Input(name, **common)
|
||||
# Opaque custom-IO fallthrough — any unknown io_type string becomes
|
||||
# an ``IO.Custom(io_type)`` socket so partner helper-config nodes
|
||||
# (OpenAIInputFiles → OPENAI_INPUT_FILES, OpenAIChatConfig →
|
||||
# OPENAI_CHAT_CONFIG, GeminiInputFiles → GEMINI_INPUT_FILES, the
|
||||
# already-shipped Recraft RECRAFT_* helpers, future partner types)
|
||||
# round-trip without per-type encoder plumbing. The descriptor's
|
||||
# original ``io_type`` string is preserved on ``IO.Custom``'s
|
||||
# ComfyType.io_type via the V3 ``@comfytype`` decorator, so the
|
||||
# frontend's connection-validity check only chains sockets when the
|
||||
# strings match — an OPENAI_INPUT_FILES output cannot connect to a
|
||||
# GEMINI_INPUT_FILES input. Encoder dispatch in
|
||||
# ``serialization`` already passes non-tensor / non-audio values
|
||||
# through ``_encode_one`` unchanged, and the deserializer treats
|
||||
# non-envelope values as already-native Python objects.
|
||||
if io_type == "SVG":
|
||||
return IO.SVG.Input(name, **common)
|
||||
if io_type == "AUTOGROW":
|
||||
# Wire shape:
|
||||
# ["AUTOGROW", {
|
||||
# "template": ["IMAGE", {...}], # nested input spec
|
||||
# "prefix": "image",
|
||||
# "min": 1,
|
||||
# "max": 5,
|
||||
# }]
|
||||
# The descriptor declares N "<prefix>0".."<prefix>{max-1}"
|
||||
# virtual inputs that the V3 frontend renders as a single
|
||||
# auto-growing block — see ``IO.Autogrow.TemplatePrefix``.
|
||||
# The first ``min`` slots are required, the rest optional;
|
||||
# the executor packs the connected ones into a dict keyed by
|
||||
# virtual id which the server provider iterates.
|
||||
template_spec = options.get("template")
|
||||
if (
|
||||
not isinstance(template_spec, (list, tuple))
|
||||
or len(template_spec) < 1
|
||||
or not isinstance(template_spec[0], str)
|
||||
):
|
||||
log.warning(
|
||||
"AUTOGROW input %s: malformed template spec %r",
|
||||
name, template_spec,
|
||||
)
|
||||
return None
|
||||
# The template is itself an ``[io_type, {...}]`` pair — reuse
|
||||
# the parser recursively so AUTOGROW inherits every IO type the
|
||||
# parser knows about (IMAGE, MASK, STRING, custom IO …).
|
||||
# The framework forbids ``DynamicInput`` as an Autogrow template
|
||||
# (``assert(not isinstance(input, DynamicInput))`` in
|
||||
# ``_AutogrowTemplate.__init__``), so passing AUTOGROW or
|
||||
# DYNAMIC_COMBO here will raise inside ``TemplatePrefix`` —
|
||||
# caught below.
|
||||
template_input = _parse_input_spec(
|
||||
"_autogrow_template", list(template_spec), optional=False,
|
||||
)
|
||||
if template_input is None:
|
||||
log.warning(
|
||||
"AUTOGROW input %s: failed to parse template %r",
|
||||
name, template_spec,
|
||||
)
|
||||
return None
|
||||
prefix = options.get("prefix")
|
||||
if not isinstance(prefix, str) or not prefix:
|
||||
return None
|
||||
min_slots = int(options.get("min", 1))
|
||||
max_slots = int(options.get("max", 1))
|
||||
if max_slots < 1 or min_slots < 0 or min_slots > max_slots:
|
||||
return None
|
||||
try:
|
||||
template = IO.Autogrow.TemplatePrefix(
|
||||
template_input,
|
||||
prefix=prefix,
|
||||
min=min_slots,
|
||||
max=max_slots,
|
||||
)
|
||||
except Exception as e: # noqa: BLE001
|
||||
log.warning(
|
||||
"AUTOGROW input %s: failed to build TemplatePrefix: %s",
|
||||
name, e,
|
||||
)
|
||||
return None
|
||||
return IO.Autogrow.Input(
|
||||
name,
|
||||
template=template,
|
||||
display_name=options.get("display_name"),
|
||||
optional=optional,
|
||||
tooltip=options.get("tooltip"),
|
||||
)
|
||||
if io_type == "DYNAMIC_COMBO":
|
||||
# Wire shape:
|
||||
# ["DYNAMIC_COMBO", {
|
||||
# "options": [
|
||||
# {"key": "recraftv4", "inputs": [
|
||||
# ["size", "COMBO", {"options": [...]}],
|
||||
# ...
|
||||
# ]},
|
||||
# {"key": "recraftv4_pro", "inputs": [...]},
|
||||
# ],
|
||||
# "default": "recraftv4",
|
||||
# "tooltip": "...",
|
||||
# }]
|
||||
# The selected key drives which sub-widget block the frontend
|
||||
# renders; the executor packs ``{"model": <key>, ...sub fields}``
|
||||
# into a dict the server consumes via
|
||||
# ``inputs["model"]["model"]`` / ``inputs["model"]["size"]``
|
||||
# idiom — same shape ``IO.DynamicCombo`` produces locally.
|
||||
opts = options.get("options")
|
||||
if not isinstance(opts, list) or not opts:
|
||||
return None
|
||||
dc_options: list[Any] = []
|
||||
for entry in opts:
|
||||
if not isinstance(entry, dict):
|
||||
return None
|
||||
key = entry.get("key")
|
||||
sub_inputs_spec = entry.get("inputs")
|
||||
if not isinstance(key, str) or not isinstance(sub_inputs_spec, list):
|
||||
return None
|
||||
sub_inputs: list[Any] = []
|
||||
for sub in sub_inputs_spec:
|
||||
if not (isinstance(sub, (list, tuple)) and len(sub) >= 2 and isinstance(sub[0], str)):
|
||||
return None
|
||||
sub_name = sub[0]
|
||||
sub_spec = list(sub[1:])
|
||||
# ``[name, io_type, options]`` flattens to the parser's
|
||||
# ``[io_type, options]`` shape with sub_name carried separately.
|
||||
sub_input = _parse_input_spec(sub_name, sub_spec, optional=False)
|
||||
if sub_input is None:
|
||||
return None
|
||||
sub_inputs.append(sub_input)
|
||||
dc_options.append(IO.DynamicCombo.Option(key, sub_inputs))
|
||||
return IO.DynamicCombo.Input(
|
||||
name,
|
||||
options=dc_options,
|
||||
display_name=options.get("display_name"),
|
||||
optional=optional,
|
||||
tooltip=options.get("tooltip"),
|
||||
)
|
||||
# Any other string io_type — partner-defined custom IO (RecraftStyle,
|
||||
# RecraftColor, Tripo3DModel, …). Built as an opaque pass-through
|
||||
# socket so the descriptor is accepted; the value flows between RNP
|
||||
# nodes verbatim and the proxy_node never inspects or serialises it.
|
||||
if isinstance(io_type, str) and io_type:
|
||||
return IO.Custom(io_type).Input(name, **common)
|
||||
return None
|
||||
@@ -735,47 +727,32 @@ def _parse_outputs(descriptor: dict[str, Any]) -> list[Any] | None:
|
||||
tip = output_tooltips[i] if i < len(output_tooltips) else None
|
||||
cls = _OUTPUT_CLASSES.get(io_type)
|
||||
if cls is None:
|
||||
# Opaque custom-IO fallthrough (mirror of
|
||||
# ``_parse_input_spec``) — any unknown output io_type
|
||||
# string becomes ``IO.Custom(io_type).Output`` so partner
|
||||
# helper-config nodes can publish their own opaque
|
||||
# outputs (RECRAFT_*, OPENAI_INPUT_FILES, GEMINI_INPUT_FILES,
|
||||
# OPENAI_CHAT_CONFIG, …). The original io_type string
|
||||
# rides on the resulting Output via V3's ``@comfytype``
|
||||
# decorator, so the frontend only chains output→input
|
||||
# sockets when the strings match.
|
||||
if isinstance(io_type, str) and io_type:
|
||||
cls = IO.Custom(io_type).Output
|
||||
else:
|
||||
# Partner-defined custom IO output (RecraftStyle, Tripo3DModel,
|
||||
# Hunyuan3DMesh, …). Built as an opaque pass-through socket so
|
||||
# the value flows downstream verbatim — no envelope decoding.
|
||||
if not isinstance(io_type, str) or not io_type:
|
||||
log.warning(
|
||||
"Skipping descriptor: unsupported output type %r",
|
||||
io_type,
|
||||
"Skipping descriptor: unsupported output type %r", io_type,
|
||||
)
|
||||
return None
|
||||
out.append(IO.Custom(io_type).Output(
|
||||
display_name=name, is_output_list=is_list, tooltip=tip,
|
||||
))
|
||||
continue
|
||||
out.append(cls(display_name=name, is_output_list=is_list, tooltip=tip))
|
||||
return out
|
||||
|
||||
|
||||
_OUTPUT_CLASSES = {
|
||||
"VIDEO": IO.Video.Output,
|
||||
"IMAGE": IO.Image.Output,
|
||||
"AUDIO": IO.Audio.Output,
|
||||
"MASK": IO.Mask.Output,
|
||||
"STRING": IO.String.Output,
|
||||
"INT": IO.Int.Output,
|
||||
"FLOAT": IO.Float.Output,
|
||||
"BOOLEAN": IO.Boolean.Output,
|
||||
# 3D MODEL output: the Tripo / Rodin / Meshy partner endpoints all
|
||||
# return a single GLB file; the server emits a ``model_3d:glb_inline``
|
||||
# envelope which ``serialization.decode_model3d_envelope`` turns
|
||||
# back into a ``File3D`` so downstream nodes (preview3d / SaveGLB /
|
||||
# Load3D) chain off the output socket unchanged. The descriptor
|
||||
# io_type string is the logical name ``"MODEL_3D"`` (consistent
|
||||
# across vendors) — the resulting V3 socket carries the upstream
|
||||
# ``"FILE_3D_GLB"`` comfytype tag from ``IO.File3DGLB`` so
|
||||
# frontend-side input/output matching works against any GLB-aware
|
||||
# node.
|
||||
"MODEL_3D": IO.File3DGLB.Output,
|
||||
"VIDEO": IO.Video.Output,
|
||||
"IMAGE": IO.Image.Output,
|
||||
"AUDIO": IO.Audio.Output,
|
||||
"MASK": IO.Mask.Output,
|
||||
"SVG": IO.SVG.Output,
|
||||
"STRING": IO.String.Output,
|
||||
"INT": IO.Int.Output,
|
||||
"FLOAT": IO.Float.Output,
|
||||
"BOOLEAN": IO.Boolean.Output,
|
||||
}
|
||||
|
||||
|
||||
@@ -1017,21 +994,28 @@ async def _execute_remote(
|
||||
timeout_s: float,
|
||||
outputs_meta: list[Any],
|
||||
hidden_names: list[str],
|
||||
hidden_forwarded: list[str],
|
||||
inputs: dict[str, Any],
|
||||
max_inline_bytes: int | None = None,
|
||||
estimated_duration_s: int | float | None = None,
|
||||
execution_mode: str = ExecutionMode.REQUEST_RESPONSE,
|
||||
execution_policy: dict[str, Any] | None = None,
|
||||
url_fetch_policy: dict[str, Any] | None = None,
|
||||
input_serialization: dict[str, list[str]] | None = None,
|
||||
local_validate: dict[str, dict[str, Any]] | None = None,
|
||||
) -> IO.NodeOutput:
|
||||
policy = execution_policy or _extract_execution_policy({})
|
||||
url_fetch_policy = url_fetch_policy or {}
|
||||
# Strip hidden inputs from the user-supplied kwargs. Auth credentials
|
||||
# (``auth_token_comfy_org`` / ``api_key_comfy_org``) travel as standard
|
||||
# HTTP headers so the server can forward them to api.comfy.org without
|
||||
# parsing the body; everything else rides in the request ``context``.
|
||||
# parsing the body; ``unique_id`` always rides in ``context`` for
|
||||
# correlation. Other markers (``prompt`` / ``extra_pnginfo`` /
|
||||
# ``dynprompt``) only forward when the descriptor explicitly opted
|
||||
# in via ``remote.hidden_forwarded`` — otherwise they're declared on
|
||||
# the schema (so the executor populates ``cls.hidden``) but stripped
|
||||
# before we build the request body, so EXTRA_PNGINFO blobs don't
|
||||
# leak by default.
|
||||
_ALWAYS_FORWARD = {"auth_token_comfy_org", "api_key_comfy_org", "unique_id"}
|
||||
forward_set = _ALWAYS_FORWARD | set(hidden_forwarded)
|
||||
context: dict[str, Any] = {}
|
||||
auth_headers: dict[str, str] = {}
|
||||
hidden_holder = getattr(cls, "hidden", None)
|
||||
@@ -1052,24 +1036,17 @@ async def _execute_remote(
|
||||
auth_headers["Authorization"] = f"Bearer {value}"
|
||||
elif hname == "api_key_comfy_org":
|
||||
auth_headers["X-API-KEY"] = str(value)
|
||||
else:
|
||||
elif hname in forward_set:
|
||||
context[hname] = value
|
||||
|
||||
# Encode heavy-typed inputs (IMAGE/MASK/AUDIO) as RNP value
|
||||
# envelopes; scalars pass through. Oversized envelopes are
|
||||
# uploaded out-of-band and rewritten to ``{..., uri: <url>}``.
|
||||
# ``input_serialization`` carries the per-input encoding allow-list
|
||||
# from the descriptor — the encoder uses it to decide whether to
|
||||
# ship a multi-frame ``png_base64_batch`` envelope or fall back
|
||||
# to single-frame ``png_base64``.
|
||||
payload_inputs = await _encode_inputs(
|
||||
inputs,
|
||||
server_url=server_url,
|
||||
max_inline_bytes=max_inline_bytes,
|
||||
auth_headers=auth_headers,
|
||||
input_serialization=input_serialization or {},
|
||||
local_validate=local_validate or {},
|
||||
node_id=node_id,
|
||||
)
|
||||
|
||||
ticker_node_id = _node_id_for_ticker(cls)
|
||||
@@ -1244,54 +1221,8 @@ async def _deserialize_output(
|
||||
# advisory here, the value already matches the native ComfyUI
|
||||
# type for STRING / INT / FLOAT / BOOLEAN.
|
||||
return value
|
||||
encoding = value.get("encoding")
|
||||
env_type = value.get("type", "")
|
||||
if encoding == "png_base64_batch_uri":
|
||||
# Externalised IMAGE batch on the response path (mirror of the
|
||||
# request-side ``serialization.maybe_externalize`` behaviour).
|
||||
# No server emits this shape today, but the decoder owns the
|
||||
# symmetry: any envelope shape the client may externalise on
|
||||
# the way out, the client must also accept on the way in.
|
||||
# Per-frame URIs fetched in parallel via ``asyncio.gather`` so
|
||||
# a 9-frame batch import is one round-trip rather than N — the
|
||||
# GET fan-out is the dominant latency for an out-of-band batch.
|
||||
# Frame ordering is preserved (gather returns positional
|
||||
# results) — critical because the IMAGE tensor's batch axis
|
||||
# carries semantic order.
|
||||
policy = url_fetch_policy.get(env_type)
|
||||
if not isinstance(policy, dict):
|
||||
raise RnpProtocolError(
|
||||
f"Server returned a URL batch envelope for output type "
|
||||
f"{env_type!r} but the descriptor declared no "
|
||||
f"remote.url_fetch policy for that type",
|
||||
code=ErrorCode.INTERNAL,
|
||||
)
|
||||
uris = value.get("uris")
|
||||
if not isinstance(uris, list) or not uris:
|
||||
raise RnpProtocolError(
|
||||
"png_base64_batch_uri envelope requires a non-empty 'uris' list",
|
||||
code=ErrorCode.INTERNAL,
|
||||
)
|
||||
import asyncio as _asyncio
|
||||
import base64 as _b64
|
||||
per_frame_envs = [
|
||||
{"type": env_type, "encoding": "png_base64", "uri": u}
|
||||
for u in uris
|
||||
]
|
||||
raws = await _asyncio.gather(*[
|
||||
_fetch_url_envelope(
|
||||
env, policy,
|
||||
auth_headers=auth_headers,
|
||||
node_cls=node_cls,
|
||||
)
|
||||
for env in per_frame_envs
|
||||
])
|
||||
# Re-stamp as the inline batch shape so the existing
|
||||
# ``_decode_image_batch_envelope`` path handles it uniformly.
|
||||
value = {k: v for k, v in value.items() if k != "uris"}
|
||||
value["encoding"] = "png_base64_batch"
|
||||
value["frames"] = [_b64.b64encode(r).decode("ascii") for r in raws]
|
||||
elif "data" not in value and "uri" in value:
|
||||
if "data" not in value and "uri" in value:
|
||||
env_type = value.get("type", "")
|
||||
policy = url_fetch_policy.get(env_type)
|
||||
if not isinstance(policy, dict):
|
||||
raise RnpProtocolError(
|
||||
@@ -1373,9 +1304,6 @@ async def _encode_inputs(
|
||||
server_url: str | None = None,
|
||||
max_inline_bytes: int | None = None,
|
||||
auth_headers: dict[str, str] | None = None,
|
||||
input_serialization: dict[str, list[str]] | None = None,
|
||||
local_validate: dict[str, dict[str, Any]] | None = None,
|
||||
node_id: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Convert in-place: replace heavy-typed input values with the
|
||||
matching RNP value envelope. Scalars pass through unchanged.
|
||||
@@ -1391,29 +1319,10 @@ async def _encode_inputs(
|
||||
When ``max_inline_bytes`` is set, any envelope whose inline payload
|
||||
exceeds the threshold is externalized via ``POST /rnp/v1/uploads``
|
||||
(see ``serialization.maybe_externalize``).
|
||||
|
||||
``input_serialization`` is the descriptor's per-input encoding
|
||||
allow-list (e.g. ``{"images": ["png_base64", "png_base64_batch"]}``).
|
||||
The encoder consults it to decide whether a multi-frame IMAGE input
|
||||
can ride a single batch envelope vs. falling back to single-frame.
|
||||
|
||||
``local_validate`` is the per-input declarative-rule map
|
||||
(e.g. ``{"images": {"image_max_batch": 9}}``). For each rule the
|
||||
encoder enforces the constraint locally and raises
|
||||
``RnpProtocolError(code=INPUT_INVALID, user_facing=True)`` so the
|
||||
proxy_node's friendly-message dispatch surfaces a per-provider
|
||||
error before any paid HTTP round trip. Lookup is defensive: a
|
||||
missing ``local_validate`` entry or missing per-rule key is a
|
||||
no-op.
|
||||
"""
|
||||
serialization_map = input_serialization or {}
|
||||
validate_map = local_validate or {}
|
||||
out: dict[str, Any] = {}
|
||||
for name, value in inputs.items():
|
||||
_enforce_local_validate(
|
||||
name, value, validate_map.get(name) or {}, node_id=node_id,
|
||||
)
|
||||
encoded = _encode_one(name, value, serialization_map.get(name))
|
||||
encoded = _encode_one(name, value)
|
||||
# Only envelope-shaped values can be externalized; scalars pass
|
||||
# straight through ``maybe_externalize`` unchanged.
|
||||
if serialization.is_envelope(encoded):
|
||||
@@ -1427,129 +1336,50 @@ async def _encode_inputs(
|
||||
return out
|
||||
|
||||
|
||||
def _enforce_local_validate(
|
||||
name: str,
|
||||
value: Any,
|
||||
rules: dict[str, Any],
|
||||
*,
|
||||
node_id: str | None,
|
||||
) -> None:
|
||||
"""Apply each ``local_validate.*`` rule to ``value`` and raise
|
||||
``RnpProtocolError(INPUT_INVALID)`` on a violation.
|
||||
def _dict_contains_tensor(value: Any) -> bool:
|
||||
"""True iff a dict (or nested dict) holds at least one torch tensor.
|
||||
|
||||
Rules currently supported:
|
||||
|
||||
* ``image_max_batch`` — for IMAGE inputs (rank-4 tensors), assert
|
||||
``tensor.shape[0] <= cap``. Mirrors the per-call cap raised on
|
||||
the server (e.g. Flux2 raises "The current maximum number of
|
||||
supported images is 9.") so a workflow with N>cap reference
|
||||
frames fails fast on the client before any paid HTTP round trip.
|
||||
|
||||
Two private composite keys are walked transparently:
|
||||
|
||||
* ``__template__`` (AUTOGROW) — ``value`` is expected to be a
|
||||
runtime Autogrow dict (``{"image0": <env>, "image1": <env>, ...}``).
|
||||
The template's rules are applied to each non-``None`` slot value.
|
||||
* ``__branches__`` (DYNAMIC_COMBO) — ``value`` is expected to be the
|
||||
DynamicCombo runtime dict whose ``value[<name>]`` carries the
|
||||
selected branch key. The matching branch's per-sub-input rules are
|
||||
applied recursively. This is what makes GPT Image V2 / Grok
|
||||
ImageEditV2 / Grok VideoReference's nested per-slot caps fire
|
||||
client-side before any HTTP round trip.
|
||||
Used to gate the recursive encode branch so we don't pointlessly
|
||||
rebuild scalar custom-IO chain dicts (RECRAFT_V3_STYLE etc.) on
|
||||
every encode pass, and so already-encoded envelope dicts pass
|
||||
straight through. Nested-dict support keeps DynamicCombo→Autogrow
|
||||
composition (and any future layered dynamic IO) round-trippable.
|
||||
"""
|
||||
if not rules:
|
||||
return
|
||||
_check_image_max_batch(name, value, rules.get("image_max_batch"), node_id=node_id)
|
||||
tmpl = rules.get("__template__")
|
||||
if isinstance(tmpl, dict) and isinstance(value, dict):
|
||||
for slot_key, slot_value in value.items():
|
||||
if slot_value is None:
|
||||
continue
|
||||
_enforce_local_validate(
|
||||
f"{name}.{slot_key}", slot_value, tmpl, node_id=node_id,
|
||||
)
|
||||
branches = rules.get("__branches__")
|
||||
if isinstance(branches, dict) and isinstance(value, dict):
|
||||
branch_key = value.get(name)
|
||||
if isinstance(branch_key, str):
|
||||
sub_rules_map = branches.get(branch_key)
|
||||
if isinstance(sub_rules_map, dict):
|
||||
for sub_name, sub_rules in sub_rules_map.items():
|
||||
if sub_name in value:
|
||||
_enforce_local_validate(
|
||||
f"{name}.{sub_name}", value[sub_name], sub_rules,
|
||||
node_id=node_id,
|
||||
)
|
||||
if isinstance(value, dict):
|
||||
return any(_dict_contains_tensor(v) for v in value.values())
|
||||
return serialization._is_torch_tensor(value)
|
||||
|
||||
|
||||
def _check_image_max_batch(
|
||||
name: str,
|
||||
value: Any,
|
||||
cap: Any,
|
||||
*,
|
||||
node_id: str | None,
|
||||
) -> None:
|
||||
"""Apply the ``image_max_batch`` rule. No-op if cap is missing /
|
||||
non-int / value is not a recognisable image tensor."""
|
||||
if cap is None:
|
||||
return
|
||||
try:
|
||||
cap_int = int(cap)
|
||||
except (TypeError, ValueError):
|
||||
return
|
||||
if not serialization._is_torch_tensor(value):
|
||||
return
|
||||
rank = value.dim()
|
||||
last_dim = int(value.shape[-1]) if rank >= 1 else 0
|
||||
if not (rank == 4 or (rank == 3 and last_dim in (3, 4))):
|
||||
return
|
||||
batch = int(value.shape[0]) if rank == 4 else 1
|
||||
if batch <= cap_int:
|
||||
return
|
||||
label = f" ({node_id})" if node_id else ""
|
||||
if cap_int == 1:
|
||||
msg = (
|
||||
f"Remote node{label}: input {name!r} expects a single image "
|
||||
f"frame, got a batch of {batch}. Wire a single-frame IMAGE "
|
||||
f"into this input."
|
||||
)
|
||||
else:
|
||||
msg = (
|
||||
f"Remote node{label}: input {name!r} accepts at most "
|
||||
f"{cap_int} image frames, got a batch of {batch}. The "
|
||||
f"current maximum number of supported images is "
|
||||
f"{cap_int}."
|
||||
)
|
||||
raise RnpProtocolError(
|
||||
msg,
|
||||
code=ErrorCode.INPUT_INVALID,
|
||||
user_facing=True,
|
||||
)
|
||||
|
||||
|
||||
def _encode_one(
|
||||
name: str,
|
||||
value: Any,
|
||||
accepted_encodings: list[str] | None = None,
|
||||
) -> Any:
|
||||
def _encode_one(name: str, value: Any) -> Any:
|
||||
if serialization.is_audio_input(value):
|
||||
return serialization.encode_audio_input(value)
|
||||
if (
|
||||
isinstance(value, dict)
|
||||
and not serialization.is_envelope(value)
|
||||
and not serialization.is_audio_input(value)
|
||||
and _dict_contains_tensor(value)
|
||||
):
|
||||
# IO.Autogrow inputs arrive grouped as a dict (e.g. ``{"image0":
|
||||
# tensor, "image1": tensor}``) keyed by the per-slot virtual id.
|
||||
# Nested dynamic types (DynamicCombo whose option inputs include
|
||||
# an Autogrow, etc.) can produce arbitrarily nested dict-of-
|
||||
# tensor shapes — recurse so every heavy-typed leaf becomes an
|
||||
# envelope. Skip ordinary scalar dicts (custom-IO chain values
|
||||
# like RECRAFT_V3_STYLE) and already-encoded envelopes; both
|
||||
# pass through unchanged via the ``return value`` tail.
|
||||
encoded_dict: dict[str, Any] = {}
|
||||
for sub_name, sub_value in value.items():
|
||||
encoded_dict[sub_name] = _encode_one(f"{name}.{sub_name}", sub_value)
|
||||
return encoded_dict
|
||||
if not serialization._is_torch_tensor(value):
|
||||
return value
|
||||
rank = value.dim()
|
||||
last_dim = int(value.shape[-1]) if rank >= 1 else 0
|
||||
if rank == 4 or (rank == 3 and last_dim in (3, 4)):
|
||||
# Per-image tensors: a length-1 batch is the common case.
|
||||
# When the descriptor advertises ``"png_base64_batch"`` for
|
||||
# this input AND the tensor has > 1 frame, ship the multi-
|
||||
# frame envelope; otherwise fall back to single-frame
|
||||
# (``tensor[0]`` for batches >1, full tensor for length-1).
|
||||
accepts_batch = bool(
|
||||
accepted_encodings and "png_base64_batch" in accepted_encodings
|
||||
)
|
||||
return serialization.encode_image_tensor(
|
||||
value, accepts_batch=accepts_batch,
|
||||
)
|
||||
# Per-image tensors: a length-1 batch is the common case; longer
|
||||
# batches send only the first frame (the framework hasn't asked
|
||||
# for batch-of-envelopes semantics yet).
|
||||
return serialization.encode_image_tensor(value)
|
||||
if rank in (2, 3):
|
||||
return serialization.encode_mask_tensor(value)
|
||||
log.warning(
|
||||
|
||||
+70
-262
@@ -17,7 +17,6 @@ tensor — no parallel dialect.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import base64
|
||||
import logging
|
||||
from io import BytesIO
|
||||
@@ -57,64 +56,14 @@ async def maybe_externalize(
|
||||
max_inline_bytes: int | None,
|
||||
auth_headers: dict[str, str] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""If the envelope's inline payload exceeds ``max_inline_bytes``,
|
||||
upload it via the presigned-PUT path and rewrite the envelope to
|
||||
reference the uploaded asset(s) instead of inlining base64.
|
||||
|
||||
Two payload shapes are recognised:
|
||||
|
||||
* Singleton ``data`` envelopes (the common single-frame IMAGE /
|
||||
MASK / VIDEO / AUDIO case). Oversize payloads upload as one
|
||||
asset and the envelope's ``data`` field is swapped for ``uri``.
|
||||
* Multi-frame ``frames=[…]`` IMAGE envelopes (encoding
|
||||
``png_base64_batch``). The original ``maybe_externalize`` only
|
||||
inspected ``data`` and so silently let a 9× 2048² Flux2 batch
|
||||
ride the wire as a 200 MB POST body — defeating the entire
|
||||
reason the cap exists. The new path computes the summed raw
|
||||
byte budget (``sum(len(b64)) * 3 // 4`` — base64 expands 4:3 so
|
||||
this is the exact payload the server would receive) and, if it
|
||||
crosses the cap, uploads each frame as an individual presigned
|
||||
PUT in parallel via :func:`asyncio.gather` then rewrites the
|
||||
envelope to ``encoding=png_base64_batch_uri`` with a parallel
|
||||
``uris`` list. ``shape`` / ``dtype`` / ``color_space`` ride
|
||||
through unchanged so the server-side decoder still sees the
|
||||
batch dimension. Per-frame URIs (NOT a tar/zip blob) compose
|
||||
with the existing per-frame fetch helpers and let the server
|
||||
fetch in parallel.
|
||||
"""If the envelope's inline data exceeds ``max_inline_bytes``,
|
||||
upload the bytes and replace ``data`` with ``uri``.
|
||||
|
||||
No-op when either argument is falsy, the envelope already has a
|
||||
``uri`` / ``uris``, or the inline payload fits under the cap.
|
||||
|
||||
Dispatch keys off the ``encoding`` string — NOT field presence
|
||||
(``frames`` vs. ``data``). An encoder that emitted ``frames`` for
|
||||
a non-batch encoding by accident, or shipped both fields together,
|
||||
used to silently take the batch branch and externalise garbage;
|
||||
keying off the canonical encoding catches that as a hard ``frames``
|
||||
check inside the batch branch itself.
|
||||
``uri``, or the inline payload fits under the cap.
|
||||
"""
|
||||
if not server_url or not max_inline_bytes:
|
||||
return envelope
|
||||
|
||||
encoding = envelope.get("encoding")
|
||||
if encoding == "png_base64_batch":
|
||||
frames = envelope.get("frames")
|
||||
if not isinstance(frames, list) or not frames:
|
||||
# Mirror the server-side make_image_batch_envelope contract:
|
||||
# an empty/missing frames list is a hard wire bug, NOT a
|
||||
# silent no-op (which used to ship through under the old
|
||||
# ``isinstance(frames, list) and frames`` dispatch and then
|
||||
# blow up on the server's resolve_image_envelope_pngs check).
|
||||
raise RnpProtocolError(
|
||||
"png_base64_batch envelope requires a non-empty 'frames' list",
|
||||
code=ErrorCode.INTERNAL,
|
||||
)
|
||||
return await _maybe_externalize_batch(
|
||||
envelope, frames,
|
||||
server_url=server_url,
|
||||
max_inline_bytes=max_inline_bytes,
|
||||
auth_headers=auth_headers,
|
||||
)
|
||||
|
||||
data = envelope.get("data")
|
||||
if not isinstance(data, str):
|
||||
return envelope
|
||||
@@ -135,122 +84,22 @@ async def maybe_externalize(
|
||||
return out
|
||||
|
||||
|
||||
async def _maybe_externalize_batch(
|
||||
envelope: dict[str, Any],
|
||||
frames: list[Any],
|
||||
*,
|
||||
server_url: str,
|
||||
max_inline_bytes: int,
|
||||
auth_headers: dict[str, str] | None,
|
||||
) -> dict[str, Any]:
|
||||
"""Externalise a multi-frame IMAGE envelope when its summed inline
|
||||
payload overflows ``max_inline_bytes``.
|
||||
|
||||
Per-frame upload (NOT a tar/zip blob): each frame becomes its own
|
||||
presigned-PUT asset so the server can fetch in parallel and the
|
||||
existing per-frame ``resolve_image_envelope_pngs`` helper picks
|
||||
up the URI shape transparently. ``asyncio.gather`` issues every
|
||||
PUT in parallel — the dominant latency for a 9-frame Flux2 batch
|
||||
is the upload, not the server hop, so serialising would multiply
|
||||
end-to-end submit time by N. Frame ordering is preserved (gather
|
||||
returns positional results) — critical because a Flux2 provider
|
||||
maps ``uris[k]`` → ``input_image_{k+1}`` 1:1.
|
||||
"""
|
||||
# Cap on raw byte count (base64 expands 4:3, so multiplying by
|
||||
# 3/4 recovers the actual payload the server would receive).
|
||||
total_raw = 0
|
||||
for f in frames:
|
||||
if isinstance(f, str):
|
||||
total_raw += len(f) * 3 // 4
|
||||
if total_raw <= max_inline_bytes:
|
||||
return envelope
|
||||
|
||||
env_type = envelope.get("type", "")
|
||||
content_type = _CONTENT_TYPE_FOR_TYPE.get(env_type, "application/octet-stream")
|
||||
|
||||
async def _upload_one(idx: int, frame_b64: Any) -> str:
|
||||
if not isinstance(frame_b64, str):
|
||||
raise RnpProtocolError(
|
||||
f"image batch frame {idx} is not a base64 string "
|
||||
f"(got {type(frame_b64).__name__})",
|
||||
code=ErrorCode.INTERNAL,
|
||||
)
|
||||
raw = base64.b64decode(frame_b64)
|
||||
return await rnp_client.upload_asset(
|
||||
server_url, raw,
|
||||
content_type=content_type,
|
||||
auth_headers=auth_headers,
|
||||
)
|
||||
|
||||
uris = await asyncio.gather(
|
||||
*(_upload_one(i, f) for i, f in enumerate(frames))
|
||||
)
|
||||
log.info(
|
||||
"RNP: externalized %s batch envelope (%d frames, %d raw bytes) "
|
||||
"-> %d presigned URIs",
|
||||
env_type, len(frames), total_raw, len(uris),
|
||||
)
|
||||
|
||||
# Strip the inline ``frames`` field; preserve ``shape`` / ``dtype``
|
||||
# / ``color_space`` so the server-side decoder still sees the
|
||||
# batch dimension and per-frame metadata. Encoding flips to the
|
||||
# URI variant so dispatch keys cleanly off the string (NOT field
|
||||
# presence — frames vs. uris is a payload field, not a shape
|
||||
# selector).
|
||||
out: dict[str, Any] = {
|
||||
k: v for k, v in envelope.items() if k != "frames"
|
||||
}
|
||||
out["encoding"] = "png_base64_batch_uri"
|
||||
out["uris"] = list(uris)
|
||||
return out
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Image encode / decode
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def encode_image_tensor(tensor: Any, *, accepts_batch: bool = False) -> dict[str, Any]:
|
||||
"""Encode a ComfyUI IMAGE tensor (B,H,W,C float32 in 0..1) as an
|
||||
image envelope.
|
||||
def encode_image_tensor(tensor: Any) -> dict[str, Any]:
|
||||
"""Encode a ComfyUI IMAGE tensor (B,H,W,C float32 in 0..1) as a
|
||||
PNG-base64 image envelope.
|
||||
|
||||
``accepts_batch`` reflects whether the descriptor's
|
||||
``input_serialization`` for this input lists ``"png_base64_batch"``.
|
||||
When True AND ``tensor.shape[0] > 1``, emit a multi-frame
|
||||
``png_base64_batch`` envelope (``frames=[<png_b64>, …]``) so the
|
||||
server decodes a real ``[B, H, W, C]`` source. Otherwise emit a
|
||||
single-frame ``png_base64`` envelope for ``tensor[0]`` — preserves
|
||||
backwards compat with descriptors that haven't migrated yet, and
|
||||
matches the wire shape of the existing ``_decode_image_batch_envelope``
|
||||
on the response path so both directions round-trip symmetrically.
|
||||
Multi-image batches send only the first frame for now; the
|
||||
framework hasn't grown batch-of-envelopes semantics yet.
|
||||
"""
|
||||
import numpy as np
|
||||
from PIL import Image
|
||||
|
||||
rank = getattr(tensor, "dim", lambda: 0)()
|
||||
if rank == 4 and accepts_batch and int(tensor.shape[0]) > 1:
|
||||
frames_arr = (tensor.cpu().clamp(0.0, 1.0).numpy() * 255).astype(np.uint8)
|
||||
b, height, width = frames_arr.shape[:3]
|
||||
channels = frames_arr.shape[3] if frames_arr.ndim == 4 else 1
|
||||
pil_mode = {1: "L", 3: "RGB", 4: "RGBA"}.get(channels, "RGB")
|
||||
frames_b64: list[str] = []
|
||||
for i in range(b):
|
||||
img = Image.fromarray(frames_arr[i], mode=pil_mode)
|
||||
buf = BytesIO()
|
||||
img.save(buf, format="PNG")
|
||||
frames_b64.append(base64.b64encode(buf.getvalue()).decode("ascii"))
|
||||
return {
|
||||
"type": "image",
|
||||
"encoding": "png_base64_batch",
|
||||
"frames": frames_b64,
|
||||
"shape": [b, height, width, channels],
|
||||
"dtype": "uint8",
|
||||
"color_space": "srgb",
|
||||
}
|
||||
|
||||
# Single-frame fallback: matches the old ``encode_image_tensor``
|
||||
# contract (sends ``tensor[0]`` for a multi-frame batch when the
|
||||
# descriptor doesn't advertise the batch encoding).
|
||||
if rank == 4:
|
||||
# Strip the batch dim if present.
|
||||
if hasattr(tensor, "dim") and tensor.dim() == 4:
|
||||
frame = tensor[0]
|
||||
else:
|
||||
frame = tensor
|
||||
@@ -272,18 +121,8 @@ def encode_image_tensor(tensor: Any, *, accepts_batch: bool = False) -> dict[str
|
||||
|
||||
|
||||
def decode_image_envelope(envelope: dict[str, Any]) -> Any:
|
||||
"""Decode an image envelope into a ComfyUI IMAGE tensor.
|
||||
|
||||
Single-frame encodings (``png_base64``, ``jpeg_base64``,
|
||||
``webp_base64``) decode into ``[1, H, W, C]``. The multi-frame
|
||||
``png_base64_batch`` encoding (a list of base64-PNG ``frames``)
|
||||
stacks each frame into ``[B, H, W, C]`` — the native ComfyUI
|
||||
image-batch shape — so providers whose upstream returns ``n > 1``
|
||||
surface as a real batched tensor in the workflow runtime.
|
||||
"""
|
||||
"""Decode an image envelope into a ComfyUI IMAGE tensor (B,H,W,C float32)."""
|
||||
encoding = envelope.get("encoding")
|
||||
if encoding == "png_base64_batch":
|
||||
return _decode_image_batch_envelope(envelope)
|
||||
if encoding not in ("png_base64", "jpeg_base64", "webp_base64"):
|
||||
raise RnpProtocolError(
|
||||
f"Unsupported image encoding: {encoding!r}",
|
||||
@@ -302,69 +141,6 @@ def decode_image_envelope(envelope: dict[str, Any]) -> Any:
|
||||
return torch.from_numpy(arr).unsqueeze(0)
|
||||
|
||||
|
||||
def _decode_image_batch_envelope(envelope: dict[str, Any]) -> Any:
|
||||
"""Decode a ``png_base64_batch`` envelope into ``[B, H, W, C]``.
|
||||
|
||||
The wire shape is ``{"type":"image", "encoding":"png_base64_batch",
|
||||
"frames":[<png_b64>, ...], "shape":[B,H,W,C], ...}``. Each frame is
|
||||
a PNG-encoded image; all frames must share the same H/W after the
|
||||
RGB conversion (the server-side encoder enforces this). A frame
|
||||
count mismatch against the ``shape[0]`` field surfaces as a hard
|
||||
parse error rather than silently dropping frames — matches the SVG
|
||||
decoder's ``count`` integrity check.
|
||||
"""
|
||||
import base64 as _b64
|
||||
import numpy as np
|
||||
import torch
|
||||
from PIL import Image
|
||||
|
||||
frames = envelope.get("frames")
|
||||
if not isinstance(frames, list) or not frames:
|
||||
raise RnpProtocolError(
|
||||
"image batch envelope missing or empty 'frames' list",
|
||||
code=ErrorCode.INTERNAL,
|
||||
)
|
||||
declared_shape = envelope.get("shape")
|
||||
if isinstance(declared_shape, list) and len(declared_shape) >= 1:
|
||||
declared_b = int(declared_shape[0])
|
||||
if declared_b != len(frames):
|
||||
raise RnpProtocolError(
|
||||
f"image batch envelope frame count {len(frames)} does not "
|
||||
f"match declared shape[0]={declared_b}",
|
||||
code=ErrorCode.INTERNAL,
|
||||
)
|
||||
|
||||
arrs: list[Any] = []
|
||||
base_hw: tuple[int, int] | None = None
|
||||
for idx, frame_b64 in enumerate(frames):
|
||||
if not isinstance(frame_b64, str):
|
||||
raise RnpProtocolError(
|
||||
f"image batch frame {idx} is not a base64 string",
|
||||
code=ErrorCode.INTERNAL,
|
||||
)
|
||||
try:
|
||||
raw = _b64.b64decode(frame_b64)
|
||||
img = Image.open(BytesIO(raw)).convert("RGB")
|
||||
except Exception as e:
|
||||
raise RnpProtocolError(
|
||||
f"image batch frame {idx} failed to decode: {e}",
|
||||
code=ErrorCode.INTERNAL,
|
||||
) from e
|
||||
arr = np.array(img).astype(np.float32) / 255.0
|
||||
if base_hw is None:
|
||||
base_hw = (arr.shape[0], arr.shape[1])
|
||||
elif (arr.shape[0], arr.shape[1]) != base_hw:
|
||||
raise RnpProtocolError(
|
||||
f"image batch frame {idx} size {arr.shape[:2]} does not "
|
||||
f"match frame 0 size {base_hw} (server should have rejected "
|
||||
f"this; likely a wire bug)",
|
||||
code=ErrorCode.INTERNAL,
|
||||
)
|
||||
arrs.append(arr)
|
||||
stacked = np.stack(arrs, axis=0)
|
||||
return torch.from_numpy(stacked)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Mask encode / decode
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -451,11 +227,7 @@ def encode_audio_input(audio: dict[str, Any]) -> dict[str, Any]:
|
||||
def decode_audio_envelope(envelope: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Decode an audio envelope back into a ComfyUI AUDIO dict."""
|
||||
encoding = envelope.get("encoding")
|
||||
if encoding not in (
|
||||
"mp3_base64", "mp3_inline",
|
||||
"wav_base64", "wav_inline",
|
||||
"opus_base64",
|
||||
):
|
||||
if encoding not in ("mp3_base64", "mp3_inline", "wav_base64", "wav_inline"):
|
||||
raise RnpProtocolError(
|
||||
f"Unsupported audio encoding: {encoding!r}",
|
||||
code=ErrorCode.INTERNAL,
|
||||
@@ -481,30 +253,66 @@ def decode_video_envelope(envelope: dict[str, Any]) -> Any:
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 3D model decode (encode lands when a remote node accepts MODEL_3D inputs)
|
||||
# SVG decode (encode lands when a remote node accepts SVG inputs; today
|
||||
# Recraft's V3/V4 vector endpoints only emit SVG, so the client is
|
||||
# decode-only).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def decode_model3d_envelope(envelope: dict[str, Any]) -> Any:
|
||||
"""Decode a 3D-model envelope into a ComfyUI ``File3D`` object.
|
||||
def decode_svg_envelope(envelope: dict[str, Any]) -> Any:
|
||||
"""Decode an SVG envelope into a ``comfy_extras.nodes_images.SVG`` instance.
|
||||
|
||||
Mirrors :func:`decode_video_envelope`: the only currently supported
|
||||
encoding is ``glb_inline`` (binary glTF 2.0 — magic ``b"glTF"`` at
|
||||
offset 0). The bytes are handed to ``File3D`` as a ``BytesIO`` so
|
||||
downstream nodes (preview3d / SaveGLB / Load3D) see exactly the
|
||||
same value shape they would from a local Tripo / Rodin / Meshy
|
||||
node. The envelope's optional ``format`` extra is forwarded as the
|
||||
second constructor arg so the materialiser picks the right file
|
||||
extension when persisting.
|
||||
The wire payload is one or more concatenated ``<svg>…</svg>`` XML
|
||||
documents; we split on the closing tag so a multi-frame Recraft
|
||||
response (n>1) hydrates a single ``SVG`` carrying a list of
|
||||
BytesIO frames — the same shape the local node produces via
|
||||
``SVG(svg_data)``.
|
||||
"""
|
||||
encoding = envelope.get("encoding")
|
||||
if encoding != "glb_inline":
|
||||
if encoding not in ("svg_xml_base64", "svg_xml_inline"):
|
||||
raise RnpProtocolError(
|
||||
f"Unsupported model_3d encoding: {encoding!r}",
|
||||
f"Unsupported svg encoding: {encoding!r}",
|
||||
code=ErrorCode.INTERNAL,
|
||||
)
|
||||
from comfy_api.latest._util import File3D
|
||||
file_format = envelope.get("format") or "glb"
|
||||
return File3D(BytesIO(decode_envelope_data(envelope)), str(file_format))
|
||||
from comfy_extras.nodes_images import SVG
|
||||
|
||||
raw = decode_envelope_data(envelope)
|
||||
text = raw.decode("utf-8", "replace")
|
||||
# The server joins per-frame SVGs with ``\n`` (see
|
||||
# ``_emit_svg_from_urls``); split on the closing tag followed by a
|
||||
# newline so an SVG document that legitimately contains ``</svg>``
|
||||
# in CDATA / a foreignObject / a comment isn't shredded mid-frame.
|
||||
# Fall back to splitting on the bare closing tag for single-frame
|
||||
# payloads or older server builds that didn't add the newline.
|
||||
delimiter = "</svg>\n"
|
||||
if delimiter not in text:
|
||||
delimiter = "</svg>"
|
||||
parts: list[BytesIO] = []
|
||||
cursor = 0
|
||||
closing = "</svg>"
|
||||
while True:
|
||||
idx = text.find(delimiter, cursor)
|
||||
if idx < 0:
|
||||
tail = text[cursor:].strip()
|
||||
if tail:
|
||||
parts.append(BytesIO(tail.encode("utf-8")))
|
||||
break
|
||||
end = idx + len(closing) # keep the closing tag on this frame
|
||||
chunk = text[cursor:end].strip()
|
||||
if chunk:
|
||||
parts.append(BytesIO(chunk.encode("utf-8")))
|
||||
cursor = idx + len(delimiter)
|
||||
declared = envelope.get("count")
|
||||
if isinstance(declared, int) and declared > 0 and declared != len(parts):
|
||||
log.warning(
|
||||
"SVG envelope count=%d but parsed %d documents — server payload may be malformed",
|
||||
declared, len(parts),
|
||||
)
|
||||
if not parts:
|
||||
raise RnpProtocolError(
|
||||
"SVG envelope payload contained no <svg> documents",
|
||||
code=ErrorCode.INTERNAL,
|
||||
)
|
||||
return SVG(parts)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -512,11 +320,11 @@ def decode_model3d_envelope(envelope: dict[str, Any]) -> Any:
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_DECODERS = {
|
||||
"image": decode_image_envelope,
|
||||
"mask": decode_mask_envelope,
|
||||
"audio": decode_audio_envelope,
|
||||
"video": decode_video_envelope,
|
||||
"model_3d": decode_model3d_envelope,
|
||||
"image": decode_image_envelope,
|
||||
"mask": decode_mask_envelope,
|
||||
"audio": decode_audio_envelope,
|
||||
"video": decode_video_envelope,
|
||||
"svg": decode_svg_envelope,
|
||||
}
|
||||
|
||||
|
||||
@@ -566,7 +374,7 @@ __all__ = [
|
||||
"encode_audio_input",
|
||||
"decode_audio_envelope",
|
||||
"decode_video_envelope",
|
||||
"decode_model3d_envelope",
|
||||
"decode_svg_envelope",
|
||||
"decode_envelope",
|
||||
"is_envelope",
|
||||
"is_image_tensor",
|
||||
|
||||
Reference in New Issue
Block a user