230 lines
11 KiB
Python
230 lines
11 KiB
Python
"""运行历史计时分解：对任意会话日志按官方计时模型独立复算并分相。
|
||
|
||
支持两种格式（自动识别）：
|
||
- 参赛队客户端日志：``scripts/capture_practice.py`` / ``SimulatorClient`` 写出的
|
||
JSONL，逐行 ``attempt`` / ``accepted`` / ``step``；
|
||
- 评测机日志：``referee-data/sessions/<case>/log.jsonl``，逐行
|
||
``session_start`` / ``action`` / ``session_end``。
|
||
|
||
复算口径与 ``scripts/verify_q3_practice.py::timing_rows`` 完全一致（题面附件2 §4.1）：
|
||
``T = Σ 距离/5 + 5·N_measure + N_switch + 5·N_clear⁺ + 3·N_clear⁻``，
|
||
起点位置 (0,0)、起始频道 1，``/clear`` 不计换频，移动按整数微秒累计。
|
||
|
||
用途：把"虚拟时间花在哪"变成一条可随时重跑的命令，而不是一次性的分析结论。
|
||
|
||
uv run python scripts/analyze_run_timing.py session.jsonl
|
||
uv run python scripts/analyze_run_timing.py referee-data/sessions/*/log.jsonl --json out.json
|
||
uv run python scripts/analyze_run_timing.py session.jsonl --plan seven # 站点/探点分类
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import math
|
||
from collections import Counter, defaultdict
|
||
from pathlib import Path
|
||
|
||
SPEED_MPS = 5.0 # [题面] 附件1 表1
|
||
MEASURE_S = 5 # [题面]
|
||
SWITCH_S = 1 # [题面]
|
||
CLEAR_SUCCESS_S = 5 # [题面]
|
||
CLEAR_FAIL_S = 3 # [题面]
|
||
START_POSITION = (0.0, 0.0)
|
||
START_CHANNEL = 1
|
||
|
||
|
||
def load_actions(path: Path) -> list[dict]:
|
||
"""读取一份 JSONL，归一化为已接受动作序列。"""
|
||
actions: list[dict] = []
|
||
for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
record = json.loads(line)
|
||
except json.JSONDecodeError:
|
||
continue
|
||
event = record.get("event")
|
||
if event == "accepted":
|
||
request, response = record.get("request", {}), record.get("response", {})
|
||
accepted = bool(response.get("accepted", True))
|
||
elif event == "action":
|
||
request, response = record.get("request", {}), record.get("response", {})
|
||
accepted = bool(response.get("accepted", record.get("status") == 200))
|
||
else:
|
||
continue
|
||
if not accepted:
|
||
continue
|
||
actions.append({
|
||
"path": record.get("path") or request.get("path") or "",
|
||
"position": request.get("position"),
|
||
"channel": request.get("channel"),
|
||
"result": response.get("measure_result") or response.get("clear_result") or "",
|
||
"recorded_s": response.get("virtual_time_s"),
|
||
"elapsed_s": record.get("elapsed_s"),
|
||
})
|
||
return actions
|
||
|
||
|
||
def recompute(actions: list[dict]) -> tuple[list[dict], dict]:
|
||
"""按官方模型逐动作复算；返回明细与汇总（微秒整数累计）。"""
|
||
position, channel, virtual_us = START_POSITION, START_CHANNEL, 0
|
||
rows: list[dict] = []
|
||
totals = Counter()
|
||
distance_total = 0.0
|
||
max_gap_us = 0
|
||
for index, action in enumerate(actions):
|
||
path = action["path"]
|
||
move_s = switches = action_s = 0.0
|
||
if path in ("/measure", "/clear") and action["position"] is not None:
|
||
target = (float(action["position"]["x"]), float(action["position"]["y"]))
|
||
distance_total += math.dist(position, target)
|
||
move_s = math.dist(position, target) / SPEED_MPS
|
||
position = target
|
||
if path == "/measure":
|
||
switches = int(channel != action["channel"])
|
||
channel = action["channel"]
|
||
action_s = MEASURE_S
|
||
else:
|
||
action_s = CLEAR_SUCCESS_S if action["result"] == "success" else CLEAR_FAIL_S
|
||
move_us = round(move_s * 1_000_000)
|
||
virtual_us += move_us + int((switches + action_s) * 1_000_000)
|
||
totals["move_us"] += move_us
|
||
totals["switch_s"] += switches
|
||
totals["action_us"] += int(action_s * 1_000_000)
|
||
totals["count"] += 1
|
||
if action["elapsed_s"]:
|
||
totals["elapsed_s"] += action["elapsed_s"]
|
||
recorded = action["recorded_s"]
|
||
gap_us = ((round(recorded * 1_000_000) - virtual_us)
|
||
if isinstance(recorded, (int, float)) else 0)
|
||
max_gap_us = max(max_gap_us, abs(gap_us))
|
||
rows.append({"index": index, "path": path, "channel": action["channel"],
|
||
"x": position[0], "y": position[1], "result": action["result"],
|
||
"move_us": move_us, "switch_s": switches, "action_s": action_s,
|
||
"recomputed_us": virtual_us, "recorded_s": recorded, "gap_us": gap_us})
|
||
summary = {
|
||
"actions": len(actions),
|
||
"virtual_s": round(virtual_us / 1_000_000, 6),
|
||
"move_s": round(totals["move_us"] / 1_000_000, 3),
|
||
"switch_s": float(totals["switch_s"]),
|
||
"action_s": round(totals["action_us"] / 1_000_000, 3),
|
||
"distance_m": round(distance_total, 1),
|
||
"real_elapsed_s": round(totals["elapsed_s"], 3),
|
||
"max_recorded_gap_s": max_gap_us / 1_000_000,
|
||
}
|
||
return rows, summary
|
||
|
||
|
||
def phase_report(rows: list[dict]) -> dict:
|
||
"""清除前后分相 + 重复复扫检测（定位"时间被重复动作吃掉"的历史问题）。"""
|
||
clear_indices = [r["index"] for r in rows if r["path"] == "/clear"]
|
||
last_clear = max(clear_indices) if clear_indices else None
|
||
measure_keys = defaultdict(int)
|
||
for row in rows:
|
||
if row["path"] == "/measure":
|
||
key = (round(row["x"] * 2), round(row["y"] * 2), row["channel"])
|
||
measure_keys[key] += 1
|
||
repeated = sum(count - 1 for count in measure_keys.values() if count > 1)
|
||
report = {
|
||
"measures": sum(1 for r in rows if r["path"] == "/measure"),
|
||
"clears": len(clear_indices),
|
||
"failed_clears": sum(1 for r in rows
|
||
if r["path"] == "/clear" and r["result"] != "success"),
|
||
"distinct_measure_stops": len(measure_keys),
|
||
"repeated_measure_actions": repeated,
|
||
"repeat_ratio": round(repeated / max(1, len(measure_keys)), 3),
|
||
}
|
||
if last_clear is not None:
|
||
tail = [r for r in rows if r["index"] > last_clear]
|
||
report["last_clear_virtual_s"] = round(rows[last_clear]["recomputed_us"] / 1_000_000, 3)
|
||
report["after_last_clear_virtual_s"] = round(
|
||
sum(r["move_us"] + (r["switch_s"] + r["action_s"]) * 1_000_000 for r in tail)
|
||
/ 1_000_000, 3)
|
||
report["after_last_clear_measures"] = sum(1 for r in tail if r["path"] == "/measure")
|
||
return report
|
||
|
||
|
||
def classify_by_plan(rows: list[dict], points, tol_m: float = 2.0) -> dict:
|
||
"""按给定站点集把 /measure 分成"站点扫频"与"专用探点"，并给出各自移动耗时。"""
|
||
buckets: dict[str, Counter] = defaultdict(Counter)
|
||
for row in rows:
|
||
if row["path"] != "/measure":
|
||
continue
|
||
nearest = min((math.dist((row["x"], row["y"]), p) for p in points), default=math.inf)
|
||
kind = "station" if nearest <= tol_m else "probe"
|
||
buckets[kind]["n"] += 1
|
||
buckets[kind]["move_us"] += row["move_us"]
|
||
return {kind: {"measures": value["n"],
|
||
"move_s": round(value["move_us"] / 1_000_000, 3)}
|
||
for kind, value in buckets.items()}
|
||
|
||
|
||
def _plan_points(name: str):
|
||
if name == "none":
|
||
return None
|
||
if name == "seven":
|
||
from mcm26b.planner.seven import seven_points
|
||
return seven_points()
|
||
if name == "49":
|
||
from mcm26b.coverage import coverage_points
|
||
return coverage_points()
|
||
raise ValueError(f"未知 --plan: {name}")
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
||
parser.add_argument("logs", nargs="+", type=Path, help="会话 JSONL（客户端或评测机格式）")
|
||
parser.add_argument("--plan", choices=("none", "seven", "49"), default="none",
|
||
help="站点集，用于把 /measure 分成站点扫频与专用探点")
|
||
parser.add_argument("--json", type=Path, default=None, help="汇总输出路径")
|
||
parser.add_argument("--top", type=int, default=6, help="逐频道表格显示前 N 个热点频道")
|
||
args = parser.parse_args()
|
||
|
||
points = _plan_points(args.plan)
|
||
payload = {}
|
||
for path in args.logs:
|
||
actions = load_actions(path)
|
||
if not actions:
|
||
print(f"[跳过] {path}：没有已接受动作")
|
||
continue
|
||
rows, summary = recompute(actions)
|
||
report = phase_report(rows)
|
||
total = summary["virtual_s"] or 1.0
|
||
print(f"\n===== {path} =====")
|
||
print(f"虚拟总时间 {summary['virtual_s']:.3f} s "
|
||
f"移动 {summary['move_s']:.1f} s ({100 * summary['move_s'] / total:.1f}%) "
|
||
f"检测/清除动作 {summary['action_s']:.1f} s ({100 * summary['action_s'] / total:.1f}%) "
|
||
f"换频 {summary['switch_s']:.0f} s ({100 * summary['switch_s'] / total:.1f}%)")
|
||
print(f"移动距离 {summary['distance_m']:.1f} m 请求 {summary['actions']} 次 "
|
||
f"现实耗时 {summary['real_elapsed_s']:.3f} s "
|
||
f"与服务器记录最大偏差 {summary['max_recorded_gap_s']:.6f} s")
|
||
print(f"检测 {report['measures']} 清除 {report['clears']}（失败 {report['failed_clears']}） "
|
||
f"不同测点 {report['distinct_measure_stops']} "
|
||
f"重复检测 {report['repeated_measure_actions']}"
|
||
f"（重复率 {report['repeat_ratio']}）")
|
||
if "last_clear_virtual_s" in report:
|
||
print(f"末次清除于 {report['last_clear_virtual_s']:.1f} s；其后又花 "
|
||
f"{report['after_last_clear_virtual_s']:.1f} s 做 "
|
||
f"{report['after_last_clear_measures']} 次检测")
|
||
per_channel = Counter(r["channel"] for r in rows if r["path"] == "/measure")
|
||
hottest = per_channel.most_common(args.top)
|
||
print("检测次数最多的频道：" + "，".join(f"ch{c}:{n}" for c, n in hottest))
|
||
if points is not None:
|
||
print("测点分类：" + json.dumps(classify_by_plan(rows, points),
|
||
ensure_ascii=False))
|
||
payload[str(path)] = {"summary": summary, "phases": report,
|
||
"per_channel_measures": dict(sorted(per_channel.items()))}
|
||
|
||
if args.json is not None:
|
||
args.json.parent.mkdir(parents=True, exist_ok=True)
|
||
args.json.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
|
||
encoding="utf-8")
|
||
print(f"\n汇总已写入 {args.json}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|