Coverage for src / tracekit / reporting / multichannel.py: 98%
124 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-11 23:04 +0000
1"""Multi-channel report generation for TraceKit.
3This module provides utilities for generating reports across multiple channels
4with channel comparison and aggregation.
7Example:
8 >>> from tracekit.reporting.multichannel import generate_multichannel_report
9 >>> report = generate_multichannel_report(channel_results, "multi_report.pdf")
10"""
12from __future__ import annotations
14from typing import Any
16from tracekit.reporting.core import Report, ReportConfig, Section
17from tracekit.reporting.tables import create_measurement_table
20def generate_multichannel_report(
21 channel_results: dict[str, dict[str, Any]],
22 *,
23 title: str = "Multi-Channel Analysis Report",
24 compare_channels: bool = True,
25 aggregate_statistics: bool = True,
26 individual_sections: bool = True,
27 **kwargs: Any,
28) -> Report:
29 """Generate report for multi-channel analysis.
31 Args:
32 channel_results: Dictionary mapping channel name to results.
33 title: Report title.
34 compare_channels: Include channel comparison section.
35 aggregate_statistics: Include aggregate statistics across channels.
36 individual_sections: Include individual channel sections.
37 **kwargs: Additional report configuration options.
39 Returns:
40 Multi-channel Report object.
42 References:
43 REPORT-007
44 """
45 config = ReportConfig(title=title, **kwargs)
46 report = Report(config=config)
48 # Add executive summary
49 summary_content = _generate_multichannel_summary(channel_results)
50 report.add_section("Executive Summary", summary_content, level=1)
52 # Add aggregate statistics
53 if aggregate_statistics:
54 stats_section = _create_aggregate_statistics_section(channel_results)
55 report.sections.append(stats_section)
57 # Add channel comparison
58 if compare_channels and len(channel_results) > 1:
59 comparison_section = _create_channel_comparison_section(channel_results)
60 report.sections.append(comparison_section)
62 # Add individual channel sections
63 if individual_sections:
64 for channel_name, results in channel_results.items():
65 channel_section = _create_channel_section(channel_name, results)
66 report.sections.append(channel_section)
68 return report
71def _generate_multichannel_summary(channel_results: dict[str, dict[str, Any]]) -> str:
72 """Generate summary for multi-channel report."""
73 summary_parts = []
75 total_channels = len(channel_results)
76 summary_parts.append(f"Analyzed {total_channels} channel(s).")
78 # Aggregate pass/fail across channels
79 total_tests = 0
80 total_passed = 0
82 for results in channel_results.values():
83 total_tests += results.get("total_count", 0)
84 total_passed += results.get("pass_count", 0)
86 if total_tests > 0:
87 total_failed = total_tests - total_passed
88 summary_parts.append(
89 f"\nOverall: {total_passed}/{total_tests} tests passed "
90 f"({total_passed / total_tests * 100:.1f}% pass rate)."
91 )
93 if total_failed > 0:
94 summary_parts.append(f"{total_failed} test(s) failed across all channels.")
96 # Channel-specific summary
97 failed_channels = []
98 for channel_name, results in channel_results.items():
99 pass_count = results.get("pass_count", 0)
100 total_count = results.get("total_count", 0)
101 if total_count > 0 and pass_count < total_count:
102 failed_channels.append(channel_name)
104 if failed_channels:
105 summary_parts.append(f"\nChannels with failures: {', '.join(failed_channels)}")
106 else:
107 summary_parts.append("\nAll channels passed all tests.")
109 return "\n".join(summary_parts)
112def _create_aggregate_statistics_section(
113 channel_results: dict[str, dict[str, Any]],
114) -> Section:
115 """Create aggregate statistics section across all channels."""
116 # Collect all measurement parameters
117 all_params = set()
118 for results in channel_results.values():
119 if "measurements" in results: 119 ↛ 118line 119 didn't jump to line 118 because the condition on line 119 was always true
120 all_params.update(results["measurements"].keys())
122 # Build aggregate table
123 import numpy as np
125 headers = ["Parameter", "Min", "Mean", "Max", "Std Dev"]
126 rows = []
128 for param in sorted(all_params):
129 values = []
130 unit = ""
132 for results in channel_results.values():
133 if "measurements" in results and param in results["measurements"]:
134 meas = results["measurements"][param]
135 if "value" in meas and meas["value"] is not None:
136 values.append(meas["value"])
137 if not unit and "unit" in meas:
138 unit = meas["unit"]
140 if values:
141 from tracekit.reporting.formatting import NumberFormatter
143 formatter = NumberFormatter()
144 rows.append(
145 [
146 param,
147 formatter.format(np.min(values), unit),
148 formatter.format(np.mean(values), unit),
149 formatter.format(np.max(values), unit),
150 formatter.format(np.std(values), unit),
151 ]
152 )
154 table = {"type": "table", "headers": headers, "data": rows}
156 return Section(
157 title="Aggregate Statistics",
158 content=[table],
159 level=1,
160 visible=True,
161 )
164def _create_channel_comparison_section(
165 channel_results: dict[str, dict[str, Any]],
166) -> Section:
167 """Create channel-to-channel comparison section."""
168 from tracekit.reporting.formatting import NumberFormatter
170 formatter = NumberFormatter()
172 # Build comparison table
173 channel_names = list(channel_results.keys())
174 headers = ["Parameter", *channel_names]
176 # Collect all parameters
177 all_params = set()
178 for results in channel_results.values():
179 if "measurements" in results: 179 ↛ 178line 179 didn't jump to line 178 because the condition on line 179 was always true
180 all_params.update(results["measurements"].keys())
182 rows = []
183 for param in sorted(all_params):
184 row = [param]
186 for channel_name in channel_names:
187 results = channel_results[channel_name]
188 if "measurements" in results and param in results["measurements"]:
189 meas = results["measurements"][param]
190 value = meas.get("value")
191 unit = meas.get("unit", "")
192 if value is not None: 192 ↛ 195line 192 didn't jump to line 195 because the condition on line 192 was always true
193 row.append(formatter.format(value, unit))
194 else:
195 row.append("-")
196 else:
197 row.append("-")
199 rows.append(row)
201 table = {"type": "table", "headers": headers, "data": rows}
203 return Section(
204 title="Channel Comparison",
205 content=[table],
206 level=1,
207 visible=True,
208 )
211def _create_channel_section(
212 channel_name: str,
213 results: dict[str, Any],
214) -> Section:
215 """Create individual channel section."""
216 subsections = []
218 # Channel summary
219 summary_parts = []
220 if "pass_count" in results and "total_count" in results:
221 pass_count = results["pass_count"]
222 total = results["total_count"]
223 summary_parts.append(
224 f"{pass_count}/{total} tests passed ({pass_count / total * 100:.1f}% pass rate)."
225 )
227 # Measurements
228 if "measurements" in results:
229 table = create_measurement_table(results["measurements"], format="dict")
230 subsections.append(
231 Section(
232 title="Measurements",
233 content=[table],
234 level=3,
235 visible=True,
236 )
237 )
239 return Section(
240 title=f"Channel: {channel_name}",
241 content="\n".join(summary_parts) if summary_parts else "",
242 level=2,
243 visible=True,
244 subsections=subsections,
245 )
248def create_channel_crosstalk_section(
249 crosstalk_results: dict[str, Any],
250) -> Section:
251 """Create channel crosstalk analysis section.
253 Args:
254 crosstalk_results: Crosstalk analysis results between channels.
256 Returns:
257 Crosstalk Section object.
259 References:
260 REPORT-007
261 """
262 from tracekit.reporting.formatting import NumberFormatter
264 formatter = NumberFormatter()
266 if "crosstalk_matrix" in crosstalk_results:
267 matrix = crosstalk_results["crosstalk_matrix"]
268 channels = crosstalk_results.get("channels", [])
270 headers = ["Aggressor → Victim", *channels]
271 rows = []
273 for i, aggressor in enumerate(channels):
274 row = [aggressor]
275 for j, _victim in enumerate(channels):
276 if i == j:
277 row.append("-")
278 else:
279 crosstalk_db = matrix[i][j]
280 row.append(formatter.format(crosstalk_db, "dB"))
281 rows.append(row)
283 table = {"type": "table", "headers": headers, "data": rows}
284 content = [
285 "Channel-to-channel crosstalk measurements:\n",
286 table,
287 ]
288 else:
289 content = "No crosstalk analysis available." # type: ignore[assignment]
291 return Section(
292 title="Channel Crosstalk Analysis",
293 content=content,
294 level=2,
295 visible=True,
296 )