1183 lines
56 KiB
Python
1183 lines
56 KiB
Python
"""docs/figures 的论文插图生成器（matplotlib 版）。
|
||
|
||
uv run python scripts/make_figures.py # 全部 11 张
|
||
uv run python scripts/make_figures.py --list # 图名
|
||
uv run python scripts/make_figures.py --only flow-overall-architecture
|
||
uv run python scripts/make_figures.py --scale 1.5 # 尺寸倍率
|
||
|
||
约定（与 docs/figures/README.md 对应）：
|
||
|
||
- **只输出 PNG**（不生成 SVG），分辨率按论文插图设置；
|
||
- **图内不放标题、图注与解释性文字**，只保留坐标轴、图例与数据标注；
|
||
图的含义由文件名与 README 清单承担，解释写在论文正文；
|
||
- 字体取自 ``docs/paper/fonts/``（宋体 / 黑体 / Times New Roman），与 main.tex 一致；
|
||
- 数据来源与 sha256 写入 ``docs/figures/manifest.json``。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import math
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Any, Callable, Iterable, Sequence
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
FONT_DIR = ROOT / "docs" / "paper" / "fonts"
|
||
OUT_DIR = ROOT / "docs" / "figures"
|
||
SESSIONS_DIR = ROOT / "referee-data" / "sessions"
|
||
ARCHIVE_DIR = ROOT / "examples" / "replay"
|
||
CACHE_DIR = ROOT / ".mpl-cache"
|
||
|
||
# matplotlib 的缓存目录：仓库内可写，避免污染用户目录
|
||
CACHE_DIR.mkdir(exist_ok=True)
|
||
os.environ.setdefault("MPLCONFIGDIR", str(CACHE_DIR))
|
||
os.environ.setdefault("XDG_CACHE_HOME", str(CACHE_DIR))
|
||
|
||
import matplotlib # noqa: E402
|
||
matplotlib.use("Agg")
|
||
import numpy as np # noqa: E402
|
||
from matplotlib import font_manager as fm # noqa: E402
|
||
from matplotlib import pyplot as plt # noqa: E402
|
||
from matplotlib.lines import Line2D # noqa: E402
|
||
from matplotlib.patches import Circle, FancyArrowPatch, FancyBboxPatch, Polygon # noqa: E402
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
|
||
# 题面常量（与 src/mcm26b/config.py 一致）
|
||
OMEGA_R = 1800.0
|
||
R_EFF = 1000.0
|
||
SPEED_MPS = 5.0
|
||
MEASURE_S = 5.0
|
||
SWITCH_S = 1.0
|
||
CLEAR_OK_S = 5.0
|
||
CLEAR_NO_S = 3.0
|
||
RING_RADIUS_M = 1159.7
|
||
SEVEN_R = 1800.0
|
||
Q4_DESIGN_R = 965.0
|
||
Q4_HEADING_MARGIN_DEG = 5.0
|
||
|
||
# 论文配色：低饱和，单色打印可辨
|
||
C = {
|
||
"ink": "#1c2430",
|
||
"muted": "#5c6673",
|
||
"grid": "#e4e9ee",
|
||
"primary": "#1f4e79",
|
||
"primary_fill": "#dce6f1",
|
||
"accent": "#b03a2e",
|
||
"accent_fill": "#f6dcd8",
|
||
"success": "#1e7a5f",
|
||
"success_fill": "#d8ece5",
|
||
"warn": "#b8860b",
|
||
"warn_fill": "#f6ecd2",
|
||
"violet": "#5b4b8a",
|
||
}
|
||
|
||
FONTS = {
|
||
"song": "SongtiSC-Regular.ttf",
|
||
"song_bold": "SongtiSC-Bold.ttf",
|
||
"hei": "HeitiSC-Light.ttf",
|
||
"hei_bold": "HeitiSC-Medium.ttf",
|
||
"times": "Times New Roman.ttf",
|
||
"times_bold": "Times New Roman Bold.ttf",
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 样式
|
||
# ---------------------------------------------------------------------------
|
||
def setup_style() -> dict[str, Any]:
|
||
"""注册自带字体并设全局样式，返回 FontProperties 字典。"""
|
||
for name in FONTS.values():
|
||
path = FONT_DIR / name
|
||
if path.is_file():
|
||
fm.fontManager.addfont(str(path))
|
||
props = {key: fm.FontProperties(fname=str(FONT_DIR / name))
|
||
for key, name in FONTS.items() if (FONT_DIR / name).is_file()}
|
||
plt.rcParams.update({
|
||
"figure.dpi": 120,
|
||
"savefig.dpi": 200,
|
||
"font.family": [props["song"].get_name(), "DejaVu Sans"],
|
||
"font.size": 8.5,
|
||
"axes.titlesize": 9,
|
||
"axes.labelsize": 8.5,
|
||
"xtick.labelsize": 8,
|
||
"ytick.labelsize": 8,
|
||
"legend.fontsize": 8,
|
||
"axes.edgecolor": C["muted"],
|
||
"axes.labelcolor": C["ink"],
|
||
"text.color": C["ink"],
|
||
"xtick.color": C["muted"],
|
||
"ytick.color": C["muted"],
|
||
"axes.grid": False,
|
||
"axes.unicode_minus": False,
|
||
"lines.linewidth": 1.2,
|
||
})
|
||
return props
|
||
|
||
|
||
def new_axes(width_mm: float, height_mm: float, *, ncols: int = 1,
|
||
nrows: int = 1, scale: float = 1.0, **kw: Any) -> tuple[Any, Any]:
|
||
"""按毫米尺寸建图（论文按 160 mm 宽排版时用 1.0 倍）。"""
|
||
fig, axes = plt.subplots(nrows, ncols,
|
||
figsize=(width_mm / 25.4 * scale, height_mm / 25.4 * scale),
|
||
layout="constrained", **kw)
|
||
return fig, axes
|
||
|
||
|
||
def axes_for_panels(panel_mm: float, aspects: Sequence[float], *, scale: float = 1.0,
|
||
width_ratios: Sequence[float] | None = None,
|
||
extra_mm: float = 30.0, **kw: Any) -> tuple[Any, Any]:
|
||
"""按"每个面板高 panel_mm、宽 = 高 × 宽高比"手工排版。
|
||
|
||
等比例坐标轴（``set_aspect('equal')``）在自动布局（constrained/tight）下会被
|
||
挤成一小块，因此这里直接按数据比例分配位置：每个子图的方框宽高比等于
|
||
数据宽高比，等比例坐标轴正好铺满方框。
|
||
"""
|
||
del kw
|
||
ratios = list(width_ratios) if width_ratios else [1.0] * len(aspects)
|
||
widths = [panel_mm * aspect * ratio for aspect, ratio in zip(aspects, ratios)]
|
||
left_mm, right_mm, top_mm = 16.0, 6.0, 6.0
|
||
gap_mm, bottom_mm = 16.0, 15.0
|
||
total_w = left_mm + sum(widths) + gap_mm * (len(widths) - 1) + right_mm
|
||
total_h = top_mm + panel_mm + bottom_mm
|
||
fig = plt.figure(figsize=(total_w / 25.4 * scale, total_h / 25.4 * scale))
|
||
|
||
def frac_x(mm: float) -> float:
|
||
return mm / total_w
|
||
|
||
def frac_y(mm: float) -> float:
|
||
return mm / total_h
|
||
|
||
axes = []
|
||
cursor = left_mm
|
||
for width in widths:
|
||
box = [frac_x(cursor), frac_y(bottom_mm), frac_x(width), frac_y(panel_mm)]
|
||
axes.append(fig.add_axes(box))
|
||
cursor += width + gap_mm
|
||
return fig, (axes if len(axes) > 1 else axes[0])
|
||
|
||
|
||
def polar_axes(ax: Any, half: float, *, grid: float = 500.0,
|
||
center: Sequence[float] = (0.0, 0.0)) -> None:
|
||
"""世界坐标视图：等比例、留白、细网格；``center`` 为视图中心。"""
|
||
cx, cy = float(center[0]), float(center[1])
|
||
step = grid
|
||
ticks = []
|
||
while step <= half * 1.2:
|
||
ticks += [-step, step]
|
||
step *= 2
|
||
ticks = sorted({round(cx + t) for t in ticks if abs(t) <= half * 1.05} |
|
||
{round(cy + t) for t in ticks if abs(t) <= half * 1.05})
|
||
# 顺序要紧：set_xticks 会在数据范围内重新 autoscale，必须最后再钉住范围
|
||
ax.set_xticks(ticks)
|
||
ax.set_yticks(ticks)
|
||
ax.set_xlim(cx - half, cx + half)
|
||
ax.set_ylim(cy - half, cy + half)
|
||
# 面板方框由 axes_for_panels 按数据宽高比分配，刻度范围也是 1:1，
|
||
# 不需要再让 matplotlib 调 aspect（它会反过来改写已设定的范围）
|
||
ax.set_aspect("auto")
|
||
ax.grid(True, color=C["grid"], lw=0.6)
|
||
for side in ("top", "right"):
|
||
ax.spines[side].set_visible(False)
|
||
ax.set_axisbelow(True)
|
||
|
||
|
||
def save(fig: Any, stem: str, sources: Sequence[dict[str, Any]], scale: float) -> dict[str, Any]:
|
||
"""落盘 PNG（仅 PNG：论文插图不再维护矢量版），返回 manifest 条目。"""
|
||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||
png = OUT_DIR / f"{stem}.png"
|
||
fig.savefig(png, format="png")
|
||
plt.close(fig)
|
||
return {"stem": stem, "png": str(png.relative_to(ROOT)), "scale": scale,
|
||
"sources": list(sources)}
|
||
|
||
|
||
def source(path: str | Path, **extra: Any) -> dict[str, Any]:
|
||
"""记录数据来源：仓库内相对路径、sha256 前 12 位、行数。"""
|
||
p = Path(path)
|
||
p = p if p.is_absolute() else ROOT / p
|
||
digest, lines = "-", 0
|
||
if p.is_file():
|
||
data = p.read_bytes()
|
||
digest = hashlib.sha256(data).hexdigest()[:12]
|
||
lines = data.count(b"\n") + (0 if data.endswith(b"\n") else 1)
|
||
rel = str(p.relative_to(ROOT)) if p.is_relative_to(ROOT) else str(p)
|
||
return {"path": rel, "sha256_12": digest, "lines": lines, **extra}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 行为日志：解析、计时复算、归档回退
|
||
# ---------------------------------------------------------------------------
|
||
def parse_jsonl(path: Path) -> dict[str, Any]:
|
||
"""兼容评测机 ``event=action`` 与客户端 ``event=accepted`` 两种日志。"""
|
||
meta: dict[str, Any] = {}
|
||
actions: list[dict[str, Any]] = []
|
||
fallback: dict[str, dict[str, Any]] = {}
|
||
seq = 0
|
||
for line in path.read_text(encoding="utf-8").splitlines():
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
ev = json.loads(line)
|
||
except json.JSONDecodeError:
|
||
continue
|
||
kind = ev.get("event")
|
||
if kind == "session_start":
|
||
meta = ev
|
||
elif kind == "action":
|
||
seq += 1
|
||
actions.append(dict(seq=ev.get("seq", seq), path=ev.get("path", ""),
|
||
request=ev.get("request") or {},
|
||
response=ev.get("response") or {},
|
||
wall_ts_ms=ev.get("wall_ts_ms")))
|
||
elif kind == "accepted":
|
||
fallback[str(ev.get("request_id", ""))] = ev
|
||
elif kind == "step" and str(ev.get("request_id", "")) not in fallback:
|
||
fallback[str(ev.get("request_id", ""))] = ev
|
||
if not actions:
|
||
for ev in fallback.values():
|
||
if not ev.get("path"):
|
||
continue
|
||
seq += 1
|
||
stamp = ev.get("logged_at_unix_s")
|
||
actions.append(dict(seq=seq, path=ev["path"], request=ev.get("request") or {},
|
||
response=ev.get("response") or {},
|
||
wall_ts_ms=stamp * 1000 if stamp else None))
|
||
if not actions:
|
||
raise ValueError(f"{path} 没有可回放的动作事件")
|
||
return dict(label=path.stem, case_code=meta.get("case_code") or path.parent.name,
|
||
module=meta.get("module") or "?", robot_id=str(meta.get("robot_id") or "?"),
|
||
truth=list(meta.get("truth") or []), actions=actions, source=path)
|
||
|
||
|
||
def discover_runs() -> list[dict[str, Any]]:
|
||
runs: list[dict[str, Any]] = []
|
||
if SESSIONS_DIR.is_dir():
|
||
for log in sorted(SESSIONS_DIR.glob("*/log.jsonl")):
|
||
try:
|
||
runs.append(parse_jsonl(log))
|
||
except ValueError:
|
||
continue
|
||
if not runs and ARCHIVE_DIR.is_dir():
|
||
for log in sorted(ARCHIVE_DIR.glob("*.jsonl")):
|
||
try:
|
||
runs.append(parse_jsonl(log))
|
||
except ValueError:
|
||
continue
|
||
return runs
|
||
|
||
|
||
def pick_run(runs: Sequence[dict[str, Any]], module: str, truth: int) -> dict[str, Any]:
|
||
cands = [r for r in runs if r["module"].startswith(module) and len(r["truth"]) >= truth
|
||
and sum(1 for a in r["actions"] if a["path"] == "/measure") > 0]
|
||
if not cands:
|
||
cands = [r for r in runs if r["module"].startswith(module)]
|
||
return sorted(cands, key=lambda r: (abs(len(r["truth"]) - truth), len(r["actions"])))[0]
|
||
|
||
|
||
def decompose(run: dict[str, Any]) -> dict[str, Any]:
|
||
"""按附件规则逐动作复算移动 / 换频 / 动作耗时。"""
|
||
pos = (0.0, 0.0)
|
||
channel: int | None = None
|
||
rows: list[dict[str, Any]] = []
|
||
for act in run["actions"]:
|
||
req, res = act["request"], act["response"]
|
||
move_m = switch = action_s = 0.0
|
||
if act["path"] in ("/measure", "/clear") and "position" in req:
|
||
tgt = (float(req["position"]["x"]), float(req["position"]["y"]))
|
||
move_m = math.dist(pos, tgt)
|
||
if move_m > 0:
|
||
pos = tgt
|
||
if act["path"] == "/measure":
|
||
ch = int(req.get("channel", 0))
|
||
switch = 0.0 if channel in (None, ch) else SWITCH_S
|
||
channel = ch
|
||
action_s = MEASURE_S
|
||
elif act["path"] == "/clear":
|
||
action_s = CLEAR_OK_S if str(res.get("clear_result")) == "success" else CLEAR_NO_S
|
||
rows.append(dict(seq=act["seq"], path=act["path"], move_s=move_m / SPEED_MPS,
|
||
switch_s=switch, action_s=action_s, move_m=move_m,
|
||
virtual_s=float(res.get("virtual_time_s") or 0.0),
|
||
hit=str(res.get("measure_result")) == "direction",
|
||
cleared=str(res.get("clear_result")) == "success",
|
||
channel=req.get("channel"),
|
||
x=req.get("position", {}).get("x"),
|
||
y=req.get("position", {}).get("y")))
|
||
wall = [r for r in run["actions"] if r.get("wall_ts_ms")]
|
||
real_s = (wall[-1]["wall_ts_ms"] - wall[0]["wall_ts_ms"]) / 1000.0 if len(wall) > 1 else 0.0
|
||
return dict(rows=rows, virtual_s=float(run["actions"][-1]["response"].get("virtual_time_s") or 0),
|
||
real_s=real_s,
|
||
move_s=sum(r["move_s"] for r in rows),
|
||
switch_s=sum(r["switch_s"] for r in rows),
|
||
action_s=sum(r["action_s"] for r in rows),
|
||
move_m=sum(r["move_m"] for r in rows))
|
||
|
||
|
||
def solved_channels(run: dict[str, Any]) -> list[int]:
|
||
return sorted({int(a["request"]["channel"]) for a in run["actions"]
|
||
if a["path"] == "/clear" and "channel" in a["request"]
|
||
and str(a["response"].get("clear_result")) == "success"})
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 几何工具
|
||
# ---------------------------------------------------------------------------
|
||
def clip_halfplane(poly: np.ndarray, a: float, b: float, c: float) -> np.ndarray:
|
||
"""Sutherland–Hodgman：半平面 a·x + b·y <= c 裁剪凸多边形。"""
|
||
if len(poly) == 0:
|
||
return poly
|
||
out: list[list[float]] = []
|
||
n = len(poly)
|
||
for i in range(n):
|
||
cur, nxt = poly[i], poly[(i + 1) % n]
|
||
dc = a * cur[0] + b * cur[1] - c
|
||
dn = a * nxt[0] + b * nxt[1] - c
|
||
if dc <= 1e-9:
|
||
out.append(list(cur))
|
||
if dn > 1e-9:
|
||
t = dc / (dc - dn)
|
||
out.append([cur[0] + t * (nxt[0] - cur[0]), cur[1] + t * (nxt[1] - cur[1])])
|
||
elif dn <= 1e-9:
|
||
t = dc / (dc - dn)
|
||
out.append([cur[0] + t * (nxt[0] - cur[0]), cur[1] + t * (nxt[1] - cur[1])])
|
||
return np.array(out) if out else np.empty((0, 2))
|
||
|
||
|
||
def wedge_polygon(station: Sequence[float], bearing: float, half_deg: float,
|
||
reach: float) -> np.ndarray:
|
||
a0, a1 = math.radians(bearing - half_deg), math.radians(bearing + half_deg)
|
||
ang = np.linspace(a0, a1, 16)
|
||
pts = [[station[0], station[1]]] + \
|
||
[[station[0] + reach * math.cos(t), station[1] + reach * math.sin(t)] for t in ang]
|
||
return np.array(pts)
|
||
|
||
|
||
def clip_convex(region: np.ndarray, a: float, b: float, c: float) -> np.ndarray:
|
||
"""半平面 a·x + b·y <= c 裁剪（对 numpy 数组的薄封装）。"""
|
||
return clip_halfplane(np.asarray(region, dtype=float), a, b, c)
|
||
|
||
|
||
def clip_wedge(region: np.ndarray, station: Sequence[float], bearing: float,
|
||
half_deg: float) -> np.ndarray:
|
||
"""用角域 W(s, θ ± δ) 裁剪区域。
|
||
|
||
角域是两条边界射线的半平面交。示向度可能跨越 0/360 度，法向的正负不能靠叉积
|
||
猜：两个半平面各试一次，用"结果中心与站点的连线是否落在示向度 ±δ 内"来判定
|
||
哪一侧是角域内部。
|
||
"""
|
||
region = np.asarray(region, dtype=float)
|
||
if len(region) == 0:
|
||
return region
|
||
for edge in (bearing - half_deg, bearing + half_deg):
|
||
ang = math.radians(edge)
|
||
a, b = -math.sin(ang), math.cos(ang)
|
||
c = a * station[0] + b * station[1]
|
||
candidates = [cand for cand in (clip_convex(region, a, b, c),
|
||
clip_convex(region, -a, -b, -c))
|
||
if len(cand) >= 3]
|
||
if not candidates:
|
||
return np.empty((0, 2))
|
||
if len(candidates) == 1:
|
||
region = candidates[0]
|
||
continue
|
||
region = min(candidates, key=lambda cand: _direction_error(cand, station,
|
||
bearing, half_deg))
|
||
return region
|
||
|
||
|
||
def _direction_error(poly: np.ndarray, station: Sequence[float], bearing: float,
|
||
half_deg: float) -> float:
|
||
"""多边形中心方向与示向度的偏差；用于挑正确的半平面一侧。"""
|
||
centre = poly.mean(axis=0)
|
||
dx, dy = centre[0] - station[0], centre[1] - station[1]
|
||
if abs(dx) < 1e-12 and abs(dy) < 1e-12:
|
||
return 0.0
|
||
return abs(angle_diff(math.degrees(math.atan2(dy, dx)), bearing))
|
||
|
||
|
||
def polygon_area(poly: np.ndarray) -> float:
|
||
if len(poly) < 3:
|
||
return 0.0
|
||
x, y = poly[:, 0], poly[:, 1]
|
||
return 0.5 * abs(float(np.dot(x, np.roll(y, -1)) - np.dot(y, np.roll(x, -1))))
|
||
|
||
|
||
def angle_diff(a: float, b: float) -> float:
|
||
return (a - b + 180.0) % 360.0 - 180.0
|
||
|
||
|
||
def wedge_clip(station: Sequence[float], bearing: float, half_deg: float,
|
||
box: Sequence[float]) -> np.ndarray:
|
||
"""角域与视窗矩形的交集，用于局部放大图。"""
|
||
x0, y0, x1, y1 = box
|
||
poly = np.array([[x0, y0], [x1, y0], [x1, y1], [x0, y1]], dtype=float)
|
||
return clip_wedge(poly, station, bearing, half_deg)
|
||
|
||
|
||
def q1_solve(stations: Sequence[Sequence[float]]) -> dict[str, Any]:
|
||
"""调用生产求解器，避免图中数字与代码不一致。"""
|
||
sys.path.insert(0, str(ROOT / "src"))
|
||
from mcm26b import Measurement, solve # noqa: PLC0415
|
||
res = solve([Measurement(x, y, b, h) for x, y, b, h in stations])
|
||
dc, mec = res["diameter_circle"], res["minimum_enclosing_circle"]
|
||
return dict(vertices=np.array(res["vertices"]), pair=res["diameter_pair"],
|
||
dc=dc, mec=mec)
|
||
|
||
|
||
def q1_practice_case() -> dict[str, Any] | None:
|
||
for path in sorted((ROOT / "output" / "q1" / "practice").glob("*_observations.json")):
|
||
doc = json.loads(path.read_text(encoding="utf-8"))
|
||
if doc.get("result"):
|
||
doc["_path"] = path
|
||
return doc
|
||
return None
|
||
|
||
|
||
def load_plan(name: str) -> dict[str, Any]:
|
||
path = ROOT / "src" / "mcm26b" / "guarantee" / "plans" / f"{name}.json"
|
||
return json.loads(path.read_text(encoding="utf-8"))
|
||
|
||
|
||
def seven_bound(a: float = RING_RADIUS_M, radius: float = SEVEN_R) -> float:
|
||
return max(a / math.sqrt(3), math.sqrt(radius ** 2 + a ** 2 - math.sqrt(3) * radius * a))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 排版小工具
|
||
# ---------------------------------------------------------------------------
|
||
def draw_boxes(ax: Any, boxes: Sequence[dict[str, Any]], font: Any) -> None:
|
||
"""画一组统一风格的圆角框（流程示意图用）。"""
|
||
for b in boxes:
|
||
ax.add_patch(FancyBboxPatch((b["x"], b["y"]), b["w"], b["h"],
|
||
boxstyle="round,pad=0.02,rounding_size=0.6",
|
||
linewidth=1.0, edgecolor=b.get("edge", C["primary"]),
|
||
facecolor=b.get("face", C["primary_fill"]),
|
||
zorder=2))
|
||
ax.text(b["x"] + b["w"] / 2, b["y"] + b["h"] / 2, b["label"],
|
||
ha="center", va="center", fontproperties=b.get("font", font["hei_bold"]),
|
||
fontsize=b.get("size", 8.2), color=b.get("color", C["ink"]),
|
||
linespacing=1.5, zorder=3)
|
||
|
||
|
||
def connect(ax: Any, p0: Sequence[float], p1: Sequence[float], *,
|
||
color: str | None = None, style: str = "-", lw: float = 1.0,
|
||
rad: float = 0.0) -> None:
|
||
ax.add_patch(FancyArrowPatch(p0, p1, arrowstyle="-|>", mutation_scale=9,
|
||
linewidth=lw, color=color or C["muted"],
|
||
linestyle=style, zorder=4,
|
||
connectionstyle=f"arc3,rad={rad}"))
|
||
|
||
|
||
def chips(ax: Any, items: Sequence[tuple[str, str]], *, x: float = 0.02, y: float = 0.02,
|
||
dy: float = 0.06, font: Any = None) -> None:
|
||
"""左下角竖排色块图例（坐标用 axes 分数）。"""
|
||
for i, (label, color) in enumerate(items):
|
||
yy = y + i * dy
|
||
ax.add_patch(FancyBboxPatch((x, yy - 0.012), 0.018, 0.024,
|
||
boxstyle="round,pad=0.002,rounding_size=0.004",
|
||
transform=ax.transAxes, facecolor=color,
|
||
edgecolor="none", zorder=5, clip_on=False))
|
||
ax.text(x + 0.026, yy, label, transform=ax.transAxes, va="center", ha="left",
|
||
fontproperties=font, fontsize=7.6, color=C["ink"], zorder=5)
|
||
|
||
|
||
# ===========================================================================
|
||
# 1. 总体思路与算法主线
|
||
# ===========================================================================
|
||
def fig_flow_overall(font: dict[str, Any]) -> Any:
|
||
fig, ax = new_axes(170, 96)
|
||
ax.set_xlim(0, 120)
|
||
ax.set_ylim(0, 56)
|
||
ax.axis("off")
|
||
|
||
def band(tag: str, y: float, items: Sequence[tuple[str, float, float]],
|
||
face: str, color: str) -> None:
|
||
ax.text(3.0, y + 3.0, tag, ha="center", va="center",
|
||
fontproperties=font["hei_bold"], fontsize=8.6, color=color)
|
||
for label, x, width in items:
|
||
draw_boxes(ax, [dict(x=x, y=y, w=width, h=6.0, label=label, face=face,
|
||
edge=color, font=font["song"], size=7.8)], font)
|
||
ax.plot([8.0, 116.0], [y - 1.6, y - 1.6], color=C["grid"], lw=0.8, zorder=1)
|
||
|
||
band("L0", 46.0, [
|
||
("区域 Ω：半径 1800 m", 9.0, 26.5),
|
||
("移动 5 m/s · 检测 5 s", 36.0, 26.5),
|
||
("换频 1 s · 清除 5/3 s", 63.0, 26.5),
|
||
("源 10~16 个 · 频道 1~20", 90.0, 25.5)], C["primary_fill"], C["primary"])
|
||
|
||
band("L1", 33.0, [
|
||
("证据层：候选集与粒子幸存集", 9.0, 33.0),
|
||
("决策层：选点与任务表", 44.0, 33.0),
|
||
("执行层：串行客户端", 79.0, 33.0)], C["primary_fill"], C["primary"])
|
||
for x in (42.6, 77.6):
|
||
connect(ax, (x, 36.0), (x + 1.4, 36.0), color=C["ink"])
|
||
|
||
band("L2", 20.0, [
|
||
("问题 1 定位解算", 9.0, 25.5),
|
||
("问题 2 第二点选择", 35.5, 25.5),
|
||
("问题 3 全向搜索", 62.0, 25.5),
|
||
("问题 4 混合定向", 88.5, 27.0)], C["accent_fill"], C["accent"])
|
||
|
||
band("L3", 7.0, [
|
||
("发现闭环", 9.0, 33.0),
|
||
("定位闭环", 44.0, 33.0),
|
||
("清除闭环", 79.0, 33.0)], C["success_fill"], C["success"])
|
||
return fig
|
||
|
||
|
||
# ===========================================================================
|
||
# 2. 执行步骤：会话状态机
|
||
# ===========================================================================
|
||
def fig_steps_state_machine(font: dict[str, Any]) -> Any:
|
||
fig, ax = new_axes(160, 34)
|
||
ax.set_xlim(0, 100)
|
||
ax.set_ylim(0, 22)
|
||
ax.axis("off")
|
||
states = [("IDLE", 3, C["grid"], C["muted"]),
|
||
("倒计时 5 s", 26, C["warn_fill"], C["warn"]),
|
||
("OPEN 接口开放", 49, C["success_fill"], C["success"]),
|
||
("CLOSED 断连", 72, C["accent_fill"], C["accent"])]
|
||
for label, x, face, edge in states:
|
||
draw_boxes(ax, [dict(x=x, y=9, w=22, h=6.5, label=label, face=face, edge=edge,
|
||
font=font["hei_bold"], size=8.4)], font)
|
||
for i in range(3):
|
||
connect(ax, (25 + i * 23 + 0.3, 12.2), (26 + i * 23 - 0.5, 12.2), color=C["ink"])
|
||
ax.text(50, 6.2, "现实预算 1200 s（程序）与测试窗口 1500 s（服务端）独立计时",
|
||
ha="center", fontproperties=font["song"], fontsize=7.6, color=C["muted"])
|
||
ax.text(83, 4.0, "剩余不足 60 s 时预留 /exit", ha="center",
|
||
fontproperties=font["song"], fontsize=7.6, color=C["accent"])
|
||
return fig
|
||
|
||
|
||
# ===========================================================================
|
||
# 3. 执行时间：附件示例逐步推演
|
||
# ===========================================================================
|
||
EXAMPLE_STEPS = [
|
||
("/enter", 0.0, 0.0, 0.0),
|
||
("/measure ch1", 100.0, 0.0, 5.0),
|
||
("/measure ch2", 0.0, 1.0, 5.0),
|
||
("/clear ch3", 80.0, 0.0, 3.0),
|
||
("/measure ch2", 0.0, 0.0, 5.0),
|
||
("/exit", 0.0, 0.0, 0.0),
|
||
]
|
||
|
||
|
||
def fig_timing_attachment(font: dict[str, Any]) -> Any:
|
||
fig, ax = new_axes(160, 46)
|
||
y = np.arange(len(EXAMPLE_STEPS))[::-1]
|
||
move = np.array([s[1] for s in EXAMPLE_STEPS])
|
||
switch = np.array([s[2] for s in EXAMPLE_STEPS])
|
||
action = np.array([s[3] for s in EXAMPLE_STEPS])
|
||
ax.barh(y, move, color=C["primary"], height=0.6, label="移动 L/5")
|
||
ax.barh(y, switch, left=move, color=C["warn"], height=0.6, label="换频 1 s")
|
||
ax.barh(y, action, left=move + switch, color=C["success"], height=0.6,
|
||
label="检测/清除动作")
|
||
cumulative = np.cumsum(move + switch + action)
|
||
for i, (yy, total, acc) in enumerate(zip(y, move + switch + action, cumulative)):
|
||
if total > 0:
|
||
ax.text(move[i] + switch[i] + action[i] + 2, yy, f"{total:.0f} s",
|
||
va="center", fontproperties=font["times"], fontsize=7.6, color=C["ink"])
|
||
ax.text(-4, yy, f"{acc:.0f}", va="center", ha="right",
|
||
fontproperties=font["times_bold"], fontsize=7.8, color=C["accent"])
|
||
ax.set_yticks(y)
|
||
ax.set_yticklabels([s[0] for s in EXAMPLE_STEPS], fontproperties=font["times"],
|
||
fontsize=7.8)
|
||
ax.set_xlabel("虚拟耗时 / s", fontproperties=font["song"])
|
||
ax.set_xlim(-14, 132)
|
||
ax.text(-4, len(EXAMPLE_STEPS) - 0.35, "累计", ha="right", va="center",
|
||
fontproperties=font["song"], fontsize=7.6, color=C["accent"])
|
||
ax.legend(prop=font["song"], frameon=False, loc="lower right", ncols=3)
|
||
for side in ("top", "right"):
|
||
ax.spines[side].set_visible(False)
|
||
ax.grid(True, axis="x", color=C["grid"], lw=0.6)
|
||
ax.set_axisbelow(True)
|
||
return fig
|
||
|
||
|
||
# ===========================================================================
|
||
# 4. 执行时间：实测日志分解
|
||
# ===========================================================================
|
||
def fig_timing_measured(run: dict[str, Any], font: dict[str, Any]) -> Any:
|
||
tl = decompose(run)
|
||
rows = tl["rows"]
|
||
seq = np.array([r["seq"] for r in rows])
|
||
move = np.cumsum([r["move_s"] for r in rows])
|
||
switch = np.cumsum([r["switch_s"] for r in rows])
|
||
action = np.cumsum([r["action_s"] for r in rows])
|
||
total = move + switch + action
|
||
parts = [tl["move_s"], tl["action_s"], tl["switch_s"]]
|
||
labels = ["移动", "检测/清除", "换频"]
|
||
colors = [C["primary"], C["success"], C["warn"]]
|
||
|
||
fig, (ax, ax2) = new_axes(170, 54, ncols=2, width_ratios=[1.45, 1.0])
|
||
ax.stackplot(seq, move, switch, action, colors=colors,
|
||
labels=["移动 L/5", "检测/清除动作", "换频 1 s"], lw=0)
|
||
ax.set_xlim(seq[0], seq[-1])
|
||
ax.set_ylim(0, total[-1] * 1.08)
|
||
ax.set_xlabel("动作序号", fontproperties=font["song"])
|
||
ax.set_ylabel("累计虚拟时间 / s", fontproperties=font["song"])
|
||
ax.legend(prop=font["song"], frameon=False, loc="upper left")
|
||
for side in ("top", "right"):
|
||
ax.spines[side].set_visible(False)
|
||
ax.grid(True, axis="y", color=C["grid"], lw=0.6)
|
||
ax.set_axisbelow(True)
|
||
|
||
# 右：构成条形 + 汇总（同一坐标下三段，末段极窄故编号标注）
|
||
ax2.set_xlim(0, 1)
|
||
ax2.set_ylim(0, 1)
|
||
ax2.axis("off")
|
||
bar_y, bar_h = 0.62, 0.14
|
||
left = 0.0
|
||
for i, (value, color) in enumerate(zip(parts, colors)):
|
||
frac = value / total[-1]
|
||
ax2.add_patch(FancyBboxPatch((left, bar_y), frac, bar_h,
|
||
boxstyle="round,pad=0,rounding_size=0.005",
|
||
facecolor=color, edgecolor="none"))
|
||
if frac > 0.06:
|
||
ax2.text(left + frac / 2, bar_y + bar_h / 2, f"{value:,.0f} s", ha="center",
|
||
va="center", fontproperties=font["song"], fontsize=7.4,
|
||
color="white")
|
||
left += frac
|
||
ax2.text(0.0, 0.84, f"虚拟 {total[-1]:,.1f} s", fontproperties=font["song"],
|
||
fontsize=7.8, color=C["ink"])
|
||
ax2.text(0.5, 0.84, f"现实 {tl['real_s']:.2f} s", fontproperties=font["song"],
|
||
fontsize=7.8, color=C["ink"])
|
||
ax2.text(1.0, 0.84, f"加速 {total[-1] / max(tl['real_s'], 1e-9):,.0f} 倍", ha="right",
|
||
fontproperties=font["song"], fontsize=7.8, color=C["accent"])
|
||
for i, (value, label, color) in enumerate(zip(parts, labels, colors)):
|
||
ax2.add_patch(FancyBboxPatch((0.0 + i * 0.36, 0.40), 0.03, 0.05,
|
||
boxstyle="round,pad=0.002,rounding_size=0.01",
|
||
facecolor=color, edgecolor="none"))
|
||
ax2.text(0.05 + i * 0.36, 0.425,
|
||
f"{i + 1} {label} {value:,.0f} s", fontproperties=font["song"],
|
||
fontsize=7.4, va="center")
|
||
ax2.text(0.0, 0.24, f"检测 {sum(1 for r in rows if r['path'] == '/measure')} 次 · "
|
||
f"清除 {sum(1 for r in rows if r['path'] == '/clear')} 次 · "
|
||
f"里程 {tl['move_m'] / 1000:.1f} km",
|
||
fontproperties=font["song"], fontsize=7.4, color=C["muted"])
|
||
return fig
|
||
|
||
|
||
# ===========================================================================
|
||
# 5. 问题一：交会定位区域
|
||
# ===========================================================================
|
||
def _q1_data() -> dict[str, Any]:
|
||
case = q1_practice_case()
|
||
if case:
|
||
return dict(subtitle=f"官方演练 {case['case_code']}",
|
||
stations=[(m["x"], m["y"], m["bearing_deg"], m["half_angle_deg"])
|
||
for m in case["measurements"]],
|
||
verts=np.array(case["result"]["vertices"]),
|
||
dc=case["result"]["diameter_circle"],
|
||
mec=case["result"]["minimum_enclosing_circle"],
|
||
src=case["_path"])
|
||
doc = json.loads((ROOT / "examples" / "q1" / "triangle.json").read_text(encoding="utf-8"))
|
||
sol = q1_solve([(m["x"], m["y"], m["bearing_deg"], m.get("half_angle_deg", 1.0))
|
||
for m in doc["measurements"]])
|
||
return dict(subtitle="反例算例 triangle.json",
|
||
stations=[(m["x"], m["y"], m["bearing_deg"], m.get("half_angle_deg", 1.0))
|
||
for m in doc["measurements"]],
|
||
verts=sol["vertices"], dc=sol["dc"], mec=sol["mec"],
|
||
src=ROOT / "examples" / "q1" / "triangle.json")
|
||
|
||
|
||
def fig_q1_meetup(font: dict[str, Any]) -> Any:
|
||
data = _q1_data()
|
||
stations = data["stations"]
|
||
xs = [s[0] for s in stations]
|
||
ys = [s[1] for s in stations]
|
||
cx, cy = (min(xs) + max(xs)) / 2, (min(ys) + max(ys)) / 2
|
||
half = max(max(xs) - min(xs), max(ys) - min(ys)) / 2 * 1.35 + 60
|
||
reach = 2.4 * half
|
||
mec_c = data["mec"]["center"]
|
||
mec_r = data["mec"]["radius_m"]
|
||
dc_c, dc_r = data["dc"]["center"], data["dc"]["radius_m"]
|
||
# 放大窗口取定位区域外接框的 1.8 倍，使多边形、直径圆、MEC 同框且尺寸可比
|
||
verts = data["verts"]
|
||
zoom_c = (float(verts[:, 0].mean()), float(verts[:, 1].mean()))
|
||
span = float(np.abs(verts - np.array(zoom_c)).max())
|
||
zoom = max(1.8 * span, 0.8 * mec_r + span, 12.0)
|
||
|
||
fig, axes = axes_for_panels(62.0, [1.0, 1.0], width_ratios=[1.0, 1.0])
|
||
ax, axz = axes
|
||
polar_axes(ax, half, grid=half / 2)
|
||
for sx, sy, bearing, hd in stations:
|
||
ax.add_patch(Polygon(wedge_polygon((sx, sy), bearing, hd, reach), closed=True,
|
||
facecolor=C["primary"], alpha=0.16, edgecolor="none"))
|
||
ax.plot([sx], [sy], marker="o", ms=4, color=C["primary"])
|
||
ax.annotate(f"({sx:.0f}, {sy:.0f})", (sx, sy), textcoords="offset points",
|
||
xytext=(6, 4), fontproperties=font["times"], fontsize=7.2,
|
||
color=C["primary"])
|
||
ax.add_patch(Polygon(data["verts"], closed=True, facecolor=C["accent"],
|
||
edgecolor=C["accent"], lw=1.0))
|
||
ax.annotate("定位区域 P →", (float(data["verts"][:, 0].mean()),
|
||
float(data["verts"][:, 1].mean())),
|
||
textcoords="offset points", xytext=(14, -18),
|
||
fontproperties=font["song"], fontsize=7.6, color=C["accent"],
|
||
arrowprops=dict(arrowstyle="-", color=C["accent"], lw=0.9))
|
||
ax.set_xlabel("x / m", fontproperties=font["song"])
|
||
ax.set_ylabel("y / m", fontproperties=font["song"])
|
||
|
||
polar_axes(axz, zoom, grid=zoom / 2, center=zoom_c)
|
||
axz.set_xticks([])
|
||
axz.set_yticks([])
|
||
axz.set_ylabel("")
|
||
# 放大面板：角域按面板的坐标范围裁剪，范围与 xlim/ylim 一致
|
||
xlim, ylim = axz.get_xlim(), axz.get_ylim()
|
||
box = (float(xlim[0]), float(ylim[0]), float(xlim[1]), float(ylim[1]))
|
||
for sx, sy, bearing, hd in stations:
|
||
poly = wedge_clip((sx, sy), bearing, hd, box)
|
||
if len(poly):
|
||
axz.add_patch(Polygon(poly, closed=True, facecolor=C["primary"],
|
||
alpha=0.28, edgecolor="none"))
|
||
region = np.array([[box[0], box[1]], [box[2], box[1]],
|
||
[box[2], box[3]], [box[0], box[3]]], dtype=float)
|
||
for sx, sy, bearing, hd in stations:
|
||
region = clip_wedge(region, (sx, sy), bearing, hd)
|
||
if len(region) >= 3:
|
||
axz.add_patch(Polygon(region, closed=True, facecolor=C["accent_fill"],
|
||
edgecolor=C["accent"], lw=1.6))
|
||
axz.add_patch(Polygon(data["verts"], closed=True, facecolor=C["accent_fill"],
|
||
edgecolor=C["accent"], lw=1.0))
|
||
axz.add_patch(Circle(dc_c, dc_r, facecolor="none", edgecolor=C["violet"],
|
||
lw=1.2, ls=(0, (5, 3))))
|
||
axz.add_patch(Circle(mec_c, mec_r, facecolor="none", edgecolor=C["success"], lw=1.4))
|
||
axz.plot([dc_c[0]], [dc_c[1]], marker="o", ms=3.4, color=C["violet"])
|
||
axz.plot([mec_c[0]], [mec_c[1]], marker="o", ms=3.4, color=C["success"])
|
||
outside = max(data["verts"], key=lambda v: math.dist(v, dc_c))
|
||
axz.annotate("", xy=outside, xytext=dc_c,
|
||
arrowprops=dict(arrowstyle="-|>", color=C["accent"], lw=1.0))
|
||
axz.annotate("E > 0", (outside[0], outside[1]), textcoords="offset points",
|
||
xytext=(10, 6), fontproperties=font["times"], fontsize=7.4,
|
||
color=C["accent"])
|
||
axz.set_xlabel(f"x / m（放大 ±{zoom:.0f} m）", fontproperties=font["song"])
|
||
chips(ax, [("前向角域 W(s, θ ± 1°)", C["primary"]), ("定位区域 P", C["accent"])])
|
||
chips(axz, [("直径圆（不覆盖）", C["violet"]), ("最小包围圆 MEC", C["success"])])
|
||
return fig
|
||
|
||
|
||
# ===========================================================================
|
||
# 6. 问题一：直径圆反例
|
||
# ===========================================================================
|
||
def fig_q1_counterexample(font: dict[str, Any]) -> Any:
|
||
doc = json.loads((ROOT / "examples" / "q1" / "triangle.json").read_text(encoding="utf-8"))
|
||
stations = [(m["x"], m["y"], m["bearing_deg"], m.get("half_angle_deg", 1.0))
|
||
for m in doc["measurements"]]
|
||
sol = q1_solve(stations)
|
||
verts = sol["vertices"]
|
||
dc, mec = sol["dc"], sol["mec"]
|
||
mec_r = mec["radius_m"]
|
||
mec_c, dc_c = mec["center"], dc["center"]
|
||
# 反例的定位区域是边长 40 m 的等边三角形：取 1.8 倍直径做放大窗口
|
||
zoom = max(1.8 * float(np.abs(verts - np.array(mec_c)).max()) * 2.0, 40.0)
|
||
|
||
fig, axes = axes_for_panels(60.0, [1.0, 1.0])
|
||
ax, axz = axes
|
||
half = 1200.0
|
||
polar_axes(ax, half, grid=400.0)
|
||
for sx, sy, bearing, hd in stations:
|
||
ax.add_patch(Polygon(wedge_polygon((sx, sy), bearing, hd, 3 * half), closed=True,
|
||
facecolor=C["primary"], alpha=0.14, edgecolor="none"))
|
||
ax.plot([sx], [sy], marker="o", ms=4, color=C["primary"])
|
||
ax.annotate(f"({sx:.0f}, {sy:.0f})", (sx, sy), textcoords="offset points",
|
||
xytext=(6, 4), fontproperties=font["times"], fontsize=7.2,
|
||
color=C["primary"])
|
||
ax.add_patch(Polygon(verts, closed=True, facecolor=C["accent"], edgecolor=C["accent"]))
|
||
ax.set_xlabel("x / m", fontproperties=font["song"])
|
||
ax.set_ylabel("y / m", fontproperties=font["song"])
|
||
|
||
polar_axes(axz, zoom, grid=zoom / 2, center=mec_c)
|
||
axz.set_xticks([])
|
||
axz.set_yticks([])
|
||
axz.set_ylabel("")
|
||
box = (mec_c[0] - zoom, mec_c[1] - zoom, mec_c[0] + zoom, mec_c[1] + zoom)
|
||
for sx, sy, bearing, hd in stations:
|
||
poly = wedge_clip((sx, sy), bearing, hd, box)
|
||
if len(poly):
|
||
axz.add_patch(Polygon(poly, closed=True, facecolor=C["primary"],
|
||
alpha=0.24, edgecolor="none"))
|
||
region = np.array([[box[0], box[1]], [box[2], box[1]],
|
||
[box[2], box[3]], [box[0], box[3]]], dtype=float)
|
||
for sx, sy, bearing, hd in stations:
|
||
region = clip_wedge(region, (sx, sy), bearing, hd)
|
||
if len(region) >= 3:
|
||
axz.add_patch(Polygon(region, closed=True, facecolor=C["accent_fill"],
|
||
edgecolor=C["accent"], lw=1.6))
|
||
axz.add_patch(Circle(dc_c, dc["radius_m"], facecolor="none", edgecolor=C["violet"],
|
||
lw=1.4, ls=(0, (5, 3))))
|
||
axz.add_patch(Circle(mec_c, mec_r, facecolor="none", edgecolor=C["success"], lw=1.4))
|
||
outside = max(verts, key=lambda v: math.dist(v, dc_c))
|
||
axz.annotate("最大超出顶点", xy=outside, xytext=(outside[0] + 6, outside[1] - 10),
|
||
fontproperties=font["song"], fontsize=7.6, color=C["accent"],
|
||
arrowprops=dict(arrowstyle="-|>", color=C["accent"], lw=1.0))
|
||
axz.set_xlabel(f"x / m（放大 ±{zoom:.0f} m）", fontproperties=font["song"])
|
||
chips(axz, [("直径圆（不覆盖）", C["violet"]), ("最小包围圆 MEC", C["success"])])
|
||
return fig
|
||
|
||
|
||
# ===========================================================================
|
||
# 7. 问题二：保证接收盘
|
||
# ===========================================================================
|
||
def fig_q2_safe_disk(font: dict[str, Any]) -> Any:
|
||
delta = math.radians(1.0)
|
||
a = 1500.0 / (2 * math.cos(delta))
|
||
safe = R_EFF - a
|
||
bearing = 30.0
|
||
c = (a * math.cos(math.radians(bearing)), a * math.sin(math.radians(bearing)))
|
||
|
||
fig, axes = axes_for_panels(64.0, [1.0, 1.0], width_ratios=[1.0, 1.0])
|
||
ax, axz = axes
|
||
half = 1900.0
|
||
polar_axes(ax, half, grid=500.0)
|
||
ax.add_patch(Circle((0, 0), OMEGA_R, facecolor="none", edgecolor=C["primary"], lw=1.2))
|
||
ax.add_patch(Polygon(wedge_polygon((0, 0), bearing, 1.0, 1500.0), closed=True,
|
||
facecolor=C["accent"], alpha=0.5, edgecolor=C["accent"], lw=1.0))
|
||
ax.plot([0], [0], marker="o", ms=4, color=C["accent"])
|
||
ax.annotate("s1", (0, 0), textcoords="offset points", xytext=(6, -10),
|
||
fontproperties=font["times"], fontsize=7.6, color=C["accent"])
|
||
ax.add_patch(Circle(c, a, facecolor="none", edgecolor=C["violet"], lw=1.0,
|
||
ls=(0, (5, 3))))
|
||
ax.add_patch(Circle(c, safe, facecolor=C["success"], alpha=0.35,
|
||
edgecolor=C["success"], lw=1.2))
|
||
ax.annotate("", xy=c, xytext=(0, 0),
|
||
arrowprops=dict(arrowstyle="-|>", color=C["violet"], lw=1.0))
|
||
ax.annotate(f"a = {a:.1f} m", ((c[0] + 0) / 2, (c[1] + 0) / 2),
|
||
textcoords="offset points", xytext=(-6, 26), ha="right",
|
||
fontproperties=font["song"], fontsize=7.6, color=C["violet"],
|
||
arrowprops=dict(arrowstyle="-", color=C["violet"], lw=0.8))
|
||
ax.annotate(f"安全盘半径 {safe:.1f} m", (c[0], c[1] - safe * 0.7),
|
||
textcoords="offset points", xytext=(14, -20), ha="left",
|
||
fontproperties=font["song"], fontsize=7.6, color=C["success"],
|
||
arrowprops=dict(arrowstyle="-", color=C["success"], lw=0.8))
|
||
ax.text(0, OMEGA_R * 0.98, "Ω：R = 1800 m", fontproperties=font["song"],
|
||
fontsize=7.6, color=C["primary"], ha="center", va="top")
|
||
ax.set_xlabel("x / m", fontproperties=font["song"])
|
||
ax.set_ylabel("y / m", fontproperties=font["song"])
|
||
|
||
# 右：安全性依据（两半径相加 = 1000 m）
|
||
axz.set_aspect("auto")
|
||
axz.axis("off")
|
||
axz.set_xlim(-1.15, 1.25)
|
||
axz.set_ylim(-0.35, 1.05)
|
||
r_safe, r_a = safe / 1000.0, a / 1000.0
|
||
axz.add_patch(Circle((0, 0.45), r_a, facecolor=C["violet"], alpha=0.16,
|
||
edgecolor=C["violet"], lw=1.0, ls=(0, (4, 3))))
|
||
axz.add_patch(Circle((0, 0.45), r_safe, facecolor=C["success"], alpha=0.35,
|
||
edgecolor=C["success"], lw=1.2))
|
||
axz.plot([0], [0.45], marker="o", ms=3.4, color=C["ink"])
|
||
ang = math.radians(20)
|
||
axz.annotate("", xy=(r_a * math.cos(ang), 0.45 + r_a * math.sin(ang)),
|
||
xytext=(0, 0.45), arrowprops=dict(arrowstyle="-", color=C["violet"], lw=1.0))
|
||
axz.annotate("", xy=(-r_safe * math.cos(ang), 0.45 - r_safe * math.sin(ang)),
|
||
xytext=(0, 0.45), arrowprops=dict(arrowstyle="-", color=C["success"], lw=1.0))
|
||
axz.text(0.30, 0.72, f"a = {a:.1f} m", fontproperties=font["song"], fontsize=7.6,
|
||
color=C["violet"])
|
||
axz.text(-0.62, 0.18, f"1000 − a = {safe:.1f} m", fontproperties=font["song"],
|
||
fontsize=7.6, color=C["success"])
|
||
chips(ax, [("前向角域 W(s1)", C["accent"]), ("截断角域外接圆 B(c, a)", C["violet"]),
|
||
("安全候选盘 B(c, 1000 − a)", C["success"])])
|
||
return fig
|
||
|
||
|
||
# ===========================================================================
|
||
# 8. 问题三：七点覆盖
|
||
# ===========================================================================
|
||
def fig_q3_seven_coverage(font: dict[str, Any]) -> Any:
|
||
a = RING_RADIUS_M
|
||
stations = [(0.0, 0.0)] + [(a * math.cos(k * math.pi / 3), a * math.sin(k * math.pi / 3))
|
||
for k in range(6)]
|
||
bound = seven_bound(a)
|
||
fig, axes = axes_for_panels(66.0, [1.0, 0.95], width_ratios=[1.0, 0.95])
|
||
ax, axr = axes
|
||
half = 2050.0
|
||
polar_axes(ax, half, grid=600.0)
|
||
ax.add_patch(Circle((0, 0), OMEGA_R, facecolor="#f7fafc", edgecolor=C["primary"], lw=1.4))
|
||
for px, py in stations:
|
||
ax.add_patch(Circle((px, py), R_EFF, facecolor=C["success"], alpha=0.28,
|
||
edgecolor=C["success"], lw=0.8))
|
||
ax.add_patch(Circle((0, 0), OMEGA_R, facecolor="none", edgecolor=C["primary"], lw=1.4))
|
||
for px, py in stations:
|
||
ax.plot([px], [py], marker="o", ms=4.2, color=C["accent"])
|
||
ax.plot([0, a], [0, 0], color=C["accent"], lw=0.9)
|
||
ax.text(a / 2, 60, f"a = {a:.1f} m", ha="center", fontproperties=font["times"],
|
||
fontsize=7.6, color=C["accent"])
|
||
ax.plot([0], [-SEVEN_R], marker="o", ms=4.6, mfc="none", mec=C["accent"], mew=1.2)
|
||
ax.annotate(f"最坏点 {bound:.2f} m", (0, -SEVEN_R), textcoords="offset points",
|
||
xytext=(-4, 14), ha="center", fontproperties=font["song"], fontsize=7.6,
|
||
color=C["accent"],
|
||
arrowprops=dict(arrowstyle="-", color=C["accent"], lw=0.8))
|
||
ax.text(0, OMEGA_R, "Ω", ha="center", va="bottom", fontproperties=font["song"],
|
||
fontsize=8, color=C["primary"])
|
||
ax.set_xlabel("x / m", fontproperties=font["song"])
|
||
ax.set_ylabel("y / m", fontproperties=font["song"])
|
||
chips(ax, [("覆盖站点（原点 + 六等分环）", C["accent"]), ("1000 m 接收盘", C["success"])])
|
||
|
||
axr.axis("off")
|
||
axr.set_xlim(0, 1)
|
||
axr.set_ylim(0, 1)
|
||
rows = [("七点解析（默认）", "7", f"{bound:.2f} m", C["success_fill"], C["success"]),
|
||
("49 点网格基线", "49", "989.95 m", "#f7f9fb", C["muted"]),
|
||
("31 点最小命中集（题 4）", "31", "990.05 m", "#f7f9fb", C["muted"])]
|
||
header = ["方案", "点数", "覆盖半径上界"]
|
||
xs = [0.02, 0.62, 0.78]
|
||
for x, head in zip(xs, header):
|
||
axr.text(x, 0.9, head, fontproperties=font["hei_bold"], fontsize=7.8,
|
||
color=C["primary"])
|
||
for i, (name, count, dist, face, edge) in enumerate(rows):
|
||
y = 0.76 - i * 0.13
|
||
axr.add_patch(FancyBboxPatch((0.0, y - 0.045), 1.0, 0.11,
|
||
boxstyle="round,pad=0.004,rounding_size=0.02",
|
||
facecolor=face, edgecolor=edge, lw=0.8))
|
||
axr.text(xs[0], y, name, fontproperties=font["song"], fontsize=7.6)
|
||
axr.text(xs[1], y, count, fontproperties=font["times"], fontsize=7.6)
|
||
axr.text(xs[2], y, dist, fontproperties=font["times"], fontsize=7.6)
|
||
return fig
|
||
|
||
|
||
# ===========================================================================
|
||
# 9. 问题四：31 点最小命中集
|
||
# ===========================================================================
|
||
def fig_q4_mincover(font: dict[str, Any]) -> Any:
|
||
plan = load_plan("q4_mincover_v2")
|
||
pts = np.array(plan["points"], dtype=float)
|
||
cert = plan["certificate"]
|
||
|
||
fig, axes = axes_for_panels(64.0, [1.0, 1.0], width_ratios=[1.0, 1.0])
|
||
ax, axr = axes
|
||
half = 2150.0
|
||
polar_axes(ax, half, grid=600.0)
|
||
ax.add_patch(Circle((0, 0), OMEGA_R, facecolor="#f7fafc", edgecolor=C["primary"], lw=1.4))
|
||
ax.plot(pts[:, 0], pts[:, 1], ls="none", marker="o", ms=4.2, color=C["accent"],
|
||
label=f"{len(pts)} 个覆盖点")
|
||
ax.legend(prop=font["song"], frameon=False, loc="lower left")
|
||
ax.text(0, OMEGA_R, "Ω", ha="center", va="bottom", fontproperties=font["song"],
|
||
fontsize=8, color=C["primary"])
|
||
ax.set_xlabel("x / m", fontproperties=font["song"])
|
||
ax.set_ylabel("y / m", fontproperties=font["song"])
|
||
ax.set_xlabel("x / m", fontproperties=font["song"])
|
||
|
||
# 右：forward_half_disk 语义与证书
|
||
axr.axis("off")
|
||
axr.set_xlim(0, 1)
|
||
axr.set_ylim(0, 1)
|
||
axr.add_patch(Circle((0.36, 0.66), 0.24, facecolor="#f7f9fb", edgecolor=C["muted"],
|
||
lw=0.9))
|
||
half_disk = [[0.36, 0.66]] + [[0.36 + 0.24 * math.cos(math.radians(-90 + 180 * i / 24)),
|
||
0.66 + 0.24 * math.sin(math.radians(-90 + 180 * i / 24))]
|
||
for i in range(25)]
|
||
axr.add_patch(Polygon(half_disk, closed=True, facecolor=C["accent"], alpha=0.32,
|
||
edgecolor="none"))
|
||
axr.annotate("", xy=(0.57, 0.66), xytext=(0.36, 0.66),
|
||
arrowprops=dict(arrowstyle="-|>", color=C["accent"], lw=1.2))
|
||
axr.plot([0.36, 0.36], [0.42, 0.90], color=C["accent"], lw=0.9, ls=(0, (4, 3)))
|
||
axr.annotate("", xy=(0.60, 0.95), xytext=(0.12, 0.95),
|
||
arrowprops=dict(arrowstyle="<->", color=C["muted"], lw=0.8))
|
||
axr.text(0.36, 0.90, "覆盖半径 1000 m", ha="center", fontproperties=font["song"],
|
||
fontsize=7.4, color=C["muted"])
|
||
axr.text(0.62, 0.60, "朝向未知\n±90° 前向半盘", ha="left", va="top",
|
||
fontproperties=font["song"], fontsize=7.4, color=C["accent"], linespacing=1.6)
|
||
return fig
|
||
# ===========================================================================
|
||
# 10/11. WebUI 回放复现
|
||
# ===========================================================================
|
||
def fig_replay(run: dict[str, Any], font: dict[str, Any], *, directional: bool) -> Any:
|
||
tl = decompose(run)
|
||
rows = tl["rows"]
|
||
truth = run["truth"]
|
||
solved = set(solved_channels(run))
|
||
|
||
fig = plt.figure(figsize=(170 / 25.4, 108 / 25.4))
|
||
ax = fig.add_axes([0.04, 0.30, 0.60, 0.66])
|
||
polar_axes(ax, 2100.0, grid=1000.0)
|
||
ax.add_patch(Circle((0, 0), OMEGA_R, facecolor="#f7fafc", edgecolor=C["primary"], lw=1.4))
|
||
for src in truth:
|
||
ch = int(src["channel"])
|
||
color = C["accent"] if ch in solved else C["warn"]
|
||
ax.plot([src["x"]], [src["y"]], marker="o", ms=5, color=color, zorder=5)
|
||
ax.annotate(f"ch{ch}", (src["x"], src["y"]), textcoords="offset points",
|
||
xytext=(5, 2), fontproperties=font["times"], fontsize=6.8, color=color)
|
||
if directional and src.get("directional"):
|
||
hd = math.radians(float(src.get("heading_deg") or 0.0))
|
||
ax.annotate("", xy=(src["x"] + 200 * math.cos(hd), src["y"] + 200 * math.sin(hd)),
|
||
xytext=(src["x"], src["y"]),
|
||
arrowprops=dict(arrowstyle="-|>", color=C["accent"], lw=1.0))
|
||
path = [(0.0, 0.0)]
|
||
for row in rows:
|
||
if row["x"] is None:
|
||
continue
|
||
if (row["x"], row["y"]) != path[-1]:
|
||
path.append((row["x"], row["y"]))
|
||
if len(path) > 1:
|
||
px, py = zip(*path)
|
||
ax.plot(px, py, color=C["primary"], lw=1.1, alpha=0.85, zorder=3)
|
||
for act, row in zip(run["actions"], rows):
|
||
if row["x"] is None:
|
||
continue
|
||
if act["path"] == "/measure" and row["hit"]:
|
||
bearing = math.radians(float(act["response"].get("svd_deg") or 0.0))
|
||
ax.annotate("", xy=(row["x"] + 220 * math.cos(bearing),
|
||
row["y"] + 220 * math.sin(bearing)),
|
||
xytext=(row["x"], row["y"]),
|
||
arrowprops=dict(arrowstyle="-|>", color=C["success"], lw=0.7,
|
||
alpha=0.55))
|
||
ax.plot([row["x"]], [row["y"]], marker="o", ms=2.6, color=C["success"])
|
||
elif act["path"] == "/clear":
|
||
ax.plot([row["x"]], [row["y"]], marker="o", ms=7, mfc="none",
|
||
mec=C["success"] if row["cleared"] else C["muted"], mew=1.2)
|
||
if row["cleared"]:
|
||
ax.plot([row["x"]], [row["y"]], marker="x", ms=5, color=C["success"],
|
||
mew=1.6)
|
||
ax.set_xlabel("x / m", fontproperties=font["song"])
|
||
ax.set_ylabel("y / m", fontproperties=font["song"])
|
||
ax.legend(handles=[
|
||
Line2D([], [], marker="o", ls="none", color=C["success"], label="已清除频道"),
|
||
Line2D([], [], marker="o", ls="none", color=C["warn"], label="未清除频道"),
|
||
Line2D([], [], color=C["accent"], lw=1.2, label="定向源朝向"),
|
||
Line2D([], [], color=C["success"], lw=0.8, label="检测示向度"),
|
||
], prop=font["song"], frameon=False, loc="upper center",
|
||
bbox_to_anchor=(0.5, -0.13), ncols=4)
|
||
|
||
# 右侧：事件流水（前 10 条）
|
||
axt = fig.add_axes([0.70, 0.30, 0.285, 0.62])
|
||
axt.axis("off")
|
||
axt.set_xlim(0, 1)
|
||
axt.set_ylim(0, 1)
|
||
axt.text(0.0, 1.0, run["case_code"], fontproperties=font["times_bold"], fontsize=7.8,
|
||
color=C["primary"], va="top")
|
||
axt.text(0.0, 0.93, f"{run['module']} · 源 {len(truth)} · "
|
||
f"全清 {len(solved)}/{len(truth)}", fontproperties=font["song"],
|
||
fontsize=7.4, color=C["muted"], va="top")
|
||
axt.text(0.0, 0.86, "动作 参数 虚拟 t/s", fontproperties=font["hei_bold"],
|
||
fontsize=7.4, color=C["primary"], va="top")
|
||
for i, (act, row) in enumerate(list(zip(run["actions"], rows))[:10]):
|
||
if act["path"] == "/measure":
|
||
detail = f"ch{row['channel']} ({row['x']:.0f},{row['y']:.0f})"
|
||
elif act["path"] == "/clear":
|
||
detail = f"ch{row['channel']} {'成功' if row['cleared'] else '未发现'}"
|
||
elif act["path"] == "/enter":
|
||
detail = "启动"
|
||
else:
|
||
detail = "退出"
|
||
axt.text(0.0, 0.78 - i * 0.075, act["path"], fontproperties=font["times"],
|
||
fontsize=7.2, va="top")
|
||
axt.text(0.45, 0.78 - i * 0.075, detail, fontproperties=font["song"],
|
||
fontsize=7.2, va="top")
|
||
axt.text(1.0, 0.78 - i * 0.075, f"{row['virtual_s']:.0f}",
|
||
fontproperties=font["times"], fontsize=7.2, va="top", ha="right")
|
||
|
||
# 底部：虚拟时间轴（检测命中 / 清除刻度）
|
||
axb = fig.add_axes([0.04, 0.13, 0.945, 0.09])
|
||
axb.set_xlim(0, max(tl["virtual_s"], 1e-9))
|
||
axb.set_ylim(0, 1)
|
||
axb.axis("off")
|
||
axb.add_patch(FancyBboxPatch((0, 0.35), tl["virtual_s"], 0.3,
|
||
boxstyle="round,pad=0,rounding_size=0",
|
||
facecolor=C["primary_fill"], edgecolor="none"))
|
||
for act, row in zip(run["actions"], rows):
|
||
if act["path"] == "/measure" and row["hit"]:
|
||
axb.plot([row["virtual_s"], row["virtual_s"]], [0.2, 0.8], color=C["accent"],
|
||
lw=0.9)
|
||
elif act["path"] == "/clear" and row["cleared"]:
|
||
axb.plot([row["virtual_s"], row["virtual_s"]], [0.05, 0.95], color=C["success"],
|
||
lw=1.4)
|
||
axb.text(0, 1.15, "0 s", fontproperties=font["times"], fontsize=7.2, color=C["muted"])
|
||
axb.text(tl["virtual_s"], 1.15, f"{tl['virtual_s']:,.0f} s", ha="right",
|
||
fontproperties=font["times"], fontsize=7.2, color=C["muted"])
|
||
axb.text(tl["virtual_s"] / 2, 1.15, "竖线：检测到信号（红）/ 成功清除（绿）",
|
||
ha="center", fontproperties=font["song"], fontsize=7.2, color=C["muted"])
|
||
return fig
|
||
|
||
|
||
# ===========================================================================
|
||
# 入口
|
||
# ===========================================================================
|
||
def build_figures(only: Sequence[str] | None, scale: float) -> list[dict[str, Any]]:
|
||
font = setup_style()
|
||
want = set(only) if only else None
|
||
runs = discover_runs()
|
||
q3 = pick_run(runs, "q3", 10) if runs else None
|
||
q4 = pick_run(runs, "q4", 12) if runs else None
|
||
if not runs:
|
||
print("[fig] 未找到 JSONL，跳过与日志相关的 3 张图")
|
||
|
||
jobs: list[tuple[str, Callable[[], Any], list[dict[str, Any]]]] = [
|
||
("flow-overall-architecture", lambda: fig_flow_overall(font),
|
||
[source("src/mcm26b/config.py"), source("docs/B题建模分析与实施计划.md")]),
|
||
("steps-state-machine", lambda: fig_steps_state_machine(font),
|
||
[source("docs/协议规则表与状态机.md"), source("src/mcm26b/protocol/client.py")]),
|
||
("timing-attachment-example-105-111-194-199", lambda: fig_timing_attachment(font),
|
||
[source("docs/协议规则表与状态机.md")]),
|
||
("geo-q1-convex-intersection-zoom", lambda: fig_q1_meetup(font), []),
|
||
("geo-q1-diameter-circle-counterexample", lambda: fig_q1_counterexample(font),
|
||
[source("examples/q1/triangle.json")]),
|
||
("geo-q2-guaranteed-reception-disk", lambda: fig_q2_safe_disk(font),
|
||
[source("src/mcm26b/config.py"), source("docs/algo-2.md")]),
|
||
("geo-q3-seven-point-coverage", lambda: fig_q3_seven_coverage(font),
|
||
[source("src/mcm26b/planner/seven.py"), source("docs/algo-3.md")]),
|
||
("geo-q4-minimum-cover-31-points", lambda: fig_q4_mincover(font),
|
||
[source("src/mcm26b/guarantee/plans/q4_mincover_v2.json"), source("docs/algo-4.md")]),
|
||
]
|
||
if q3:
|
||
jobs.append(("timing-measured-q3-10-sources", lambda: fig_timing_measured(q3, font),
|
||
[source(q3["source"], case_code=q3["case_code"], module=q3["module"])]))
|
||
jobs.append(("replay-webui-q3-omnidirectional",
|
||
lambda: fig_replay(q3, font, directional=False),
|
||
[source(q3["source"], case_code=q3["case_code"])]))
|
||
if q4:
|
||
jobs.append(("replay-webui-q4-directional",
|
||
lambda: fig_replay(q4, font, directional=True),
|
||
[source(q4["source"], case_code=q4["case_code"])]))
|
||
|
||
entries: list[dict[str, Any]] = []
|
||
for stem, maker, sources in jobs:
|
||
if want and stem not in want:
|
||
continue
|
||
fig = maker()
|
||
if stem == "geo-q1-convex-intersection-zoom":
|
||
sources = [source(_q1_data()["src"])]
|
||
entry = save(fig, stem, sources, scale)
|
||
entries.append(entry)
|
||
print(f"[fig] {stem:<44} → {entry['png']}")
|
||
return entries
|
||
|
||
|
||
def main() -> int:
|
||
names = ["flow-overall-architecture", "steps-state-machine",
|
||
"timing-attachment-example-105-111-194-199", "timing-measured-q3-10-sources",
|
||
"geo-q1-convex-intersection-zoom", "geo-q1-diameter-circle-counterexample",
|
||
"geo-q2-guaranteed-reception-disk", "geo-q3-seven-point-coverage",
|
||
"geo-q4-minimum-cover-31-points", "replay-webui-q3-omnidirectional",
|
||
"replay-webui-q4-directional"]
|
||
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
||
parser.add_argument("--only", default="", help="逗号分隔的图名子集")
|
||
parser.add_argument("--scale", type=float, default=1.0, help="尺寸倍率")
|
||
parser.add_argument("--list", action="store_true")
|
||
args = parser.parse_args()
|
||
if args.list:
|
||
print("\n".join(names))
|
||
return 0
|
||
only = [s.strip() for s in args.only.split(",") if s.strip()]
|
||
unknown = [s for s in only if s not in names]
|
||
if unknown:
|
||
parser.error(f"未知图名 {unknown}")
|
||
entries = build_figures(only or None, args.scale)
|
||
manifest = {"generator": "scripts/make_figures.py",
|
||
"command": "uv run python scripts/make_figures.py"
|
||
+ (f" --only {args.only}" if args.only else ""),
|
||
"palette": C, "fonts": {"dir": "docs/paper/fonts", "map": FONTS},
|
||
"figures": entries}
|
||
path = OUT_DIR / "manifest.json"
|
||
path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n",
|
||
encoding="utf-8")
|
||
print(f"[fig] manifest → {path.relative_to(ROOT)}（{len(entries)} 张）")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|