File size: 4,360 Bytes
1d842e3 baa1d92 1d842e3 baa1d92 1d842e3 baa1d92 1d842e3 baa1d92 1d842e3 | 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 | #!/usr/bin/env python3
"""Fail closed unless expert event and qrel annotation gates both pass."""
from __future__ import annotations
import argparse
import hashlib
import json
import sys
from datetime import datetime, timezone
from pathlib import Path
import pandas as pd
REPO_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO_ROOT / "src"))
from telen.annotation import ( # noqa: E402
PROTOCOL_VERSION,
audit_event_annotations,
audit_event_qrel_links,
audit_question_annotations,
audit_question_event_links,
audit_qrel_annotations,
)
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for block in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()
def run(args: argparse.Namespace) -> dict:
event_frame = pd.read_csv(args.events, dtype=str).fillna("")
question_frame = pd.read_csv(args.questions, dtype=str).fillna("")
qrel_frame = pd.read_csv(args.qrels, dtype=str).fillna("")
event_audit = audit_event_annotations(
event_frame, minimum_kappa=args.minimum_event_kappa
)
qrel_audit = audit_qrel_annotations(
qrel_frame,
minimum_weighted_kappa=args.minimum_qrel_kappa,
minimum_event_groups=args.minimum_event_groups,
)
question_audit = audit_question_annotations(
question_frame, minimum_kappa=args.minimum_question_kappa
)
question_link_audit = audit_question_event_links(event_frame, question_frame)
package_link_audit = audit_event_qrel_links(event_frame, qrel_frame)
learned_authorized = bool(
event_audit.authorized
and question_audit.authorized
and question_link_audit["authorized"]
and qrel_audit.authorized
and package_link_audit["authorized"]
)
result = {
"schema_version": 2,
"artifact_type": "expert_annotation_gate_audit",
"created_at": datetime.now(timezone.utc).isoformat(),
"protocol_version": PROTOCOL_VERSION,
"inputs": {
"events": str(args.events),
"events_sha256": sha256_file(args.events),
"questions": str(args.questions),
"questions_sha256": sha256_file(args.questions),
"qrels": str(args.qrels),
"qrels_sha256": sha256_file(args.qrels),
},
"event_gate": event_audit.to_dict(),
"question_gate": question_audit.to_dict(),
"question_event_link_gate": question_link_audit,
"qrel_gate": qrel_audit.to_dict(),
"package_link_gate": package_link_audit,
"learned_residual_gate": {
"authorized": learned_authorized,
"rule": (
"event_gate.authorized AND question_gate.authorized AND "
"question_event_link_gate.authorized AND qrel_gate.authorized "
"AND package_link_gate.authorized"
),
},
}
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(
json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
)
return result
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser()
root = REPO_ROOT / "tmp/telen/temporal-pilot"
parser.add_argument(
"--events", type=Path, default=root / "official_verification_template.csv"
)
parser.add_argument(
"--qrels", type=Path, default=root / "qrel_judgment_template.csv"
)
parser.add_argument(
"--questions", type=Path, default=root / "expert_annotation_template.csv"
)
parser.add_argument(
"--output",
type=Path,
default=REPO_ROOT / "tmp/telen/expert-annotation-gate-audit.json",
)
parser.add_argument("--minimum-event-kappa", type=float, default=0.80)
parser.add_argument("--minimum-qrel-kappa", type=float, default=0.70)
parser.add_argument("--minimum-question-kappa", type=float, default=0.70)
parser.add_argument("--minimum-event-groups", type=int, default=5)
return parser.parse_args()
def main() -> None:
args = parse_args()
result = run(args)
print(json.dumps(result, ensure_ascii=False, indent=2))
if not result["learned_residual_gate"]["authorized"]:
raise SystemExit(2)
if __name__ == "__main__":
main()
|