108 lines
4.4 KiB
Python
108 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
TEMP = ROOT / "Temp"
|
|
OUTPUT = TEMP / "wbs_4_1_7_1_status_v1.json"
|
|
|
|
|
|
def _load_json(path: Path) -> dict[str, Any]:
|
|
if not path.exists():
|
|
return {}
|
|
try:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
return payload if isinstance(payload, dict) else {}
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def _pick_top_candidates(priority: dict[str, Any], limit: int = 5) -> list[dict[str, Any]]:
|
|
rows = priority.get("priority_list") if isinstance(priority.get("priority_list"), list) else []
|
|
out: list[dict[str, Any]] = []
|
|
for row in rows[:limit]:
|
|
if isinstance(row, dict):
|
|
out.append(
|
|
{
|
|
"calibration_id": row.get("calibration_id"),
|
|
"source": row.get("source"),
|
|
"sample_n": row.get("sample_n"),
|
|
"urgency_score": row.get("urgency_score"),
|
|
"current_value": row.get("current_value"),
|
|
"owner_formula": row.get("owner_formula"),
|
|
}
|
|
)
|
|
return out
|
|
|
|
|
|
def main() -> int:
|
|
live_gate = _load_json(TEMP / "live_data_activation_gate_v1.json")
|
|
op_queue = _load_json(TEMP / "operational_eval_queue_v1.json")
|
|
calib_reg = _load_json(TEMP / "calibration_registry_v1.json")
|
|
calib_priority = _load_json(TEMP / "calibration_priority_v1.json")
|
|
pred_acc = _load_json(TEMP / "prediction_accuracy_harness_v2.json")
|
|
|
|
live_t20_count = int(live_gate.get("live_t20_count") or 0)
|
|
live_t20_threshold = int(live_gate.get("live_t20_threshold") or 30)
|
|
live_progress_pct = float(live_gate.get("progress_pct") or 0.0)
|
|
live_gate_state = str(live_gate.get("gate") or "PENDING")
|
|
|
|
eval_metrics = op_queue.get("metrics") if isinstance(op_queue.get("metrics"), dict) else {}
|
|
records_total = int(eval_metrics.get("records_total") or 0)
|
|
t20_evaluated_count = int(eval_metrics.get("t20_evaluated_count") or 0)
|
|
t20_due_capture_count = int(eval_metrics.get("t20_due_capture_count") or 0)
|
|
|
|
reg_counts = {
|
|
"total_thresholds": int(calib_reg.get("total_thresholds") or 0),
|
|
"calibrated_count": int(calib_reg.get("calibrated_count") or 0),
|
|
"provisional_count": int(calib_reg.get("provisional_count") or 0),
|
|
"expert_prior_count": int(calib_reg.get("expert_prior_count") or 0),
|
|
"spec_derived_count": int(calib_reg.get("spec_derived_count") or 0),
|
|
"unregistered_threshold_count": int(calib_reg.get("unregistered_threshold_count") or 0),
|
|
"overclaimed_count": int(calib_reg.get("overclaimed_count") or 0),
|
|
}
|
|
|
|
pred_t5_sample = int(pred_acc.get("t5_sample") or 0)
|
|
pred_t5_rate = pred_acc.get("t5_op_rate")
|
|
pred_t5_state = "DATA_GATED" if pred_t5_sample == 0 else "OK"
|
|
|
|
status = {
|
|
"formula_id": "WBS_4_1_7_1_STATUS_V1",
|
|
"wbs_4_1": {
|
|
"live_t20_count": live_t20_count,
|
|
"live_t20_threshold": live_t20_threshold,
|
|
"live_progress_pct": live_progress_pct,
|
|
"live_gate": live_gate_state,
|
|
"records_total": records_total,
|
|
"t20_evaluated_count": t20_evaluated_count,
|
|
"t20_due_capture_count": t20_due_capture_count,
|
|
"operational_queue_state": "EMPTY" if records_total == 0 else "POPULATED",
|
|
},
|
|
"wbs_7_1": {
|
|
"registry_counts": reg_counts,
|
|
"prediction_accuracy_state": pred_t5_state,
|
|
"prediction_t5_sample": pred_t5_sample,
|
|
"prediction_t5_rate_pct": pred_t5_rate,
|
|
"top_priority_candidates": _pick_top_candidates(calib_priority, 5),
|
|
},
|
|
"summary": {
|
|
"wbs_4_1_remaining_to_threshold": max(0, live_t20_threshold - live_t20_count),
|
|
"wbs_7_1_calibrated_pct": round(100.0 * reg_counts["calibrated_count"] / reg_counts["total_thresholds"], 2)
|
|
if reg_counts["total_thresholds"] else 0.0,
|
|
"wbs_7_1_unvalidated_pct": round(100.0 * (reg_counts["spec_derived_count"] + reg_counts["expert_prior_count"]) / reg_counts["total_thresholds"], 2)
|
|
if reg_counts["total_thresholds"] else 0.0,
|
|
},
|
|
}
|
|
|
|
OUTPUT.write_text(json.dumps(status, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
print(json.dumps(status, ensure_ascii=False, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|