| |
| """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()) |
|
|
