129 lines
3.7 KiB
Python
129 lines
3.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate and evaluate one independent human-held-out G7 voice-gain pack.
|
|
|
|
The command emits only aggregate metrics and PII-safe JSON pointers. It never
|
|
echoes input paths, participant/labeler keys, labels, or raw validation values.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
|
|
from pydantic import ValidationError
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
API_ROOT = REPO_ROOT / "apps/api"
|
|
if str(API_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(API_ROOT))
|
|
|
|
from app.contracts.g7_external_evidence import ( # noqa: E402
|
|
G7HumanVoiceGainEvidencePack,
|
|
)
|
|
from app.services.g7_voice_gain_evidence import ( # noqa: E402
|
|
evaluate_human_voice_gain,
|
|
)
|
|
|
|
|
|
def _json_pointer(location: tuple[int | str, ...]) -> str:
|
|
if not location:
|
|
return "/"
|
|
parts = []
|
|
for item in location:
|
|
value = str(item).replace("~", "~0").replace("/", "~1")
|
|
parts.append(value)
|
|
return "/" + "/".join(parts)
|
|
|
|
|
|
def _base_report() -> dict[str, Any]:
|
|
return {
|
|
"schema_version": "vignette.g7-human-voice-gain-check.v1",
|
|
"passed": False,
|
|
"clinical_claim_allowed": False,
|
|
"privacy_boundary": {
|
|
"input_path_logged": False,
|
|
"participant_keys_logged": False,
|
|
"labeler_keys_logged": False,
|
|
"labels_logged": False,
|
|
"raw_validation_values_logged": False,
|
|
},
|
|
"validation_errors": [],
|
|
"result": {},
|
|
}
|
|
|
|
|
|
def validate_payload(payload: object) -> dict[str, Any]:
|
|
report = _base_report()
|
|
try:
|
|
pack = G7HumanVoiceGainEvidencePack.model_validate(payload)
|
|
except ValidationError as exc:
|
|
report["validation_errors"] = [
|
|
{
|
|
"pointer": _json_pointer(tuple(item["loc"])),
|
|
"type": item["type"],
|
|
}
|
|
for item in exc.errors(
|
|
include_url=False,
|
|
include_context=False,
|
|
include_input=False,
|
|
)
|
|
]
|
|
return report
|
|
|
|
try:
|
|
result = evaluate_human_voice_gain(pack)
|
|
except Exception as exc:
|
|
report["validation_errors"] = [
|
|
{"pointer": "/", "type": f"evaluation:{type(exc).__name__}"}
|
|
]
|
|
return report
|
|
|
|
report["passed"] = result.passed
|
|
report["result"] = result.model_dump(mode="json")
|
|
return report
|
|
|
|
|
|
def parser() -> argparse.ArgumentParser:
|
|
result = argparse.ArgumentParser(description=__doc__)
|
|
mode = result.add_mutually_exclusive_group(required=True)
|
|
mode.add_argument("--input", type=Path, help="deidentified human pack JSON")
|
|
mode.add_argument(
|
|
"--print-schema",
|
|
action="store_true",
|
|
help="print the authoritative JSON Schema and exit",
|
|
)
|
|
return result
|
|
|
|
|
|
def main(argv: Iterable[str] | None = None) -> int:
|
|
args = parser().parse_args(list(argv) if argv is not None else None)
|
|
if args.print_schema:
|
|
print(
|
|
json.dumps(
|
|
G7HumanVoiceGainEvidencePack.model_json_schema(),
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
sort_keys=True,
|
|
)
|
|
)
|
|
return 0
|
|
|
|
try:
|
|
payload = json.loads(args.input.read_text(encoding="utf-8"))
|
|
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
|
|
report = _base_report()
|
|
report["validation_errors"] = [
|
|
{"pointer": "/", "type": f"input:{type(exc).__name__}"}
|
|
]
|
|
else:
|
|
report = validate_payload(payload)
|
|
print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True))
|
|
return 0 if report["passed"] is True else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|