chris.dumas/DumasNodes
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Add H3 latent upscale refinement stage

Browse Source
This commit is contained in:
chris.dumas
2026-09-03 12:53:57 +00:00
parent 306816534b
commit c1d937e0e2
7 changed files with 761 additions and 253 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_dumas_h3_longvideos.py
+67 -22
View File
@@ -24,6 +24,7 @@ class DumasH3LongVideosHelperTests(unittest.TestCase):
"PIL.Image",
"folder_paths",
"dumas_image_nodes",
"dumas_h3_latent_upscale",
"dumas_h3_longvideos",
)
}
@@ -207,7 +208,7 @@ class DumasH3LongVideosHelperTests(unittest.TestCase):
"retry_elapsed": 1.2,
"attempts": 2,
"sample": 8.0,
"detail_sample": 0.5,
"latent_upscale_sample": 0.5,
"decode_video": 2.1,
"decode_audio": 0.4,
"cleanup": 0.2,
@@ -230,11 +231,11 @@ class DumasH3LongVideosHelperTests(unittest.TestCase):
self.assertIn("decode audio 0.7s", note)
self.assertIn("cleanup 0.3s", note)
self.assertIn("retry elapsed 1.2s", note)
self.assertIn("detail 0.5s", note)
self.assertIn("latent upscale 0.5s", note)
self.assertIn("retries 1", note)
self.assertIn("slowest shot 1 12.4s", note)
def test_detail_pass_refines_video_but_preserves_audio(self):
def test_latent_upscale_refines_video_but_preserves_audio(self):
class FakeTensor:
def __init__(self, name):
self.name = name
@@ -263,6 +264,8 @@ class DumasH3LongVideosHelperTests(unittest.TestCase):
original_decode_video = self.module._decode_video
original_decode_audio = self.module._decode_audio
original_cleanup = self.module._deep_cleanup
original_upscale = self.module._upscale_latent_video
original_copy_sample = self.module._copy_sample_latent
original_nested = getattr(self.module.comfy.nested_tensor, "NestedTensor", None)
try:
self.module.comfy.nested_tensor.NestedTensor = FakeNestedTensor
@@ -277,6 +280,8 @@ class DumasH3LongVideosHelperTests(unittest.TestCase):
{"samples": FakeNestedTensor((FakeTensor("basev"), FakeTensor("basea")))},
)
self.module._evict_all_but = lambda *_args, **_kwargs: None
self.module._upscale_latent_video = lambda video, param: (FakeTensor("upv"), 8, 16)
self.module._copy_sample_latent = lambda sampled: sampled["samples"].unbind()
self.module._decode_video = lambda _vae, out_latent, *_args, **_kwargs: out_latent
self.module._decode_audio = lambda _vae, out_latent: out_latent
self.module._deep_cleanup = lambda: None
@@ -298,18 +303,22 @@ class DumasH3LongVideosHelperTests(unittest.TestCase):
tiled=False,
sa=(123, 20, 1.0, "res_multistep", "simple", 1.0),
handoff=None,
detail_pass=True,
detail_sampler_name="euler",
detail_scheduler="beta",
detail_steps=5,
detail_denoise=0.4,
latent_upscale_param={
"mode": "model",
"model_name": "upscale.safetensors",
"width": 256,
"height": 128,
"device": "cpu",
"precision": "fp16",
"refine_denoise": 0.4,
},
)
self.assertEqual(len(calls), 2)
self.assertIsNot(calls[1][0][8], first_out)
self.assertIs(calls[1][0][8]["samples"], first_out["samples"])
self.assertEqual(calls[1][0][4], "euler")
self.assertEqual(calls[1][0][5], "beta")
self.assertEqual(calls[1][0][8]["samples"].unbind()[0].name, "upv")
self.assertEqual(calls[1][0][4], "res_multistep")
self.assertEqual(calls[1][0][5], "simple")
self.assertAlmostEqual(calls[1][1]["denoise"], 0.4)
self.assertEqual(result[1], first_out)
self.assertEqual(result[2][0].name, "v2")
@@ -323,12 +332,14 @@ class DumasH3LongVideosHelperTests(unittest.TestCase):
self.module._decode_video = original_decode_video
self.module._decode_audio = original_decode_audio
self.module._deep_cleanup = original_cleanup
self.module._upscale_latent_video = original_upscale
self.module._copy_sample_latent = original_copy_sample
if original_nested is None:
delattr(self.module.comfy.nested_tensor, "NestedTensor")
else:
self.module.comfy.nested_tensor.NestedTensor = original_nested
def test_detail_pass_decodes_audio_before_video_and_cleans_up(self):
def test_latent_upscale_decodes_audio_before_video_and_cleans_up(self):
class FakeTensor:
def __init__(self, name):
self.name = name
@@ -357,12 +368,14 @@ class DumasH3LongVideosHelperTests(unittest.TestCase):
original_decode_video = self.module._decode_video
original_decode_audio = self.module._decode_audio
original_cleanup = self.module._deep_cleanup
original_upscale = self.module._upscale_latent_video
original_copy_sample = self.module._copy_sample_latent
original_nested = getattr(self.module.comfy.nested_tensor, "NestedTensor", None)
try:
self.module.comfy.nested_tensor.NestedTensor = FakeNestedTensor
def common_ksampler(*args, **kwargs):
order.append("detail_sample" if len(order) else "sample")
order.append("latent_upscale_sample" if len(order) else "sample")
return (first_out if len([x for x in order if x.endswith("sample")]) == 1 else second_out,)
def decode_audio(_vae, out_latent):
@@ -385,6 +398,8 @@ class DumasH3LongVideosHelperTests(unittest.TestCase):
{"samples": FakeNestedTensor((FakeTensor("basev"), FakeTensor("basea")))},
)
self.module._evict_all_but = lambda *_args, **_kwargs: None
self.module._upscale_latent_video = lambda video, param: (FakeTensor("upv"), 8, 16)
self.module._copy_sample_latent = lambda sampled: sampled["samples"].unbind()
self.module._decode_video = decode_video
self.module._decode_audio = decode_audio
self.module._deep_cleanup = cleanup
@@ -406,15 +421,19 @@ class DumasH3LongVideosHelperTests(unittest.TestCase):
tiled=False,
sa=(123, 20, 1.0, "res_multistep", "simple", 1.0),
handoff=None,
detail_pass=True,
detail_sampler_name="euler",
detail_scheduler="beta",
detail_steps=5,
detail_denoise=0.4,
latent_upscale_param={
"mode": "model",
"model_name": "upscale.safetensors",
"width": 256,
"height": 128,
"device": "cpu",
"precision": "fp16",
"refine_denoise": 0.4,
},
)
self.assertEqual(order[0], "sample")
self.assertEqual(order[1], "detail_sample")
self.assertEqual(order[1], "latent_upscale_sample")
self.assertLess(order.index("audio"), order.index("video"))
self.assertEqual(order[-1], "cleanup")
finally:
@@ -424,12 +443,14 @@ class DumasH3LongVideosHelperTests(unittest.TestCase):
self.module._decode_video = original_decode_video
self.module._decode_audio = original_decode_audio
self.module._deep_cleanup = original_cleanup
self.module._upscale_latent_video = original_upscale
self.module._copy_sample_latent = original_copy_sample
if original_nested is None:
delattr(self.module.comfy.nested_tensor, "NestedTensor")
else:
self.module.comfy.nested_tensor.NestedTensor = original_nested
def test_detail_pass_treats_falsey_strings_as_disabled(self):
def test_latent_upscale_off_skips_second_pass(self):
calls = []
original_common_ksampler = self.module.nodes.common_ksampler
original_build = self.module._build_shot_conditioning
@@ -437,6 +458,8 @@ class DumasH3LongVideosHelperTests(unittest.TestCase):
original_decode_video = self.module._decode_video
original_decode_audio = self.module._decode_audio
original_cleanup = self.module._deep_cleanup
original_upscale = self.module._upscale_latent_video
original_copy_sample = self.module._copy_sample_latent
try:
self.module.nodes.common_ksampler = lambda *args, **kwargs: (calls.append((args, kwargs)) or {"samples": "latent"},)
self.module._build_shot_conditioning = lambda *_args, **_kwargs: ("cond", {"samples": "base"})
@@ -444,6 +467,8 @@ class DumasH3LongVideosHelperTests(unittest.TestCase):
self.module._decode_video = lambda _vae, out_latent, *_args, **_kwargs: out_latent
self.module._decode_audio = lambda _vae, out_latent: out_latent
self.module._deep_cleanup = lambda: None
self.module._upscale_latent_video = lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("should not run"))
self.module._copy_sample_latent = lambda sampled: sampled
self.module.H3LongVideos()._render(
model=object(),
@@ -459,7 +484,7 @@ class DumasH3LongVideosHelperTests(unittest.TestCase):
tiled=False,
sa=(123, 20, 1.0, "res_multistep", "simple", 1.0),
handoff=None,
detail_pass="false",
latent_upscale_param={"mode": "off"},
)
self.assertEqual(len(calls), 1)
@@ -470,6 +495,8 @@ class DumasH3LongVideosHelperTests(unittest.TestCase):
self.module._decode_video = original_decode_video
self.module._decode_audio = original_decode_audio
self.module._deep_cleanup = original_cleanup
self.module._upscale_latent_video = original_upscale
self.module._copy_sample_latent = original_copy_sample
def test_distribute_generations_canonicalizes_per_shot_audio_and_anchor_directives(self):
generations = self.module.distribute_generations(
@@ -736,6 +763,12 @@ class DumasH3LongVideosHelperTests(unittest.TestCase):
self.assertNotIn(f"ref_image_{index}", optional)
self.assertNotIn("per_beat_length", optional)
self.assertNotIn("cleanup_between_shots", optional)
self.assertNotIn("detail_pass", optional)
self.assertNotIn("detail_sampler_name", optional)
self.assertNotIn("detail_scheduler", optional)
self.assertNotIn("detail_steps", optional)
self.assertNotIn("detail_denoise", optional)
self.assertIn("latent_upscale_param", optional)
def test_shot_seconds_tooltip_describes_ceiling_behavior(self):
optional = self.module.H3LongVideos.INPUT_TYPES()["optional"]
@@ -743,7 +776,7 @@ class DumasH3LongVideosHelperTests(unittest.TestCase):
self.assertIn("GLOBAL per-shot maximum", tooltip)
self.assertIn("A beat's own `seconds:` directive can still ask for less", tooltip)
self.assertIn("honoring it; may spill to system RAM (slow) or OOM", tooltip)
self.assertIn("let the render fail instead of shrinking it", tooltip)
def test_resolve_shot_frames_honors_forced_request_over_budget(self):
original_estimate_shot_frames = self.module.estimate_shot_frames
@@ -1061,6 +1094,18 @@ class DumasH3LongVideosHelperTests(unittest.TestCase):
{"DumasH3LongVideos": "Dumas H3 Long Videos (FL2VA + REF2VA)"},
)
def test_latent_upscale_params_node_is_exposed(self):
latent = importlib.import_module("dumas_h3_latent_upscale")
self.assertEqual(
latent.NODE_CLASS_MAPPINGS,
{"DumasH3LatentUpscaleParams": latent.H3LatentUpscaleParams},
)
self.assertEqual(
latent.NODE_DISPLAY_NAME_MAPPINGS,
{"DumasH3LatentUpscaleParams": "Dumas H3 Latent Upscale Params"},
)
def test_compose_persistent_does_not_expand_ambiguous_plural_to_full_cast(self):
active = self.module.parse_wardrobe(
"Maya = she, red jacket\n"