File size: 3,587 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 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 | #!/usr/bin/env python3
"""Build and verify the frozen eight-source objective-only evaluation."""
from __future__ import annotations
import argparse
import hashlib
import json
from pathlib import Path
from typing import Sequence
from music3lab.eval import (
canonical_json_bytes,
evaluate_objective_only,
load_objective_config,
make_synthetic_candidate,
prepare_corpus,
publication_order,
publish_evaluation,
verify_evaluation_bundle,
)
def _parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description=(
"Freeze eight opaque audio sources, run deterministic objective "
"metrics/canaries, and publish canonical JSON plus Markdown."
)
)
parser.add_argument("--config", required=True)
parser.add_argument(
"--source",
action="append",
required=True,
help="Input audio path; pass exactly eight times.",
)
parser.add_argument("--ffmpeg", default="/usr/bin/ffmpeg")
parser.add_argument("--corpus-root", required=True)
parser.add_argument("--candidate-root", required=True)
parser.add_argument("--bundle-root", required=True)
parser.add_argument("--staging-root", required=True)
parser.add_argument(
"--corruption",
default="identity",
choices=(
"identity",
"clipping",
"noise",
"silence",
"phase_inversion",
"lowpass",
"duplicate",
"shuffle",
"nan",
"short",
"missing",
),
)
return parser
def run(arguments: argparse.Namespace) -> dict[str, object]:
loaded = load_objective_config(arguments.config)
optional = tuple(
sorted(
loaded.config.metric_policy.optional_evaluators.model_dump(
mode="json"
)
)
)
with prepare_corpus(
sources=arguments.source,
output_root=arguments.corpus_root,
ffmpeg_path=arguments.ffmpeg,
config_path=arguments.config,
) as authority:
candidate = make_synthetic_candidate(
authority,
output_root=arguments.candidate_root,
corruption=arguments.corruption,
)
result = evaluate_objective_only(
authority=authority,
candidate_root=candidate,
protected_baseline=None,
optional_evaluators=optional,
)
published = publish_evaluation(
result=result,
output_root=arguments.bundle_root,
staging_root=arguments.staging_root,
)
verified = verify_evaluation_bundle(published, authority=authority)
report = Path(published, "report.md").read_bytes()
return {
"bundle_root": str(published),
"candidate_semantic_digest": verified.candidate_semantic_digest,
"corpus_aggregate_sha256": verified.corpus_aggregate_sha256,
"optional_evaluators": verified.optional,
"promotion": verified.promotion,
"publication_order": publication_order(published),
"report_sha256": hashlib.sha256(report).hexdigest(),
"result_semantic_digest": verified.semantic_digest,
"status": verified.status,
}
def main(argv: Sequence[str] | None = None) -> int:
arguments = _parser().parse_args(argv)
print(canonical_json_bytes(run(arguments)).decode("utf-8"), end="")
return 0
if __name__ == "__main__":
raise SystemExit(main())
|