1  """
  2  Benchmark reporting utilities for AI System Optimization Series.
  3  Generates Markdown tables and performance charts.
  4  """
  5  
  6  import json
  7  from dataclasses import asdict, dataclass, field
  8  from datetime import datetime
  9  
 10  from .timer import TimingResult
 11  
 12  
 13  @dataclass
 14  class BenchmarkEntry:
 15      """Single benchmark entry."""
 16  
 17      module_name: str
 18      test_name: str
 19      hardware: str
 20      latency_ms: float
 21      throughput: float | None = None
 22      memory_mb: float | None = None
 23      timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
 24  
 25  
 26  @dataclass
 27  class BenchmarkReport:
 28      """Complete benchmark report."""
 29  
 30      entries: list[BenchmarkEntry] = field(default_factory=list)
 31      environment: dict[str, str] = field(default_factory=dict)
 32  
 33      def add_entry(self, entry: BenchmarkEntry) -> None:
 34          self.entries.append(entry)
 35  
 36      def to_dict(self) -> dict:
 37          """Serialize report to a JSON-compatible dictionary."""
 38          return {
 39              "entries": [asdict(e) for e in self.entries],
 40              "environment": self.environment,
 41          }
 42  
 43      def to_json(self, indent: int = 2) -> str:
 44          """Serialize report to a JSON string."""
 45          return json.dumps(self.to_dict(), indent=indent, ensure_ascii=False)
 46  
 47      def save_json(self, path: str) -> None:
 48          """Save report to a JSON file."""
 49          with open(path, "w", encoding="utf-8") as f:
 50              f.write(self.to_json())
 51  
 52      @classmethod
 53      def from_json(cls, path: str) -> "BenchmarkReport":
 54          """Load report from a JSON file."""
 55          with open(path, encoding="utf-8") as f:
 56              data = json.load(f)
 57          entries = [BenchmarkEntry(**e) for e in data.get("entries", [])]
 58          environment = data.get("environment", {})
 59          return cls(entries=entries, environment=environment)
 60  
 61      def to_markdown_table(self) -> str:
 62          """Generate Markdown table from entries."""
 63          if not self.entries:
 64              return "No benchmark entries."
 65  
 66          lines = [
 67              "| Module | Test | Hardware | Latency (ms) | Throughput | Memory (MB) |",
 68              "|--------|------|----------|--------------|------------|-------------|",
 69          ]
 70  
 71          for e in self.entries:
 72              throughput = f"{e.throughput:.2f}" if e.throughput is not None else "N/A"
 73              memory = f"{e.memory_mb:.1f}" if e.memory_mb is not None else "N/A"
 74              lines.append(
 75                  f"| {e.module_name} | {e.test_name} | {e.hardware} | "
 76                  f"{e.latency_ms:.3f} | {throughput} | {memory} |"
 77              )
 78  
 79          return "\n".join(lines)
 80  
 81  
 82  def generate_comparison_table(
 83      results: dict[str, TimingResult], baseline_key: str = "pytorch_eager"
 84  ) -> str:
 85      """
 86      Generate Markdown comparison table from timing results.
 87  
 88      Args:
 89          results: Dict mapping implementation name to TimingResult
 90          baseline_key: Key for baseline implementation (for speedup calculation)
 91  
 92      Returns:
 93          Markdown formatted table string
 94      """
 95      if not results:
 96          return "No results to display."
 97  
 98      baseline = results.get(baseline_key)
 99      baseline_ms = baseline.mean_ms if baseline else None
100  
101      lines = [
102          "| Implementation | Mean (ms) | Std (ms) | Min (ms) | Max (ms) | Speedup |",
103          "|----------------|-----------|----------|----------|----------|---------|",
104      ]
105  
106      for name, result in results.items():
107          if baseline_ms and baseline_ms > 0:
108              speedup = f"{baseline_ms / result.mean_ms:.2f}x"
109          else:
110              speedup = "1.00x" if name == baseline_key else "N/A"
111  
112          lines.append(
113              f"| {name} | {result.mean_ms:.3f} | {result.std_ms:.3f} | "
114              f"{result.min_ms:.3f} | {result.max_ms:.3f} | {speedup} |"
115          )
116  
117      return "\n".join(lines)
118  
119  
120  def generate_performance_chart(
121      results: dict[str, TimingResult],
122      output_path: str = "performance_chart.png",
123      title: str = "Performance Comparison",
124  ) -> bool:
125      """
126      Generate performance comparison bar chart.
127  
128      Args:
129          results: Dict mapping implementation name to TimingResult
130          output_path: Path to save the chart
131          title: Chart title
132  
133      Returns:
134          True if chart was generated successfully, False otherwise
135      """
136      try:
137          import matplotlib.pyplot as plt
138          import numpy as np
139      except ImportError:
140          print("matplotlib not available, skipping chart generation")
141          return False
142  
143      if not results:
144          return False
145  
146      names = list(results.keys())
147      means = [r.mean_ms for r in results.values()]
148      stds = [r.std_ms for r in results.values()]
149  
150      fig, ax = plt.subplots(figsize=(10, 6))
151  
152      x = np.arange(len(names))
153      bars = ax.bar(x, means, yerr=stds, capsize=5, color="steelblue", alpha=0.8)
154  
155      ax.set_xlabel("Implementation")
156      ax.set_ylabel("Latency (ms)")
157      ax.set_title(title)
158      ax.set_xticks(x)
159      ax.set_xticklabels(names, rotation=45, ha="right")
160  
161      # Add value labels on bars
162      for bar, mean in zip(bars, means):
163          height = bar.get_height()
164          ax.annotate(
165              f"{mean:.2f}",
166              xy=(bar.get_x() + bar.get_width() / 2, height),
167              xytext=(0, 3),
168              textcoords="offset points",
169              ha="center",
170              va="bottom",
171              fontsize=9,
172          )
173  
174      plt.tight_layout()
175      plt.savefig(output_path, dpi=150)
176      plt.close()
177  
178      return True
179  
180  
181  def format_environment_info() -> dict[str, str]:
182      """Collect environment information for benchmark reports."""
183      import platform
184      import sys
185  
186      env = {
187          "python_version": sys.version.split()[0],
188          "platform": platform.platform(),
189          "processor": platform.processor(),
190      }
191  
192      # Try to get CUDA info
193      try:
194          import torch
195  
196          if torch.cuda.is_available():
197              env["cuda_version"] = torch.version.cuda or "N/A"
198              env["gpu_name"] = torch.cuda.get_device_name(0)
199              env["gpu_count"] = str(torch.cuda.device_count())
200      except ImportError:
201          pass
202  
203      return env