Spaces:
Running
Running
| from __future__ import annotations | |
| import json | |
| from dataclasses import dataclass | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| from typing import Dict, List | |
| class Signal: | |
| name: str | |
| weight: int | |
| SIGNALS: List[Signal] = [ | |
| Signal("has_tests", 10), | |
| Signal("has_validation_script", 10), | |
| Signal("has_greedy_baseline", 10), | |
| Signal("has_training_pipeline", 15), | |
| Signal("has_submission_bundle_script", 15), | |
| Signal("has_judge_brief", 15), | |
| Signal("has_replay_endpoint_docs", 10), | |
| Signal("has_judge_pack_endpoint_docs", 10), | |
| Signal("has_tuned_metrics_artifact", 15), | |
| ] | |
| PROJECT_DIR_CANDIDATES = [ | |
| "gdpr_eraseops", | |
| "incident-commander", | |
| "llmfleet-sre-main", | |
| "quantam_main", | |
| "finops_arb_env", | |
| ] | |
| def _read_text_if_exists(path: Path) -> str: | |
| if not path.exists(): | |
| return "" | |
| return path.read_text(encoding="utf-8", errors="ignore") | |
| def _detect_signals(project_root: Path) -> Dict[str, bool]: | |
| readme = _read_text_if_exists(project_root / "README.md") | |
| scripts_dir = project_root / "scripts" | |
| artifacts_dir = project_root / "artifacts" | |
| return { | |
| "has_tests": (project_root / "tests").exists(), | |
| "has_validation_script": (scripts_dir / "validate_project.py").exists() | |
| or (scripts_dir / "validate-submission.sh").exists() | |
| or (scripts_dir / "validate-submission.ps1").exists(), | |
| "has_greedy_baseline": (project_root / "greedy_baseline.py").exists(), | |
| "has_training_pipeline": (scripts_dir / "train_sb3_agent.py").exists(), | |
| "has_submission_bundle_script": (scripts_dir / "generate_submission_bundle.py").exists(), | |
| "has_judge_brief": (project_root / "JUDGE_BRIEF.md").exists(), | |
| "has_replay_endpoint_docs": "/replay" in readme, | |
| "has_judge_pack_endpoint_docs": "/judge_pack" in readme, | |
| "has_tuned_metrics_artifact": (artifacts_dir / "ppo_tuned_v3" / "evaluation_summary.json").exists(), | |
| } | |
| def _score(signals: Dict[str, bool]) -> int: | |
| total = 0 | |
| for signal in SIGNALS: | |
| if signals.get(signal.name, False): | |
| total += signal.weight | |
| return total | |
| def _project_kind(project_name: str) -> str: | |
| if project_name == "gdpr_eraseops": | |
| return "target" | |
| return "peer" | |
| def main() -> None: | |
| script_path = Path(__file__).resolve() | |
| gdpr_root = script_path.parents[1] | |
| workspace_root = gdpr_root.parent | |
| results = [] | |
| for candidate in PROJECT_DIR_CANDIDATES: | |
| project_root = workspace_root / candidate | |
| if not project_root.exists() or not project_root.is_dir(): | |
| continue | |
| signals = _detect_signals(project_root) | |
| score = _score(signals) | |
| results.append( | |
| { | |
| "project": candidate, | |
| "kind": _project_kind(candidate), | |
| "score": score, | |
| "signals": signals, | |
| } | |
| ) | |
| results.sort(key=lambda item: item["score"], reverse=True) | |
| output_dir = gdpr_root / "artifacts" / "readiness" | |
| output_dir.mkdir(parents=True, exist_ok=True) | |
| report = { | |
| "generated_at": datetime.now(timezone.utc).isoformat(), | |
| "rubric": [{"name": signal.name, "weight": signal.weight} for signal in SIGNALS], | |
| "results": results, | |
| } | |
| json_path = output_dir / "cohort_comparison.json" | |
| json_path.write_text(json.dumps(report, indent=2), encoding="utf-8") | |
| md_lines = [ | |
| "# Cohort Readiness Comparison", | |
| "", | |
| f"Generated at: {report['generated_at']}", | |
| "", | |
| "## Leaderboard", | |
| "", | |
| "| Rank | Project | Score |", | |
| "|---:|---|---:|", | |
| ] | |
| for idx, row in enumerate(results, start=1): | |
| md_lines.append(f"| {idx} | {row['project']} | {row['score']} |") | |
| md_lines.extend( | |
| [ | |
| "", | |
| "## Signals", | |
| "", | |
| "Legend: Y = signal present, N = signal absent", | |
| "", | |
| ] | |
| ) | |
| header = "| Project | " + " | ".join(signal.name for signal in SIGNALS) + " |" | |
| separator = "|---|" + "|".join(["---"] * len(SIGNALS)) + "|" | |
| md_lines.append(header) | |
| md_lines.append(separator) | |
| for row in results: | |
| cells = ["Y" if row["signals"].get(signal.name, False) else "N" for signal in SIGNALS] | |
| md_lines.append("| " + row["project"] + " | " + " | ".join(cells) + " |") | |
| md_lines.extend( | |
| [ | |
| "", | |
| "## Recommendation", | |
| "", | |
| "Use this report alongside JUDGE_BRIEF.md and submission_report.md during review.", | |
| ] | |
| ) | |
| md_path = output_dir / "cohort_comparison.md" | |
| md_path.write_text("\n".join(md_lines) + "\n", encoding="utf-8") | |
| print(json.dumps({"json": str(json_path), "markdown": str(md_path), "leader": results[0] if results else None}, indent=2)) | |
| if __name__ == "__main__": | |
| main() | |