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
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-11 23:04 +0000
1"""Comparison report generation for TraceKit.
3This module provides utilities for comparing multiple traces or test runs
4and generating comparison reports with diff visualization.
7Example:
8 >>> from tracekit.reporting.comparison import generate_comparison_report
9 >>> report = generate_comparison_report(baseline, current, "comparison.pdf")
10"""
12from __future__ import annotations
14from typing import Any, Literal
16from tracekit.reporting.core import Report, ReportConfig, Section
17from tracekit.reporting.tables import create_comparison_table
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.
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.
41 Returns:
42 Comparison Report object.
44 References:
45 REPORT-008
46 """
47 config = ReportConfig(title=title, **kwargs)
48 report = Report(config=config)
50 # Add summary
51 summary = _generate_comparison_summary(baseline, current)
52 report.add_section("Comparison Summary", summary, level=1)
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)
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)
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)
79 return report
82def _generate_comparison_summary(
83 baseline: dict[str, Any],
84 current: dict[str, Any],
85) -> str:
86 """Generate comparison summary."""
87 summary_parts = []
89 # Count changes
90 baseline_meas = baseline.get("measurements", {})
91 current_meas = current.get("measurements", {})
93 all_params = set(baseline_meas.keys()) | set(current_meas.keys())
94 changed_params = []
95 improved_params = []
96 degraded_params = []
98 for param in all_params:
99 base_val = baseline_meas.get(param, {}).get("value")
100 curr_val = current_meas.get(param, {}).get("value")
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)
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)
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)
115 summary_parts.append(f"Comparing {len(all_params)} parameter(s) between baseline and current.")
117 if changed_params:
118 summary_parts.append(f"\n{len(changed_params)} measurement(s) changed significantly (>5%).")
120 if improved_params:
121 summary_parts.append(
122 f"\n✓ {len(improved_params)} parameter(s) improved (failures → passes)."
123 )
125 if degraded_params:
126 summary_parts.append(
127 f"\n✗ {len(degraded_params)} parameter(s) degraded (passes → failures)."
128 )
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)
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
141 summary_parts.append(
142 f"\nPass rate: {baseline_rate:.1f}% → {current_rate:.1f}% ({delta:+.1f}% change)"
143 )
145 return "\n".join(summary_parts)
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", {})
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 )
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)
177 table["data"] = filtered_rows # type: ignore[index]
179 return Section(
180 title="Measurement Changes",
181 content=[table],
182 level=1,
183 visible=True,
184 )
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", [])}
195 content_parts = []
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}")
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}")
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}")
218 if not content_parts:
219 content_parts.append("No violations in either baseline or current.")
221 content = "\n".join(content_parts)
223 return Section(
224 title="Violations Comparison",
225 content=content,
226 level=1,
227 visible=True,
228 )
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
240 formatter = NumberFormatter()
242 content_parts = []
244 all_params = sorted(set(baseline_meas.keys()) | set(current_meas.keys()))
246 for param in all_params:
247 base = baseline_meas.get(param, {})
248 curr = current_meas.get(param, {})
250 base_val = base.get("value")
251 curr_val = curr.get("value")
252 unit = base.get("unit", curr.get("unit", ""))
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
258 base_str = formatter.format(base_val, unit)
259 curr_str = formatter.format(curr_val, unit)
260 delta_str = formatter.format(delta, unit)
262 # Determine if improved/degraded
263 base_passed = base.get("passed", True)
264 curr_passed = curr.get("passed", True)
266 status = ""
267 if not base_passed and curr_passed:
268 status = " ✓ IMPROVED"
269 elif base_passed and not curr_passed:
270 status = " ✗ DEGRADED"
272 content_parts.append(
273 f"**{param}:** {base_str} → {curr_str} (Δ {delta_str}, {pct_change:+.1f}%){status}"
274 )
276 content = "\n\n".join(content_parts) if content_parts else "No measurements to compare."
278 return Section(
279 title="Detailed Comparison",
280 content=content,
281 level=1,
282 visible=True,
283 collapsible=True,
284 )
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.
293 Args:
294 baseline_signal: Baseline waveform data.
295 current_signal: Current waveform data.
297 Returns:
298 Dictionary with comparison metrics.
300 References:
301 REPORT-008
302 """
303 import numpy as np
305 comparison = {
306 "correlation": None,
307 "rms_difference": None,
308 "max_difference": None,
309 "mean_difference": None,
310 }
312 base_data = baseline_signal.get("data")
313 curr_data = current_signal.get("data")
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]
321 # Correlation
322 comparison["correlation"] = np.corrcoef(base_data, curr_data)[0, 1]
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)
330 return comparison