Files
mcm-26-b/scripts/plot_paper_figures.py
2026-09-13 18:43:25 +08:00

616 lines
28 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- coding: utf-8 -*-
from __future__ import annotations
import csv, json, math, re
from pathlib import Path
import numpy as np
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
from matplotlib.patches import Arc, Circle, FancyArrowPatch, Polygon, Rectangle, Wedge
ROOT = Path(__file__).resolve().parents[1]
FIG = ROOT / "docs" / "figures"
FIG.mkdir(parents=True, exist_ok=True)
INK = "#234C70"
GRAY = "#676767"
LIGHT = "#9DB7CA"
PALE = "#D9E4EC"
GOLD = "#B58B3A"
GREEN = "#4E7A55"
plt.rcParams.update({"font.sans-serif": ["Microsoft YaHei", "SimHei", "DejaVu Sans"], "axes.unicode_minus": False,
"figure.dpi": 120, "savefig.dpi": 300, "axes.edgecolor": GRAY,
"axes.labelcolor": "#222222", "text.color": "#222222"})
def save(fig, name):
try:
fig.tight_layout()
except Exception:
pass
fig.savefig(FIG / name, bbox_inches="tight", facecolor="white")
plt.close(fig)
def load_formal():
rows = []
path = ROOT / "docs" / "paper" / "data" / "formal-results.csv"
with path.open(encoding="utf-8-sig") as f:
for row in csv.DictReader(f):
for key in ("problem_no", "formal_index", "cleared_jammer_count",
"measure_accepted_count", "channel_switch_count",
"clear_failure_count"):
row[key] = int(row[key])
for key in ("virtual_time_s", "program_run_duration_s",
"inferred_movement_time_s"):
row[key] = float(row[key])
rows.append(row)
return rows
def fig_formal_time_bars():
rows = load_formal()
labels = ["Q%d-%d" % (r["problem_no"], r["formal_index"]) for r in rows]
move = np.array([r["inferred_movement_time_s"] for r in rows])
measure = np.array([5.0 * r["measure_accepted_count"] for r in rows])
switch = np.array([1.0 * r["channel_switch_count"] for r in rows])
clear = np.array([5.0 * r["cleared_jammer_count"] + 3.0 * r["clear_failure_count"] for r in rows])
fig, ax = plt.subplots(figsize=(9.0, 3.8))
ax.barh(labels, move, color=INK, label="移动")
ax.barh(labels, measure, left=move, color=LIGHT, label="检测")
ax.barh(labels, switch, left=move + measure, color=GRAY, label="换频")
ax.barh(labels, clear, left=move + measure + switch, color=GOLD, label="清除")
ax.set_xlabel("累计虚拟时间 / s")
ax.grid(axis="x", color="#E5E5E5", linewidth=0.7)
ax.set_axisbelow(True)
ax.legend(loc="lower right", frameon=False, ncol=4)
save(fig, "formal-time-bars.png")
def practice_macro(name: str) -> str:
"""从 ``data/practice-macros.tex`` 读取正文主局案例编码。
正文与插图必须引用同一次采集的同一局,因此这里不写死案例编码,
而是读 ``build_practice_data.py`` 生成的宏,避免两处口径分叉。
"""
path = ROOT / "docs" / "paper" / "data" / "practice-macros.tex"
match = re.search(r"\\newcommand\{\\" + name + r"\}\{([^}]*)\}",
path.read_text(encoding="utf-8"))
if match is None:
raise KeyError(f"宏 {name} 不存在;请先运行 scripts/build_practice_data.py")
return match.group(1)
def box(ax, x, y, w, h, text, fc=PALE, ec=INK, size=8):
"""画一个流程框;``size`` 用 pt 指定字号,避免缩放后文字过小或过大。"""
ax.add_patch(Rectangle((x, y), w, h, facecolor=fc, edgecolor=ec, linewidth=1.0))
ax.text(x + w/2, y + h/2, text, ha="center", va="center", fontsize=size, wrap=True)
def arrow(ax, p, q, color=GRAY):
ax.add_patch(FancyArrowPatch(p, q, arrowstyle="-|>", mutation_scale=10,
linewidth=1.0, color=color))
def fig_framework():
"""闭环决策图:观测—候选集—任务—动作的循环,自下而上闭合。
箭头只画在框与框之间的空白处:起点取框边、终点取框边,
不在框内穿过,也不在一点挤成一团。因此这里给四周留出布线通道,
并用一条水平汇流线把"任务选择"到"退出/报告"的支路接出去。
"""
fig, ax = plt.subplots(figsize=(7.6, 5.15))
ax.axis("off")
box(ax, 0.3, 2.8, 2.0, 0.7, "观测\n示向度 / 近距离 / 无信号", size=8.5)
box(ax, 3.1, 2.8, 2.0, 0.7, "候选集更新\n几何 / 网格 / 朝向", size=8.5)
box(ax, 3.1, 1.4, 2.0, 0.7, "任务选择\n覆盖 / 定位 / 清除", size=8.5)
box(ax, 0.3, 1.4, 2.0, 0.7, "动作执行\n移动与计时", size=8.5)
box(ax, 3.1, 0.25, 2.0, 0.6, "退出 / 报告", fc="#EFEFEF", size=8.5)
# 回路:观测 →(右)候选集更新 →(下)任务选择 →(左)动作执行 →(上)观测
arrow(ax, (2.3, 3.15), (3.1, 3.15))
arrow(ax, (4.1, 2.8), (4.1, 2.1))
arrow(ax, (3.1, 1.75), (2.3, 1.75))
arrow(ax, (1.3, 2.1), (1.3, 2.8))
# 支路:任务选择下行到"退出/报告",箭头端点都落在框边,不进入框内
arrow(ax, (4.1, 1.4), (4.1, 0.85))
ax.set_xlim(0.12, 5.28); ax.set_ylim(0.13, 3.62)
save(fig, "framework-decision-loop.png")
ax.set_xlim(0, 5.6); ax.set_ylim(0.3, 3.7)
save(fig, "framework-decision-loop.png")
def fig_state_machine():
fig, ax = plt.subplots(figsize=(7.0, 2.2))
ax.axis("off")
xs = [(0.1, "未进入"), (1.8, "进入\n5 s 倒计时"), (3.8, "接口开放"),
(5.8, "关闭\n退出 / 超时")]
for x, t in xs:
box(ax, x, 0.6, 1.5, 0.8, t)
for x in (1.6, 3.6, 5.6):
arrow(ax, (x, 1.0), (x + 0.2, 1.0))
ax.set_xlim(0, 7.4); ax.set_ylim(0, 2.0)
save(fig, "steps-state-machine.png")
def fig_timing_attachment():
steps = ["步骤 1", "步骤 2", "步骤 3", "步骤 4"]
move = np.array([100.0, 0.0, 80.0, 0.0])
action = np.array([5.0, 5.0, 3.0, 5.0])
switch = np.array([0.0, 1.0, 0.0, 0.0])
fig, ax = plt.subplots(figsize=(6.0, 3.2))
ax.bar(steps, move, color=INK, label="移动")
ax.bar(steps, action, bottom=move, color=LIGHT, label="检测 / 清除")
ax.bar(steps, switch, bottom=move + action, color=GRAY, label="换频")
for i, total in enumerate(move + action + switch):
ax.text(i, total + 2, "%.0f" % total, ha="center", fontsize=8)
ax.set_ylabel("虚拟时间 / s")
ax.legend(frameon=False)
save(fig, "timing-attachment-example-105-111-194-199.png")
def clip_halfplane(poly, a, b, c):
if len(poly) == 0:
return poly
out = []
for i in range(len(poly)):
p = poly[i]; q = poly[(i + 1) % len(poly)]
fp = a * p[0] + b * p[1] - c
fq = a * q[0] + b * q[1] - c
if fp <= 1e-9:
out.append(p)
if (fp < -1e-9 and fq > 1e-9) or (fp > 1e-9 and fq < -1e-9):
t = fp / (fp - fq)
out.append(p + t * (q - p))
return np.array(out, dtype=float)
def wedge_halfplanes(sx, sy, theta_deg, delta_deg):
theta = math.radians(theta_deg); delta = math.radians(delta_deg)
a1 = math.sin(theta - delta); b1 = -math.cos(theta - delta)
c1 = a1 * sx + b1 * sy
a2 = -math.sin(theta + delta); b2 = math.cos(theta + delta)
c2 = a2 * sx + b2 * sy
a3 = -math.cos(theta); b3 = -math.sin(theta)
c3 = a3 * sx + b3 * sy
return [(a1, b1, c1), (a2, b2, c2), (a3, b3, c3)]
def intersection_polygon(meas, bound=2500.0):
poly = np.array([[-bound, -bound], [bound, -bound], [bound, bound], [-bound, bound]], dtype=float)
for m in meas:
for a, b, c in wedge_halfplanes(m["x"], m["y"], m["bearing_deg"], m["half_angle_deg"]):
poly = clip_halfplane(poly, a, b, c)
if len(poly) == 0:
return poly
return poly
def pair_circle(p, q):
c = (p + q) / 2.0
r = np.linalg.norm(p - q) / 2.0
return c, r
def contains_all(c, r, pts, tol=1e-6):
return all(np.linalg.norm(p - c) <= r + tol for p in pts)
def circumcircle(p, q, r):
ax, ay = p; bx, by = q; cx, cy = r
d = 2.0 * (ax * (by - cy) + bx * (cy - ay) + cx * (ay - by))
if abs(d) < 1e-12:
return None
ux = ((ax*ax + ay*ay) * (by - cy) + (bx*bx + by*by) * (cy - ay) + (cx*cx + cy*cy) * (ay - by)) / d
uy = ((ax*ax + ay*ay) * (cx - bx) + (bx*bx + by*by) * (ax - cx) + (cx*cx + cy*cy) * (bx - ax)) / d
c = np.array([ux, uy])
return c, np.linalg.norm(c - p)
def min_enclosing_circle(pts):
pts = np.asarray(pts, dtype=float)
best_c = pts[0]; best_r = 0.0
for i in range(len(pts)):
for j in range(i + 1, len(pts)):
c, r = pair_circle(pts[i], pts[j])
if contains_all(c, r, pts) and r < best_r or (best_r == 0.0 and contains_all(c, r, pts)):
best_c, best_r = c, r
for i in range(len(pts)):
for j in range(i + 1, len(pts)):
for k in range(j + 1, len(pts)):
cc = circumcircle(pts[i], pts[j], pts[k])
if cc is None:
continue
c, r = cc
if contains_all(c, r, pts) and r < best_r:
best_c, best_r = c, r
return best_c, best_r
def load_q1_example():
"""问题一交会算例(scripts/build_practice_data.py 复算自建模拟器观测)。"""
path = ROOT / "docs" / "paper" / "data" / "practice-geometry.json"
return json.loads(path.read_text(encoding="utf-8"))
def set_equal_view(ax, centre, half_width, box_aspect):
"""设定等比例视野,并让轴框高宽比等于 ``box_aspect``。
``set_box_aspect`` 与固定坐标范围同时使用会让 matplotlib 丢弃范围并告警,
因此这里直接按目标高宽比放大较窄的一维:三幅取同一 ``box_aspect`` 与同一宽度后,
轴框尺寸完全一致,版面高度自然严格相等。
"""
half_height = half_width * box_aspect
ax.set_xlim(centre[0] - half_width, centre[0] + half_width)
ax.set_ylim(centre[1] - half_height, centre[1] + half_height)
ax.set_aspect("equal")
def fig_q1_geometry():
"""问题一几何图:角域交会、定位区域放大、等边三角形反例,三幅等高并排。
三幅必须等高才能并排读,因此用 :func:`set_equal_view` 把三个轴框
固定为同一高宽比;三幅取相同宽度后,版面高度严格相等。
视野由观测点与定位区域联合确定,避免只按观测点设限导致区域跑出画面。
反例图原本是独立插图,这里并入同一张图,省去一次浮动与一个图注。
"""
example = load_q1_example()
meas = example["measurements"]
poly = np.array(example["vertices"], dtype=float)
box_aspect = 1.15
fig, (ax, ax2, ax3) = plt.subplots(1, 3, figsize=(9.6, 4.0))
stations = np.array([[m["x"], m["y"]] for m in meas], dtype=float)
for m in meas:
x, y = m["x"], m["y"]
th = math.radians(m["bearing_deg"])
d = math.radians(example["half_angle_deg"])
for sgn in (-1.0, 1.0):
ang = th + sgn * d
end = np.array([x, y]) + 900.0 * np.array([math.cos(ang), math.sin(ang)])
ax.plot([x, end[0]], [y, end[1]], color=LIGHT, linewidth=0.9, zorder=1)
ax.scatter([x], [y], color=GRAY, s=16, zorder=3)
ax.annotate("S", (x, y), textcoords="offset points", xytext=(6, 5), fontsize=8)
ax.add_patch(Polygon(poly, closed=True, facecolor=PALE, edgecolor=INK,
linewidth=1.4, zorder=2))
# 视野取"观测点 ∪ 定位区域"的外包框,再补边距,保证两者同时入画。
stacked = np.vstack([stations, poly])
lo, hi = stacked.min(axis=0), stacked.max(axis=0)
set_equal_view(ax, (lo + hi) / 2.0, max(hi - lo) * 0.54, box_aspect)
ax.grid(True, color="#EEEEEE")
ax.set_title("(a) 角域交会", fontsize=9)
ax.set_xlabel("x / m"); ax.set_ylabel("y / m")
ax2.add_patch(Polygon(poly, closed=True, facecolor=PALE, edgecolor=INK, linewidth=1.4))
ax2.add_patch(Circle(example["diameter_circle_centre"], example["diameter_circle_radius_m"],
fill=False, edgecolor=GOLD, linestyle="--", linewidth=1.2,
label="直径圆"))
ax2.add_patch(Circle(example["mec_centre"], example["mec_radius_m"],
fill=False, edgecolor=GREEN, linewidth=1.4, label="最小包围圆"))
ax2.scatter(poly[:, 0], poly[:, 1], color=INK, s=16, label="定位区域顶点")
lo, hi = poly.min(axis=0), poly.max(axis=0)
set_equal_view(ax2, (lo + hi) / 2.0, max(hi - lo) * 0.62, box_aspect)
ax2.grid(True, color="#EEEEEE")
ax2.set_title("(b) 定位区域放大", fontsize=9)
ax2.set_xlabel("x / m"); ax2.set_ylabel("y / m")
ax2.legend(frameon=False, fontsize=7, loc="lower left", bbox_to_anchor=(0.0, 0.0))
tri = np.array([[0.0, 0.0], [40.0, 0.0], [20.0, 20.0 * math.sqrt(3.0)]])
ax3.add_patch(Polygon(tri, closed=True, fill=False, edgecolor=INK, linewidth=1.4))
ax3.add_patch(Circle((20.0, 0.0), 20.0, fill=False, edgecolor=GOLD,
linestyle="--", linewidth=1.2, label="直径圆 ($r=20$ m)"))
ax3.add_patch(Circle((20.0, 20.0 / math.sqrt(3.0)), 40.0 / math.sqrt(3.0), fill=False,
edgecolor=GREEN, linewidth=1.4,
label="最小包围圆 ($\\rho=40/\\sqrt{3}$ m)"))
ax3.scatter(tri[:, 0], tri[:, 1], color=INK, s=16)
for (px, py), name, dx, dy in ((tri[0], "A", -11.0, -3.0), (tri[1], "B", 7.0, -3.0),
(tri[2], "C", -2.0, 9.0)):
ax3.annotate(name, (px, py), textcoords="offset points", xytext=(dx, dy), fontsize=9)
set_equal_view(ax3, (20.0, 15.0), 30.0, box_aspect)
ax3.grid(True, color="#EEEEEE")
ax3.set_title("(c) 等边三角形反例", fontsize=9)
ax3.set_xlabel("x / m"); ax3.set_ylabel("y / m")
ax3.legend(frameon=False, fontsize=6.5, loc="upper right")
save(fig, "geo-q1-geometry.png")
def fig_q2_safe_disk():
delta = math.radians(1.0); rp = 1500.0; a = rp / (2.0 * math.cos(delta)); safe = 1000.0 - a
fig, axes = plt.subplots(1, 2, figsize=(8.0, 3.6))
ax = axes[0]
ax.plot([0, rp * math.cos(delta)], [0, rp * math.sin(delta)], color=GRAY, linewidth=0.8)
ax.plot([0, rp * math.cos(delta)], [0, -rp * math.sin(delta)], color=GRAY, linewidth=0.8)
ax.add_patch(Circle((a, 0), a, fill=False, edgecolor=LIGHT, linestyle="--", linewidth=1.2))
ax.add_patch(Circle((a, 0), safe, facecolor=PALE, edgecolor=INK, linewidth=1.2, alpha=0.8))
ax.scatter([0], [0], color=GRAY, s=12); ax.scatter([a], [0], color=INK, s=12)
ax.text(a + 20, 20, "c", fontsize=9); ax.text(10, 15, "s", fontsize=9)
ax.set_aspect("equal"); ax.grid(True, color="#EEEEEE")
ax.set_xlim(-80, rp + 80); ax.set_ylim(-320, 320)
ax.set_xlabel("x / m"); ax.set_ylabel("y / m"); ax.set_title("截断角域与安全候选盘", fontsize=9)
ax = axes[1]
ax.plot([0, 0], [0, 1], color=GRAY, linewidth=2)
ax.plot([0, 1], [0, 0], color=INK, linewidth=4)
ax.text(0.5, -0.08, "a = 1500/(2cosδ)", ha="center", fontsize=8)
ax.text(0.5, 0.55, "安全半径 = 1000 - a", ha="center", fontsize=8)
ax.axis("off")
save(fig, "geo-q2-guaranteed-reception-disk.png")
def fig_q4_plan():
"""31 点最小命中集:站点分布 + 圆域外站点 + 定向源的读图约定。
原图把一枚前向半盘楔形压在站点与轨迹附近,读者无法判断它在说什么。
这里改为:圆外用空心标记单独区分,并在"圆外站点"上给出定向源示例——
绿箭头为朝向、浅绿细线扇形为 ±90° 发射范围、红色虚线为该点实测到的示向度,
三者构成一条完整的"朝向 → 可见范围 → 观测量"读图链。
"""
data = json.loads((ROOT / "src" / "mcm26b" / "guarantee" / "plans"
/ "q4_mincover_v2.json").read_text(encoding="utf-8"))
pts = np.array(data["points"], dtype=float)
inside = np.linalg.norm(pts, axis=1) <= 1800.0
fig, ax = plt.subplots(figsize=(7.6, 6.6))
ax.add_patch(Circle((0, 0), 1800.0, facecolor=PALE, edgecolor=INK,
linewidth=1.1, alpha=0.45, zorder=1))
ax.text(0, 1840, "Ω", ha="center", va="bottom", fontsize=9)
ax.scatter(pts[inside, 0], pts[inside, 1], color=INK, s=18,
label=f"圆域内检测站({int(inside.sum())} 个)", zorder=4)
ax.scatter(pts[~inside, 0], pts[~inside, 1], facecolor="white", edgecolor=INK,
linewidth=1.1, s=26, label=f"圆域外检测站({int((~inside).sum())} 个)",
zorder=4)
ax.scatter([0], [0], color=GOLD, marker="*", s=60, label="原点(出发位置)", zorder=5)
# 定向源示例挂在"离圆域最近的圆外站点"上,扇形朝外,不再压住站点云。
outside = pts[~inside]
anchor = outside[np.argmin(np.linalg.norm(outside, axis=1))] if len(outside) else pts[0]
bearing = 20.0
draw_source(ax, {"x": anchor[0], "y": anchor[1], "heading_deg": bearing,
"directional": True}, fan_radius=560.0, annotate=True)
# 源放在该站的前向半盘内,并画出它到检测站的实测示向度,补全读图链。
src = np.array([anchor[0] + 820.0 * math.cos(math.radians(bearing - 30.0)),
anchor[1] + 820.0 * math.sin(math.radians(bearing - 30.0))])
ax.scatter([src[0]], [src[1]], marker="X", s=110, color="#B33A3A",
label="定向源读图示例(非真实源)", zorder=6)
ax.plot([anchor[0], src[0]], [anchor[1], src[1]], color="#B33A3A",
linestyle=(0, (4, 3)), linewidth=1.1, zorder=3)
measured = math.degrees(math.atan2(src[1] - anchor[1], src[0] - anchor[0]))
ax.annotate(f"实测示向度 {measured:.0f}°",
xy=((anchor[0] + src[0]) / 2, (anchor[1] + src[1]) / 2),
xytext=(src[0] + 320, src[1] + 380), fontsize=7.5, color="#B33A3A",
arrowprops=dict(arrowstyle="->", color="#B33A3A", linewidth=0.8))
draw_reading_guide(ax, "定向源读图约定:绿箭头=朝向,浅绿细线扇形\n"
"=该源 ±90° 发射范围,红色虚线=检测站测得\n"
"的示向度(有朝向遮挡时才可能出现无信号)",
anchor=(0.015, 0.30))
ax.set_aspect("equal")
ax.set_xlim(-3100, 3100)
ax.set_ylim(-3100, 3100)
ax.set_xlabel("x / m")
ax.set_ylabel("y / m")
ax.grid(True, color="#EEEEEE", linewidth=0.6)
ax.legend(frameon=False, fontsize=8, loc="center left", bbox_to_anchor=(1.02, 0.5))
save(fig, "geo-q4-minimum-cover-31-points.png")
def load_practice():
"""自建模拟器演练局复算结果(scripts/build_practice_data.py 生成)。"""
path = ROOT / "docs" / "paper" / "data" / "practice-results.json"
return json.loads(path.read_text(encoding="utf-8"))["records"]
def fig_practice_total():
"""演练局虚拟时间:逐局散点 + 全清局均值,用对数轴容纳四次方的跨度。"""
rows = load_practice()
fig, ax = plt.subplots(figsize=(7.6, 3.6))
styles = {3: (INK, "o", "问题三:全向源"), 4: (GOLD, "s", "问题四:混合源")}
for problem, (color, marker, label) in styles.items():
subset = [r for r in rows if r["problem_no"] == problem]
times = [r["virtual_time_s"] for r in subset]
ax.scatter(times, [problem + (0.13 if r["complete"] else -0.13) for r in subset],
color=color, marker=marker, s=34, label=label, zorder=3)
ok = [r["virtual_time_s"] for r in subset if r["complete"]]
mean = sum(ok) / len(ok)
ax.plot([mean, mean], [problem - 0.28, problem + 0.28], color=color, linewidth=2.2, zorder=4)
ax.text(mean, problem + 0.34, "均值 %.0f s" % mean, color=color, fontsize=8, ha="center")
ax.set_xscale("log")
ax.set_yticks([3, 4])
ax.set_yticklabels(["问题三\n(全向源)", "问题四\n(混合源)"])
ax.set_ylim(2.5, 4.6)
ax.set_xlabel("总虚拟时间 / s")
ax.grid(axis="x", color="#E5E5E5", linewidth=0.7)
ax.set_axisbelow(True)
ax.legend(frameon=False, fontsize=8, ncol=2, loc="lower center",
bbox_to_anchor=(0.5, -0.46))
save(fig, "practice-virtual-time.png")
def main():
fig_formal_time_bars()
fig_practice_total()
fig_framework()
fig_state_machine()
fig_timing_attachment()
fig_q1_geometry()
fig_q2_safe_disk()
fig_q4_plan()
fig_selfbuilt_path(practice_macro("PractQThreeMainCase"),
"自建模拟器:全向源搜索与清除路径",
"selfbuilt-q3-search-clear-path.png")
fig_selfbuilt_path(
practice_macro("PractQFourMainCase"),
"自建模拟器:混合源搜索与清除路径(含定向源前向半盘)",
"selfbuilt-q4-search-clear-path.png",
guide="定向源读图约定:绿箭头=发射朝向,\n"
"浅绿细线扇形=该源 ±90° 可见范围;\n"
"每条细射线=一次检测得到的示向度")
def load_selfbuilt_session(case_code):
paths = [ROOT / "referee-data" / "sessions" / case_code / "log.jsonl"]
replay_dir = ROOT / "examples" / "replay"
if replay_dir.exists():
paths += sorted(replay_dir.glob(case_code + "*.jsonl"))
for path in paths:
if path.exists():
data = []
for line in path.read_text(encoding="utf-8").splitlines():
if line.strip():
data.append(json.loads(line))
truth = next((ev.get("truth", []) for ev in data if ev.get("event") == "session_start"), [])
actions = [ev for ev in data if ev.get("event") == "action"]
return truth, actions
raise FileNotFoundError(case_code)
RAY_MAX_M = 620.0 # 示向度射线显示长度上限:过长会盖住轨迹,只画到能读出方向为止
FAN_RADIUS_M = 260.0 # 检测点处示意接收角域的扇形半径
SRC_FAN_RADIUS_M = 300.0 # 定向源前向半盘示意半径:过大即与站点/轨迹重叠
def selfbuilt_path_data(actions):
"""把行为日志整理成画图所需的序列。
返回轨迹、逐次检测(位置、结果、示向度)、成功与失败清除点。
检测结果是画图的关键:``direction`` 反映测到了示向度,
``near`` 表示已进入 5 m 可直接清除,``no_signal`` 表示该点收不到信号。
"""
path = [(0.0, 0.0)]
measures = []
clears_ok = []
clears_fail = []
pos = (0.0, 0.0)
for ev in actions:
req = ev.get("request") or {}
res = ev.get("response") or {}
if "position" in req:
pos = (float(req["position"]["x"]), float(req["position"]["y"]))
if path[-1] != pos:
path.append(pos)
if ev.get("path") == "/measure":
measures.append({"xy": pos,
"result": res.get("measure_result", ""),
"bearing": res.get("svd_deg"),
"channel": req.get("channel")})
elif ev.get("path") == "/clear":
if res.get("clear_result") == "success":
clears_ok.append(pos)
elif "clear_result" in res:
clears_fail.append(pos)
return np.array(path, dtype=float), measures, clears_ok, clears_fail
def draw_source(ax, src, fan_radius=SRC_FAN_RADIUS_M, annotate=False):
"""画一个干扰源;定向源额外画出朝向箭头与其 ±90° 前向半盘。
``annotate`` 只在需要解释读图约定时置真,且每个源只写一次角度标注,
避免同一句话在图上重复出现。
"""
x, y = float(src["x"]), float(src["y"])
if not src.get("directional"):
return
heading = float(src.get("heading_deg", 0.0))
ax.add_patch(Wedge((x, y), fan_radius, heading - 90, heading + 90,
facecolor=GREEN, edgecolor=GREEN, alpha=0.16,
linewidth=0.7, zorder=2))
ax.add_patch(FancyArrowPatch((x, y),
(x + fan_radius * 0.95 * math.cos(math.radians(heading)),
y + fan_radius * 0.95 * math.sin(math.radians(heading))),
arrowstyle="-|>", mutation_scale=9, linewidth=1.2,
color=GREEN, zorder=4))
if annotate:
ax.annotate(f"{heading:.0f}°", xy=(x, y),
xytext=(x - 70, y + fan_radius * 1.12),
fontsize=7.5, color=GREEN, ha="right",
arrowprops=dict(arrowstyle="-", color=GREEN, linewidth=0.6,
shrinkA=1, shrinkB=3))
def draw_reading_guide(ax, text, anchor=(0.015, 0.985)):
"""在坐标区一角写读图约定,替代散落在图上的大段解释文字。"""
ax.text(anchor[0], anchor[1], text, transform=ax.transAxes,
fontsize=7.5, color=GREEN, va="top", ha="left",
bbox=dict(boxstyle="round,pad=0.35", facecolor="white",
edgecolor=GREEN, linewidth=0.6, alpha=0.92), zorder=10)
def draw_measures(ax, measures, ray_color=None):
"""画检测点与示向度射线,并按检测结果区分标记。
射线长度取固定上限:既能读出方向,又不会像原来那样铺满整张图掩盖轨迹。
线宽与不透明度按印刷可辨识度选取——早期版本用 0.5 pt/0.5 太细,
缩到论文栏宽后几乎看不见,这里统一加粗到 1.1 pt 并提高对比度。
"""
ray_color = ray_color or "#7FA6C4"
for item in measures:
x, y = item["xy"]
if item["result"] == "direction" and item["bearing"] is not None:
ang = math.radians(float(item["bearing"]))
unit = np.array([math.cos(ang), math.sin(ang)])
ax.plot([x, x + unit[0] * RAY_MAX_M], [y, y + unit[1] * RAY_MAX_M],
color=ray_color, linewidth=1.1, alpha=0.75, zorder=1,
solid_capstyle="round")
kinds = {
"direction": dict(marker="o", size=16, color=INK, label="检测到示向度(含射线)"),
"near": dict(marker="D", size=18, color=GOLD, label="检测到近距离"),
"no_signal": dict(marker="x", size=12, color="#9A9A9A", label="检测为无信号"),
}
for result, style in kinds.items():
subset = [m for m in measures if m["result"] == result]
if not subset:
continue
xs = [m["xy"][0] for m in subset]
ys = [m["xy"][1] for m in subset]
ax.scatter(xs, ys, marker=style["marker"], s=style["size"], color=style["color"],
linewidths=0.9, label=f"{style['label']}({len(subset)} 次)", zorder=6)
def fig_selfbuilt_path(case_code, title, outname, guide=None):
"""画一局演练的搜索与清除路径。
``guide`` 只在需要解释定向源读图约定时传入;全向源局不画扇形符号,
以免出现图上没有对应元素的图例文字。
"""
truth, actions = load_selfbuilt_session(case_code)
path, measures, clears_ok, clears_fail = selfbuilt_path_data(actions)
fig, ax = plt.subplots(figsize=(8.6, 7.4))
ax.add_patch(Circle((0, 0), 1800, fill=False, edgecolor=INK, linewidth=1.2, zorder=2))
ax.text(0, 1830, "Ω", ha="center", va="bottom", fontsize=9)
for src in truth:
draw_source(ax, src, annotate=bool(src.get("directional")))
ax.scatter([float(s["x"]) for s in truth], [float(s["y"]) for s in truth],
marker="X", s=70, color=GOLD, label="干扰源真值", zorder=7)
ax.plot(path[:, 0], path[:, 1], color=INK, linewidth=1.4, alpha=0.95,
label="机器狗轨迹", zorder=5)
ax.scatter(path[0, 0], path[0, 1], marker="o", s=45, color=GREEN, label="起点", zorder=8)
ax.scatter(path[-1, 0], path[-1, 1], marker="s", s=45, color=GRAY, label="终点", zorder=8)
draw_measures(ax, measures)
if guide:
draw_reading_guide(ax, guide)
if clears_ok:
xy = np.array(clears_ok)
ax.scatter(xy[:, 0], xy[:, 1], marker="*", s=95, color=GREEN,
edgecolor="white", linewidth=0.5, label="成功清除点", zorder=9)
if clears_fail:
xy = np.array(clears_fail)
ax.scatter(xy[:, 0], xy[:, 1], marker="x", s=45, color="#B33A3A",
label="清除失败点", zorder=9)
ax.set_aspect("equal")
ax.set_xlim(-2150, 2150)
ax.set_ylim(-2150, 2150)
ax.set_xlabel("x / m")
ax.set_ylabel("y / m")
ax.set_title(title, fontsize=11)
ax.grid(True, color="#EEEEEE", linewidth=0.6)
ax.legend(loc="center left", bbox_to_anchor=(1.02, 0.5), frameon=False, fontsize=8)
save(fig, outname)
if __name__ == "__main__":
main()