diff --git a/src/mcm26b/sim/__init__.py b/src/mcm26b/sim/__init__.py deleted file mode 100644 index c3f656b..0000000 --- a/src/mcm26b/sim/__init__.py +++ /dev/null @@ -1,8 +0,0 @@ -"""自建模拟器与测试侧传输：只对测试端开放真值，策略端仅见观测。""" - -from .playback import PlaybackTransport -from .simulator import LocalSimulator, SimulatorClosed, Source -from .transport import LocalSimTransport, ScriptedFaultTransport - -__all__ = ["LocalSimTransport", "LocalSimulator", "PlaybackTransport", - "ScriptedFaultTransport", "Source", "SimulatorClosed"] diff --git a/src/mcm26b/sim/playback.py b/src/mcm26b/sim/playback.py deleted file mode 100644 index c5a2469..0000000 --- a/src/mcm26b/sim/playback.py +++ /dev/null @@ -1,38 +0,0 @@ -"""回放传输：把既有演练日志作为服务器响应原样回放给客户端。 - -用于 P1 验收"本地状态机与官方响应一致"：官方演练记录 -（output/q1/practice/*_actions.json）逐条喂给客户端，客户端重建的 -状态必须与当时服务器返回的虚拟时刻逐步吻合。 - -按录制顺序匹配（客户端的 request_id 生成规则与采集器不同， -故不参与匹配；故障重试场景请用 ScriptedFaultTransport）。 -""" - -from __future__ import annotations - -import json -from urllib.parse import urlsplit - -from ..protocol.transport import Transport, TransportResponse - - -class PlaybackTransport(Transport): - """events: 演练日志 [{path, request, response, ...}, ...]。""" - - def __init__(self, events: list[dict]): - self.events = list(events) - self.index = 0 - self.served: list[dict] = [] - - def post(self, url, body, headers, timeout): - if self.index >= len(self.events): - raise AssertionError("回放日志已耗尽仍有请求") - event = self.events[self.index] - path = urlsplit(url).path - if path != event["path"]: - raise AssertionError( - f"回放顺序不一致: 客户端请求 {path}, 录制预期 {event['path']}") - self.index += 1 - self.served.append(event) - return TransportResponse( - 200, json.dumps(event["response"], ensure_ascii=False).encode("utf-8")) diff --git a/src/mcm26b/sim/scenes.py b/src/mcm26b/sim/scenes.py deleted file mode 100644 index 9a7f8d4..0000000 --- a/src/mcm26b/sim/scenes.py +++ /dev/null @@ -1,66 +0,0 @@ -"""自检场景生成器（测试端专用）。 - -真值只进入测试与 CLI ``--self-test`` 入口，策略端仅见模拟器响应； -默认采用满足题面范围的自建随机场景：源总数 10~16 随机、位置在目标圆域内均匀分布、 -频道互异（1..20 抽样）、有效接收半径 [1000,1500] 均匀。 -这些均匀分布是测试假设，题面并未规定官方场景的概率分布。 -边界情形（近圆域边界源、接收半径两档、定向朝外源）由专门测试显式构造。 -""" - -from __future__ import annotations - -import math -from random import Random - -from ..config import ProblemConfig -from .simulator import Source - - -def random_omni_scene(seed: int = 20260910, count: int | None = None, - config: ProblemConfig = ProblemConfig()) -> list[Source]: - """全向源场景 [题面]：count=None 时源总数在 10..16 均匀随机。 - - 位置在半径 omega_radius_m 圆域内按面积均匀（r = R*sqrt(u)）， - 有效接收半径在 [r_eff_min, r_eff_max] 均匀，频道从 1..20 无重复抽样。 - """ - field = config.field - rng = Random(seed) - if count is None: - count = rng.randint(field.source_count_min, field.source_count_max) - if not 0 <= count <= field.channel_max: - raise ValueError(f"源数量须在 0..{field.channel_max}") - channels = rng.sample(range(field.channel_min, field.channel_max + 1), count) - sources = [] - for channel in channels: - radius = field.omega_radius_m * math.sqrt(rng.random()) - angle = rng.uniform(0.0, 2 * math.pi) - r_eff = rng.uniform(field.r_eff_min_m, field.r_eff_max_m) - sources.append(Source(channel=channel, - x=radius * math.cos(angle), - y=radius * math.sin(angle), - r_eff_m=r_eff)) - return sources - - -def mixed_scene(seed: int = 20260910, count: int | None = None, - directional: int = 3, - config: ProblemConfig = ProblemConfig()) -> list[Source]: - """问题4混合场景：count 个源中 directional 个为定向源。 - - 改造 random_omni_scene 的前 directional 个源：近圆域边界的前两个定向源 - 朝向取"背离圆心"（发射半平面基本朝向域外，覆盖网格仍保证可发现， - 即"朝外边界源不漏"验收情形），其余定向源朝向随机。 - """ - sources = list(random_omni_scene(seed=seed, count=count, config=config)) - directional = min(directional, len(sources)) - rng = Random(seed + 1) - for index in range(directional): - source = sources[index] - if index < 2 and len(sources) > 1 and math.hypot(source.x, source.y) > 1700.0: - heading = math.degrees(math.atan2(source.y, source.x)) % 360.0 - else: - heading = rng.uniform(0.0, 360.0) - sources[index] = source.__class__(channel=source.channel, x=source.x, - y=source.y, r_eff_m=source.r_eff_m, - directional=True, heading_deg=heading) - return sources diff --git a/src/mcm26b/sim/simulator.py b/src/mcm26b/sim/simulator.py deleted file mode 100644 index a76f391..0000000 --- a/src/mcm26b/sim/simulator.py +++ /dev/null @@ -1,324 +0,0 @@ -"""自建本地模拟器：按附件1/2语义复现官方模拟器的可观测行为。 - -用途（P1 验收"自建回放测试"）：不依赖官方模拟器即可测试客户端、 -故障恢复与后续策略。真值 Source 只对测试端开放，策略端只能拿到响应。 - -与官方实现的边界：复现公开规定的响应字段、耗时、截止、幂等与拒绝语义， -不复现TCP限流、UI、上传生命周期；示向度误差分布是测试假设，不代表官方实现。 -误差场为逐点固定：同一（源频道，位置）的示向度误差由种子确定性决定， -重复测量读数不变 [题面/官方演练实测]；不同位置误差独立（白噪声假设）。 -""" - -from __future__ import annotations - -import hashlib -import json -from dataclasses import dataclass -from math import atan2, degrees, hypot, isfinite -from time import monotonic, time -from threading import Event, Lock -from unicodedata import category -from copy import deepcopy - -from ..config import ProblemConfig -from ..protocol.rules import (ACTION_FIELDS, CLEAR_NO_TARGET, CLEAR_RESULTS, - CLEAR_SUCCESS, COMMON_FIELDS, DIRECTION, - EXIT_USER, MEASURE_RESULTS, NEAR, NO_SIGNAL, - POSITION_FIELDS) -from ..protocol.timing import clear_cost, measure_cost -from ..protocol.transport import TransportError - - -class SimulatorClosed(TransportError): - """测试结束，官方接口此时关闭连接，不再返回业务响应。""" - - -@dataclass(frozen=True) -class Source: - """一局真值。每个频道最多一个源 [题面]；构造时强制检查。""" - - channel: int - x: float - y: float - r_eff_m: float # 有效接收半径，题目范围 1000..1500 - directional: bool = False - heading_deg: float = 0.0 # 定向源朝向；全向源忽略 - - -class LocalSimulator: - """handle返回状态码及JSON；测试结束后抛出SimulatorClosed。 - - 默认执行官方时限；stress_mode允许配置更长预算和窗口。 - clock为单调秒时钟，wall_clock为Unix秒时钟；窗口起点默认为接口开放时刻。 - rtt_ms仅保留旧调用兼容，不再制造响应时间戳或延迟。 - """ - - def __init__(self, sources: list[Source], *, robot_id: str = "2026001", - config: ProblemConfig = ProblemConfig(), - error_half_deg: float | None = None, error_seed: int = 20260910, - rtt_ms: float = 40.0, clock=monotonic, wall_clock=time, - window_started_at: float | None = None, stress_mode: bool = False): - channels = [s.channel for s in sources] - if len(channels) != len(set(channels)): - raise ValueError("每个频道最多对应一个干扰源") - for s in sources: - if not (config.field.channel_min <= s.channel <= config.field.channel_max): - raise ValueError("频道超出 1..20") - if not s.directional and s.heading_deg != 0.0: - raise ValueError("全向源不应有朝向") - self.sources = list(sources) - self.robot_id = robot_id - self.config = config - self.field = config.field - self.cost = config.cost - self.error_half_deg = (self.field.bearing_error_half_deg - if error_half_deg is None else error_half_deg) - self.error_seed = error_seed - self._clock = clock - self._wall_clock = wall_clock - self.stress_mode = stress_mode - self._real_limit = (self.field.max_real_duration_s if stress_mode - else min(1200, self.field.max_real_duration_s)) - start = clock() if window_started_at is None else window_started_at - self._deadline = start + (max(self.field.test_window_s, self._real_limit) - if stress_mode else min(1500, self.field.test_window_s)) - self._lock = Lock() - self._pending = None - self.closed_reason = None - self.entered = False - self.exited = False - self.position = (0.0, 0.0) - self.channel = self.field.channel_min - self.virtual_time_s = 0.0 - self.cleared: set[int] = set() - self._idempotency: dict[str, tuple[str, int, dict]] = {} # id -> (指纹, 状态码, 响应) - - # ------------------------------------------------------------------ 主入口 - def handle(self, path: str, payload) -> tuple[int, dict]: - # 只在登记/发布时持锁；动作执行期间不同请求必须立即得到409。 - fingerprint = self._fingerprint(path, payload) - request_id = payload.get("request_id") if isinstance(payload, dict) else None - with self._lock: - self.ensure_open() - if self._pending is not None: - pending = self._pending - if pending["id"] != request_id or pending["fingerprint"] != fingerprint: - return 409, self._reject(path) - owner = False - else: - pending = {"id": request_id, "fingerprint": fingerprint, - "done": Event(), "result": None, "error": None} - self._pending = pending - self._registered_at = self._clock() - owner = True - if not owner: - pending["done"].wait() - if pending["error"] is not None: - raise pending["error"] - return deepcopy(pending["result"]) - try: - result = self._handle(path, payload) - pending["result"] = deepcopy(result) - return result - except BaseException as exc: - pending["error"] = exc - raise - finally: - with self._lock: - try: - self.ensure_open() - except SimulatorClosed: - pass # 已登记动作仍可取得完整响应。 - self._pending = None - pending["done"].set() - - def ensure_open(self): - if self.closed_reason is None: - if self.exited: - self.closed_reason = "user_exit" - elif self.virtual_time_s >= self.field.max_virtual_duration_s: - self.closed_reason = "virtual_time_limit" - elif self._clock() >= self._deadline: - self.closed_reason = "real_time_limit" - if self.closed_reason is not None: - raise SimulatorClosed(self.closed_reason) - - def _handle(self, path: str, payload) -> tuple[int, dict]: - if path not in ACTION_FIELDS: - return 404, self._reject(path) - if not isinstance(payload, dict): - return 400, self._reject(path) - # 1) 公共字段存在性与类型（缺失/类型错 -> 400 [题面]） - for name, spec in COMMON_FIELDS.items(): - value = payload.get(name) - if (not isinstance(value, str) or not value - or any(category(c) in ("Cc", "Cf", "Cs") for c in value) - or (spec.max_bytes and len(value.encode("utf-8")) > spec.max_bytes)): - return 400, self._reject(path) - # 2) 动作字段存在性 -> 400 - spec = ACTION_FIELDS[path] - for name in spec: - if name not in payload: - return 400, self._reject(path) - # 3) 未知字段 -> 200 accepted=false（不占用 request_id [题面]） - known = set(COMMON_FIELDS) | set(spec) - if set(payload) != known: - return 200, self._not_accepted() - # 4) arena / robot 不匹配 -> 200 accepted=false（不占用 request_id [题面]） - if payload["arena_id"] != "default" or payload["robot_id"] != self.robot_id: - return 200, self._not_accepted() - # 4.5) position 下的未知字段同样 -> 200 accepted=false [题面 7.4] - if path in ("/measure", "/clear"): - position = payload.get("position") - if isinstance(position, dict) and set(position) - set(POSITION_FIELDS): - return 200, self._not_accepted() - # 5) 幂等：同一 ID 同一内容回放首次响应；不同内容 409 [题面] - request_id = payload["request_id"] - fingerprint = self._fingerprint(path, payload) - if request_id in self._idempotency: - seen_fp, seen_status, seen_body = self._idempotency[request_id] - if seen_fp == fingerprint: - return seen_status, dict(seen_body) - return 409, self._reject(path) - # 6) 业务状态机 - status, body = self._execute(path, payload) - if status == 200 and body.get("accepted") is True: - self._idempotency[request_id] = (fingerprint, status, dict(body)) - return status, body - - # ------------------------------------------------------------------ 动作 - def _execute(self, path: str, payload: dict) -> tuple[int, dict]: - if path == "/enter": - if self.entered or self.exited: - return 200, self._not_accepted() - self.entered = True - self._deadline = min(self._deadline, self._registered_at + self._real_limit) - return 200, self._ok({ - "max_virtual_duration_s": self.field.max_virtual_duration_s, - "max_real_duration_s": self._real_limit, - "remaining_real_duration_s": int(self._deadline - self._registered_at), - }) - if not self.entered or self.exited: - return 200, self._not_accepted() - if path == "/exit": - self.exited = True - return 200, self._ok({"exit_reason": EXIT_USER}) - # /measure 与 /clear 共同的字段校验 -> 400 [题面] - position = payload.get("position") - if not isinstance(position, dict) or not set(POSITION_FIELDS) <= set(position): - return 400, self._reject(path) - for key in ("x", "y"): - value = position[key] - if isinstance(value, bool) or not isinstance(value, (int, float)) \ - or abs(value) > self.field.coord_abs_limit_m or not isfinite(value): - return 400, self._reject(path) - channel = payload.get("channel") - if isinstance(channel, bool) or not isinstance(channel, (int, float)) \ - or not self.field.channel_min <= channel <= self.field.channel_max \ - or not float(channel).is_integer(): - return 400, self._reject(path) - channel = int(channel) - pos = (float(position["x"]), float(position["y"])) - - if path == "/measure": - return self._measure(pos, channel) - return self._clear(pos, channel) - - def _measure(self, pos, channel: int) -> tuple[int, dict]: - breakdown = measure_cost(self.position, pos, self.channel, channel, self.cost) - result, svd = self._observe(pos, channel) - self.position = pos - self.channel = channel # 只有合法 /measure 更新频道 [题面] - self._advance(breakdown.total_s) - body = {"measure_result": result} - if result == DIRECTION: - body["svd_deg"] = svd - return 200, self._ok(body) - - def _clear(self, pos, channel: int) -> tuple[int, dict]: - source = self._source(channel) - success = (source is not None and source.channel not in self.cleared - and hypot(pos[0] - source.x, pos[1] - source.y) - <= self.field.clear_radius_m) - breakdown = clear_cost(self.position, pos, success, self.cost) - self.position = pos # /clear 不改变频道 [题面] - if success: - self.cleared.add(channel) - self._advance(breakdown.total_s) - result = CLEAR_SUCCESS if success else CLEAR_NO_TARGET - return 200, self._ok({"clear_result": result}) - - # ------------------------------------------------------------------ 观测 - def _observe(self, pos, channel: int): - source = self._source(channel) - if source is None or channel in self.cleared: - return NO_SIGNAL, None - distance = hypot(pos[0] - source.x, pos[1] - source.y) - if distance > source.r_eff_m: - return NO_SIGNAL, None - if source.directional: - bearing_to_point = degrees(atan2(pos[1] - source.y, pos[0] - source.x)) - if self._angular_gap(bearing_to_point, source.heading_deg) \ - > self.field.directional_half_angle_deg: - return NO_SIGNAL, None - if distance <= self.field.near_radius_m: - return NEAR, None - true_bearing = degrees(atan2(source.y - pos[1], source.x - pos[0])) - error = self._bearing_error(channel, pos) - svd = round(true_bearing + error, self.field.bearing_decimals) % 360.0 - return DIRECTION, svd - - def _bearing_error(self, channel: int, pos) -> float: - """逐点固定误差场：同一（频道, 位置）误差恒定 [题面：同一地点误差不变]。 - - 坐标按 1e-6 m 量化后与种子、频道一起哈希，映射到 - [-error_half_deg, +error_half_deg)；不同位置近似独立（白噪声假设）。 - """ - x, y = (round(float(value), 6) or 0.0 for value in pos) - key = f"{self.error_seed}|{channel}|{x}|{y}" - digest = hashlib.sha256(key.encode("utf-8")).digest() - unit = int.from_bytes(digest[:8], "big") / 2.0 ** 64 - return (2.0 * unit - 1.0) * self.error_half_deg - - def _source(self, channel: int): - for source in self.sources: - if source.channel == channel: - return source - return None - - @staticmethod - def _angular_gap(a: float, b: float) -> float: - return abs((a - b + 180.0) % 360.0 - 180.0) - - # ------------------------------------------------------------------ 计时与响应 - def _advance(self, seconds: float) -> None: - self._virtual_us += round(seconds * 1_000_000) - - @property - def virtual_time_s(self): - return self._virtual_us / 1_000_000 - - @virtual_time_s.setter - def virtual_time_s(self, seconds): - self._virtual_us = round(seconds * 1_000_000) - - def _ok(self, extra: dict) -> dict: - body = {"accepted": True, - "real_timestamp_ms": int(self._wall_clock() * 1000), - "virtual_time_s": self.virtual_time_s, **extra} - return body - - def _not_accepted(self) -> dict: - return {"accepted": False, "real_timestamp_ms": int(self._wall_clock() * 1000), - "virtual_time_s": 0} # accepted=false 时恒为 0 [题面] - - def _reject(self, path: str) -> dict: - return {"accepted": False, "real_timestamp_ms": int(self._wall_clock() * 1000), - "virtual_time_s": 0, "detail": f"{path} 请求被拒绝"} - - @staticmethod - def _fingerprint(path: str, payload) -> str: - canonical = json.dumps({"path": path, "payload": payload}, - sort_keys=True, ensure_ascii=False, - separators=(",", ":")) - return hashlib.sha256(canonical.encode("utf-8", errors="surrogatepass")).hexdigest() diff --git a/src/mcm26b/sim/transport.py b/src/mcm26b/sim/transport.py deleted file mode 100644 index 1d2c316..0000000 --- a/src/mcm26b/sim/transport.py +++ /dev/null @@ -1,105 +0,0 @@ -"""测试侧传输：把 LocalSimulator 挂到 Transport 接口 + 故障注入脚本。 - -故障脚本用于 P1 验收"失败重试不污染状态"： -- ``"drop_before"`` 连接失败，动作未送达服务器（可安全重试）； -- ``"timeout_after"`` 服务器已受理并生效，但响应丢失（必须靠幂等回放）； -- ``("status", n)`` 服务器返回指定 HTTP 状态码（如 500/429）； -- ``None`` 正常放行。 -脚本耗尽后一律放行。 -""" - -from __future__ import annotations - -import json -from urllib.parse import urlsplit - -from ..protocol.transport import Transport, TransportError, TransportResponse -from .simulator import LocalSimulator -from ..protocol.rules import ACTION_FIELDS - - -def _object(pairs): - result = {} - for key, value in pairs: - if key in result: - raise ValueError("duplicate JSON key") - result[key] = value - return result - - -def _invalid_constant(value): - raise ValueError("non-finite JSON number") - - -def _depth(value): - if isinstance(value, dict): - return 1 + max((_depth(v) for v in value.values()), default=0) - if isinstance(value, list): - return 1 + max((_depth(v) for v in value), default=0) - return 0 - - -class LocalSimTransport(Transport): - """内存回环：不经过网络栈的本地模拟器传输。""" - - def __init__(self, simulator: LocalSimulator): - self.simulator = simulator - self.received: list[tuple[str, dict]] = [] - - def post(self, url, body, headers, timeout): - return self.request("POST", url, body, headers, timeout) - - def request(self, method, url, body, headers, timeout=None): - """验证原始请求边界；内存回环不模拟TCP、UI或连接限流。""" - self.simulator.ensure_open() - target = urlsplit(url) - path = target.path - def reject(status): - return TransportResponse(status, json.dumps( - self.simulator._reject(path), ensure_ascii=False).encode("utf-8")) - if path not in ACTION_FIELDS or "?" in url or "#" in url: - return reject(404) - if method != "POST": - return reject(405) - normalized = {key.lower(): value.strip().lower() for key, value in headers.items()} - if len(normalized) != len(headers): - return reject(415) - media = [part.strip() for part in normalized.get("content-type", "").split(";")] - if (media[0] != "application/json" or len(media) > 2 - or (len(media) == 2 and media[1].replace(" ", "") - not in ('charset=utf-8', 'charset="utf-8"')) - or normalized.get("content-encoding", "identity") != "identity"): - return reject(415) - if len(body) > 65536: - return reject(413) - try: - payload = json.loads(body.decode("utf-8"), object_pairs_hook=_object, - parse_constant=_invalid_constant) - if not isinstance(payload, dict) or _depth(payload) > 16: - return reject(400) - except (ValueError, UnicodeError, RecursionError): - return reject(400) - self.received.append((path, payload)) - status, response = self.simulator.handle(path, payload) - return TransportResponse(status, json.dumps(response, ensure_ascii=False).encode("utf-8")) - - -class ScriptedFaultTransport(Transport): - """按脚本注入故障后转发给内层传输。""" - - def __init__(self, inner: Transport, script: list): - self.inner = inner - self.script = list(script) - self.sent_requests: list[bytes] = [] - - def post(self, url, body, headers, timeout): - self.sent_requests.append(body) - fault = self.script.pop(0) if self.script else None - if fault == "drop_before": - raise TransportError("injected: connection refused before send") - if fault == "timeout_after": - self.inner.post(url, body, headers, timeout) # 服务器已受理 - raise TransportError("injected: read timeout after server applied") - if isinstance(fault, tuple) and fault and fault[0] == "status": - return TransportResponse(int(fault[1]), b'{"accepted": false, "real_timestamp_ms": 0, "virtual_time_s": 0}') - return self.inner.post(url, body, headers, timeout)