File size: 8,163 Bytes
bc50ea0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 | #!/usr/bin/env python3
"""Attach a legacy zero-context take, then output only its mixed-size chapter."""
import copy
import importlib.util
import json
from pathlib import Path
import tempfile
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location(
"attributed_chapter_helpers", ROOT / "tests/_checkpoint_local_output_unit_test.py")
h = importlib.util.module_from_spec(spec)
spec.loader.exec_module(h)
chain = h.chain
def main():
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
h.h.folder_paths.output_directory = directory
run = root / "h3_chains" / "attributed_chapter"
metadata, lineage = [], []
for scene in range(1, 10):
compatibility = {
"width":1344 if scene < 8 else 960,
"height":768 if scene < 8 else 544, "fps":24,
"audio_mode":"generated_audio", "context_length":39,
"audio_context_length":39, "continuation_mode":"audio_feathered_av",
"generation_fingerprint":f"catalog-{scene}",
}
item, _ = h.h.write_revision(
run, scene, str(scene) * 32, scene, active=True,
predecessor=metadata[-1] if metadata else None,
run_name=run.name, compatibility=compatibility,
context_length=5 if scene == 9 else 0, audio_context_length=0)
metadata.append(item)
lineage.append({"scene":scene, "revision":str(scene) * 32})
previous, _ = h.h.write_revision(
run, 9, "a" * 32, 90, run_name=run.name,
predecessor=metadata[7], compatibility=metadata[8]["compatibility"],
context_length=0, audio_context_length=0)
candidate, _ = h.h.write_revision(
run, 10, "b" * 32, 100, run_name=run.name, predecessor=previous,
compatibility={**metadata[8]["compatibility"],
"generation_fingerprint":"older-catalog"},
context_length=0, audio_context_length=39, generated_continuity="on",
recovery_archive=True)
# Legacy AV saves retain a 39f audio default although the zero-video
# early return consumed neither predecessor video nor audio.
candidate["segment"].update({
"resolved_context_length":0, "resolved_audio_context_length":39,
"continuation_mode":"audio_feathered_av",
"raw_frames":141, "delivered_frames":141})
(root / candidate["segment"]["revision_metadata"]).write_text(json.dumps(candidate))
(run / "plan.json").write_text(json.dumps({
"run_name":run.name, "shots":[{"id":f"scene_{i}"} for i in range(1, 13)]}))
# Modern saves share their immutable recovery snapshot when attached.
# The alias must resolve this original snapshot, not demand a copy
# under its new lineage id or silently use today's Run-level Plan.
archived_plan = root / candidate["archives"]["plan"]
archived_plan.write_text((run / "plan.json").read_text())
(run / "plan.json").write_text('{"shots":[]}')
(run / "editorial.json").write_text(json.dumps({
"scene_order":[{"scene":i, "scene_id":f"scene_{i}"} for i in range(1, 13)],
"chapters":[{"id":"one", "start_scene_id":"scene_1"},
{"id":"two", "start_scene_id":"scene_8"}],
}))
manager = chain.CheckpointGraphManager(directory)
graph = manager.graph(run.name)
before = h.snapshot(run)
parent_branch = next(b for b in graph["branches"] if b["leaf_revision"] == "9" * 32)
assert {"scene":10, "revision":"b" * 32} in parent_branch["attribution_slot"]["candidates"]
attached = manager.attribute(run.name, 9, "9" * 32, 10, "b" * 32)
assert attached["created"]
after = h.snapshot(run)
assert len(after) == len(before) + 1, sorted(set(after) - set(before))
assert all(after[path] == value for path, value in before.items()), \
"attachment must not rewrite media, original metadata or active pointers"
alias = json.loads((run / "checkpoints" / f"clip_0010.{attached['revision']}.json").read_text())
assert alias["segment"]["segment"] == candidate["segment"]["segment"]
assert alias["segment"]["checkpoint"] == candidate["segment"]["checkpoint"]
selection = {"run_name":run.name, "output_mode":"workflow_local",
"output_scope":"chapter", "scope_start_scene":8, "scope_end_scene":12,
"lineage":[*lineage, {"scene":10, "revision":attached["revision"]}]}
# Calling the real node, not only the graph's candidate listing, is the
# regression: chapter 1 must not participate in the geometry check.
with patch.object(chain, "_atomic_json", side_effect=AssertionError("output wrote project files")):
result = chain.MiniMaxH3ChainCheckpointManager().passthrough(json.dumps(selection))[0]
assert [s["index"] for s in result["segments"]] == [8, 9, 10]
assert result["segments"][-1]["revision"] == attached["revision"]
assert result["segments"][-1]["delivered_frames"] == 141
assert result["compatibility"]["width"] == 960
assert result["archives"] == candidate["archives"]
assert h.snapshot(run) == after
# Ordinary saves still cannot claim another take's archive. An alias
# can share only the origin explicitly recorded by attribution.
def archives(value):
return chain._checkpoint_run_archives({"run_name":run.name}, value)
invalid = copy.deepcopy(alias)
invalid.pop("adoption")
h.rejected(lambda: archives(invalid), "different revision")
invalid = copy.deepcopy(alias)
invalid["adoption"]["source_revision"] = "c" * 32
h.rejected(lambda: archives(invalid), "origin is inconsistent")
invalid = copy.deepcopy(alias)
invalid["adoption"]["source_scene"] = 9
h.rejected(lambda: archives(invalid), "origin is inconsistent")
invalid = copy.deepcopy(alias)
invalid["archives"]["workflow"] = invalid["archives"]["workflow"].replace("b" * 32, "c" * 32)
h.rejected(lambda: archives(invalid), "outside revision")
h.rejected(lambda: chain._checkpoint_run_archives({"run_name":"another_run"}, alias),
"outside revision")
workflow_archive = root / alias["archives"]["workflow"]
workflow_bytes = workflow_archive.read_bytes()
workflow_archive.unlink()
h.rejected(lambda: archives(alias), "workflow is missing")
workflow_archive.write_bytes(workflow_bytes)
# Deleting an inactive original sidecar must not strand surviving
# aliases: they already carry the origin and share its artifacts.
candidate_path = root / candidate["segment"]["revision_metadata"]
candidate_bytes = candidate_path.read_bytes()
candidate_path.unlink()
assert archives(alias) == candidate["archives"]
candidate_path.write_bytes(candidate_bytes)
assert h.snapshot(run) == after
h.rejected(lambda: chain._checkpoint_selection_manifest({
**selection, "output_scope":"project"}), "compatibility")
# The unattached original and a genuinely different canvas remain
# invalid; fixing scope must not remove lineage or geometry checks.
h.rejected(lambda: chain._checkpoint_selection_manifest({
**selection, "lineage":[*lineage, {"scene":10, "revision":"b" * 32}]}),
"depends on")
alias_path = run / "checkpoints" / f"clip_0010.{attached['revision']}.json"
alias["compatibility"]["width"] = 1024
alias_path.write_text(json.dumps(alias))
h.rejected(lambda: chain._checkpoint_selection_manifest(selection), "compatible takes within this chapter")
print("Attributed chapter output: 960x544 context-free candidate after 5f take, mixed earlier chapter, immutable media and compatibility guards pass")
if __name__ == "__main__":
main()
|