325 lines
16 KiB
Python
325 lines
16 KiB
Python
"""生成论文附录的"核心函数"源码清单（数据源：单文件核心代码 ipynb）。
|
||
|
||
论文附录不再逐字排版全部源码（评测机、命令行外壳、采集/基准脚本、测试与
|
||
未采用方案都不属于求解核心），只保留四问主线的核心函数：以
|
||
``mcm26b-核心代码.ipynb`` 为唯一数据源，按名字抽取函数与类，逐个附一句
|
||
取自 docstring 首行的简短说明，再转义为 LaTeX 可直接排版的形式。
|
||
|
||
用法::
|
||
|
||
python scripts/build_code_appendix.py # 生成 data/source-listings.tex
|
||
python scripts/build_code_appendix.py --check # 只校验并报告规模，不写文件
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import ast
|
||
import json
|
||
import re
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
PAPER = ROOT / "docs/paper"
|
||
NOTEBOOK = ROOT / "mcm26b-核心代码.ipynb"
|
||
|
||
# XeLaTeX + ctex 下 lstlisting 可直接输出中文；这里只把数学符号转成 LaTeX 命令。
|
||
# 每段 = (小节标题, 一句话说明, 要抽取的名字)
|
||
# 名字既可以是模块级函数/类，也可以是 "类名.方法名"，只取被点名的成员。
|
||
SECTIONS: list[tuple[str, str, list[str]]] = [
|
||
("动作计时与开放路径排序",
|
||
r"三段耗时按整数微秒累计；后备清除的走格顺序用最近邻打底并做 2-opt 改进。",
|
||
["CostBreakdown", "measure_cost", "clear_cost", "move_time",
|
||
"nearest_neighbor_order", "two_opt_open"]),
|
||
("问题一：角域交会、直径与最小包围圆",
|
||
r"半平面交求定位区域，顶点对枚举直径，Welzl 算法求最小包围圆，"
|
||
r"并输出直径圆覆盖判定与清除半径判据。",
|
||
["Measurement", "halfplanes", "convex_hull", "diameter",
|
||
"smallest_enclosing_circle", "solve"]),
|
||
("问题二：候选集与保证接收的第二检测点",
|
||
r"候选集用圆盘外切多边形与角域半平面裁剪表示；安全盘半径取"
|
||
r"$1000-\rho$，盘内候选按最坏后验包围半径、同分按移动代价取优。",
|
||
["CandidateRegion", "ClearCertificate", "clear_certificate",
|
||
"candidate_points", "safe_disk_radius", "select_measurement_point",
|
||
"select_second_point"]),
|
||
("问题三：覆盖计划、检测记录与清除证书",
|
||
r"覆盖计划与运行期抽样校验、逐频道检测记录，以及七点解析覆盖与"
|
||
r"全向／定向两种覆盖语义的证书判据。",
|
||
["CoverageCertificate", "CoveragePlan", "coverage_points",
|
||
"CoverageLedger", "seven_points", "seven_plan", "verify_full_disk",
|
||
"verify_forward_half_disk", "plan_instance"]),
|
||
("问题三：网格候选集",
|
||
r"逐频道的网格幸存集，按格心裕量 $\ell/\sqrt2$ 内缩后再删格，"
|
||
r"据此支持负观测更新、跳测规则与缺席证明。",
|
||
["ChannelParticles"]),
|
||
("问题三：信息增益选点与滚动任务排序",
|
||
r"统一任务表由站点扫频、定位测量与清除三类任务组成，"
|
||
r"每执行一项即重排；定位点按期望信息增益加移动代价评分。",
|
||
["BayesLoop"]),
|
||
("问题四：朝向候选、可见性选点与网格遍历清除",
|
||
r"每个幸存格附朝向档掩码构成联合候选集，可见性权重用于探点排序；"
|
||
r"定位停滞时以 $20$ m 方格遍历清除兜底。",
|
||
["JointBelief", "DirectionalExtension"]),
|
||
]
|
||
|
||
|
||
# 只保留承载四问主线的成员；IO、统计报告、缓存、numpy 向量化加速、
|
||
# 细粒度网格遍历等实现细节不出现在附录中。
|
||
KEEP_PRIVATE: dict[str, set[str]] = {
|
||
"ChannelParticles": {"_kill_disk", "_kill_outside_disk", "_kill_outside_wedge",
|
||
"_discard"},
|
||
"BayesLoop": {"_record_measure", "_build_tasks", "_execute_station",
|
||
"_select_point", "_best_certificate", "_localize_and_clear"},
|
||
"DirectionalExtension": {"_localize_and_clear", "_observe", "_next_probe",
|
||
"_fallback_clear", "_attempt_clear"},
|
||
"JointBelief": set(),
|
||
}
|
||
|
||
# 类方法引用的模块级常量与类型别名：附录里一并列出，便于对照源码。
|
||
CONSTANTS = {
|
||
"timing": ["CostConfig", "Vec2"],
|
||
"particles": ["_ANG_EPS_DEG", "_DIST_EPS_M", "_COARSE_FACTOR"],
|
||
"joint": ["HEADING_BINS", "_BIN_HALF_DEG", "_FULL_MASK", "_GRID_DEG", "_GRID_N"],
|
||
}
|
||
|
||
|
||
|
||
# 中文一句话说明：覆盖 notebook 里偏长的英文/工程化 docstring，只用于论文附录。
|
||
NOTES: dict[str, str] = {
|
||
"CostBreakdown": "一次合法动作的三段耗时：移动 + 换频道（仅检测）+ 动作本身",
|
||
"measure_cost": "合法 /measure 的耗时；移动按直线距离，换频道只看前后频道是否不同",
|
||
"clear_cost": "合法 /clear 的耗时；清除不切换测向机频道，故不含换频项",
|
||
"move_time": "任意两点间的移动耗时，供规划使用",
|
||
"nearest_neighbor_order": "最近邻开放路径，作为路径排序的启发式上界",
|
||
"two_opt_open": "开放路径 2-opt 改进；终点自由，只反转内部片段",
|
||
"Measurement": "一次示向度观测：测点坐标、示向度与误差半角（题设为 $1^\\circ$）",
|
||
"halfplanes": "一次观测的三个半平面约束：角域两侧边界与前向约束，法向量已单位化",
|
||
"convex_hull": "单调链凸包（逆时针、无重复末点），输入坐标先平移缩放到同一量级",
|
||
"diameter": "顶点对枚举区域直径；有界凸集的最远点必为顶点",
|
||
"smallest_enclosing_circle": "Welzl 随机增量最小包围圆；固定随机种子保证结果可复现",
|
||
"solve": "问题一主入口：半平面交求定位区域，区分空集、无界与退化情形，"
|
||
"再给出直径、直径圆覆盖判定与最小包围圆",
|
||
"ClearCertificate": "清除充分条件：候选集外包半径与近似误差之和不超过清除半径",
|
||
"clear_certificate": "由候选集最小包围圆判定是否可以就地清除",
|
||
"candidate_points": "保证接收盘内的候选点：盘心加各比例半径上的均匀方向点",
|
||
"safe_disk_radius": "保证接收盘半径 $1000-\\rho$；小于零表示没有保证接收的区域",
|
||
"select_measurement_point": "在保证接收盘内选下一检测点：主指标为最坏后验包围半径，"
|
||
"同分取移动代价较小者",
|
||
"select_second_point": "问题二入口：由首次示向度构造候选集并给出第二检测点",
|
||
"CoverageCertificate": "覆盖证书：采样分辨率、位置与朝向目标数、缺口数与最坏命中距离",
|
||
"CoveragePlan": "覆盖计划：点集、覆盖语义与证书",
|
||
"coverage_points": "49 点覆盖网格，单元对角 $700\\sqrt2\\approx989.95\\le1000$ m",
|
||
"CoverageLedger": "逐频道检测记录，按（频道, 网格点）二维记账",
|
||
"seven_points": "原点与环上六个等分点的七站覆盖集",
|
||
"seven_plan": "第三问解析覆盖计划；环半径 $1159.7$ m 时最坏覆盖距离不超过 $984.539$ m",
|
||
"verify_full_disk": "第三问覆盖语义：圆域内每点都落在某个被访点的最小接收圆盘内",
|
||
"verify_forward_half_disk": "第四问覆盖语义：任意位置与朝向都存在前向半盘内的被访点",
|
||
"plan_instance": "按名称取覆盖计划；证书计划在运行期按同一分辨率复核",
|
||
"ChannelParticles": "单个频道的网格候选集：格心裕量 $\\ell/\\sqrt2$，"
|
||
"所有删除判定内缩该裕量，故真源所在格永不被删",
|
||
"BayesLoop": "问题三闭环：网格候选集的正负观测更新、信息增益选点与滚动任务排序",
|
||
"JointBelief": "单频道的位置与朝向联合候选集；朝向分 $24$ 档，"
|
||
"负观测只在整格必落在最小接收半径内时才约束朝向",
|
||
"DirectionalExtension": "问题四闭环：定向覆盖计划、遮挡解释与网格遍历清除兜底",
|
||
}
|
||
|
||
def notebook_sources(path: Path) -> list[str]:
|
||
notebook = json.loads(path.read_text(encoding="utf-8"))
|
||
return ["".join(cell["source"]) for cell in notebook["cells"]
|
||
if cell["cell_type"] == "code"]
|
||
|
||
|
||
def index_definitions(sources: list[str]) -> dict[str, str]:
|
||
"""把单元里的模块级定义拆开，得到 名字 -> 源码 的映射。"""
|
||
table: dict[str, str] = {}
|
||
for source in sources:
|
||
tree = ast.parse(source)
|
||
lines = source.splitlines()
|
||
for node in tree.body:
|
||
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef,
|
||
ast.ClassDef)):
|
||
continue
|
||
node.end_lineno = getattr(node, "end_lineno", None) or node.lineno
|
||
body = "\n".join(lines[node.lineno - 1:node.end_lineno])
|
||
table[node.name] = body.rstrip() + "\n"
|
||
return table
|
||
|
||
|
||
def _indent_of(line: str) -> int:
|
||
return len(line) - len(line.lstrip())
|
||
|
||
|
||
def extract_member(block: str, member: str) -> str:
|
||
"""从类源码里取出单个方法的源码（含其装饰器行）。"""
|
||
lines = block.splitlines()
|
||
tree = ast.parse(block)
|
||
for node in tree.body[0].body: # type: ignore[attr-defined]
|
||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) \
|
||
and node.name == member:
|
||
start = node.lineno - 1
|
||
decorators = getattr(node, "decorator_list", [])
|
||
if decorators:
|
||
start = min(start, min(item.lineno for item in decorators) - 1)
|
||
end = getattr(node, "end_lineno", None) or node.lineno
|
||
return "\n".join(lines[start:end]).rstrip() + "\n"
|
||
# 类体里没有：退化为"类.方法"全名查找（如嵌套类或模块级同名函数）
|
||
for node in tree.body:
|
||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) \
|
||
and node.name == member:
|
||
end = getattr(node, "end_lineno", None) or node.lineno
|
||
return "\n".join(lines[node.lineno - 1:end]).rstrip() + "\n"
|
||
raise KeyError(f"类中没有成员 {member!r}")
|
||
|
||
|
||
def compress_docstring(source: str, limit: int = 58) -> str:
|
||
"""把函数/类的 docstring 压成一行（首句），避免附录被历史记录淹没。"""
|
||
lines = source.splitlines()
|
||
tree = ast.parse(source)
|
||
node = tree.body[0]
|
||
doc = ast.get_docstring(node)
|
||
if not doc:
|
||
return source
|
||
first = re.sub(r"\s+", " ", doc.strip().splitlines()[0]).strip()
|
||
if len(first) > limit:
|
||
first = first[:limit].rstrip() + "…"
|
||
start = node.body[0].lineno - 1 # docstring 所在行
|
||
end = start
|
||
if lines[start].lstrip().startswith(('"""', "r, ")):
|
||
quote = lines[start].lstrip()[:3]
|
||
if not (lines[start].strip().endswith(quote) and len(lines[start].strip()) > 3):
|
||
while end < len(lines) and not lines[end].rstrip().endswith(quote):
|
||
end += 1
|
||
indent = " " * _indent_of(lines[start])
|
||
return "\n".join(lines[:start] + [indent + '"""' + first + '"""']
|
||
+ lines[end + 1:])
|
||
|
||
|
||
def note_for(name: str, block: str) -> str:
|
||
"""论文附录用的中文一句话说明：优先查 NOTES，否则取 docstring 首句。"""
|
||
if name in NOTES:
|
||
return NOTES[name]
|
||
return shorten_docstring(block)
|
||
|
||
|
||
def shorten_docstring(block: str, limit: int = 46) -> str:
|
||
"""取 docstring 的首句作为一句话说明（超长则截断）。"""
|
||
tree = ast.parse(block)
|
||
node = tree.body[0]
|
||
doc = ast.get_docstring(node) or ""
|
||
first = doc.strip().splitlines()[0].strip() if doc.strip() else ""
|
||
first = re.sub(r"\s+", " ", first).rstrip("。")
|
||
if len(first) > limit:
|
||
first = first[:limit].rstrip() + "…"
|
||
return first
|
||
|
||
|
||
INLINE_ESCAPES = {"\\": r"\textbackslash{}", "{": r"\{", "}": r"\}",
|
||
"$": r"\$", "&": r"\&", "%": r"\%", "#": r"\#",
|
||
"^": r"\^{}", "~": r"\~{}"}
|
||
|
||
|
||
def latex_escape_text(text: str) -> str:
|
||
"""行内说明的转义；下划线在行内模式必须转义。"""
|
||
out = [INLINE_ESCAPES.get(ch, ch) for ch in text]
|
||
return "".join(out).replace("_", r"\_")
|
||
|
||
|
||
|
||
def class_skeleton(block: str, keep_private: set[str] | None = None) -> str:
|
||
"""类的精简版：保留类声明、类字段与公开成员（``keep_private`` 内的私有成员也保留）。"""
|
||
keep_private = keep_private or set()
|
||
lines = block.splitlines()
|
||
tree = ast.parse(block)
|
||
cls = tree.body[0]
|
||
head_end = cls.body[0].lineno - 1 if cls.body else cls.lineno
|
||
head = lines[:head_end]
|
||
doc = ast.get_docstring(cls)
|
||
if doc:
|
||
head += [' """' + doc.strip().splitlines()[0] + '"""']
|
||
head.append("")
|
||
out = list(head)
|
||
for node in cls.body:
|
||
if isinstance(node, (ast.Assign, ast.AnnAssign)): # 数据类字段与类常量
|
||
end = getattr(node, "end_lineno", None) or node.lineno
|
||
out += lines[node.lineno - 1:end]
|
||
continue
|
||
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||
continue
|
||
public = not node.name.startswith("_") or node.name in keep_private
|
||
if not public:
|
||
continue
|
||
start = node.lineno - 1
|
||
decorators = getattr(node, "decorator_list", [])
|
||
if decorators:
|
||
start = min(start, min(item.lineno for item in decorators) - 1)
|
||
end = getattr(node, "end_lineno", None) or node.lineno
|
||
out += lines[start:end] + [""]
|
||
while out and not out[-1].strip():
|
||
out.pop()
|
||
return "\n".join(out) + "\n"
|
||
|
||
|
||
def collect() -> list[tuple[str, str, list[tuple[str, str]]]]:
|
||
"""返回 [(标题, 说明, [(函数名, 源码)])]，类按公开成员精简。"""
|
||
table = index_definitions(notebook_sources(NOTEBOOK))
|
||
sections = []
|
||
for title, note, names in SECTIONS:
|
||
blocks: list[tuple[str, str]] = []
|
||
for name in names:
|
||
if "." in name:
|
||
owner, member = name.split(".", 1)
|
||
blocks.append((name, extract_member(table[owner], member)))
|
||
elif name in table:
|
||
block = table[name]
|
||
skeleton = (class_skeleton(block, KEEP_PRIVATE.get(name))
|
||
if block.startswith("class ") else block)
|
||
blocks.append((name, compress_docstring(skeleton)))
|
||
else:
|
||
raise KeyError(f"notebook 中没有定义 {name!r}")
|
||
sections.append((title, note, blocks))
|
||
return sections
|
||
|
||
|
||
def render(sections) -> str:
|
||
r"""渲染成可直接 \input 的 LaTeX 片段（样式 corecode 在 main.tex 导言区定义）。"""
|
||
lines = [
|
||
r"% Generated by scripts/build_code_appendix.py from mcm26b-核心代码.ipynb.",
|
||
r"% Do not edit by hand: rerun the generator after the notebook changes.",
|
||
r"\begingroup\linespread{1.0}\selectfont",
|
||
]
|
||
for title, note, blocks in sections:
|
||
lines += [rf"\subsection*{{{title}}}", note, ""]
|
||
for name, source in blocks:
|
||
summary = note_for(name, source)
|
||
label = rf"\texttt{{{latex_escape_text(name)}}}"
|
||
caption = f"{label}：{summary}。" if summary else f"{label}。"
|
||
lines += [rf"\noindent\small {caption}",
|
||
r"\begin{lstlisting}[style=corecode]",
|
||
source.rstrip(), r"\end{lstlisting}", ""]
|
||
lines.append(r"\endgroup")
|
||
return "\n".join(lines).rstrip() + "\n"
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("--check", action="store_true",
|
||
help="只统计规模，不写文件")
|
||
args = parser.parse_args()
|
||
sections = collect()
|
||
text = render(sections)
|
||
code_lines = sum(len(source.splitlines())
|
||
for _, _, blocks in sections for _, source in blocks)
|
||
print(f"核心函数：{sum(len(b) for _, _, b in sections)} 个，"
|
||
f"代码 {code_lines} 行，LaTeX {len(text.splitlines())} 行")
|
||
for title, _, blocks in sections:
|
||
print(f" {title}: " + "、".join(name for name, _ in blocks))
|
||
if args.check:
|
||
return
|
||
target = PAPER / "data/source-listings.tex"
|
||
target.write_text(text, encoding="utf-8")
|
||
print(f"已写出 {target.relative_to(ROOT)}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|