baseline
This commit is contained in:
Horldsence
@@ -0,0 +1,14 @@
|
||||
"""问题二：第二检测点选择策略。
|
||||
|
||||
- ``optimal.py`` ：生产选点器 ``mcm26b.planner.select_measurement_point``
|
||||
（保证接收盘内最坏后验半径最小化，采样启发式）；
|
||||
- ``baseline.py`` ："垂直 1000 m 最大交会角"朴素启发式（忽略接收保证），
|
||||
以及"退到候选集 MEC 圆心"的保守基线；
|
||||
- ``reference.py``：在保证接收盘内稠密网格枚举的近似最优参考；
|
||||
- ``cases.py`` ：随机首次示向度算例；
|
||||
- ``compare.py`` ：同案配对对比并落盘。
|
||||
"""
|
||||
|
||||
from . import baseline, cases, optimal, reference
|
||||
|
||||
__all__ = ["optimal", "baseline", "reference", "cases"]
|
||||
@@ -0,0 +1,71 @@
|
||||
"""问题二基线实现：两个朴素但常见的对照。
|
||||
|
||||
- ``candidate_perp``：沿首次示向度垂直方向、距检测点约 1000 m 取点。
|
||||
这是"尽量把交会角做成 90°"的教科书启发式，**但它忽略接收距离保证**：
|
||||
真源可能落在首次角域的远端（≤1500 m），此时第二点距离可超过 1000 m，
|
||||
第二次测量直接 ``no_signal``，几何收益落空。
|
||||
- ``candidate_mec``：退到候选集最小包围圆圆心。它确实落在保证接收盘内
|
||||
（ρ≤1000 时），但交会角可能接近 0°、后验几乎不收缩，代表"最保守"一端。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
|
||||
from mcm26b.config import ProblemConfig
|
||||
from mcm26b.planner import safe_disk_radius
|
||||
|
||||
from project.common import Method, problem_config
|
||||
from project.q2.region import first_bearing_region
|
||||
|
||||
PERP = Method(
|
||||
key="baseline_perp",
|
||||
name="垂直 1000 m 最大交会角",
|
||||
kind="baseline",
|
||||
note="忽略接收保证的几何启发式",
|
||||
)
|
||||
MEC = Method(
|
||||
key="baseline_mec",
|
||||
name="候选集 MEC 圆心",
|
||||
kind="baseline",
|
||||
note="保证接收但交会角最差的保守基线",
|
||||
)
|
||||
METHODS = (PERP, MEC)
|
||||
|
||||
|
||||
def _region(case: dict, config: ProblemConfig):
|
||||
return first_bearing_region(case, config)
|
||||
|
||||
|
||||
def candidate_perp(case: dict, config: ProblemConfig | None = None) -> dict:
|
||||
config = config or problem_config()
|
||||
region = _region(case, config)
|
||||
station = (case["station"][0], case["station"][1])
|
||||
theta = math.radians(case["bearing_deg"] + 90.0)
|
||||
distance = config.field.r_eff_min_m
|
||||
position = (station[0] + distance * math.cos(theta),
|
||||
station[1] + distance * math.sin(theta))
|
||||
return {
|
||||
"method": PERP.key,
|
||||
"position": [round(position[0], 4), round(position[1], 4)],
|
||||
"worst_posterior_radius_m": None, # 由 compare 统一评估
|
||||
"move_cost_s": None,
|
||||
"safe_disk_radius_m": safe_disk_radius(region, config),
|
||||
"initial_radius_m": region.mec()[1],
|
||||
"reason": "沿首次示向度法向 1000 m 取 90° 交会点",
|
||||
}
|
||||
|
||||
|
||||
def candidate_mec(case: dict, config: ProblemConfig | None = None) -> dict:
|
||||
config = config or problem_config()
|
||||
region = _region(case, config)
|
||||
center, radius = region.mec()
|
||||
return {
|
||||
"method": MEC.key,
|
||||
"position": [round(center[0], 4), round(center[1], 4)],
|
||||
"worst_posterior_radius_m": None,
|
||||
"move_cost_s": None,
|
||||
"safe_disk_radius_m": safe_disk_radius(region, config),
|
||||
"initial_radius_m": radius,
|
||||
"reason": "取候选集最小包围圆圆心，保证接收但交会角可能最差",
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
"""问题二算例：随机首次示向度（全向源）。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
import random
|
||||
|
||||
from mcm26b.config import ProblemConfig
|
||||
|
||||
|
||||
def random_case(seed: int, config: ProblemConfig | None = None) -> dict:
|
||||
"""随机检测点 s 与示向度 θ：默认让真源与读数一致（误差 ±1° 内）。"""
|
||||
config = config or ProblemConfig()
|
||||
rng = random.Random(seed)
|
||||
radius = config.field.omega_radius_m
|
||||
# 检测点可在目标域内或域外；这里取域内，符合题目常规场景
|
||||
angle_s = rng.uniform(0, 2 * math.pi)
|
||||
distance_s = radius * math.sqrt(rng.random()) * 0.9
|
||||
station = (distance_s * math.cos(angle_s), distance_s * math.sin(angle_s))
|
||||
# 真源与检测点距离落在 [60, 1500) 且仍在 Ω 内
|
||||
while True:
|
||||
bearing_true = rng.uniform(0, 2 * math.pi)
|
||||
source_distance = rng.uniform(60, config.field.r_eff_max_m - 1)
|
||||
target = (station[0] + source_distance * math.cos(bearing_true),
|
||||
station[1] + source_distance * math.sin(bearing_true))
|
||||
if math.hypot(*target) <= radius:
|
||||
break
|
||||
error = math.radians(rng.uniform(-0.95, 0.95))
|
||||
bearing_deg = math.degrees(bearing_true + error) % 360.0
|
||||
return {"id": f"q2-{seed}", "station": [round(station[0], 2), round(station[1], 2)],
|
||||
"bearing_deg": round(bearing_deg, 2), "half_angle_deg": 1.0,
|
||||
"true_source": [round(target[0], 2), round(target[1], 2)]}
|
||||
|
||||
|
||||
def random_cases(count: int, base_seed: int = 4200,
|
||||
config: ProblemConfig | None = None) -> list[dict]:
|
||||
return [random_case(base_seed + index, config) for index in range(count)]
|
||||
@@ -0,0 +1,136 @@
|
||||
"""问题二同案配对对比：最优 / 两个基线 / 网格参考，用同一评估器打分。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
|
||||
from mcm26b.config import ProblemConfig
|
||||
from mcm26b.planner import _worst_posterior_radius
|
||||
|
||||
from project.common import (Method, RESULTS_DIR, markdown_table, mean, ratio,
|
||||
write_json, write_text)
|
||||
from project.q2 import baseline, cases, optimal, reference
|
||||
from project.q2.region import first_bearing_region
|
||||
|
||||
METHODS: tuple[Method, ...] = (optimal.METHOD, baseline.PERP, baseline.MEC,
|
||||
reference.METHOD)
|
||||
EVAL_WITNESS_SPACING_M = 60.0
|
||||
|
||||
|
||||
def _evaluate(case: dict, config: ProblemConfig) -> list[dict]:
|
||||
region = first_bearing_region(case, config)
|
||||
station = (case["station"][0], case["station"][1])
|
||||
initial_radius = region.mec()[1]
|
||||
half_angle = (config.field.bearing_error_half_deg
|
||||
+ config.search.bearing_rounding_margin_deg)
|
||||
witnesses = region.witnesses(EVAL_WITNESS_SPACING_M)
|
||||
|
||||
plans = [optimal.candidate(case, config),
|
||||
baseline.candidate_perp(case, config),
|
||||
baseline.candidate_mec(case, config),
|
||||
reference.candidate(case, config)]
|
||||
rows = []
|
||||
for plan in plans:
|
||||
q = (plan["position"][0], plan["position"][1])
|
||||
guarantee_distance = max(math.dist(q, vertex) for vertex in region.polygon)
|
||||
worst = _worst_posterior_radius(
|
||||
region, q, witnesses, half_angle,
|
||||
config.field.near_radius_m, config.search.polygon_sides)
|
||||
move_cost = math.dist(station, q) / config.cost.speed_mps
|
||||
rows.append({
|
||||
"case_id": case["id"],
|
||||
"method": plan["method"],
|
||||
"initial_radius_m": round(initial_radius, 4),
|
||||
"q": [round(q[0], 2), round(q[1], 2)],
|
||||
"guaranteed_reception": bool(guarantee_distance
|
||||
<= config.field.r_eff_min_m + 1e-6),
|
||||
"guarantee_distance_m": round(guarantee_distance, 3),
|
||||
"worst_posterior_radius_m": round(worst, 4),
|
||||
"improvement_m": round(initial_radius - worst, 4),
|
||||
"final_ratio": round(worst / initial_radius, 4) if initial_radius > 0 else None,
|
||||
"move_cost_s": round(move_cost, 4),
|
||||
"reason": plan["reason"],
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
def _aggregate(rows: list[dict], labels: list[str]) -> dict:
|
||||
summary: dict[str, dict] = {}
|
||||
initial = mean(r["initial_radius_m"] for r in rows) if rows else 0.0
|
||||
for label in labels:
|
||||
part = [r for r in rows if r["method"] == label]
|
||||
if not part:
|
||||
continue
|
||||
summary[label] = {
|
||||
"cases": len(part),
|
||||
"guaranteed_rate": ratio(sum(r["guaranteed_reception"] for r in part),
|
||||
len(part)),
|
||||
"worst_posterior_mean_m": mean(r["worst_posterior_radius_m"] for r in part),
|
||||
"worst_posterior_median_m": sorted(
|
||||
r["worst_posterior_radius_m"] for r in part)[len(part) // 2],
|
||||
"improvement_mean_m": mean(r["improvement_m"] for r in part),
|
||||
"final_ratio_mean": mean(r["final_ratio"] for r in part
|
||||
if r["final_ratio"] is not None),
|
||||
"move_cost_mean_s": mean(r["move_cost_s"] for r in part),
|
||||
}
|
||||
summary["_initial_radius_mean_m"] = initial
|
||||
return summary
|
||||
|
||||
|
||||
def _render(rows: list[dict], aggregate: dict, labels: list[str]) -> str:
|
||||
headers = ["算例", "方法", "初始 ρ(m)", "第二点 q", "保证接收",
|
||||
"q 到 K₁ 最远(m)", "最坏后验 ρ'(m)", "收缩(m)", "移动(s)"]
|
||||
body = []
|
||||
order = {label: index for index, label in enumerate(labels)}
|
||||
for row in sorted(rows, key=lambda r: (r["case_id"], order.get(r["method"], 99))):
|
||||
body.append([
|
||||
row["case_id"], row["method"], _f(row["initial_radius_m"]),
|
||||
f"({row['q'][0]:.0f},{row['q'][1]:.0f})",
|
||||
"是" if row["guaranteed_reception"] else "否",
|
||||
_f(row["guarantee_distance_m"]), _f(row["worst_posterior_radius_m"]),
|
||||
_f(row["improvement_m"]), _f(row["move_cost_s"]),
|
||||
])
|
||||
text = ["# 问题二：第二检测点选择对比", "",
|
||||
"初始 ρ 为首次示向度候选集 K₁ 的最小包围半径；最坏后验按"
|
||||
"目标边界采样 × 读数区间端点统一评估。", "",
|
||||
markdown_table(headers, body), "", "## 逐方法汇总", ""]
|
||||
agg_headers = ["方法", "保证接收率", "最坏后验均值(m)", "中位(m)",
|
||||
"平均收缩(m)", "终值比 ρ'/ρ", "移动均值(s)"]
|
||||
agg_rows = []
|
||||
for label in labels:
|
||||
s = aggregate.get(label)
|
||||
if not s:
|
||||
continue
|
||||
agg_rows.append([label, f"{s['guaranteed_rate']*100:.1f}%",
|
||||
_f(s["worst_posterior_mean_m"]),
|
||||
_f(s["worst_posterior_median_m"]),
|
||||
_f(s["improvement_mean_m"]),
|
||||
_f(s["final_ratio_mean"], 3),
|
||||
_f(s["move_cost_mean_s"])])
|
||||
text.append(markdown_table(agg_headers, agg_rows))
|
||||
text.append("")
|
||||
text.append(f"初始 ρ 均值：{aggregate['_initial_radius_mean_m']:.2f} m")
|
||||
return "\n".join(text) + "\n"
|
||||
|
||||
|
||||
def _f(value, digits: int = 2) -> str:
|
||||
if value is None:
|
||||
return "—"
|
||||
if isinstance(value, bool):
|
||||
return "是" if value else "否"
|
||||
if isinstance(value, float):
|
||||
return f"{value:.{digits}f}"
|
||||
return str(value)
|
||||
|
||||
|
||||
def run(quick: bool = False) -> dict:
|
||||
config = ProblemConfig()
|
||||
case_list = cases.random_cases(8 if quick else 30, config=config)
|
||||
rows = [row for case in case_list for row in _evaluate(case, config)]
|
||||
labels = [m.key for m in METHODS]
|
||||
aggregate = _aggregate(rows, labels)
|
||||
payload = {"question": 2, "methods": [m.__dict__ for m in METHODS],
|
||||
"cases": case_list, "rows": rows, "aggregate": aggregate}
|
||||
write_json(RESULTS_DIR / "q2.json", payload)
|
||||
write_text(RESULTS_DIR / "q2.md", _render(rows, aggregate, labels))
|
||||
return payload
|
||||
@@ -0,0 +1,42 @@
|
||||
"""问题二最优实现：保证接收盘内的最坏后验半径最小化选点（生产实现）。
|
||||
|
||||
复用 ``mcm26b.planner.select_measurement_point``：
|
||||
|
||||
1. 在"保证接收盘" B(mec_center, 1000-ρ) 内环形采样候选点；
|
||||
2. 对每个候选，用"目标边界点 × 读数区间端点（+near 分支）"估计
|
||||
最坏后验最小包围半径，并以移动时间同分决胜；
|
||||
3. 选定后按 1/4 间距加密复核，返回位置、后验半径与理由。
|
||||
|
||||
采样是启发式（不是连续问题全局最优证明），但保证接收是严格的。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from mcm26b.config import ProblemConfig
|
||||
from mcm26b.planner import safe_disk_radius, select_measurement_point
|
||||
|
||||
from project.common import Method, problem_config
|
||||
from project.q2.region import first_bearing_region
|
||||
|
||||
METHOD = Method(
|
||||
key="optimal",
|
||||
name="保证接收盘内最小最坏后验半径",
|
||||
kind="optimal",
|
||||
note="生产实现 mcm26b.planner.select_measurement_point",
|
||||
)
|
||||
|
||||
|
||||
def candidate(case: dict, config: ProblemConfig | None = None) -> dict:
|
||||
config = config or problem_config()
|
||||
region = first_bearing_region(case, config)
|
||||
station = (case["station"][0], case["station"][1])
|
||||
plan = select_measurement_point(region, station, config)
|
||||
return {
|
||||
"method": METHOD.key,
|
||||
"position": [round(plan.position[0], 4), round(plan.position[1], 4)],
|
||||
"worst_posterior_radius_m": plan.worst_posterior_radius_m,
|
||||
"move_cost_s": plan.move_cost_s,
|
||||
"safe_disk_radius_m": safe_disk_radius(region, config),
|
||||
"initial_radius_m": region.mec()[1],
|
||||
"reason": plan.reason,
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
"""问题二近似最优参考：在保证接收盘内做稠密网格枚举。
|
||||
|
||||
目标函数与生产实现一致（最坏后验包围半径 + 移动时间同分决胜），但候选点
|
||||
取自**均匀网格**而非环形采样族，因此可用来检验生产采样是否接近该目标下的
|
||||
网格最优。注意这是"同一目标下的网格参考"，不是连续问题的严格全局最优。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
|
||||
from mcm26b.config import ProblemConfig
|
||||
from mcm26b.planner import (_worst_posterior_radius, safe_disk_radius)
|
||||
|
||||
from project.common import Method, problem_config
|
||||
from project.q2.region import first_bearing_region
|
||||
|
||||
METHOD = Method(
|
||||
key="reference",
|
||||
name="保证接收盘内网格枚举",
|
||||
kind="reference",
|
||||
note="同一目标下的稠密网格参考，非连续全局最优",
|
||||
)
|
||||
GRID_SIDES = 9
|
||||
WITNESS_SPACING_M = 150.0
|
||||
|
||||
|
||||
def candidate(case: dict, config: ProblemConfig | None = None) -> dict:
|
||||
config = config or problem_config()
|
||||
region = first_bearing_region(case, config)
|
||||
station = (case["station"][0], case["station"][1])
|
||||
center = region.mec()[0]
|
||||
safe = safe_disk_radius(region, config)
|
||||
if safe < 0:
|
||||
return _payload(center, region, config, station, "无保证接收盘：退回 MEC 圆心")
|
||||
|
||||
field, search = config.field, config.search
|
||||
half_angle = field.bearing_error_half_deg + search.bearing_rounding_margin_deg
|
||||
witnesses = region.witnesses(WITNESS_SPACING_M)
|
||||
step = 2.0 * safe / (GRID_SIDES - 1)
|
||||
best_key, best_point, best_worst = None, center, None
|
||||
for i in range(GRID_SIDES):
|
||||
for j in range(GRID_SIDES):
|
||||
point = (center[0] - safe + i * step, center[1] - safe + j * step)
|
||||
if math.dist(point, center) > safe + 1e-9:
|
||||
continue
|
||||
worst = _worst_posterior_radius(region, point, witnesses, half_angle,
|
||||
field.near_radius_m, search.polygon_sides)
|
||||
move = math.dist(station, point) / config.cost.speed_mps
|
||||
key = (round(worst, 6), round(move, 6))
|
||||
if best_key is None or key < best_key:
|
||||
best_key, best_point, best_worst = key, point, worst
|
||||
return _payload(best_point, region, config, station,
|
||||
f"安全盘网格 {GRID_SIDES}×{GRID_SIDES} 取最坏后验最小",
|
||||
worst_value=best_worst)
|
||||
|
||||
|
||||
def _payload(point, region, config, station, reason, worst_value=None) -> dict:
|
||||
return {
|
||||
"method": METHOD.key,
|
||||
"position": [round(point[0], 4), round(point[1], 4)],
|
||||
"worst_posterior_radius_m": worst_value,
|
||||
"move_cost_s": math.dist(station, point) / config.cost.speed_mps,
|
||||
"safe_disk_radius_m": safe_disk_radius(region, config),
|
||||
"initial_radius_m": region.mec()[1],
|
||||
"reason": reason,
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
"""问题二候选集 K1 的统一构造（三个方法共用同一输入，保证配对公平）。
|
||||
|
||||
K1 = Ω ∩ W(s, θ ± δ_eff) ∩ B(s, 1500)，δ_eff = 题设 1° + 示向度舍入裕量。
|
||||
生产实现用"圆盘外切多边形 ∩ 半平面"的保守外包（``CandidateRegion``），
|
||||
圆盘用外切多边形严格外包，保证真源不丢。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from mcm26b.belief import CandidateRegion
|
||||
from mcm26b.config import ProblemConfig, Vec2
|
||||
|
||||
from project.common import problem_config
|
||||
|
||||
|
||||
def first_bearing_region(case: dict,
|
||||
config: ProblemConfig | None = None) -> CandidateRegion:
|
||||
config = config or problem_config()
|
||||
field, search = config.field, config.search
|
||||
half_angle = field.bearing_error_half_deg + search.bearing_rounding_margin_deg
|
||||
station: Vec2 = (case["station"][0], case["station"][1])
|
||||
region = CandidateRegion.initial(station, config)
|
||||
region.clip_wedge(station, case["bearing_deg"], half_angle)
|
||||
return region
|
||||
Reference in New Issue
Block a user