src/services/reporting.py

"""Optimisation report + Model Performance Certificate (spec §7).
JSON always; CSV flat metrics; PDF via reportlab (lazy import)."""

import csv
import io
import json
import platform
import time


def _lock_fingerprint():
    """Runtime dependency fingerprint for the environment block (spec §7.2)."""
    import hashlib
    vers = []
    for mod in ("gradio", "transformers", "peft", "trl", "torch", "huggingface_hub", "pydantic"):
        try:
            vers.append(f"{mod}=={__import__(mod).__version__}")
        except Exception:  # noqa: BLE001
            vers.append(f"{mod}=absent")
    return {"packages": vers,
            "fingerprint": hashlib.sha256(";".join(vers).encode()).hexdigest()[:12],
            "python": platform.python_version()}


def environment_block(manifest, backend_used: str, accelerator: str, quant: str,
                      seed: int, n_items: int, demo_run: bool) -> dict:
    return {"backend": backend_used, "accelerator": accelerator, "quantization": quant,
            "model_revision": manifest.model_revision,
            "dataset_fingerprint": manifest.dataset_fingerprint,
            "dependency_lock": _lock_fingerprint(),
            "evaluation_seed": seed, "sample_size": n_items,
            "run_type": "demo" if demo_run else "full"}


def confidence_stars(baseline: dict, post: dict) -> tuple[int, str]:
    """Reflects evaluation comprehensiveness, never model quality (spec §7.2.4)."""
    score = 0
    n = min(baseline.get("n_items", 0), post.get("n_items", 0))
    if n >= 25: score += 1
    if n >= 100: score += 1
    if n >= 200: score += 1
    if baseline.get("seed") == post.get("seed"): score += 1
    if baseline.get("full_benchmark_executed") and post.get("full_benchmark_executed"): score += 1
    rubric = (f"n={n} paired items; same seed: {baseline.get('seed') == post.get('seed')}; "
              f"full benchmark: {'yes' if score == 5 else 'no'}. "
              "Stars reflect evaluation comprehensiveness, not model quality.")
    return max(score, 1), rubric


def strengths_weaknesses(comparison: dict) -> tuple[list, list]:
    s, w = [], []
    nice = {"accuracy": "factual accuracy", "bleu": "BLEU overlap", "rougeL": "ROUGE-L coverage",
            "token_f1": "answer consistency", "unsupported_claim_rate": "hallucination estimate",
            "latency_s": "response latency"}
    for r in comparison["rows"]:
        if not r["significant"]:
            continue
        label = nice.get(r["metric"], r["metric"])
        pct = f"{abs(r['change']):.3f}"
        if r["direction"] == "improved":
            s.append(f"Improved {label} ({'+' if r['change'] > 0 else '-'}{pct})")
        elif r["direction"] == "degraded":
            w.append(f"Worse {label} ({r['change']:+.3f})")
    return s or ["No statistically significant strengths detected"], \
           w or ["No statistically significant weaknesses detected"]


def deployment_recommendation(overall: str, diags: list, n_samples: int) -> str:
    if overall == "Improved":
        return "Ready for Deployment"
    if overall == "Degraded":
        return "Do Not Deploy"
    if any(d["reason"] == "Dataset too small" for d in diags) or n_samples < 500:
        return "Needs Better Dataset"
    return "Needs More Training"


def build_certificate(manifest, ds_summary, training_log, baseline, post,
                      comparison, diags, env, hardware_rows) -> dict:
    stars, rubric = confidence_stars(baseline, post)
    s, w = strengths_weaknesses(comparison)
    rec = deployment_recommendation(comparison["overall"], diags, ds_summary.get("samples", 0))
    summary_lines = []
    for r in comparison["rows"]:
        if r["significant"]:
            summary_lines.append(f"{r['metric']}: {r['baseline']:.3f} → {r['finetuned']:.3f} "
                                 f"({r['change']:+.3f}, p={r['p_value']})")
    if not summary_lines:
        summary_lines.append("No statistically significant metric changes at alpha=0.05.")
    summary_lines.append(f"Overall recommendation: {rec}.")
    return {
        "title": "MODEL PERFORMANCE CERTIFICATE",
        "platform": "MLOL — MultiDomain LLM Optimisation Lab",
        "section_1_identity": {
            "model": manifest.title or manifest.run_id, "base_model": manifest.model_repo,
            "adapter": training_log.get("adapter_dir") or "(demo/mock run)",
            "date": time.strftime("%Y-%m-%d"),
            "training_time_s": training_log.get("train_seconds"),
            "dataset": ds_summary.get("source_file"),
            "dataset_fingerprint": manifest.dataset_fingerprint},
        "section_2_performance": {k: post["metrics"].get(k) for k in
                                  ("accuracy", "bleu", "rougeL", "token_f1", "latency_s")} |
                                 {"hallucination_estimate": post["hallucination_estimate"]},
        "section_3_overall": {"Improved": "✓ Improved", "Neutral": "⚠ Neutral",
                              "Degraded": "✗ Degraded"}[comparison["overall"]],
        "section_4_confidence": {"stars": "★" * stars + "☆" * (5 - stars), "rubric": rubric},
        "section_5_strengths": s,
        "section_6_weaknesses": w,
        "section_7_deployment": rec,
        "section_8_hardware": hardware_rows,
        "section_9_research_summary": summary_lines,
        "environment": env,
        "statistical_note": comparison["method"] + f"; {comparison['n_paired_items']} paired items; "
                            "sampled evaluation — full benchmark: "
                            + ("executed" if post.get("full_benchmark_executed") else "NOT executed"),
    }


def certificate_csv(cert: dict) -> str:
    buf = io.StringIO()
    w = csv.writer(buf)
    w.writerow(["field", "value"])
    for k, v in cert["section_1_identity"].items():
        w.writerow([k, v])
    for k, v in cert["section_2_performance"].items():
        w.writerow([k, json.dumps(v)])
    w.writerow(["overall", cert["section_3_overall"]])
    w.writerow(["confidence", cert["section_4_confidence"]["stars"]])
    w.writerow(["deployment", cert["section_7_deployment"]])
    return buf.getvalue()


def certificate_pdf(cert: dict) -> bytes:
    from reportlab.lib.pagesizes import A4  # lazy
    from reportlab.lib.styles import getSampleStyleSheet
    from reportlab.lib.units import cm
    from reportlab.platypus import Paragraph, SimpleDocTemplate, Spacer, Table, TableStyle
    from reportlab.lib import colors

    buf = io.BytesIO()
    doc = SimpleDocTemplate(buf, pagesize=A4, topMargin=1.5 * cm, bottomMargin=1.5 * cm)
    ss = getSampleStyleSheet()
    el = [Paragraph(cert["title"], ss["Title"]),
          Paragraph(cert["platform"], ss["Italic"]), Spacer(1, 12)]

    def sec(title, rows):
        el.append(Paragraph(title, ss["Heading2"]))
        t = Table(rows, colWidths=[6 * cm, 10 * cm])
        t.setStyle(TableStyle([("GRID", (0, 0), (-1, -1), 0.4, colors.grey),
                               ("FONTSIZE", (0, 0), (-1, -1), 8),
                               ("VALIGN", (0, 0), (-1, -1), "TOP")]))
        el.append(t)
        el.append(Spacer(1, 8))

    sec("1 · Identity", [[k, str(v)] for k, v in cert["section_1_identity"].items()])
    perf = []
    for k, v in cert["section_2_performance"].items():
        if isinstance(v, dict) and "mean" in v:
            perf.append([k, f"{v['mean']} (95% CI {v['ci_low']}–{v['ci_high']}, n={v['n']})"])
        else:
            perf.append([k, json.dumps(v)[:220]])
    sec("2 · Performance", perf)
    sec("3–4 · Result & Confidence", [["Overall", cert["section_3_overall"]],
                                      ["Confidence", cert["section_4_confidence"]["stars"]],
                                      ["Rubric", cert["section_4_confidence"]["rubric"]]])
    sec("5 · Strengths", [[str(i + 1), s] for i, s in enumerate(cert["section_5_strengths"])])
    sec("6 · Weaknesses", [[str(i + 1), s] for i, s in enumerate(cert["section_6_weaknesses"])])
    sec("7 · Deployment", [["Recommendation", cert["section_7_deployment"]]])
    sec("8 · Hardware", [[r["hardware"], f"{r['verdict']} — {r['note']}"] for r in cert["section_8_hardware"]])
    sec("9 · Research Summary", [[str(i + 1), s] for i, s in enumerate(cert["section_9_research_summary"])])
    env = cert["environment"]
    sec("Environment", [["backend", env["backend"]], ["accelerator", env["accelerator"]],
                        ["quantization", env["quantization"]],
                        ["model revision", env["model_revision"]],
                        ["dataset fingerprint", env["dataset_fingerprint"]],
                        ["dependency lock", env["dependency_lock"]["fingerprint"]],
                        ["eval seed / n", f"{env['evaluation_seed']} / {env['sample_size']}"],
                        ["run type", env["run_type"]]])
    el.append(Paragraph(cert["statistical_note"], ss["Italic"]))
    doc.build(el)
    return buf.getvalue()