File size: 2,661 Bytes
90884df | 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 | #!/usr/bin/env python3
"""Run or verify Tier-A captured-state continuation."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
from music3lab.editing.continue_capture import ArbitraryWavContinuationUnavailableError
from music3lab.editing.music3_continuation import (
run_captured_continuation,
verify_continuation_bundle,
)
def parser() -> argparse.ArgumentParser:
result = argparse.ArgumentParser()
commands = result.add_subparsers(dest="command", required=True)
run = commands.add_parser("run")
for name in (
"config",
"snapshot",
"base-manifest",
"diffusers-root",
"phase0-artifacts",
"output-root",
):
run.add_argument(f"--{name}", required=True)
verify = commands.add_parser("verify")
verify.add_argument("--output-root", required=True)
blocked = commands.add_parser("arbitrary-wav")
blocked.add_argument("--wav", required=True)
return result
def main(argv: list[str] | None = None) -> int:
args = parser().parse_args(argv)
if args.command == "run":
verified = run_captured_continuation(
config_path=Path(args.config),
snapshot=Path(args.snapshot),
base_manifest=Path(args.base_manifest),
diffusers_root=Path(args.diffusers_root),
phase0_artifacts=Path(args.phase0_artifacts),
output_root=Path(args.output_root),
)
print(json.dumps({
"captured_state_continuation_pass": verified["metrics"][
"captured_state_continuation_pass"
],
"metrics_semantic_digest": verified["metrics"]["semantic_digest"],
"output_root": str(Path(args.output_root).absolute()),
}, sort_keys=True))
return 0
if args.command == "verify":
verified = verify_continuation_bundle(Path(args.output_root))
print(json.dumps({
"captured_state_continuation_pass": True,
"metrics_semantic_digest": verified["metrics"]["semantic_digest"],
}, sort_keys=True))
return 0
try:
raise ArbitraryWavContinuationUnavailableError(
"arbitrary WAV continuation is BLOCKED; a verified Phase-0 "
"native capture is required"
)
except ArbitraryWavContinuationUnavailableError as error:
print(json.dumps({
"capability": "arbitrary_wav_continuation",
"status": "BLOCKED",
"detail": str(error),
"wav": args.wav,
}, sort_keys=True))
return 2
if __name__ == "__main__":
raise SystemExit(main())
|