This commit is contained in:
Horldsence
2026-09-11 11:35:47 +08:00
parent f3dfa9c2a1
commit 2d2a9a4e6a
5 changed files with 0 additions and 541 deletions
-8
View File
@@ -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"]
-38
View File
@@ -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"))
-66
View File
@@ -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
-324
View File
@@ -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()
-105
View File
@@ -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)