"""P1 协议基线:串行、幂等、带现实预算的模拟器客户端。 协议硬规则 [题面] 与实现对应: - 每个新动作使用新的 request_id;仅在网络失败后重试"完全相同的动作"时, 复用原 request_id 与原请求体 —— 本类通过一次性序列化出 bytes 并原样重发保证; - 必须同时检查 HTTP 状态与 accepted;accepted=false 不更新状态, 其 virtual_time_s=0 不当作当前虚拟时刻; - 逐次等待响应、不得并发;本类完全同步,一个动作未决不发起下一个; - 现实截止用 /enter 响应的 remaining_real_duration_s 建立本地单调时钟, 预留退出余量;/exit 不做预算拦截。 """ from __future__ import annotations import json import uuid from dataclasses import dataclass from time import monotonic, sleep from ..config import ProblemConfig, Vec2 from .log import SessionLog from .rules import (CLEAR_RESULTS, DIRECTION, EXIT_USER, MEASURE_RESULTS, POSITION_FIELDS, STATUS_OK, STATUS_RETRYABLE) from .state import RobotState from .timing import CostBreakdown from .transport import RequestsTransport, Transport, TransportError class ProtocolError(RuntimeError): """协议层失败,含 HTTP 状态与响应体;默认不重试(确定性错误)。""" def __init__(self, message: str, status: int | None = None, body=None): super().__init__(message) self.status = status self.body = body class ActionRejected(ProtocolError): """HTTP 200 但 accepted=false:动作未生效,状态不得更新。""" def __init__(self, message: str, body=None): super().__init__(message, status=STATUS_OK, body=body) class ProtocolViolation(ProtocolError): """响应与协议约定不符(字段缺失、值非法、虚拟时刻回退等)。""" class DeadlineExceeded(RuntimeError): """现实预算不足:预留退出余量后剩余时间不够执行新动作。""" @dataclass(frozen=True) class MeasureOutcome: result: str svd_deg: float | None virtual_time_s: float predicted: CostBreakdown raw: dict @dataclass(frozen=True) class ClearOutcome: result: str virtual_time_s: float predicted: CostBreakdown raw: dict @dataclass(frozen=True) class ExitInfo: reason: str virtual_time_s: float raw: dict def _validate_identifier(name: str, value: str, max_bytes: int) -> None: if not isinstance(value, str) or not value: raise ValueError(f"{name} 必须是非空字符串") encoded = value.encode("utf-8") if not 1 <= len(encoded) <= max_bytes: raise ValueError(f"{name} 的 UTF-8 长度须在 1..{max_bytes} 字节") if any(code < 0x20 or code == 0x7F for code in encoded): raise ValueError(f"{name} 不能包含控制字符") def _validate_position(position: dict, limit: float) -> None: if set(position) != set(POSITION_FIELDS): raise ValueError("position 字段必须是且只能是 x、y") for key in ("x", "y"): value = position[key] if isinstance(value, bool) or not isinstance(value, (int, float)): raise ValueError(f"position.{key} 必须是数值") if value != value or value in (float("inf"), float("-inf")): raise ValueError(f"position.{key} 必须是有限数值") if abs(value) > limit: raise ValueError(f"position.{key} 绝对值不得超过 {limit} 米") def _validate_channel(channel) -> int: # 写成 JSON number 1.0 且恰为整数也接受 [题面];1.5 拒绝。 if isinstance(channel, bool) or not isinstance(channel, (int, float)): raise ValueError("channel 必须是数值") if isinstance(channel, float) and not channel.is_integer(): raise ValueError("channel 必须是整数") channel = int(channel) if not 1 <= channel <= 20: raise ValueError("channel 必须在 1..20 范围内") return channel class SimulatorClient: """四条指令的同步客户端。状态只在 accepted=true 后推进。""" def __init__(self, robot_id: str, *, config: ProblemConfig = ProblemConfig(), transport: Transport | None = None, log: SessionLog | None = None, clock=monotonic): _validate_identifier("robot_id", robot_id, config.session.robot_id_max_bytes) self.robot_id = robot_id self.config = config self.session = config.session self.transport = transport if transport is not None else RequestsTransport() self.log = log if log is not None else SessionLog() self.clock = clock self.state = RobotState() self._prefix = uuid.uuid4().hex[:12] self._seq = 0 self._deadline: float | None = None # 本地单调时钟截止 self._server_deadline_ms: float | None = None # 服务器时间戳交叉核对 self._enter_monotonic: float | None = None # 实机计时起点(/enter 成功) # ------------------------------------------------------------ 公共动作 API def enter(self) -> dict: started = self.clock() payload = self._payload("/enter") response = self._call("/enter", payload) self.state.apply_enter(response) remaining = response["remaining_real_duration_s"] if not isinstance(remaining, (int, float)) or remaining <= 0: raise ProtocolViolation("remaining_real_duration_s 缺失或非法", body=response) self._deadline = self.clock() + remaining self._server_deadline_ms = (response["real_timestamp_ms"] + remaining * 1000.0) self._enter_monotonic = self.clock() self._log_step("/enter", payload, response, started, predicted=CostBreakdown(0.0, 0.0, 0.0)) return response def measure(self, position: Vec2, channel: int) -> MeasureOutcome: started = self.clock() payload = self._payload("/measure", position=position, channel=channel) response = self._call("/measure", payload) predicted = self.state.apply_measure(payload, response, cost=self.config.cost) result = response.get("measure_result") if result not in MEASURE_RESULTS: raise ProtocolViolation(f"measure_result 非法: {result!r}", body=response) svd = response.get("svd_deg") if result == DIRECTION: if not isinstance(svd, (int, float)) or isinstance(svd, bool): raise ProtocolViolation("direction 响应缺少数值 svd_deg", body=response) self._log_step("/measure", payload, response, started, result=result, predicted=predicted, channel=payload["channel"]) return MeasureOutcome(result, float(svd) if svd is not None else None, response["virtual_time_s"], predicted, response) def clear(self, position: Vec2, channel: int) -> ClearOutcome: started = self.clock() payload = self._payload("/clear", position=position, channel=channel) response = self._call("/clear", payload) predicted = self.state.apply_clear(payload, response, cost=self.config.cost) result = response.get("clear_result") if result not in CLEAR_RESULTS: raise ProtocolViolation(f"clear_result 非法: {result!r}", body=response) self._log_step("/clear", payload, response, started, result=result, predicted=predicted, channel=payload["channel"]) return ClearOutcome(result, response["virtual_time_s"], predicted, response) def exit(self) -> ExitInfo: started = self.clock() payload = self._payload("/exit") response = self._call("/exit", payload) self.state.apply_exit(response) reason = response.get("exit_reason") if reason != EXIT_USER: raise ProtocolViolation(f"exit_reason 非法: {reason!r}", body=response) self._log_step("/exit", payload, response, started, predicted=CostBreakdown(0.0, 0.0, 0.0)) return ExitInfo(reason, response["virtual_time_s"], response) # ---------------------------------------------------------------- 预算控制 @property def real_remaining_s(self) -> float | None: """按本地单调时钟估计的剩余现实秒数(未 enter 时为 None)。""" return None if self._deadline is None else self._deadline - self.clock() def _log_step(self, path: str, payload: dict, response: dict, started: float, *, result: str | None = None, predicted: CostBreakdown, channel: int | None = None) -> None: """每步计时留档 [用户决策 2026-09-11]:实机墙钟耗时 + 比赛虚拟耗时分项 (移动=直线距离/5、切频道 1s、检测 5s、清除成功 5s/未发现 3s)。 step 事件不含 response 字段,不会混入 action_events / 虚拟时刻复算。 """ now = self.clock() entry: dict = {"event": "step", "path": path, "request_id": payload["request_id"], "real_elapsed_s": round(now - started, 6), "virtual_time_s": response["virtual_time_s"], "virtual_cost": {"move_s": round(predicted.move_s, 6), "switch_s": round(predicted.switch_s, 6), "action_s": round(predicted.action_s, 6), "total_s": round(predicted.total_s, 6)}} if self._enter_monotonic is not None: entry["real_since_enter_s"] = round(now - self._enter_monotonic, 6) if "position" in payload: entry["position"] = {"x": payload["position"]["x"], "y": payload["position"]["y"]} if channel is not None: entry["channel"] = channel if result is not None: entry["result"] = result self.log.append(entry) def _check_budget(self, path: str) -> None: if path == "/exit" or self._deadline is None: return margin = self.session.reserve_exit_s if self.real_remaining_s <= margin: raise DeadlineExceeded( f"剩余现实时间 {self.real_remaining_s:.1f}s 不足以预留 {margin:.1f}s 退出余量") # ---------------------------------------------------------------- 内部机制 def _payload(self, path: str, *, position: Vec2 | None = None, channel: int | None = None) -> dict: payload = {"arena_id": self.session.arena_id, "robot_id": self.robot_id, "request_id": self._next_request_id()} if position is not None: x, y = float(position[0]), float(position[1]) _validate_position({"x": x, "y": y}, self.config.field.coord_abs_limit_m) payload["position"] = {"x": x, "y": y} if channel is not None: payload["channel"] = _validate_channel(channel) return payload def _next_request_id(self) -> str: self._seq += 1 request_id = f"{self._prefix}-{self._seq:04d}" _validate_identifier("request_id", request_id, self.session.request_id_max_bytes) return request_id def _call(self, path: str, payload: dict) -> dict: """串行执行一个动作;网络失败原样重发,HTTP 确定性错误立即抛出。""" self._check_budget(path) body = json.dumps(payload, ensure_ascii=False, allow_nan=False).encode("utf-8") if len(body) > self.session.request_body_max_bytes: raise ValueError("请求体超过 65536 字节") headers = {"Content-Type": "application/json"} url = self.session.base_url + path attempts, last_error = 0, None while attempts < self.session.max_attempts: attempts += 1 started = self.clock() self.log.append({"event": "attempt", "path": path, "request_id": payload["request_id"], "attempt": attempts, "request": payload}) try: result = self.transport.post( url, body, headers, timeout=(self.session.connect_timeout_s, self.session.read_timeout_s)) except TransportError as exc: last_error = exc self.log.append({"event": "transport_error", "path": path, "request_id": payload["request_id"], "attempt": attempts, "error": str(exc), "elapsed_s": self.clock() - started}) if attempts < self.session.max_attempts: sleep(self.session.retry_backoff_s * (2 ** (attempts - 1))) continue if result.status in STATUS_RETRYABLE: self.log.append({"event": "retryable_status", "path": path, "request_id": payload["request_id"], "attempt": attempts, "status": result.status}) if attempts < self.session.max_attempts: sleep(self.session.retry_backoff_s * (2 ** (attempts - 1))) continue if result.status != STATUS_OK: raise ProtocolError( f"{path} 返回 HTTP {result.status}", status=result.status, body=self._safe_json(result.body)) response = self._safe_json(result.body) if response is None or response.get("accepted") is not True: raise ActionRejected( f"{path} 未被接受: {response}", body=response) self._verify_clock(response) self.log.append({"event": "accepted", "path": path, "request_id": payload["request_id"], "request": payload, "response": response, "attempts": attempts, "elapsed_s": self.clock() - started}) return response raise TransportError( f"{path} 重试 {attempts} 次仍失败: {last_error}") from last_error def _verify_clock(self, response: dict) -> None: """accepted=true 的虚拟时刻必须单调不减;现实时间戳用于漂移核对。""" virtual = response.get("virtual_time_s") if not isinstance(virtual, (int, float)) or isinstance(virtual, bool): raise ProtocolViolation("accepted 响应缺少 virtual_time_s", body=response) if virtual < self.state.virtual_time_s - 1e-9: raise ProtocolViolation( f"虚拟时刻回退: {self.state.virtual_time_s} -> {virtual}", body=response) if self._server_deadline_ms is not None: drift = (self._server_deadline_ms - response["real_timestamp_ms"]) / 1000.0 if drift < self.session.reserve_exit_s / 2: # 服务器视角余量告急(本地时钟漂移或请求排队);仅记录,交由上层决策。 self.log.append({"event": "server_time_warning", "server_remaining_s": drift}) @staticmethod def _safe_json(body: bytes): try: return json.loads(body.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): return None