Files
JWeBot/tools/performance-baseline.py
T

765 lines
23 KiB
Python
Executable File

#!/usr/bin/env python3
"""Measure local JWeBot latency and summarize runtime telemetry."""
from __future__ import annotations
import argparse
import gc
import importlib
import json
import math
import os
import platform
import shutil
import subprocess
import sys
import tempfile
import time
from contextlib import contextmanager
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Iterable, Iterator, Mapping, Sequence
from PIL import Image, ImageDraw
PROJECT_ROOT = Path(__file__).resolve().parents[1]
CONFIG_SRC = PROJECT_ROOT / "packages" / "jwebot-config" / "src"
SDK_SRC = PROJECT_ROOT / "packages" / "jwebot-sdk" / "src"
KERNEL_SRC = PROJECT_ROOT / "packages" / "jwebot-kernel" / "src"
HOST_SRC = PROJECT_ROOT / "packages" / "jwebot-extension-host" / "src"
OFFICIAL_SRC = (
PROJECT_ROOT
/ "packages"
/ "jwebot-official-extensions"
/ "src"
)
sys.path[:0] = [
str(PROJECT_ROOT),
str(CONFIG_SRC),
str(SDK_SRC),
str(OFFICIAL_SRC),
str(KERNEL_SRC),
str(HOST_SRC),
]
os.environ["JWEBOT_CONFIG_ROOT"] = str(
PROJECT_ROOT / ".benchmark-config-does-not-exist"
)
import wechat_local_ocr as ocr # noqa: E402
from jwebot_kernel import ( # noqa: E402
ExtensionHostClient,
build_release,
install_release,
sign_release_record,
)
from jwebot_kernel import edge_runtime # noqa: E402
REPORT_SCHEMA_VERSION = 1
PUBLISHER = "performance-baseline"
def percentile(values: Sequence[float], quantile: float) -> float:
"""Return an interpolated percentile for a non-empty sample."""
if not values:
raise ValueError("percentile requires at least one value")
if not 0 <= quantile <= 100:
raise ValueError("percentile must be between 0 and 100")
ordered = sorted(float(value) for value in values)
position = (len(ordered) - 1) * quantile / 100
lower = math.floor(position)
upper = math.ceil(position)
if lower == upper:
return ordered[lower]
fraction = position - lower
return ordered[lower] + (ordered[upper] - ordered[lower]) * fraction
def summarize_ms(values: Sequence[float]) -> dict[str, float | int]:
if not values:
raise ValueError("latency summary requires at least one value")
return {
"samples": len(values),
"mean_ms": round(sum(values) / len(values), 3),
"min_ms": round(min(values), 3),
"p50_ms": round(percentile(values, 50), 3),
"p95_ms": round(percentile(values, 95), 3),
"max_ms": round(max(values), 3),
}
def measure_case(
name: str,
operation: Callable[[], object],
*,
iterations: int,
warmups: int,
category: str,
) -> dict[str, Any]:
for _ in range(warmups):
operation()
samples: list[float] = []
gc_enabled = gc.isenabled()
gc.disable()
try:
for _ in range(iterations):
started = time.perf_counter_ns()
operation()
samples.append((time.perf_counter_ns() - started) / 1_000_000)
finally:
if gc_enabled:
gc.enable()
return {
"name": name,
"category": category,
"status": "measured",
**summarize_ms(samples),
}
def skipped_case(name: str, category: str, reason: str) -> dict[str, Any]:
return {
"name": name,
"category": category,
"status": "skipped",
"reason": reason[:160],
}
def _synthetic_bubble(width: int = 500, height: int = 180) -> Image.Image:
image = Image.new("RGB", (width, height), ocr.INCOMING_COLOR)
draw = ImageDraw.Draw(image)
for top in (22, 58, 94, 130):
draw.rectangle((24, top, width - 28, top + 12), fill=(38, 38, 38))
return image
def _synthetic_message_view(width: int, height: int) -> Image.Image:
image = Image.new("RGB", (width, height), (245, 245, 245))
draw = ImageDraw.Draw(image)
for index, top in enumerate((60, 180, 310, 455)):
bubble_height = 72 + index * 8
draw.rounded_rectangle(
(64, top, 430, top + bubble_height),
radius=6,
fill=ocr.INCOMING_COLOR,
)
draw.rectangle(
(86, top + 20, 380, top + 31),
fill=(35, 35, 35),
)
return image
def benchmark_ocr(iterations: int, warmups: int) -> list[dict[str, Any]]:
bubble = _synthetic_bubble()
messages = [
ocr.IncomingMessage(
sender=f"sender-{index}",
text=f"message-{index}",
confidence=0.99,
sender_confidence=0.99,
body_confidence=0.99,
y=index * 30,
)
for index in range(18)
]
previous = [message.signature for message in messages[:-1]]
frame_image = Image.new("RGB", (2, 2), "white")
sequence = 0
def buffer_round_trip() -> None:
nonlocal sequence
buffer = ocr.CaptureFrameBuffer(12)
for _ in range(12):
sequence += 1
buffer.offer(
ocr.CapturedFrame(
image=frame_image,
layout_hash=f"layout-{sequence}",
visual_hash=f"visual-{sequence}",
observed_at="2026-10-14T00:00:00+00:00",
captured_monotonic=float(sequence),
sequence=sequence,
latest_synced=True,
)
)
buffer.get_latest()
results = [
measure_case(
"ocr.text_line_regions",
lambda: ocr.text_line_regions(bubble),
iterations=iterations,
warmups=warmups,
category="ocr-preprocess",
),
measure_case(
"ocr.sequence_suffix",
lambda: ocr.new_sequence_suffix(previous, messages[1:]),
iterations=iterations,
warmups=warmups,
category="ocr-sequence",
),
measure_case(
"ocr.capture_buffer_latest",
buffer_round_trip,
iterations=iterations,
warmups=warmups,
category="capture-queue",
),
]
with tempfile.TemporaryDirectory(
prefix="jwebot-performance-ocr-queue-"
) as value:
original_queue = ocr.QUEUE_DIR
ocr.QUEUE_DIR = Path(value)
counter = 0
def durable_spool() -> None:
nonlocal counter
counter += 1
ocr.queue_event(
{
"event_id": f"queue-{counter}",
"observed_at": "2026-10-14T00:00:00+00:00",
"sender": "PerformanceUser",
"text": "@JWeBot /翻硬币",
"mention_bot": True,
"group_fingerprint": "performance-group",
}
)
try:
results.append(
measure_case(
"ocr.local_event.durable_spool",
durable_spool,
iterations=iterations,
warmups=warmups,
category="ocr-persistence",
)
)
finally:
ocr.QUEUE_DIR = original_queue
ocr.QUEUE_WAKE_EVENT.clear()
try:
importlib.import_module("cv2")
except ModuleNotFoundError:
results.extend(
[
skipped_case(
"ocr.frame_fingerprint.full_view",
"ocr-preprocess",
"opencv-python-headless is not installed",
),
skipped_case(
"ocr.frame_fingerprint.roi",
"ocr-preprocess",
"opencv-python-headless is not installed",
),
]
)
return results
message_width = ocr.MESSAGE_BOX[2] - ocr.MESSAGE_BOX[0]
message_height = ocr.MESSAGE_BOX[3] - ocr.MESSAGE_BOX[1]
full_view = _synthetic_message_view(message_width, message_height)
region = _synthetic_message_view(540, 220)
results.extend(
[
measure_case(
"ocr.frame_fingerprint.full_view",
lambda: ocr.incoming_frame_hashes(full_view),
iterations=iterations,
warmups=warmups,
category="ocr-preprocess",
),
measure_case(
"ocr.frame_fingerprint.roi",
lambda: ocr.incoming_frame_hashes(region),
iterations=iterations,
warmups=warmups,
category="ocr-preprocess",
),
]
)
return results
def _edge_config(root: Path) -> edge_runtime.EdgeConfig:
return edge_runtime.EdgeConfig(
socket_path=root / "edge.sock",
database_path=root / "edge.sqlite3",
cloud_host="",
cloud_agent="wechat-responder",
deepseek_key="",
deepseek_base_url="https://api.deepseek.com",
deepseek_model=edge_runtime.EDGE_FLASH_MODEL,
memory_remote_path="/home/node/.openclaw/workspace/MEMORY.md",
delivery_enabled=False,
steam_api_key="",
steam_id64="",
steam_proxy_url=edge_runtime.DEFAULT_STEAM_MIHOMO_PROXY_URL,
extension_mode="in-process",
extension_state_root=root / "extensions",
extension_allowed_signers=root / "allowed_signers",
extension_socket_root=root / "sockets",
telemetry_enabled=False,
)
def _event(identifier: str, text: str) -> dict[str, object]:
return {
"event_id": identifier,
"group_fingerprint": "performance-group",
"sender": "PerformanceUser",
"text": text,
"observed_at": "2026-10-14T00:00:00+00:00",
"mention_bot": True,
"confidence": 0.99,
"sender_confidence": 0.99,
"body_confidence": 0.99,
"source": "performance-baseline",
}
@contextmanager
def _edge_application() -> Iterator[edge_runtime.EdgeApplication]:
with tempfile.TemporaryDirectory(prefix="jwebot-performance-edge-") as value:
application = edge_runtime.EdgeApplication(_edge_config(Path(value)))
try:
yield application
finally:
application.close()
def benchmark_edge(iterations: int, warmups: int) -> list[dict[str, Any]]:
total = iterations + warmups
results: list[dict[str, Any]] = []
with _edge_application() as application:
counter = 0
def local_command() -> None:
nonlocal counter
counter += 1
response = application.accept(
_event(f"local-{counter}", "@JWeBot /翻硬币")
)
if response.get("status") != "reply":
raise RuntimeError("local benchmark did not return a reply")
results.append(
measure_case(
"edge.local_command.in_process",
local_command,
iterations=iterations,
warmups=warmups,
category="edge-ledger",
)
)
with _edge_application() as application:
counter = 0
def prepare_ack() -> None:
nonlocal counter
counter += 1
response = application.accept(
_event(f"ack-prepare-{counter}", "@JWeBot benchmark chat")
)
if response.get("fast_ack") is not True:
raise RuntimeError("chat benchmark did not prepare an ACK")
results.append(
measure_case(
"edge.chat_ack.prepare",
prepare_ack,
iterations=iterations,
warmups=warmups,
category="edge-ledger",
)
)
with _edge_application() as application:
identifiers = [f"ack-confirm-{index}" for index in range(total)]
for identifier in identifiers:
application.accept(
_event(identifier, "@JWeBot benchmark confirmation")
)
iterator = iter(identifiers)
def confirm_ack() -> None:
response = application.confirm_fast_acknowledgement(
next(iterator),
delivered=True,
)
if response.get("acknowledged") is not True:
raise RuntimeError("chat benchmark did not confirm an ACK")
results.append(
measure_case(
"edge.chat_ack.confirm",
confirm_ack,
iterations=iterations,
warmups=warmups,
category="edge-ledger",
)
)
with _edge_application() as application:
identifiers = [f"queue-claim-{index}" for index in range(total)]
for identifier in identifiers:
application.accept(_event(identifier, "@JWeBot queued benchmark"))
application.confirm_fast_acknowledgement(
identifier,
delivered=True,
)
def claim_queue() -> None:
row = application.ledger.claim_direct_chat()
if row is None:
raise RuntimeError("chat benchmark queue is empty")
results.append(
measure_case(
"edge.direct_queue.claim",
claim_queue,
iterations=iterations,
warmups=warmups,
category="edge-ledger",
)
)
return results
def _publisher_key(root: Path) -> tuple[Path, Path]:
executable = shutil.which("ssh-keygen")
if executable is None:
raise RuntimeError("ssh-keygen is unavailable")
private_key = root / "publisher"
completed = subprocess.run(
[
executable,
"-q",
"-t",
"ed25519",
"-N",
"",
"-f",
str(private_key),
],
capture_output=True,
check=False,
)
if completed.returncode != 0:
raise RuntimeError("cannot create the benchmark publisher key")
allowed_signers = root / "allowed_signers"
allowed_signers.write_text(
f"{PUBLISHER} "
+ private_key.with_suffix(".pub").read_text(encoding="utf-8"),
encoding="utf-8",
)
return private_key, allowed_signers
def _host_wrapper(root: Path) -> Path:
paths = [str(SDK_SRC), str(KERNEL_SRC), str(HOST_SRC)]
try:
import tomli
except ModuleNotFoundError:
pass
else:
paths.append(str(Path(tomli.__file__).resolve().parent.parent))
wrapper = root / "host-wrapper.py"
wrapper.write_text(
"\n".join(
(
"import runpy, sys",
f"sys.path[:0] = {paths!r}",
"runpy.run_module('jwebot_extension_host', run_name='__main__')",
)
)
+ "\n",
encoding="utf-8",
)
return wrapper
@contextmanager
def _signed_coin_host() -> Iterator[
tuple[ExtensionHostClient, Path, Mapping[str, object]]
]:
with tempfile.TemporaryDirectory(prefix="jwebot-performance-host-") as value:
root = Path(value)
private_key, allowed_signers = _publisher_key(root)
artifact, record_path, _ = build_release(
PROJECT_ROOT / "extensions" / "alice" / "coin",
root / "dist",
)
signature_path, _ = sign_release_record(
record_path,
private_key,
PUBLISHER,
)
release_dir = install_release(
artifact,
record_path,
root / "state" / "coin",
signature=signature_path,
allowed_signers=allowed_signers,
expected_publisher=PUBLISHER,
health_check=lambda release: None,
)
client = ExtensionHostClient(
command=(sys.executable, str(_host_wrapper(root))),
timeout_seconds=8,
allowed_signers=allowed_signers,
expected_publisher=PUBLISHER,
plugin_id="coin",
)
request = {
"op": "invoke_command",
"command": "/翻硬币",
"arguments": "",
"event": _event("host-benchmark", "@JWeBot /翻硬币"),
}
yield client, release_dir, request
def benchmark_host(iterations: int, warmups: int) -> list[dict[str, Any]]:
host_iterations = min(iterations, 25)
host_warmups = min(warmups, 3)
try:
with _signed_coin_host() as (client, release_dir, request):
health = measure_case(
"host.signed_process.health",
lambda: client.invoke(release_dir, {"op": "health"}),
iterations=host_iterations,
warmups=host_warmups,
category="extension-host",
)
def invoke_command() -> None:
response = client.invoke(
release_dir,
request,
service_handler=(
lambda profile, operation, payload: {
"text": "benchmark reply"
}
),
)
if response.get("status") != "ok":
raise RuntimeError("signed Host benchmark failed")
command = measure_case(
"host.signed_process.command",
invoke_command,
iterations=host_iterations,
warmups=host_warmups,
category="extension-host",
)
return [health, command]
except Exception as error:
reason = f"{type(error).__name__}: {error}"
return [
skipped_case(
"host.signed_process.health",
"extension-host",
reason,
),
skipped_case(
"host.signed_process.command",
"extension-host",
reason,
),
]
def parse_telemetry_line(line: str) -> dict[str, Any] | None:
text = line.strip()
marker = "telemetry "
if marker in text:
text = text.split(marker, 1)[1].strip()
elif not text.startswith("{"):
return None
try:
value = json.loads(text)
except json.JSONDecodeError:
return None
if not isinstance(value, dict) or not isinstance(value.get("name"), str):
return None
return value
def read_telemetry(paths: Iterable[str]) -> list[dict[str, Any]]:
records: list[dict[str, Any]] = []
for value in paths:
if value == "-":
lines = sys.stdin
records.extend(
record
for line in lines
if (record := parse_telemetry_line(line)) is not None
)
continue
with Path(value).open("r", encoding="utf-8", errors="replace") as source:
records.extend(
record
for line in source
if (record := parse_telemetry_line(line)) is not None
)
return records
def summarize_telemetry(records: Sequence[Mapping[str, Any]]) -> dict[str, Any]:
latencies: dict[str, list[float]] = {}
statuses: dict[str, int] = {}
event_hashes: set[str] = set()
for record in records:
name = str(record.get("name", "unknown"))[:80]
status = str(record.get("status", "unknown"))[:24]
statuses[f"{name}:{status}"] = statuses.get(f"{name}:{status}", 0) + 1
event_hash = record.get("event_hash")
if isinstance(event_hash, str) and event_hash:
event_hashes.add(event_hash)
elapsed = record.get("elapsed_ms")
if (
isinstance(elapsed, (int, float))
and not isinstance(elapsed, bool)
and math.isfinite(float(elapsed))
and float(elapsed) >= 0
):
latencies.setdefault(name, []).append(float(elapsed))
stages = [
{"name": name, **summarize_ms(values)}
for name, values in latencies.items()
]
stages.sort(
key=lambda value: (
-float(value["p95_ms"]),
str(value["name"]),
)
)
return {
"records": len(records),
"events": len(event_hashes),
"stages": stages,
"statuses": dict(sorted(statuses.items())),
}
def build_report(
*,
iterations: int,
warmups: int,
include_local: bool,
include_host: bool,
telemetry_paths: Sequence[str],
) -> dict[str, Any]:
benchmarks: list[dict[str, Any]] = []
if include_local:
benchmarks.extend(benchmark_ocr(iterations, warmups))
benchmarks.extend(benchmark_edge(iterations, warmups))
if include_host:
benchmarks.extend(benchmark_host(iterations, warmups))
report: dict[str, Any] = {
"schema_version": REPORT_SCHEMA_VERSION,
"generated_at": datetime.now(timezone.utc).isoformat(
timespec="seconds"
),
"version": (PROJECT_ROOT / "VERSION").read_text(
encoding="utf-8"
).strip(),
"environment": {
"platform": platform.platform(),
"python": platform.python_version(),
"machine": platform.machine(),
},
"parameters": {
"iterations": iterations,
"warmups": warmups,
"host_enabled": include_host,
},
"configured_latency_windows_ms": {
"damage_debounce": round(
ocr.DAMAGE_DEBOUNCE_SECONDS * 1000
),
"damage_max_burst": round(
ocr.DAMAGE_MAX_BURST_SECONDS * 1000
),
"edge_idle_worker_poll": 1000,
"follow_latest_settle": round(
ocr.FOLLOW_LATEST_SETTLE_SECONDS * 1000
),
"ocr_responder_fallback_poll": 1000,
"window_layout_settle": round(
ocr.WINDOW_LAYOUT_SETTLE_SECONDS * 1000
),
},
"benchmarks": benchmarks,
}
if telemetry_paths:
report["telemetry"] = summarize_telemetry(
read_telemetry(telemetry_paths)
)
return report
def main(argv: Sequence[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--iterations", type=int, default=50)
parser.add_argument("--warmups", type=int, default=5)
parser.add_argument(
"--skip-host",
action="store_true",
help="Skip signed Extension Host process measurements.",
)
parser.add_argument(
"--telemetry",
action="append",
default=[],
metavar="PATH",
help="Read a telemetry log; use '-' for standard input.",
)
parser.add_argument(
"--telemetry-only",
action="store_true",
help="Analyze telemetry without running local microbenchmarks.",
)
parser.add_argument("--output", type=Path)
args = parser.parse_args(argv)
if args.iterations < 1:
parser.error("--iterations must be positive")
if args.warmups < 0:
parser.error("--warmups cannot be negative")
if args.telemetry_only and not args.telemetry:
parser.error("--telemetry-only requires at least one --telemetry path")
report = build_report(
iterations=args.iterations,
warmups=args.warmups,
include_local=not args.telemetry_only,
include_host=not args.skip_host,
telemetry_paths=args.telemetry,
)
payload = json.dumps(
report,
ensure_ascii=False,
indent=2,
sort_keys=True,
)
if args.output is not None:
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(payload + "\n", encoding="utf-8")
print(payload)
return 0
if __name__ == "__main__":
raise SystemExit(main())