765 lines
23 KiB
Python
Executable File
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())
|