"""热路径基准：给「优化前后」提供可对比的数字。

为什么需要它
------------
这个服务同时扛着三条高频路径，而它们**都在重复序列化整局状态**：

    * 每次 ``POST /api/actions``（TUI 客户端固定带 ``full=1``）
    * ``Publisher`` 每 250ms 一次的 SSE 推送
    * ``GET /api/world`` / ``/api/farms`` 的排行榜（遍历所有牧场）

改任何一处序列化代码，代价都会乘以上面这些倍数。所以先有数字，再动手。

跑法::

    uv run python tools/bench.py
    uv run python tools/bench.py --repeat 5    # 每项取最快一轮，抗系统噪声

输出两段：

1. **重复计数**——一次请求到底触发了几次 ``GameState.to_dict`` /
   ``net_worth`` / ``available_actions``。这是优化最直接的证据。
2. **耗时**——各场景的 µs/op 与 ops/s。

参考数字（16 块地 + 8 只牲畜的最坏形态，本机实测）
--------------------------------------------------
2026-09-19 那次优化（把 brief/summary 改成读标量、net_worth 只算一次）前后::

    场景                       优化前      优化后
    brief(session, hub)        63.6 µs     15.4 µs      -76%
    session.summary()          60.2 µs     13.8 µs      -77%
    leaderboard(32 farms)    1981.7 µs    445.8 µs      -78%
    engine.snapshot()          57.3 µs     45.0 µs      -21%
    POST /api/actions full=1  464.7 µs    390.2 µs      -16%
    GET  /api/actions         398.3 µs    333.2 µs      -16%
    GET  /api/world           193.6 µs    133.5 µs      -31%

一次 ``POST /api/actions?full=1`` 的重复工作量：``to_dict`` 2 → 1 次、
``snapshot`` 2 → 1 次、``net_worth`` 6 → 2 次。
"""

from __future__ import annotations

import argparse
import asyncio
import json
import os
import sys
import tempfile
import time
from collections.abc import Callable
from typing import Any

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from farm import config, models
from farm.engine import FarmEngine
from farm.models import Animal, Plot
from farm.server.hub import ConnectionHub, build_state_message
from farm.server.session import SessionManager

try:
    from farm.server.app import ServerConfig, brief, create_app
except ImportError as exc:  # pragma: no cover - 环境没装服务器依赖时
    raise SystemExit(f"本基准需要服务器依赖（fastapi 等）：{exc}") from None

SPEED_FARM = "bench"
ACTION_BODY = b'{"action":"pause","params":{}}'


# --------------------------------------------------------------------- 造数据
def fill_endgame(engine: FarmEngine) -> None:
    """把一个引擎填成"后期农场"：地种满一半、牲畜拉满。

    挑后期形态是因为它才是序列化的最坏情况——16 块地 + 8 只牲畜。
    """
    state = engine.state
    state.level = 6  # 6 + 2*(6-1) = 16，正好顶到 MAX_PLOTS
    state.gold = 50000
    state.ap = config.START_AP
    state.feed = 40
    state.plots = [Plot(i) for i in range(config.MAX_PLOTS)]
    for i, plot in enumerate(state.plots):
        if i % 2 == 0:  # 一半种着、一半空着，两条分支都走到
            plot.crop = "corn"
            plot.moisture = config.WATER_CAPACITY
            plot.growth = 2.0
    state.animals = [Animal("cow", aid=i) for i in range(config.MAX_ANIMALS)]
    for i, animal in enumerate(state.animals):
        animal.fed = i % 2 == 0
        animal.pending = 1 if i % 3 == 0 else 0


def make_session(save_dir: str) -> Any:
    manager = SessionManager(save_dir=save_dir, tick_seconds=config.DEFAULT_TICK_SECONDS)
    session = manager.get(SPEED_FARM, create=True)
    fill_endgame(session.engine)
    return session


# --------------------------------------------------------------------- 计时
def measure(func: Callable[[], Any], repeat: int = 5, target: float = 0.15) -> float:
    """返回单次调用耗时（秒）。先估一次，再决定循环几轮让总时长接近 target。"""
    start = time.perf_counter()
    func()
    single = max(time.perf_counter() - start, 1e-9)
    number = max(1, min(20000, int(target / single)))
    best = float("inf")
    for _ in range(repeat):
        start = time.perf_counter()
        for _ in range(number):
            func()
        best = min(best, time.perf_counter() - start)
    return best / number


class Scenario:
    """一条被测路径。"""

    def __init__(self, name: str, note: str, func: Callable[[], Any]) -> None:
        self.name = name
        self.note = note
        self.func = func
        self.per_op = 0.0

    def run(self, repeat: int) -> None:
        self.per_op = measure(self.func, repeat=repeat)


# --------------------------------------------------------------------- ASGI 直连
async def asgi_call(
    app: Any, method: str, path: str, query: str = "", body: bytes = b""
) -> tuple[int, dict[str, Any]]:
    """不经过任何 HTTP 客户端，直接把请求喂给 ASGI 应用。

    这样基准里既能测到真实路由/中间件开销，又不需要 httpx 之类的依赖。

    注意：这条路径**不会跑 lifespan**，所以它测不到时钟与推送线程的启动；
    那部分由 ``tests/test_server.py`` 的真机用例覆盖。
    """
    scope: dict[str, Any] = {
        "type": "http",
        "asgi": {"version": "3.0", "spec_version": "2.3"},
        "http_version": "1.1",
        "method": method,
        "scheme": "http",
        "path": path,
        "raw_path": path.encode(),
        "query_string": query.encode(),
        "root_path": "",
        "headers": [
            (b"host", b"127.0.0.1"),
            (b"content-type", b"application/json"),
            (b"content-length", str(len(body)).encode()),
        ],
        "client": ("127.0.0.1", 41234),
        "server": ("127.0.0.1", 8765),
    }
    sent: list[dict[str, Any]] = []

    async def receive() -> dict[str, Any]:
        return {"type": "http.request", "body": body, "more_body": False}

    async def send(message: dict[str, Any]) -> None:
        sent.append(message)

    await app(scope, receive, send)
    status = next(m["status"] for m in sent if m["type"] == "http.response.start")
    raw = b"".join(m.get("body", b"") for m in sent if m["type"] == "http.response.body")
    return status, json.loads(raw) if raw else {}


def call_once(app: Any, method: str, path: str, query: str, body: bytes) -> None:
    """同步跑一次 ASGI 请求（给"只跑一次"的计数场景用）。"""
    loop = asyncio.new_event_loop()
    try:
        status, _ = loop.run_until_complete(asgi_call(app, method, path, query, body))
        if status >= 400:
            raise RuntimeError(f"{method} {path} 返回 {status}")
    finally:
        loop.close()


def measure_http(
    app: Any, method: str, path: str, query: str, body: bytes, repeat: int = 5, number: int = 300
) -> float:
    """批量跑 HTTP 调用；循环放在事件循环内部，避免把建循环的开销算进去。"""

    async def burst() -> None:
        for _ in range(number):
            status, _ = await asgi_call(app, method, path, query, body)
            if status >= 400:
                raise RuntimeError(f"{method} {path} 返回 {status}")

    loop = asyncio.new_event_loop()
    try:
        loop.run_until_complete(burst())  # 预热，顺带把牧场建出来
        best = float("inf")
        for _ in range(repeat):
            start = time.perf_counter()
            loop.run_until_complete(burst())
            best = min(best, time.perf_counter() - start)
        return best / number
    finally:
        loop.close()


# --------------------------------------------------------------------- 重复计数
def count_calls(func: Callable[[], Any], patches: list[tuple[Any, str, str]]) -> dict[str, int]:
    """给若干方法/属性装计数器，跑一次 ``func()``，返回各标签的命中次数。

    ``patches`` 每项是 ``(类, 属性名, 标签)``；属性会自动识破 property 与普通方法。
    """
    counters: dict[str, int] = {label: 0 for _, _, label in patches}
    originals: list[tuple[Any, str, Any]] = []

    for cls, attr, label in patches:
        original = getattr(cls, attr)
        if isinstance(original, property):
            getter = original.fget

            def count_property(self: Any, _getter: Any = getter, _label: str = label) -> Any:
                counters[_label] += 1
                return _getter(self)

            replacement: Any = property(count_property)
        else:

            def count_method(
                self: Any, *args: Any, _orig: Any = original, _label: str = label, **kwargs: Any
            ) -> Any:
                counters[_label] += 1
                return _orig(self, *args, **kwargs)

            replacement = count_method

        originals.append((cls, attr, original))
        setattr(cls, attr, replacement)

    try:
        func()
    finally:
        for cls, attr, original in originals:
            setattr(cls, attr, original)
    return counters


# --------------------------------------------------------------------- 主流程
def main() -> int:
    parser = argparse.ArgumentParser(description="SimpleFarm 热路径基准")
    parser.add_argument("--repeat", type=int, default=5, help="每个场景取最快的一轮（默认 5）")
    parser.add_argument("--farms", type=int, default=32, help="排行榜场景的牧场数（默认 32）")
    args = parser.parse_args()

    save_dir = tempfile.mkdtemp(prefix="farm-bench-")
    session = make_session(save_dir)
    hub = ConnectionHub()
    engine = session.engine

    cfg = ServerConfig(save_dir=save_dir, tick_seconds=config.DEFAULT_TICK_SECONDS)
    app = create_app(cfg)
    fill_endgame(app.state.manager.get(SPEED_FARM, create=True).engine)

    # ---------------------------------------------------------- 重复计数
    query = f"farm={SPEED_FARM}&full=1"
    print("=" * 74)
    print("重复工作量")
    print("=" * 74)
    per_action = count_calls(
        lambda: call_once(app, "POST", "/api/actions", query, ACTION_BODY),
        [
            (models.GameState, "to_dict", "to_dict"),
            (FarmEngine, "snapshot", "snapshot"),
            (FarmEngine, "brief_source", "brief_source"),
        ],
    )
    expected_per_action = {"to_dict": 1, "snapshot": 1, "brief_source": 1}
    print("  一次 POST /api/actions?full=1:")
    for label, value in per_action.items():
        want = expected_per_action[label]
        mark = "OK" if value == want else f"!! 期望 {want}"
        print(f"    {label:16} {value:3d} 次   {mark}")

    # 动作回执与完整快照各需要一次 net_worth，所以单独量"一次 to_dict 里算几次"
    per_to_dict = count_calls(engine.state.to_dict, [(models.GameState, "net_worth", "net_worth")])
    count = per_to_dict["net_worth"]
    print(
        f"  一次 GameState.to_dict():  net_worth {count} 次   {'OK' if count == 1 else '!! 应为 1'}"
    )
    print()

    # ---------------------------------------------------------- 耗时
    scenarios = [
        Scenario("engine.snapshot()", "完整序列化整局（16 地 + 8 牲畜）", engine.snapshot),
        Scenario("state.net_worth", "净资产属性（每次访问都重算）", lambda: engine.state.net_worth),
        Scenario("engine.available_actions()", "GET /api/actions 的主体", engine.available_actions),
        Scenario(
            "brief(session, hub)", "动作回执的精简状态（只读标量）", lambda: brief(session, hub)
        ),
        Scenario("session.snapshot()", "full=1 时回执附带的完整快照", session.snapshot),
        Scenario(
            "build_state_message()",
            "SSE 每 250ms 一次的推送",
            lambda: build_state_message(session, 1),
        ),
        Scenario("session.summary()", "排行榜单行摘要（只读标量）", session.summary),
    ]

    # 排行榜：N 个牧场，每个都走一遍 summary()
    league_dir = tempfile.mkdtemp(prefix="farm-league-")
    league = SessionManager(save_dir=league_dir, tick_seconds=config.DEFAULT_TICK_SECONDS)
    for i in range(args.farms):
        fill_endgame(league.get(f"farm-{i:03d}", create=True).engine)
    scenarios.append(
        Scenario(f"leaderboard({args.farms} farms)", "GET /api/world 的排行榜", league.leaderboard)
    )

    print("=" * 74)
    print(f"耗时（每项取 {args.repeat} 轮最快，越大越慢）")
    print("=" * 74)
    print(f"  {'场景':<32} {'µs/op':>10} {'ops/s':>10}   说明")
    for scenario in scenarios:
        scenario.run(args.repeat)
        us = scenario.per_op * 1e6
        print(f"  {scenario.name:<32} {us:10.1f} {1 / scenario.per_op:10.0f}   {scenario.note}")

    # ---------------------------------------------------------- HTTP 端到端
    print()
    print("=" * 74)
    print("HTTP 端到端（真实路由 + 响应装配；不含 lifespan）")
    print("=" * 74)
    http_cases = [
        ("GET  /api/actions", "GET", "/api/actions", f"farm={SPEED_FARM}", b""),
        ("POST /api/actions full=1", "POST", "/api/actions", query, ACTION_BODY),
        ("GET  /api/state", "GET", "/api/state", f"farm={SPEED_FARM}", b""),
        ("GET  /api/world", "GET", "/api/world", f"farm={SPEED_FARM}", b""),
    ]
    print(f"  {'场景':<32} {'µs/op':>10} {'ops/s':>10}")
    for label, method, path, case_query, body in http_cases:
        per_op = measure_http(app, method, path, case_query, body, repeat=args.repeat)
        print(f"  {label:<32} {per_op * 1e6:10.1f} {1 / per_op:10.0f}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())