"""Analysis service: compute metrics and produce charts/JSON for frontend.""" from typing import Any, Dict, List from uuid import UUID from sqlalchemy.orm import Session from app.analysis.charts import boxplot, grouped_bar, heatmap from app.analysis.likert import aggregate_likert_by_model_and_level from app.analysis.parser import compare_findings, parse_output from app.analysis.statistics import anova, descriptive_stats, paired_t_test from app.models import Experiment, ExperimentRun, Result class AnalysisService: def __init__(self, db: Session): self.db = db def get_experiment_results(self, experiment_id: str) -> List[Dict[str, Any]]: runs = ( self.db.query(ExperimentRun) .filter_by(experiment_id=UUID(experiment_id)) .all() ) return [ { "run_id": str(run.id), "model_id": run.model_id, "level": run.template_version.template.level, "sample_id": str(run.sample_id), "repeat_index": run.repeat_index, "status": run.status, "result": self._serialize_result(run.result) if run.result else None, } for run in runs ] def _serialize_result(self, result: Result) -> Dict[str, Any]: return { "raw_output": result.raw_output, "token_usage": result.token_usage, "latency_ms": result.latency_ms, "parsed_findings": result.parsed_findings, "detection_rate": result.detection_rate, "false_positive_rate": result.false_positive_rate, "coverage_rate": result.coverage_rate, "stability_score": result.stability_score, "likert_score": result.likert_score, } def compute_metrics_for_run(self, run_id: str) -> Dict[str, Any]: run = self.db.query(ExperimentRun).filter_by(id=UUID(run_id)).first() if not run or not run.result: return {"error": "Run or result not found"} level = run.template_version.template.level findings = parse_output(run.result.raw_output or "", level) gt = [ { "id": str(d.id), "defect_type": d.defect_type, "line_start": d.line_start, "line_end": d.line_end, } for d in run.sample.defects ] metrics = compare_findings(findings, gt) run.result.parsed_findings = [self._finding_to_dict(f) for f in findings] run.result.detection_rate = metrics["detection_rate"] run.result.false_positive_rate = metrics["false_positive_rate"] run.result.coverage_rate = metrics["coverage_rate"] self.db.commit() return { "run_id": run_id, "findings": [self._finding_to_dict(f) for f in findings], **metrics, } def _finding_to_dict(self, finding) -> Dict[str, Any]: return { "defect_type": finding.defect_type, "line_start": finding.line_start, "line_end": finding.line_end, "description": finding.description, } def aggregate_metrics(self, experiment_id: str) -> Dict[str, Any]: runs = ( self.db.query(ExperimentRun) .filter_by(experiment_id=UUID(experiment_id), status="done") .all() ) detection_rates = [] fp_rates = [] coverage_rates = [] for run in runs: if not run.result: continue if run.result.detection_rate is not None: detection_rates.append(run.result.detection_rate) if run.result.false_positive_rate is not None: fp_rates.append(run.result.false_positive_rate) if run.result.coverage_rate is not None: coverage_rates.append(run.result.coverage_rate) return { "detection_rate": descriptive_stats(detection_rates), "false_positive_rate": descriptive_stats(fp_rates), "coverage_rate": descriptive_stats(coverage_rates), } def likert_aggregation(self) -> Dict[str, Any]: return aggregate_likert_by_model_and_level(self.db) def generate_charts(self, experiment_id: str) -> Dict[str, str]: runs = ( self.db.query(ExperimentRun) .filter_by(experiment_id=UUID(experiment_id), status="done") .all() ) heatmap_data: Dict[str, Dict[str, float]] = {} box_groups: Dict[str, List[float]] = {} bar_data: Dict[str, Dict[str, float]] = {} for run in runs: if not run.result: continue model = run.model_id level = run.template_version.template.level dr = run.result.detection_rate or 0.0 heatmap_data.setdefault(model, {}) bar_data.setdefault(model, {}) heatmap_data[model][level] = heatmap_data[model].get(level, 0.0) + dr box_groups.setdefault(f"{model}-{level}", []).append(dr) bar_data[model][level] = bar_data[model].get(level, 0.0) + dr # average heatmap and bar values counts: Dict[str, Dict[str, int]] = {} for run in runs: if not run.result: continue model = run.model_id level = run.template_version.template.level counts.setdefault(model, {}).setdefault(level, 0) counts[model][level] += 1 for model in heatmap_data: for level in heatmap_data[model]: heatmap_data[model][level] /= counts[model][level] bar_data[model][level] /= counts[model][level] return { "heatmap": heatmap(heatmap_data, title="Detection Rate Heatmap"), "boxplot": boxplot(box_groups, title="Detection Rate Distribution"), "grouped_bar": grouped_bar(bar_data, title="Detection Rate by Model and Level"), } def run_anova(self, experiment_id: str) -> Dict[str, Any]: runs = ( self.db.query(ExperimentRun) .filter_by(experiment_id=UUID(experiment_id), status="done") .all() ) groups: Dict[str, List[float]] = {} for run in runs: if not run.result or run.result.detection_rate is None: continue key = f"{run.model_id}-{run.template_version.template.level}" groups.setdefault(key, []).append(run.result.detection_rate) return anova(list(groups.values())) def run_paired_t_test(self, group_a_key: str, group_b_key: str, experiment_id: str) -> Dict[str, Any]: runs = ( self.db.query(ExperimentRun) .filter_by(experiment_id=UUID(experiment_id), status="done") .all() ) groups: Dict[str, List[float]] = {} for run in runs: if not run.result or run.result.detection_rate is None: continue key = f"{run.model_id}-{run.template_version.template.level}" groups.setdefault(key, []).append(run.result.detection_rate) return paired_t_test(groups.get(group_a_key, []), groups.get(group_b_key, []))