digital-garbageandClaude Opus 5
134cee78b7
Core already turns images into a video and writes the file, and handles containers, codecs and muxing properly. Writing our own would be a worse copy of tested code, so there is no FunPack save node and no reason for VHS either. What core cannot generalise is decoding a latent that is not one tensor: an AV latent has two branches wanting different VAEs, and their arrangement is the model's business. Same rule as the empty latent -- a model module may claim it, and H3 now does. A nested latent nobody claims is refused rather than handed to vae.decode, which would raise somewhere confusing or quietly decode one branch as the picture. A missing audio VAE is named, because returning no audio looks exactly like a model that makes none. With this the pipeline closes end to end using FunPack's eight nodes and core's samplers and video nodes -- verified by loading both together. That matters more than the node count: until now nothing FunPack produced could be measured, only reasoned about. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
118 lines
4.1 KiB
Python
118 lines
4.1 KiB
Python
"""Decoding: the ordinary path, and the one a model has to claim."""
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _needs_comfy(comfyui):
|
|
"""Imports comfy."""
|
|
|
|
|
|
class _Vae:
|
|
def __init__(self, out=None):
|
|
self.seen = []
|
|
self._out = out
|
|
|
|
def decode(self, latent):
|
|
import torch
|
|
self.seen.append(latent)
|
|
return self._out if self._out is not None else torch.zeros(4, 8, 8, 3)
|
|
|
|
|
|
def test_a_single_tensor_latent_decodes_normally():
|
|
import torch
|
|
from modules.output.decode.nodes import FunPackDecode
|
|
|
|
vae = _Vae()
|
|
out = FunPackDecode.execute({"samples": torch.zeros(1, 4, 8, 8)}, vae)
|
|
images, audio, status = out.result
|
|
|
|
assert audio is None and "single latent" in status
|
|
assert vae.seen and images.shape[-1] == 3
|
|
|
|
|
|
def test_a_batch_of_clips_becomes_one_strip_of_frames():
|
|
"""A 5-D decode is [B, T, H, W, C]; downstream wants frames."""
|
|
import torch
|
|
from modules.output.decode.nodes import FunPackDecode
|
|
|
|
vae = _Vae(out=torch.zeros(2, 5, 8, 8, 3))
|
|
images, _audio, _status = FunPackDecode.execute({"samples": torch.zeros(1, 4, 8, 8)}, vae).result
|
|
assert images.shape == (10, 8, 8, 3)
|
|
|
|
|
|
def test_a_nested_latent_nobody_claims_is_refused_not_decoded_as_one_tensor(monkeypatch):
|
|
"""vae.decode on a NestedTensor would either raise somewhere confusing or
|
|
quietly decode one branch as if it were the picture.
|
|
|
|
H3 is installed and claims any nested latent, so proving this path needs its
|
|
claim taken away -- which is what a machine without that module would look
|
|
like.
|
|
"""
|
|
import torch
|
|
from comfy.nested_tensor import NestedTensor
|
|
from core import registry as registry_mod
|
|
from modules.output.decode.nodes import FunPackDecode
|
|
|
|
for spec in registry_mod.current().specs.values():
|
|
if "decode" in spec.provides:
|
|
monkeypatch.delitem(spec.provides, "decode")
|
|
|
|
nested = NestedTensor((torch.zeros(1, 24, 2, 8, 8), torch.zeros(1, 32, 2, 16)))
|
|
with pytest.raises(RuntimeError, match="more than one part"):
|
|
FunPackDecode.execute({"samples": nested}, _Vae())
|
|
|
|
|
|
def test_h3_claims_a_nested_latent_and_uses_both_vaes(monkeypatch):
|
|
import torch
|
|
from comfy.nested_tensor import NestedTensor
|
|
from modules.models import minimax_h3
|
|
|
|
video_vae = _Vae(out=torch.zeros(1, 3, 8, 8, 3))
|
|
audio_vae = _Vae()
|
|
|
|
monkeypatch.setattr("comfy_extras.nodes_audio.vae_decode_audio",
|
|
lambda vae, samples: {"waveform": "decoded", "sample_rate": 32000})
|
|
|
|
nested = NestedTensor((torch.zeros(1, 24, 2, 8, 8), torch.zeros(1, 32, 2, 16)))
|
|
images, audio = minimax_h3.decode(nested, vae=video_vae, audio_vae=audio_vae)
|
|
|
|
assert images.shape == (3, 8, 8, 3)
|
|
assert audio["waveform"] == "decoded"
|
|
|
|
|
|
def test_h3_does_not_speak_for_a_plain_latent():
|
|
import torch
|
|
from modules.models import minimax_h3
|
|
assert minimax_h3.decode(torch.zeros(1, 4, 8, 8), vae=_Vae()) is None
|
|
|
|
|
|
def test_a_missing_audio_vae_is_named_rather_than_silently_silent():
|
|
"""Returning no audio would look exactly like a model that makes none."""
|
|
import torch
|
|
from comfy.nested_tensor import NestedTensor
|
|
from modules.models import minimax_h3
|
|
|
|
nested = NestedTensor((torch.zeros(1, 24, 2, 8, 8), torch.zeros(1, 32, 2, 16)))
|
|
with pytest.raises(RuntimeError, match="audio VAE"):
|
|
minimax_h3.decode(nested, vae=_Vae(out=torch.zeros(1, 3, 8, 8, 3)), audio_vae=None)
|
|
|
|
|
|
def test_a_claiming_module_that_breaks_stops_the_decode(monkeypatch):
|
|
import torch
|
|
from comfy.nested_tensor import NestedTensor
|
|
from core import registry as registry_mod
|
|
from modules.output.decode.nodes import FunPackDecode
|
|
|
|
def broken(latent, vae, audio_vae=None):
|
|
if not getattr(latent, "is_nested", False):
|
|
return None
|
|
raise RuntimeError("branch order changed")
|
|
|
|
spec = registry_mod.current().specs["model_minimax_h3"]
|
|
monkeypatch.setitem(spec.provides, "decode", broken)
|
|
|
|
nested = NestedTensor((torch.zeros(1, 24, 2, 8, 8), torch.zeros(1, 32, 2, 16)))
|
|
with pytest.raises(RuntimeError, match="Refusing to decode"):
|
|
FunPackDecode.execute({"samples": nested}, _Vae())
|