Something went wrong. Try again.
forked from zzstoatzz.io/atproto-bench
Something went wrong. Try again.
10 kB · 349 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350#!/usr/bin/env python3# /// script# requires-python = ">=3.11"# dependencies = ["plotext"]# ///"""generate verify timing charts from `just verify` output.
produces two SVGs: verify-compute.svg — compute-only breakdown (the interesting part) verify-total.svg — full timeline including network
usage: just verify pfrazee.com 2>&1 | uv run scripts/verify_chart.py just verify pfrazee.com 2>&1 | uv run scripts/verify_chart.py --dir charts/"""
import reimport sysfrom dataclasses import dataclass, fieldfrom pathlib import Path
import plotext as plt
@dataclassclass VerifyResult: name: str handle_ms: float = 0 did_ms: float = 0 fetch_ms: float = 0 car_ms: float = 0 sig_ms: float = 0 walk_ms: float = 0 rebuild_ms: float = 0 total_ms: float = 0 record_count: int = 0 block_count: int = 0
@property def network_ms(self) -> float: return self.handle_ms + self.did_ms + self.fetch_ms
@property def compute_ms(self) -> float: return self.car_ms + self.sig_ms + self.walk_ms + self.rebuild_ms
def parse_output(text: str) -> list[VerifyResult]: results = [] current = None
for line in text.splitlines(): if "=== zat:" in line: current = VerifyResult(name="zig (zat)") elif "=== go" in line: current = VerifyResult(name="go (indigo)") elif "=== rust" in line: current = VerifyResult(name="rust (RustCrypto)")
if current is None: continue
# extract trailing timing value m = re.search(r"(\d+\.?\d*)\s*ms\s*$", line) if not m: m = re.search(r"(\d+\.?\d*)\s*s\s*$", line) if m: ms = float(m.group(1)) * 1000 else: ms = None else: ms = float(m.group(1))
if ms is not None: if "handle" in line.lower() or ("→ did:" in line and "DID doc" not in line): if current.handle_ms == 0: current.handle_ms = ms elif "signing key" in line: current.did_ms = ms elif "CAR" in line and ("parse" in line or "load" in line): current.car_ms = ms elif ("CAR" in line and "MB" in line) or ("CAR" in line and "KB" in line): current.fetch_ms = ms elif "signature" in line.lower() or "VERIFIED" in line: current.sig_ms = ms elif "walk" in line.lower(): current.walk_ms = ms rc = re.search(r"(\d+)\s+records", line) if rc: current.record_count = int(rc.group(1)) elif "rebuild" in line.lower() or "root CID" in line: current.rebuild_ms = ms
# total line tm = re.match( r"total\s+(\d+\.?\d*)(ms|s)\s+\(network:\s+(\d+\.?\d*)(ms|s),\s+compute:\s+(\d+\.?\d*)(ms|s)\)", line.strip(), ) if tm: current.total_ms = float(tm.group(1)) * (1000 if tm.group(2) == "s" else 1) results.append(current) current = None
bc = re.search(r"(\d+)\s+blocks", line) if bc and current: current.block_count = int(bc.group(1))
return results
def show_terminal(results: list[VerifyResult]) -> None: """plotext chart in terminal.""" names = [r.name for r in results] compute = [r.compute_ms for r in results]
plt.clear_figure() plt.theme("dark") plt.title("compute breakdown (ms)") plt.bar(names, compute, color="orange", orientation="h") plt.xlabel("time (ms)") plt.plotsize(80, max(10, len(results) * 3 + 4)) plt.show() print()
for r in results: print( f" {r.name:<20s}" f" car={r.car_ms:>6.1f}ms" f" sig={r.sig_ms:>5.1f}ms" f" walk={r.walk_ms:>6.1f}ms" f" rebuild={r.rebuild_ms:>6.1f}ms" f" total={r.compute_ms:>7.1f}ms" f" (network={r.network_ms:>8.0f}ms)" ) print()
# --- SVG ---
COLORS = { "handle": "#6ea8d9", "DID doc": "#5b9bd5", "fetch": "#4472c4", "CAR parse": "#e8944a", "sig verify": "#ed7d31", "MST walk": "#c55a11", "MST rebuild": "#a04000",}
def generate_svg( results: list[VerifyResult], path: str, title: str, segments_fn, subtitle: str = "",) -> None: bar_h = 38 bar_gap = 20 left_margin = 160 right_margin = 110 top_margin = 55 if subtitle else 45 bottom_margin = 58 chart_w = 520
n = len(results) chart_h = n * (bar_h + bar_gap) - bar_gap w = left_margin + chart_w + right_margin h = top_margin + chart_h + bottom_margin
all_segments = [segments_fn(r) for r in results] max_ms = max(sum(ms for _, ms in segs) for segs in all_segments) if max_ms == 0: max_ms = 1
lines = [] lines.append( f'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 {w} {h}"' f" font-family=\"'SF Mono', 'Fira Code', 'Cascadia Code', Menlo, monospace\">" ) lines.append(f'<rect width="{w}" height="{h}" fill="#1a1a2e" rx="8"/>')
# title lines.append( f'<text x="{w / 2}" y="28" text-anchor="middle" fill="#e0e0e0"' f' font-size="15" font-weight="600">{title}</text>' ) if subtitle: lines.append( f'<text x="{w / 2}" y="44" text-anchor="middle" fill="#666"' f' font-size="11">{subtitle}</text>' )
# gridlines tick_count = 5 for i in range(tick_count + 1): x = left_margin + chart_w * i / tick_count lines.append( f'<line x1="{x:.1f}" y1="{top_margin - 2}"' f' x2="{x:.1f}" y2="{top_margin + chart_h + 2}"' f' stroke="#262640" stroke-width="1"/>' )
# bars for i, (r, segs) in enumerate(zip(results, all_segments)): y = top_margin + i * (bar_h + bar_gap)
# label lines.append( f'<text x="{left_margin - 14}" y="{y + bar_h / 2 + 5}"' f' text-anchor="end" fill="#c0c0c0" font-size="13">{r.name}</text>' )
# stacked segments x = left_margin for label, ms in segs: if ms <= 0: continue seg_w = max(1, ms / max_ms * chart_w) color = COLORS.get(label, "#888") lines.append( f'<rect x="{x:.1f}" y="{y}" width="{seg_w:.1f}"' f' height="{bar_h}" fill="{color}" rx="3"/>' ) if seg_w > 55: lines.append( f'<text x="{x + seg_w / 2:.1f}" y="{y + bar_h / 2 + 4}"' f' text-anchor="middle" fill="white" font-size="10"' f' font-weight="500">{label}</text>' ) x += seg_w
# total label total = sum(ms for _, ms in segs) lines.append( f'<text x="{x + 10:.1f}" y="{y + bar_h / 2 + 5}"' f' fill="#a0a0a0" font-size="12" font-weight="500">{format_ms(total)}</text>' )
# x-axis axis_y = top_margin + chart_h + 18 for i in range(tick_count + 1): ms = max_ms * i / tick_count x = left_margin + chart_w * i / tick_count lines.append( f'<text x="{x:.1f}" y="{axis_y}" text-anchor="middle"' f' fill="#606060" font-size="10">{format_ms(ms)}</text>' )
# legend legend_y = top_margin + chart_h + 40 used_labels = set() for segs in all_segments: for label, ms in segs: if ms > 0: used_labels.add(label)
lx = left_margin for label in ["handle", "DID doc", "fetch", "CAR parse", "sig verify", "MST walk", "MST rebuild"]: if label not in used_labels: continue color = COLORS[label] lines.append( f'<rect x="{lx:.0f}" y="{legend_y - 8}" width="10" height="10"' f' fill="{color}" rx="2"/>' ) lx += 14 lines.append( f'<text x="{lx:.0f}" y="{legend_y}" fill="#808080" font-size="10">{label}</text>' ) lx += len(label) * 6.2 + 12
lines.append("</svg>") Path(path).write_text("\n".join(lines)) print(f" saved: {path}")
def format_ms(ms: float) -> str: if ms == 0: return "0" if ms >= 1000: return f"{ms / 1000:.1f}s" if ms >= 1: return f"{ms:.0f}ms" return f"{ms:.1f}ms"
def compute_segments(r: VerifyResult): return [ ("CAR parse", r.car_ms), ("sig verify", r.sig_ms), ("MST walk", r.walk_ms), ("MST rebuild", r.rebuild_ms), ]
def total_segments(r: VerifyResult): return [ ("handle", r.handle_ms), ("DID doc", r.did_ms), ("fetch", r.fetch_ms), ("CAR parse", r.car_ms), ("sig verify", r.sig_ms), ("MST walk", r.walk_ms), ("MST rebuild", r.rebuild_ms), ]
def main() -> None: import argparse
parser = argparse.ArgumentParser(description="chart verify output") parser.add_argument("--dir", default=".", help="output directory for SVGs") args = parser.parse_args()
text = sys.stdin.read() results = parse_output(text)
if not results: print("error: no verify results found in input") print("usage: just verify pfrazee.com 2>&1 | uv run scripts/verify_chart.py") sys.exit(1)
show_terminal(results)
out = Path(args.dir) out.mkdir(parents=True, exist_ok=True)
records = results[0].record_count subtitle = f"{records:,} records" if records else ""
generate_svg( results, str(out / "verify-compute.svg"), "AT Protocol trust chain — compute", compute_segments, subtitle=subtitle, ) generate_svg( results, str(out / "verify-total.svg"), "AT Protocol trust chain — total (network + compute)", total_segments, subtitle=subtitle, )
if __name__ == "__main__": main()