82 lines
3.5 KiB
Python
82 lines
3.5 KiB
Python
"""评估与统计：从自身动作日志离线复算虚拟时刻与统计量。
|
||
|
||
"演练统计"验收要求：总时间与分项耗时可独立复算 [题面/计划 §6]。
|
||
recompute_virtual_time 是回放测试与本模块共用的唯一复算实现，
|
||
客户端日志格式为 SessionLog 中 accepted 事件的 {path, request, response}。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from ..config import CostConfig
|
||
from ..protocol.rules import CLEAR_SUCCESS
|
||
from ..protocol.timing import clear_cost, measure_cost
|
||
|
||
|
||
def recompute_virtual_time(events: list[dict],
|
||
cost: CostConfig = CostConfig()) -> dict:
|
||
"""从动作日志复算逐步虚拟时刻。
|
||
|
||
events 每项形如 {path, request, response}（accepted=true）。
|
||
返回逐步表与最终时刻；任何一步与服务器响应偏差超过容差记入 mismatches。
|
||
"""
|
||
position = (0.0, 0.0)
|
||
channel = 1
|
||
virtual = 0.0
|
||
steps, mismatches = [], []
|
||
for index, event in enumerate(events):
|
||
path, request = event["path"], event["request"]
|
||
response = event["response"]
|
||
if response.get("accepted") is not True:
|
||
mismatches.append({"index": index, "issue": "响应未被接受"})
|
||
continue
|
||
if path == "/enter":
|
||
position, channel, virtual = (0.0, 0.0), 1, response["virtual_time_s"]
|
||
elif path == "/measure":
|
||
pos = (request["position"]["x"], request["position"]["y"])
|
||
predicted = measure_cost(position, pos, channel, request["channel"], cost)
|
||
position, channel = pos, request["channel"]
|
||
virtual += predicted.total_s
|
||
elif path == "/clear":
|
||
pos = (request["position"]["x"], request["position"]["y"])
|
||
success = response.get("clear_result") == CLEAR_SUCCESS
|
||
predicted = clear_cost(position, pos, success, cost)
|
||
position = pos
|
||
virtual += predicted.total_s
|
||
elif path == "/exit":
|
||
pass # /exit 不推进虚拟时钟 [题面]
|
||
else:
|
||
mismatches.append({"index": index, "issue": f"未知路径 {path}"})
|
||
continue
|
||
delta = response["virtual_time_s"] - virtual
|
||
if abs(delta) > 1e-6:
|
||
mismatches.append({"index": index, "path": path,
|
||
"recomputed_s": virtual,
|
||
"recorded_s": response["virtual_time_s"],
|
||
"gap_s": delta})
|
||
steps.append({"index": index, "path": path,
|
||
"predicted_s": round(virtual, 9),
|
||
"recorded_s": response["virtual_time_s"]})
|
||
return {"steps": steps, "final_virtual_time_s": virtual,
|
||
"mismatches": mismatches}
|
||
|
||
|
||
def summarize(events: list[dict]) -> dict:
|
||
"""动作计数摘要：检测/清除/移动距离等基础统计。"""
|
||
position = (0.0, 0.0)
|
||
measures, clears, cleared, distance_m = 0, 0, 0, 0.0
|
||
for event in events:
|
||
request = event["request"]
|
||
if event["path"] in ("/measure", "/clear"):
|
||
pos = (request["position"]["x"], request["position"]["y"])
|
||
distance_m += ((pos[0] - position[0]) ** 2
|
||
+ (pos[1] - position[1]) ** 2) ** 0.5
|
||
position = pos
|
||
if event["path"] == "/measure":
|
||
measures += 1
|
||
elif event["path"] == "/clear":
|
||
clears += 1
|
||
if event["response"].get("clear_result") == CLEAR_SUCCESS:
|
||
cleared += 1
|
||
return {"measures": measures, "clears": clears, "cleared": cleared,
|
||
"distance_m": distance_m}
|