| |
| """Older active-only Ref2V checkpoints remain usable in Checkpoint Manager.""" |
|
|
| import hashlib |
| import importlib.util |
| import json |
| import pathlib |
| import sys |
| import tempfile |
| import types |
|
|
|
|
| ROOT = pathlib.Path(__file__).resolve().parents[1] |
| PACKAGE = "h3_legacy_checkpoint_adoption_unit" |
|
|
| folder_paths = types.ModuleType("folder_paths") |
| folder_paths.output_directory = str(ROOT) |
| folder_paths.get_output_directory = lambda: folder_paths.output_directory |
| folder_paths.get_temp_directory = lambda: folder_paths.output_directory |
| folder_paths.get_input_directory = lambda: folder_paths.output_directory |
| folder_paths.get_annotated_filepath = lambda value: str(value) |
| sys.modules["folder_paths"] = folder_paths |
|
|
| server = types.ModuleType("server") |
| server.PromptServer = type("PromptServer", (), {"instance": None}) |
| sys.modules["server"] = server |
|
|
| package = types.ModuleType(PACKAGE) |
| package.__path__ = [str(ROOT)] |
| sys.modules[PACKAGE] = package |
|
|
| shared_nodes = types.ModuleType(PACKAGE + ".nodes") |
| shared_nodes.MiniMaxH3MotionContext = object |
| shared_nodes._claim_inline_patch_ownership = lambda _conditioning=None: "test patch owner" |
| shared_nodes._prepare_native_guide_conditioning = lambda *args: None |
| shared_nodes._resize = lambda *args: None |
| shared_nodes._streams_from_latent = lambda *args: None |
| sys.modules[shared_nodes.__name__] = shared_nodes |
|
|
| spec = importlib.util.spec_from_file_location( |
| PACKAGE + ".chain_nodes", ROOT / "chain_nodes.py") |
| chain = importlib.util.module_from_spec(spec) |
| sys.modules[spec.name] = chain |
| spec.loader.exec_module(chain) |
|
|
|
|
| def digest(path): |
| return hashlib.sha256(path.read_bytes()).hexdigest() |
|
|
|
|
| def write_legacy_active(run, scene, token, compatibility): |
| segments = run / "segments" |
| checkpoints = run / "checkpoints" |
| segments.mkdir(parents=True, exist_ok=True) |
| checkpoints.mkdir(parents=True, exist_ok=True) |
| stem = "clip_%04d.%s" % (scene, token) |
| segment_path = segments / (stem + ".mp4") |
| checkpoint_path = checkpoints / (stem + ".safetensors") |
| segment_path.write_bytes(("legacy-video-%d" % scene).encode()) |
| checkpoint_path.write_bytes(("legacy-latent-%d" % scene).encode()) |
| segment = { |
| "index": scene, |
| "id": "scene_%d" % scene, |
| "segment": str(segment_path.relative_to(folder_paths.output_directory)), |
| "checkpoint": str( |
| checkpoint_path.relative_to(folder_paths.output_directory)), |
| "metadata": str((checkpoints / ("clip_%04d.json" % scene)).relative_to( |
| folder_paths.output_directory)), |
| "raw_frames": 29, |
| "delivered_frames": 29, |
| "history_hash": "legacy-history-%d" % scene, |
| "prompt_prefix": "shared", |
| "scene_prompt": "legacy prompt %d" % scene, |
| "prompt": "shared\nlegacy prompt %d" % scene, |
| "prompt_hash": hashlib.sha256( |
| ("shared\nlegacy prompt %d" % scene).encode()).hexdigest(), |
| "seed": str(100 + scene), |
| "steps": 8, |
| "segment_sha256": digest(segment_path), |
| "checkpoint_sha256": digest(checkpoint_path), |
| } |
| metadata = { |
| "format": "h3_chain_segment_v2", |
| "run_name": "legacy_test", |
| "plan_hash": "legacy-plan", |
| "history_hash": segment["history_hash"], |
| "compatibility": compatibility, |
| "segment": segment, |
| } |
| pointer = checkpoints / ("clip_%04d.json" % scene) |
| pointer.write_text(json.dumps(metadata), encoding="utf-8") |
| return pointer, segment_path, checkpoint_path |
|
|
|
|
| def main(): |
| with tempfile.TemporaryDirectory() as temporary: |
| folder_paths.output_directory = temporary |
| run = pathlib.Path(temporary) / "h3_chains" / "legacy_test" |
| compatibility = { |
| "width": 64, |
| "height": 64, |
| "fps": 24, |
| "context_length": 0, |
| "audio_context_length": 0, |
| "audio_mode": "source_track", |
| "continuation_mode": "guide", |
| } |
| plan = { |
| "format": "h3_chain_plan_v2", |
| "run_name": "legacy_test", |
| "plan_hash": "legacy-plan", |
| "prompt_prefix": "shared", |
| "compatibility": compatibility, |
| "shots": [ |
| {"id": "scene_1", "prompt": "legacy prompt 1"}, |
| {"id": "scene_2", "prompt": "legacy prompt 2"}, |
| ], |
| } |
| run.mkdir(parents=True) |
| (run / "plan.json").write_text(json.dumps(plan), encoding="utf-8") |
| first = "1" * 32 |
| second = "2" * 32 |
| artifacts = [ |
| write_legacy_active(run, 1, first, compatibility), |
| write_legacy_active(run, 2, second, compatibility), |
| ] |
| pointer_bytes = [item[0].read_bytes() for item in artifacts] |
| media_bytes = [(item[1].read_bytes(), item[2].read_bytes()) |
| for item in artifacts] |
|
|
| payload = chain._saved_checkpoint_listing("legacy_test") |
| assert [item["revision"] for item in payload["checkpoints"]] == [ |
| first, second] |
| assert payload["summary"] == { |
| "scene_count": 2, |
| "revision_count": 2, |
| "branch_count": 1, |
| "bytes": payload["summary"]["bytes"], |
| "broken_count": 0, |
| } |
| assert payload["branches"][0]["active"] |
| assert payload["branches"][0]["path"] == [ |
| {"scene": 1, "revision": first}, |
| {"scene": 2, "revision": second}, |
| ] |
| indexed = {(item["scene"], item["revision"]): item |
| for item in payload["revisions"]} |
| assert indexed[(2, second)]["parent"] == { |
| "scene": 1, "revision": first} |
|
|
| sidecars = [ |
| run / "checkpoints" / ("clip_0001.%s.json" % first), |
| run / "checkpoints" / ("clip_0002.%s.json" % second), |
| ] |
| assert all(path.is_file() for path in sidecars) |
| assert all(json.loads(path.read_text())["legacy_adoption"]["version"] == 1 |
| for path in sidecars) |
| assert [item[0].read_bytes() for item in artifacts] == pointer_bytes |
| assert [(item[1].read_bytes(), item[2].read_bytes()) |
| for item in artifacts] == media_bytes |
|
|
| selected = chain._checkpoint_selection_manifest({ |
| "run_name": "legacy_test", |
| "lineage": [ |
| {"scene": 1, "revision": first}, |
| {"scene": 2, "revision": second}, |
| ], |
| "scope_start_scene": 1, |
| "scope_end_scene": 2, |
| }) |
| assert selected["clip_count"] == 2 |
| assert selected["selection_complete"] is True |
| assert [item["revision"] for item in selected["segments"]] == [ |
| first, second] |
|
|
| |
| refreshed = chain._saved_checkpoint_listing("legacy_test") |
| assert refreshed["summary"]["revision_count"] == 2 |
| assert sorted((run / "checkpoints").glob("clip_*.json")) == sorted([ |
| artifacts[0][0], artifacts[1][0], *sidecars]) |
|
|
| print("H3 legacy checkpoints: active-only Ref2V saves are adopted without copying artifacts") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
