Coverage for src / tracekit / reporting / comparison.py: 99%

133 statements  

« prev     ^ index     » next       coverage.py v7.13.1, created at 2026-01-11 23:04 +0000

1"""Comparison report generation for TraceKit. 

2 

3This module provides utilities for comparing multiple traces or test runs 

4and generating comparison reports with diff visualization. 

5 

6 

7Example: 

8 >>> from tracekit.reporting.comparison import generate_comparison_report 

9 >>> report = generate_comparison_report(baseline, current, "comparison.pdf") 

10""" 

11 

12from __future__ import annotations 

13 

14from typing import Any, Literal 

15 

16from tracekit.reporting.core import Report, ReportConfig, Section 

17from tracekit.reporting.tables import create_comparison_table 

18 

19 

20def generate_comparison_report( 

21 baseline: dict[str, Any], 

22 current: dict[str, Any], 

23 *, 

24 title: str = "Comparison Report", 

25 mode: Literal["side_by_side", "inline"] = "side_by_side", 

26 show_only_changes: bool = False, 

27 highlight_changes: bool = False, 

28 **kwargs: Any, 

29) -> Report: 

30 """Generate comparison report between baseline and current results. 

31 

32 Args: 

33 baseline: Baseline results dictionary. 

34 current: Current results dictionary. 

35 title: Report title. 

36 mode: Comparison mode (side_by_side or inline). 

37 show_only_changes: Only show changed measurements. 

38 highlight_changes: Highlight changes in output. Reserved for future use. 

39 **kwargs: Additional report configuration options. 

40 

41 Returns: 

42 Comparison Report object. 

43 

44 References: 

45 REPORT-008 

46 """ 

47 config = ReportConfig(title=title, **kwargs) 

48 report = Report(config=config) 

49 

50 # Add summary 

51 summary = _generate_comparison_summary(baseline, current) 

52 report.add_section("Comparison Summary", summary, level=1) 

53 

54 # Add change details 

55 changes_section = _create_changes_section( 

56 baseline, 

57 current, 

58 show_only_changes=show_only_changes, 

59 ) 

60 report.sections.append(changes_section) 

61 

62 # Add violations comparison 

63 if "violations" in baseline or "violations" in current: 

64 violations_section = _create_violations_comparison_section( 

65 baseline, 

66 current, 

67 ) 

68 report.sections.append(violations_section) 

69 

70 # Add detailed comparison 

71 if "measurements" in baseline or "measurements" in current: 

72 detailed_section = _create_detailed_comparison_section( 

73 baseline.get("measurements", {}), 

74 current.get("measurements", {}), 

75 mode=mode, 

76 ) 

77 report.sections.append(detailed_section) 

78 

79 return report 

80 

81 

82def _generate_comparison_summary( 

83 baseline: dict[str, Any], 

84 current: dict[str, Any], 

85) -> str: 

86 """Generate comparison summary.""" 

87 summary_parts = [] 

88 

89 # Count changes 

90 baseline_meas = baseline.get("measurements", {}) 

91 current_meas = current.get("measurements", {}) 

92 

93 all_params = set(baseline_meas.keys()) | set(current_meas.keys()) 

94 changed_params = [] 

95 improved_params = [] 

96 degraded_params = [] 

97 

98 for param in all_params: 

99 base_val = baseline_meas.get(param, {}).get("value") 

100 curr_val = current_meas.get(param, {}).get("value") 

101 

102 if base_val is not None and curr_val is not None: 

103 if abs(curr_val - base_val) / abs(base_val) > 0.05: # >5% change 

104 changed_params.append(param) 

105 

106 # Determine if improved or degraded 

107 base_passed = baseline_meas.get(param, {}).get("passed", True) 

108 curr_passed = current_meas.get(param, {}).get("passed", True) 

109 

110 if not base_passed and curr_passed: 

111 improved_params.append(param) 

112 elif base_passed and not curr_passed: 

113 degraded_params.append(param) 

114 

115 summary_parts.append(f"Comparing {len(all_params)} parameter(s) between baseline and current.") 

116 

117 if changed_params: 

118 summary_parts.append(f"\n{len(changed_params)} measurement(s) changed significantly (>5%).") 

119 

120 if improved_params: 

121 summary_parts.append( 

122 f"\n✓ {len(improved_params)} parameter(s) improved (failures → passes)." 

123 ) 

124 

125 if degraded_params: 

126 summary_parts.append( 

127 f"\n✗ {len(degraded_params)} parameter(s) degraded (passes → failures)." 

128 ) 

129 

130 # Pass/fail comparison 

131 baseline_pass = baseline.get("pass_count", 0) 

132 baseline_total = baseline.get("total_count", 0) 

133 current_pass = current.get("pass_count", 0) 

134 current_total = current.get("total_count", 0) 

135 

136 if baseline_total > 0 and current_total > 0: 

137 baseline_rate = baseline_pass / baseline_total * 100 

138 current_rate = current_pass / current_total * 100 

139 delta = current_rate - baseline_rate 

140 

141 summary_parts.append( 

142 f"\nPass rate: {baseline_rate:.1f}% → {current_rate:.1f}% ({delta:+.1f}% change)" 

143 ) 

144 

145 return "\n".join(summary_parts) 

146 

147 

148def _create_changes_section( 

149 baseline: dict[str, Any], 

150 current: dict[str, Any], 

151 *, 

152 show_only_changes: bool = False, 

153) -> Section: 

154 """Create section detailing changes.""" 

155 baseline_meas = baseline.get("measurements", {}) 

156 current_meas = current.get("measurements", {}) 

157 

158 # Create comparison table 

159 table = create_comparison_table( 

160 baseline_meas, 

161 current_meas, 

162 format="dict", 

163 show_delta=True, 

164 show_percent_change=True, 

165 ) 

166 

167 # Filter to only changes if requested 

168 if show_only_changes: 

169 filtered_rows = [] 

170 for row in table["data"]: # type: ignore[index] 

171 # Check if delta is significant 

172 if len(row) >= 4: # Has delta column 172 ↛ 170line 172 didn't jump to line 170 because the condition on line 172 was always true

173 delta_str = str(row[3]) 

174 if delta_str not in {"-", "0"}: 174 ↛ 170line 174 didn't jump to line 170 because the condition on line 174 was always true

175 filtered_rows.append(row) 

176 

177 table["data"] = filtered_rows # type: ignore[index] 

178 

179 return Section( 

180 title="Measurement Changes", 

181 content=[table], 

182 level=1, 

183 visible=True, 

184 ) 

185 

186 

187def _create_violations_comparison_section( 

188 baseline: dict[str, Any], 

189 current: dict[str, Any], 

190) -> Section: 

191 """Create section comparing violations.""" 

192 baseline_violations = {v.get("parameter") for v in baseline.get("violations", [])} 

193 current_violations = {v.get("parameter") for v in current.get("violations", [])} 

194 

195 content_parts = [] 

196 

197 # New violations 

198 new_violations = current_violations - baseline_violations 

199 if new_violations: 

200 content_parts.append("**New Violations:**") 

201 for param in sorted(new_violations): 

202 content_parts.append(f"- {param}") 

203 

204 # Resolved violations 

205 resolved_violations = baseline_violations - current_violations 

206 if resolved_violations: 

207 content_parts.append("\n**Resolved Violations:**") 

208 for param in sorted(resolved_violations): 

209 content_parts.append(f"- {param}") 

210 

211 # Persistent violations 

212 persistent_violations = baseline_violations & current_violations 

213 if persistent_violations: 

214 content_parts.append("\n**Persistent Violations:**") 

215 for param in sorted(persistent_violations): 

216 content_parts.append(f"- {param}") 

217 

218 if not content_parts: 

219 content_parts.append("No violations in either baseline or current.") 

220 

221 content = "\n".join(content_parts) 

222 

223 return Section( 

224 title="Violations Comparison", 

225 content=content, 

226 level=1, 

227 visible=True, 

228 ) 

229 

230 

231def _create_detailed_comparison_section( 

232 baseline_meas: dict[str, Any], 

233 current_meas: dict[str, Any], 

234 *, 

235 mode: str = "side_by_side", 

236) -> Section: 

237 """Create detailed measurement comparison section.""" 

238 from tracekit.reporting.formatting import NumberFormatter 

239 

240 formatter = NumberFormatter() 

241 

242 content_parts = [] 

243 

244 all_params = sorted(set(baseline_meas.keys()) | set(current_meas.keys())) 

245 

246 for param in all_params: 

247 base = baseline_meas.get(param, {}) 

248 curr = current_meas.get(param, {}) 

249 

250 base_val = base.get("value") 

251 curr_val = curr.get("value") 

252 unit = base.get("unit", curr.get("unit", "")) 

253 

254 if base_val is not None and curr_val is not None: 

255 delta = curr_val - base_val 

256 pct_change = (delta / base_val * 100) if base_val != 0 else 0 

257 

258 base_str = formatter.format(base_val, unit) 

259 curr_str = formatter.format(curr_val, unit) 

260 delta_str = formatter.format(delta, unit) 

261 

262 # Determine if improved/degraded 

263 base_passed = base.get("passed", True) 

264 curr_passed = curr.get("passed", True) 

265 

266 status = "" 

267 if not base_passed and curr_passed: 

268 status = " ✓ IMPROVED" 

269 elif base_passed and not curr_passed: 

270 status = " ✗ DEGRADED" 

271 

272 content_parts.append( 

273 f"**{param}:** {base_str}{curr_str}{delta_str}, {pct_change:+.1f}%){status}" 

274 ) 

275 

276 content = "\n\n".join(content_parts) if content_parts else "No measurements to compare." 

277 

278 return Section( 

279 title="Detailed Comparison", 

280 content=content, 

281 level=1, 

282 visible=True, 

283 collapsible=True, 

284 ) 

285 

286 

287def compare_waveforms( 

288 baseline_signal: dict[str, Any], 

289 current_signal: dict[str, Any], 

290) -> dict[str, Any]: 

291 """Compare two waveforms and extract differences. 

292 

293 Args: 

294 baseline_signal: Baseline waveform data. 

295 current_signal: Current waveform data. 

296 

297 Returns: 

298 Dictionary with comparison metrics. 

299 

300 References: 

301 REPORT-008 

302 """ 

303 import numpy as np 

304 

305 comparison = { 

306 "correlation": None, 

307 "rms_difference": None, 

308 "max_difference": None, 

309 "mean_difference": None, 

310 } 

311 

312 base_data = baseline_signal.get("data") 

313 curr_data = current_signal.get("data") 

314 

315 if base_data is not None and curr_data is not None: 

316 # Ensure same length 

317 min_len = min(len(base_data), len(curr_data)) 

318 base_data = base_data[:min_len] 

319 curr_data = curr_data[:min_len] 

320 

321 # Correlation 

322 comparison["correlation"] = np.corrcoef(base_data, curr_data)[0, 1] 

323 

324 # Differences 

325 diff = curr_data - base_data 

326 comparison["rms_difference"] = np.sqrt(np.mean(diff**2)) 

327 comparison["max_difference"] = np.max(np.abs(diff)) 

328 comparison["mean_difference"] = np.mean(diff) 

329 

330 return comparison