186 lines
7.1 KiB
Python
186 lines
7.1 KiB
Python
"""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, []))
|