130 lines
4.2 KiB
Python
130 lines
4.2 KiB
Python
"""Parse model outputs and compare against Ground Truth."""
|
|
|
|
import re
|
|
from dataclasses import dataclass
|
|
from typing import Dict, List, Optional
|
|
|
|
|
|
@dataclass
|
|
class Finding:
|
|
defect_type: str
|
|
line_start: Optional[int]
|
|
line_end: Optional[int]
|
|
description: str
|
|
|
|
|
|
def parse_output(output: str, level: str) -> List[Finding]:
|
|
"""Parse model output into structured findings.
|
|
|
|
Does not guess: if output is empty or unparseable, returns empty list.
|
|
"""
|
|
if not output or not output.strip():
|
|
return []
|
|
|
|
findings = []
|
|
if level == "L1":
|
|
# Expect bullet list of defect types or short descriptions
|
|
for line in output.splitlines():
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
if line.startswith(("-", "*", "•", "1.", "2.", "3.")):
|
|
item = re.sub(r"^[-*•0-9.\s]+", "", line)
|
|
findings.append(
|
|
Finding(
|
|
defect_type=item.split(":", 1)[0].strip(),
|
|
line_start=None,
|
|
line_end=None,
|
|
description=item,
|
|
)
|
|
)
|
|
else:
|
|
# Parse L2/L3 structured blocks
|
|
current: Dict[str, str] = {}
|
|
for raw in output.splitlines():
|
|
line = raw.strip()
|
|
if line.startswith(("-", "*", "•")):
|
|
if current:
|
|
findings.append(_build_finding(current))
|
|
current = {}
|
|
key, _, value = line.lstrip("-*• ").partition(":")
|
|
current[key.strip().lower()] = value.strip()
|
|
elif line and current:
|
|
key, _, value = line.partition(":")
|
|
current[key.strip().lower()] = value.strip()
|
|
if current:
|
|
findings.append(_build_finding(current))
|
|
|
|
return findings
|
|
|
|
|
|
def _build_finding(fields: Dict[str, str]) -> Finding:
|
|
defect_type = fields.get("type", "unknown")
|
|
lines = fields.get("lines", "")
|
|
line_start, line_end = None, None
|
|
if lines:
|
|
parts = re.split(r"[-,\s]+", lines)
|
|
try:
|
|
line_start = int(parts[0])
|
|
line_end = int(parts[-1]) if len(parts) > 1 else line_start
|
|
except ValueError:
|
|
pass
|
|
description = fields.get("explanation", fields.get("fix", ""))
|
|
return Finding(defect_type, line_start, line_end, description)
|
|
|
|
|
|
def compare_findings(
|
|
findings: List[Finding],
|
|
ground_truth: List[dict],
|
|
line_tolerance: int = 3,
|
|
) -> Dict[str, float]:
|
|
"""Compare parsed findings to Ground Truth defects.
|
|
|
|
Returns detection_rate, false_positive_rate, coverage_rate.
|
|
"""
|
|
if not ground_truth:
|
|
return {"detection_rate": 0.0, "false_positive_rate": 0.0, "coverage_rate": 0.0}
|
|
|
|
detected = set()
|
|
false_positives = 0
|
|
|
|
for finding in findings:
|
|
matched = False
|
|
for gt in ground_truth:
|
|
type_match = finding.defect_type.lower() in gt["defect_type"].lower() or gt[
|
|
"defect_type"
|
|
].lower() in finding.defect_type.lower()
|
|
line_match = False
|
|
if finding.line_start is not None and gt.get("line_start") is not None:
|
|
gt_start = gt["line_start"]
|
|
gt_end = gt.get("line_end", gt_start)
|
|
if (
|
|
min(finding.line_start, finding.line_end or finding.line_start) - line_tolerance
|
|
<= gt_end
|
|
and max(finding.line_start, finding.line_end or finding.line_start)
|
|
+ line_tolerance
|
|
>= gt_start
|
|
):
|
|
line_match = True
|
|
if type_match or line_match:
|
|
matched = True
|
|
detected.add(gt.get("id", id(gt)))
|
|
break
|
|
if not matched:
|
|
false_positives += 1
|
|
|
|
tp = len(detected)
|
|
fp = false_positives
|
|
fn = len(ground_truth) - tp
|
|
|
|
detection_rate = tp / len(ground_truth)
|
|
precision = tp / (tp + fp) if (tp + fp) > 0 else 0.0
|
|
false_positive_rate = 1.0 - precision
|
|
coverage_rate = tp / len(ground_truth)
|
|
|
|
return {
|
|
"detection_rate": round(detection_rate, 4),
|
|
"false_positive_rate": round(false_positive_rate, 4),
|
|
"coverage_rate": round(coverage_rate, 4),
|
|
}
|