| |
| """Generate an SVG throughput graph from a ds4-bench CSV file. |
| |
| The benchmark intentionally reports instantaneous throughput at each context |
| frontier. This script keeps the plot equally direct: one line for incremental |
| prefill t/s, one line for greedy generation t/s, and separate y axes because |
| the two values live on very different scales. |
| """ |
|
|
| import argparse |
| import csv |
| import html |
| import math |
| from pathlib import Path |
|
|
|
|
| PREFILL_COLOR = "#2563eb" |
| GEN_COLOR = "#dc2626" |
| TEXT_COLOR = "#1f2933" |
| MUTED_COLOR = "#64748b" |
| GRID_COLOR = "#e2e8f0" |
| AXIS_COLOR = "#334155" |
|
|
|
|
| def nice_ceil(value): |
| """Round a positive axis maximum up to a human-friendly tick boundary.""" |
| if value <= 0: |
| return 1.0 |
|
|
| magnitude = 10 ** math.floor(math.log10(value)) |
| normalized = value / magnitude |
| for step in (1, 2, 2.5, 3, 4, 5, 10): |
| if normalized <= step: |
| return step * magnitude |
| return 10 * magnitude |
|
|
|
|
| def nice_step(span, target_ticks): |
| """Return a readable tick spacing close to span / target_ticks.""" |
| if span <= 0: |
| return 1.0 |
|
|
| raw = span / target_ticks |
| magnitude = 10 ** math.floor(math.log10(raw)) |
| normalized = raw / magnitude |
| for step in (1, 2, 2.5, 5, 10): |
| if normalized <= step: |
| return step * magnitude |
| return 10 * magnitude |
|
|
|
|
| def fmt_tick(value): |
| if abs(value) >= 1000: |
| return f"{value / 1000:g}k" |
| return f"{value:g}" |
|
|
|
|
| def read_points(path): |
| rows = [] |
| with path.open("r", encoding="utf-8-sig", newline="") as fp: |
| reader = csv.DictReader(fp) |
| required = {"ctx_tokens", "prefill_tps", "gen_tps"} |
| missing = required.difference(reader.fieldnames or ()) |
| if missing: |
| missing_list = ", ".join(sorted(missing)) |
| raise SystemExit(f"{path}: missing CSV column(s): {missing_list}") |
|
|
| for row in reader: |
| rows.append( |
| ( |
| int(row["ctx_tokens"]), |
| float(row["prefill_tps"]), |
| float(row["gen_tps"]), |
| ) |
| ) |
|
|
| if len(rows) < 2: |
| raise SystemExit(f"{path}: need at least two data rows") |
|
|
| rows.sort(key=lambda item: item[0]) |
| return rows |
|
|
|
|
| def derive_title(csv_path): |
| words = csv_path.stem.replace("_", " ").replace("-", " ").split() |
| return " ".join(word.upper() if word[0:1].lower() == "m" and word[1:].isdigit() else word.capitalize() for word in words) + " t/s" |
|
|
|
|
| def points_to_polyline(points, x_min, x_max, y_max, plot): |
| left, top, width, height = plot |
|
|
| def project(point): |
| x, y = point |
| px = left + (x - x_min) / (x_max - x_min) * width |
| py = top + height - y / y_max * height |
| return f"{px:.2f},{py:.2f}" |
|
|
| return " ".join(project(point) for point in points) |
|
|
|
|
| def render_svg(rows, title, width, height): |
| margin_left = 82 |
| margin_right = 82 |
| margin_top = 66 |
| margin_bottom = 72 |
| plot = ( |
| margin_left, |
| margin_top, |
| width - margin_left - margin_right, |
| height - margin_top - margin_bottom, |
| ) |
| left, top, plot_width, plot_height = plot |
| right = left + plot_width |
| bottom = top + plot_height |
|
|
| ctx_values = [row[0] for row in rows] |
| prefill_values = [row[1] for row in rows] |
| gen_values = [row[2] for row in rows] |
| x_min = 0 |
| x_max = max(ctx_values) |
| prefill_max = nice_ceil(max(prefill_values) * 1.05) |
| gen_max = nice_ceil(max(gen_values) * 1.05) |
|
|
| x_step = nice_step(x_max - x_min, 6) |
| x_ticks = [] |
| tick = math.ceil(x_min / x_step) * x_step |
| while tick <= x_max: |
| x_ticks.append(tick) |
| tick += x_step |
|
|
| prefill_step = nice_step(prefill_max, 5) |
| gen_step = nice_step(gen_max, 5) |
| prefill_ticks = [tick for tick in frange(0, prefill_max, prefill_step)] |
| gen_ticks = [tick for tick in frange(0, gen_max, gen_step)] |
|
|
| prefill_points = [(row[0], row[1]) for row in rows] |
| gen_points = [(row[0], row[2]) for row in rows] |
| prefill_poly = points_to_polyline(prefill_points, x_min, x_max, prefill_max, plot) |
| gen_poly = points_to_polyline(gen_points, x_min, x_max, gen_max, plot) |
|
|
| parts = [ |
| '<?xml version="1.0" encoding="UTF-8"?>', |
| f'<svg xmlns="http://www.w3.org/2000/svg" width="{width}" height="{height}" viewBox="0 0 {width} {height}">', |
| "<style>", |
| "text { font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', sans-serif; }", |
| ".title { font-size: 26px; font-weight: 700; fill: #1f2933; }", |
| ".axis-label { font-size: 14px; font-weight: 600; fill: #334155; }", |
| ".tick { font-size: 12px; fill: #64748b; }", |
| ".legend { font-size: 13px; font-weight: 600; fill: #1f2933; }", |
| "</style>", |
| f'<rect width="{width}" height="{height}" fill="#ffffff"/>', |
| f'<text class="title" x="{width / 2:.1f}" y="34" text-anchor="middle">{html.escape(title)}</text>', |
| ] |
|
|
| |
| for tick in prefill_ticks: |
| y = bottom - tick / prefill_max * plot_height |
| parts.append(f'<line x1="{left}" y1="{y:.2f}" x2="{right}" y2="{y:.2f}" stroke="{GRID_COLOR}" stroke-width="1"/>') |
| parts.append(f'<text class="tick" x="{left - 12}" y="{y + 4:.2f}" text-anchor="end">{fmt_tick(tick)}</text>') |
|
|
| |
| for tick in gen_ticks: |
| y = bottom - tick / gen_max * plot_height |
| parts.append(f'<text class="tick" x="{right + 12}" y="{y + 4:.2f}" text-anchor="start">{fmt_tick(tick)}</text>') |
|
|
| for tick in x_ticks: |
| x = left + (tick - x_min) / (x_max - x_min) * plot_width |
| parts.append(f'<line x1="{x:.2f}" y1="{top}" x2="{x:.2f}" y2="{bottom}" stroke="{GRID_COLOR}" stroke-width="1"/>') |
| parts.append(f'<text class="tick" x="{x:.2f}" y="{bottom + 24}" text-anchor="middle">{fmt_tick(tick)}</text>') |
|
|
| parts.extend( |
| [ |
| f'<line x1="{left}" y1="{top}" x2="{left}" y2="{bottom}" stroke="{AXIS_COLOR}" stroke-width="1.4"/>', |
| f'<line x1="{right}" y1="{top}" x2="{right}" y2="{bottom}" stroke="{AXIS_COLOR}" stroke-width="1.4"/>', |
| f'<line x1="{left}" y1="{bottom}" x2="{right}" y2="{bottom}" stroke="{AXIS_COLOR}" stroke-width="1.4"/>', |
| f'<text class="axis-label" x="{width / 2:.1f}" y="{height - 20}" text-anchor="middle">ctx size</text>', |
| f'<text class="axis-label" x="22" y="{top + plot_height / 2:.1f}" text-anchor="middle" transform="rotate(-90 22 {top + plot_height / 2:.1f})">prefill t/s</text>', |
| f'<text class="axis-label" x="{width - 22}" y="{top + plot_height / 2:.1f}" text-anchor="middle" transform="rotate(90 {width - 22} {top + plot_height / 2:.1f})">generation t/s</text>', |
| f'<polyline fill="none" stroke="{PREFILL_COLOR}" stroke-width="3" stroke-linecap="round" stroke-linejoin="round" points="{prefill_poly}"/>', |
| f'<polyline fill="none" stroke="{GEN_COLOR}" stroke-width="3" stroke-linecap="round" stroke-linejoin="round" points="{gen_poly}"/>', |
| ] |
| ) |
|
|
| legend_x = right - 170 |
| legend_y = top + 18 |
| parts.extend( |
| [ |
| f'<rect x="{legend_x - 14}" y="{legend_y - 18}" width="176" height="62" rx="6" fill="#ffffff" stroke="#cbd5e1"/>', |
| f'<rect x="{legend_x}" y="{legend_y - 6}" width="12" height="12" fill="{PREFILL_COLOR}"/>', |
| f'<text class="legend" x="{legend_x + 22}" y="{legend_y + 5}">prefill</text>', |
| f'<rect x="{legend_x}" y="{legend_y + 20}" width="12" height="12" fill="{GEN_COLOR}"/>', |
| f'<text class="legend" x="{legend_x + 22}" y="{legend_y + 31}">generation</text>', |
| ] |
| ) |
|
|
| parts.append("</svg>") |
| return "\n".join(parts) + "\n" |
|
|
|
|
| def frange(start, stop, step): |
| value = start |
| |
| while value <= stop + step * 0.001: |
| yield round(value, 10) |
| value += step |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser(description="Plot ds4-bench throughput CSV data as SVG.") |
| parser.add_argument("csv", type=Path, help="input CSV produced by ds4-bench") |
| parser.add_argument("-o", "--output", type=Path, help="output SVG path") |
| parser.add_argument("--title", help="graph title; defaults to a title derived from the CSV name") |
| parser.add_argument("--width", type=int, default=960, help="SVG width in pixels") |
| parser.add_argument("--height", type=int, default=540, help="SVG height in pixels") |
| args = parser.parse_args() |
|
|
| output = args.output |
| if output is None: |
| output = args.csv.with_name(f"{args.csv.stem}_ts.svg") |
|
|
| rows = read_points(args.csv) |
| title = args.title or derive_title(args.csv) |
| output.write_text(render_svg(rows, title, args.width, args.height), encoding="utf-8") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
