| """Kod bazında gelişim: aynı kodun (hazır örnek, eklenen dosya ya da yapıştırılan kod) tüm koşuları bir arada. |
| |
| Koşular kodun içeriğine göre gruplanır (dil + boşlukları kırpılmış kod → kısa SHA-1). Kod değiştirilirse yeni bir |
| grup oluşur; böylece her grup tam olarak aynı kod üzerindeki denemeleri karşılaştırır. Çalışma alanı, Otomasyon, |
| Değerlendirme ve Deneyler'deki koşuların hepsi sayılır. |
| """ |
| from __future__ import annotations |
|
|
| import hashlib |
| import re |
| from collections import defaultdict |
| from typing import Any, Optional |
|
|
| from . import store |
| from .eval import benchmark |
| from .llm.providers import provider_of |
| from .models import JobState |
|
|
| PREFIX = re.compile(r"^\[(bench|oto|auto)\]\s*", re.I) |
|
|
|
|
| def code_key(language: str, code: str) -> str: |
| norm = "\n".join(line.rstrip() for line in (code or "").strip().splitlines()) |
| return hashlib.sha1(f"{language}\n{norm}".encode("utf-8")).hexdigest()[:12] |
|
|
|
|
| def _known() -> dict[str, dict[str, Any]]: |
| """Kod anahtarı → hazır örnek ya da kitaplık öğesi (başlık ve kimlik için).""" |
| out: dict[str, dict[str, Any]] = {} |
| for it in benchmark.load_manifest(): |
| out[code_key(it["language"], it["code"])] = {"source": "sample", "sample_id": it["id"], "title": it["title"], |
| "title_en": it.get("title_en") or it["title"]} |
| try: |
| from .automation import LIB_PREFIX, library |
| for d in library(): |
| out.setdefault(code_key(d["language"], d["code"]), {"source": "library", "sample_id": LIB_PREFIX + d["id"], |
| "title": d["name"], "title_en": d["name"]}) |
| except Exception: |
| pass |
| return out |
|
|
|
|
| def _origin(job: JobState) -> str: |
| m = PREFIX.match(job.request.title or "") |
| if not m: |
| return "workspace" |
| return "evaluation" if m.group(1).lower() == "bench" else "automation" |
|
|
|
|
| def _row(job: JobState) -> dict[str, Any]: |
| r = job.result |
| fr = r.final_run if r else None |
| tokens = sum(int(e.data.get("tokens_out") or 0) for e in job.events if e.type == "llm") |
| model = job.request.model or "" |
| return {"job_id": job.id, "created_at": job.created_at, "status": job.status, "origin": _origin(job), |
| "model": model or None, "provider": provider_of(model) if model else "ollama", |
| "passed": fr.passed if fr else None, "total": fr.total if fr else None, |
| "coverage": fr.coverage.percent if fr else None, |
| "mutation": r.mutation.score if r and r.mutation else None, |
| "mutation_before": r.mutation_before.score if r and r.mutation_before else None, |
| "iterations": len(r.iterations) if r else 0, "duration_s": round(r.duration_s, 1) if r else None, |
| "tokens_out": tokens or None, "energy_wh": (r.compute or {}).get("energy_wh") if r else None, |
| "max_iterations": job.request.max_iterations, "mutation_rounds": job.request.mutation_rounds, |
| "title": PREFIX.sub("", job.request.title or "") or None, |
| "summary": (r.analysis.summary if r else "")[:160]} |
|
|
|
|
| def _green(x: dict[str, Any]) -> bool: |
| return x["status"] == "done" and bool(x["total"]) and x["passed"] == x["total"] |
|
|
|
|
| def _best(rows: list[dict[str, Any]]) -> Optional[dict[str, Any]]: |
| """En iyi koşu: önce tüm testleri geçenler, sonra mutasyon skoru, sonra kapsama.""" |
| done = [x for x in rows if x["status"] == "done" and x["total"]] |
| if not done: |
| return None |
| return max(done, key=lambda x: (_green(x), x["mutation"] or -1, x["coverage"] or -1, -(x["duration_s"] or 0))) |
|
|
|
|
| def _avg(xs: list[Optional[float]]) -> Optional[float]: |
| v = [x for x in xs if x is not None] |
| return round(sum(v) / len(v), 2) if v else None |
|
|
|
|
| def _groups() -> dict[str, dict[str, Any]]: |
| known = _known() |
| groups: dict[str, dict[str, Any]] = {} |
| for job in store.all_jobs(): |
| req = job.request |
| key = code_key(req.language, req.code) |
| g = groups.get(key) |
| if g is None: |
| k = known.get(key, {}) |
| g = groups[key] = {"key": key, "language": req.language, "kind": req.test_kind, "source": k.get("source", "custom"), |
| "sample_id": k.get("sample_id"), "title": k.get("title"), "title_en": k.get("title_en"), |
| "source_name": req.source_name, "lines": req.code.strip().count("\n") + 1, "rows": []} |
| g["rows"].append(_row(job)) |
| for g in groups.values(): |
| g["rows"].sort(key=lambda x: x["created_at"]) |
| if not g["title"]: |
| last = next((x for x in reversed(g["rows"]) if x["title"] or x["summary"]), None) |
| g["title"] = g["title_en"] = g["source_name"] or (last and (last["title"] or last["summary"][:70])) or g["key"] |
| return groups |
|
|
|
|
| def overview() -> list[dict[str, Any]]: |
| out = [] |
| for g in _groups().values(): |
| rows = g["rows"] |
| best = _best(rows) |
| done = [x for x in rows if x["status"] == "done" and x["total"]] |
| out.append({k: g[k] for k in ("key", "language", "kind", "source", "sample_id", "title", "title_en", "lines")} |
| | {"runs": len(rows), "first_at": rows[0]["created_at"], "last_at": rows[-1]["created_at"], |
| "green": sum(_green(x) for x in rows), "models": sorted({x["model"] or "" for x in rows} - {""}), |
| "best": {k: best[k] for k in ("job_id", "coverage", "mutation", "passed", "total", "model")} if best else None, |
| "last": {k: rows[-1][k] for k in ("job_id", "status", "coverage", "mutation", "passed", "total", "model")}, |
| "trend": [x["mutation"] if x["mutation"] is not None else x["coverage"] for x in done][-12:]}) |
| return sorted(out, key=lambda g: g["last_at"], reverse=True) |
|
|
|
|
| def detail(key: str) -> Optional[dict[str, Any]]: |
| g = _groups().get(key) |
| if not g: |
| return None |
| rows = g["rows"] |
| by_model: dict[str, list[dict[str, Any]]] = defaultdict(list) |
| for x in rows: |
| by_model[x["model"] or "—"].append(x) |
| models = [] |
| for m, xs in by_model.items(): |
| dn = [x for x in xs if x["status"] == "done" and x["total"]] |
| models.append({"model": m, "runs": len(xs), "green": sum(_green(x) for x in xs), |
| "failed": sum(1 for x in xs if x["status"] == "failed"), |
| "avg_coverage": _avg([x["coverage"] for x in dn]), "avg_mutation": _avg([x["mutation"] for x in dn]), |
| "best_mutation": max((x["mutation"] for x in dn if x["mutation"] is not None), default=None), |
| "avg_duration_s": _avg([x["duration_s"] for x in dn])}) |
| models.sort(key=lambda m: (-(m["avg_mutation"] or -1), -m["runs"])) |
| best = _best(rows) |
| dn = [x for x in rows if x["status"] == "done" and x["total"]] |
| return {**{k: v for k, v in g.items() if k != "rows"}, "runs": rows, "models": models, |
| "best": best, "stats": {"runs": len(rows), "green": sum(_green(x) for x in rows), |
| "failed": sum(1 for x in rows if x["status"] == "failed"), |
| "best_coverage": max((x["coverage"] for x in dn if x["coverage"] is not None), default=None), |
| "best_mutation": max((x["mutation"] for x in dn if x["mutation"] is not None), default=None), |
| "avg_duration_s": _avg([x["duration_s"] for x in dn])}} |
|
|
