File size: 6,863 Bytes
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 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 | #!/usr/bin/env python3
"""Freeze a chronological, law-family-disjoint L3 train/development split."""
from __future__ import annotations
import argparse
import hashlib
import json
from datetime import datetime, timezone
from pathlib import Path
import pandas as pd
REPO_ROOT = Path(__file__).resolve().parents[1]
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 resolve_recorded_path(value: str, receipt_path: Path) -> Path:
candidate = Path(value)
if candidate.is_absolute():
return candidate
repo_candidate = REPO_ROOT / candidate
return repo_candidate if repo_candidate.exists() else receipt_path.parent / candidate
def audit(args: argparse.Namespace) -> dict[str, object]:
annotation = json.loads(args.annotation_audit.read_text(encoding="utf-8"))
inputs = annotation.get("inputs", {})
events_path = resolve_recorded_path(str(inputs.get("events", "")), args.annotation_audit)
events_hash_matches = bool(
events_path.is_file()
and sha256_file(events_path) == inputs.get("events_sha256")
)
annotation_authorized = bool(
annotation.get("artifact_type") == "expert_annotation_gate_audit"
and annotation.get("learned_residual_gate", {}).get("authorized") is True
and annotation.get("event_gate", {}).get("authorized") is True
and events_hash_matches
)
assignments = pd.read_csv(args.assignments, dtype=str).fillna("")
required = {"event_group_id", "split", "effective_date", "law_family"}
missing = sorted(required - set(assignments.columns))
if missing:
raise ValueError(f"split assignments are missing columns: {missing}")
duplicate_groups = bool(assignments["event_group_id"].duplicated().any())
allowed_splits = bool(assignments["split"].isin({"train", "dev"}).all())
dates = pd.to_datetime(
assignments["effective_date"], format="%Y-%m-%d", errors="coerce"
)
valid_dates = bool(dates.notna().all())
nonempty_family = bool(assignments["law_family"].str.strip().ne("").all())
gold_event_groups: set[str] = set()
if events_path.is_file():
event_frame = pd.read_csv(events_path, dtype=str).fillna("")
if {"event_group_id", "verification_status"}.issubset(event_frame.columns):
gold_event_groups = set(
event_frame.loc[
event_frame["verification_status"].eq("adjudicated_gold"),
"event_group_id",
]
)
assigned_groups = set(assignments["event_group_id"])
exact_gold_partition = bool(gold_event_groups and assigned_groups == gold_event_groups)
train = assignments.loc[assignments["split"].eq("train")].copy()
dev = assignments.loc[assignments["split"].eq("dev")].copy()
minimum_counts_pass = bool(
train["event_group_id"].nunique() >= args.minimum_train_events
and dev["event_group_id"].nunique() >= args.minimum_dev_events
)
chronological = False
if valid_dates and not train.empty and not dev.empty:
train_dates = pd.to_datetime(train["effective_date"], format="%Y-%m-%d")
dev_dates = pd.to_datetime(dev["effective_date"], format="%Y-%m-%d")
chronological = bool(train_dates.max() < dev_dates.min())
train_families = set(train["law_family"])
dev_families = set(dev["law_family"])
family_overlap = sorted(train_families & dev_families)
family_disjoint = not family_overlap
authorized = bool(
annotation_authorized
and not duplicate_groups
and allowed_splits
and valid_dates
and nonempty_family
and exact_gold_partition
and minimum_counts_pass
and chronological
and family_disjoint
)
receipt: dict[str, object] = {
"schema_version": 1,
"artifact_type": "telen_l3_event_split_audit",
"created_at": datetime.now(timezone.utc).isoformat(),
"authorized": authorized,
"inputs": {
"annotation_audit": str(args.annotation_audit),
"annotation_audit_sha256": sha256_file(args.annotation_audit),
"annotation_events": str(events_path),
"annotation_events_sha256": sha256_file(events_path)
if events_path.is_file()
else None,
"assignments": str(args.assignments),
"assignments_sha256": sha256_file(args.assignments),
},
"configuration": {
"minimum_train_events": args.minimum_train_events,
"minimum_dev_events": args.minimum_dev_events,
"final_test_included": False,
},
"counts": {
"gold_event_groups": len(gold_event_groups),
"assigned_event_groups": len(assigned_groups),
"train_event_groups": int(train["event_group_id"].nunique()),
"development_event_groups": int(dev["event_group_id"].nunique()),
},
"gates": {
"expert_annotation_authorized": annotation_authorized,
"unique_event_assignment": not duplicate_groups,
"train_dev_only": allowed_splits,
"valid_effective_dates": valid_dates,
"nonempty_law_family": nonempty_family,
"exact_gold_event_partition": exact_gold_partition,
"minimum_event_counts": minimum_counts_pass,
"strict_chronology": chronological,
"law_family_disjoint": family_disjoint,
"final_test_sealed": True,
},
"law_family_overlap": family_overlap,
"rule": (
"authorized expert events AND every gold event assigned exactly once "
"to train/dev AND max(train effective_date) < min(dev effective_date) "
"AND no law_family overlap AND minimum event counts pass"
),
}
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(
json.dumps(receipt, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
)
return receipt
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--annotation-audit", type=Path, required=True)
parser.add_argument("--assignments", type=Path, required=True)
parser.add_argument("--minimum-train-events", type=int, default=5)
parser.add_argument("--minimum-dev-events", type=int, default=5)
parser.add_argument("--output", type=Path, required=True)
return parser.parse_args()
def main() -> None:
result = audit(parse_args())
print(json.dumps(result, ensure_ascii=False, indent=2))
if not result["authorized"]:
raise SystemExit(2)
if __name__ == "__main__":
main()
|