554ebc239a
用户要求：Create 与 Aeronautics 会引入大量原版没有的方块与危险区，
寻路必须一起优化。用户裁定：危险区用「表 + 命名推测并标记」，
动态机械用「警告 + 提高代价」。
## ★ 关键实测结论（本轮最重要）
我先按「给 mod 方块建语义表」做，实测覆盖率：
Create 643 个方块 → 显式表命中 9、命名规则推测 21、完全不识 613
覆盖率 4.7%
⇒ **手写表是死路**（64 mod / 1465 方块永远追不上）。
⇒ 但寻路真正需要的属性，**运行时全都有**：
Java 侧 blockinfo 已在读 BuiltInRegistries.BLOCK（**含 mod 运行时注册的方块**），
并返回 isSolid / destroySpeed / lightEmission / isAir / hasBlockEntity。
⇒ 正确做法是**问**，不是猜。
## 改动
新增 mcfold/solid.py：
- SolidSource 抽象：RegistrySolidSource（本机，快但无 mod）
vs RuntimeSolidSource（问 JVM，**含 mod**，准但要 JVM）
- 按 **state_id** 缓存（属性只依赖 state，不依赖坐标）
- attach_runtime() 批量探测（一次问一格要一次往返，批量才可用）
mcfold/pathfind.py：Terrain 改为「**先运行时缓存，再本机注册表**」；
can_dig 用 destroySpeed 判可挖（基岩官方约定是 -1 = 不可破坏），
不再靠名字猜「基岩挖不动」
新增 mcfold/modblocks.py（危险/会动这部分的表，配套覆盖率报告）：
- 语义取���：ground/passable/hazard/breakable/solid/dynamic/carrier/unknown
- 三级来源：显式表 → 命名规则（**标记 inferred**）→ **如实说不认识**
- ★ vanilla 与 mod **分开算覆盖率**：vanilla 语义由注册表给，
把它算成「未知」会让人误以为「连石头都认不出」
WorldState：加 solid_cache 字段
## 判据
test_pathfind 19 → 24 项，新增 5 项：
- mod 方块优先用运行时 isSolid（本机注册表确实认不出）
- 不可破坏吃 destroySpeed（含基岩 -1 约定）
- 两边都无数据 ⇒ 不知道（**不谎称可走**）
- 覆盖率报告如实（且区分 mod/vanilla）
- 语义查询永远返回带 inferred 标记的结构
★ 写判据时踩了两个坑（都是 fixture 形状错，不是代码错）：
1. `Chunk.sections[i]` 对应 y = MIN_Y + 16*i，MIN_Y=-64
⇒ 我按「第 0 个 section = y=0」写，查 y=0 拿到的是空气
2. 同一 section 内 16 格**只能一个值** ⇒ mod 方块与空气落同一 section 时，
空气格会变成 mod 方块（看着像代码 bug，其实是 fixture 形状问题）
⇒ 教训：**判据的 fixture 形状必须与被测代码的真实输入结构一致**。
226 项全通过。
623 lines
27 KiB
Python
623 lines
27 KiB
Python
"""寻路：两个**独立**的工具（用户裁定：可达性判断 与 完整版寻路 分开）
|
||
|
||
## 为什么要分成两个
|
||
|
||
| 工具 | 回答 | 成本 |
|
||
|---|---|---|
|
||
| `reachable(x,y,z)` | **到得了吗 / 大概多远** | 便宜（可以只做可达性、不返回具体路线） |
|
||
| `find_path(x,y,z)` | **具体怎么走**（含每步类型、要不要挖/放） | 贵（要展开搜索、回溯、排序） |
|
||
|
||
agent 很多时候只想知道「那边能过去吗」—— 那时不必付完整寻路的代价，
|
||
也不必把一长串路线塞进上下文。反过来真要走过去时才需要详细路线。
|
||
|
||
## 移动词表（对齐 mineflayer 的 movements.js）
|
||
|
||
| 动作 | 含义 | 需要改世界？ |
|
||
|---|---|---|
|
||
| `forward` | 平地走一格 | 否 |
|
||
| `step_up` | 上一格台阶（脚下 +1） | 否 |
|
||
| `step_down` | 下一格 | 否 |
|
||
| `drop` | 掉落到更低处 | 否 |
|
||
| `jump` | 跳上 1 格（前方挡路时） | 否 |
|
||
| `dig` | 挖开挡路的方块 | **是** |
|
||
| `place` | 搭桥 / 垫脚 | **是** |
|
||
|
||
## ★ 三条硬纪律
|
||
|
||
1. **未加载 = 走到边界 + 如实报告**，绝不乐观推断
|
||
（把「没见过」当「空旷」是这类工具最危险的错）。
|
||
2. **危险方块不能踩**：`lava`/`fire` 的 `solid=False`（能穿过去），
|
||
但碰了会死 ⇒ 判定「可站立」不能只看 solid。
|
||
3. **碰撞粒度有限且如实告知**：`blocks.json` 的 `boundingBox` 只有
|
||
`block`/`empty` 两档，半砖与楼梯都会被当整块 ⇒ 可能多绕路，但不会撞墙。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import heapq
|
||
import math
|
||
from typing import Any, Iterable
|
||
|
||
|
||
# ---------- 危险方块（solid=False 但致命） ----------
|
||
|
||
#: 这些方块 `solid=False`（碰撞上能穿过去）但**碰到就有害**，
|
||
#: 所以「可站立」判定必须单独排除它们 —— 只看 solid 会把它们当地面。
|
||
#: 数据来源：实测 blocks.json（1.21.1）
|
||
HAZARD_NAMES = frozenset({
|
||
"lava", "fire", "soul_fire", "nether_portal", "end_portal", "magma_block",
|
||
"cactus", "sweet_berry_bush", "wither_rose", "powder_snow", "campfire",
|
||
"soul_campfire", "pointed_dripstone", "scaffolding", "end_portal_frame",
|
||
"tnt", "powder_snow_block", "structure_void", "end_gateway",
|
||
})
|
||
|
||
#: 液体：能站进去但会减速/溺水，不作为「地面」也不作为「可站」目标
|
||
LIQUID_NAMES = frozenset({"water", "flowing_water", "bubble_column"})
|
||
|
||
#: 能当「地面」承重的方块（挖/放时的目标）
|
||
DEFAULT_HAZARD: str = "unknown" # 认不出的方块按「可能有危害」处理更安全？→ 见下
|
||
# 注：**不能**默认当危险。认不出的方块多半是 mod 方块，
|
||
# 把所有未知方块都当致命会让寻路在 mod 环境里完全瘫痪。
|
||
# 折中：未知方块按「可站立但记 unknown」处理，并在结果里提示。
|
||
|
||
# ---------- 动作定义 ----------
|
||
|
||
|
||
class Move:
|
||
"""一步移动的完整描述（agent 照着走，或由 follow 执行）。"""
|
||
|
||
__slots__ = ("action", "dx", "dy", "dz", "needs_world_change", "detail",
|
||
"dig_count")
|
||
|
||
def __init__(self, action: str, dx: int, dy: int, dz: int,
|
||
needs_world_change: bool = False, detail: str = "",
|
||
dig_count: int = 1):
|
||
self.action = action # forward/step_up/step_down/drop/jump/dig/place
|
||
self.dx, self.dy, self.dz = dx, dy, dz
|
||
self.needs_world_change = needs_world_change
|
||
self.detail = detail
|
||
self.dig_count = dig_count # 挖穿一列要几格
|
||
|
||
def as_dict(self) -> dict[str, Any]:
|
||
return {
|
||
"action": self.action,
|
||
"d": [self.dx, self.dy, self.dz],
|
||
"world_change": self.needs_world_change,
|
||
**({"detail": self.detail} if self.detail else {}),
|
||
**({"dig_count": self.dig_count} if self.action == "dig" else {}),
|
||
}
|
||
|
||
def __repr__(self) -> str:
|
||
return (f"{self.action}({self.dx:+d},{self.dy:+d},{self.dz:+d})")
|
||
|
||
|
||
#: 动作的相对代价。
|
||
#:
|
||
#: ★ 调这两个数字时踩过的坑（实测）：
|
||
#: 1 格台阶（石墙）与目标同高时，`step_up`(1.2) 和 `dig`(6.0) 同时可用，
|
||
#: 但因为 step_up 会让 y 变成 1、而目标 y=0，路径总代价算下来
|
||
#: 「挖 6 格平路」<「上 1 级再下 1 级」⇒ A* 选了荒谬的挖路。
|
||
#: 真实原因是我给**高度差**的权重（1.2/格）低于 dig 的常数 6.0，
|
||
#: 于是一级台阶的「绕」比「挖」便宜。
|
||
#: 修法：把 dig/place 的代价提到明显高于「多绕 1~2 级台阶」的水平
|
||
#: （挖要真花时间、可能失败、还会改世界），这样先绕后挖。
|
||
COSTS = {
|
||
"forward": 1.0,
|
||
"step_up": 1.5, # 上一级台阶
|
||
"step_down": 1.2, # 下一级台阶
|
||
"drop": 0.7, # 掉落最快
|
||
"jump": 2.2, # 跳最慢
|
||
"dig": 12.0, # 挖方块：慢 + 可能失败 + 改世界
|
||
"place": 8.0, # 放方块：也要时间 + 改世界
|
||
}
|
||
#: 高度差权重（用于 A* 启发式与代价）：高出/低于目标一格的「心理代价」
|
||
HEIGHT_PENALTY = 2.0
|
||
#: 挖穿一列时最多挖几格（再高就当"挖不动"⇒ 建议绕路或搭桥）
|
||
MAX_DIG_DEPTH = 5
|
||
|
||
|
||
# ---------- 地形判定（只读，不改） ----------
|
||
|
||
class Terrain:
|
||
"""把 WorldState 包成寻路要的三个问题（带安全默认）。
|
||
|
||
★ 三个问题都必须能回答「不知道」：
|
||
- 能不能站（solid + 非危险）
|
||
- 能不能穿过（non-solid）
|
||
- 能不能承载（solid 且非危险，用作 dig/place 的目标）
|
||
"""
|
||
|
||
def __init__(self, world, allow_world_change: bool = True):
|
||
self.w = world
|
||
self.allow_world_change = allow_world_change
|
||
self._hazards: set[int] = set()
|
||
self._liquids: set[int] = set()
|
||
self._partial: dict[int, str | None] = {}
|
||
reg = world.world.registry
|
||
if reg is not None:
|
||
for sid, name in reg.by_state.items():
|
||
if name in HAZARD_NAMES:
|
||
self._hazards.add(sid)
|
||
elif name in LIQUID_NAMES:
|
||
self._liquids.add(sid)
|
||
|
||
def _reg(self):
|
||
return self.w.world.registry
|
||
|
||
def at(self, x: int, y: int, z: int) -> dict[str, Any]:
|
||
"""该格的方块信息（沿用状态层的四态：registry/unknown-state/unloaded/…）。"""
|
||
return self.w.block_at(x, y, z)
|
||
|
||
def known(self, x: int, y: int, z: int) -> bool:
|
||
return self.at(x, y, z)["source"] in ("registry", "unknown-state")
|
||
|
||
def is_ground(self, x: int, y: int, z: int) -> bool | None:
|
||
"""这一格能否**作为承重的实心方块**。None = 不知道。
|
||
|
||
★★ 这是我把两个谓词混成一个之后才分出来的（真 bug）：
|
||
is_ground —— 「这一格本身是实心底」
|
||
can_stand —— 「玩家的**脚**能停在这一格」
|
||
两者差一格。混用会让邻居生成永远找不到落脚点
|
||
（表现为「平地上也走不了，只能搭桥」）。
|
||
"""
|
||
h = self.at(x, y, z)
|
||
if h["source"] not in ("registry", "unknown-state"):
|
||
return None
|
||
if h["state_id"] in self._hazards or h["state_id"] in self._liquids:
|
||
return False # 危险/液体：不能承重
|
||
if h["name"] == "air" or h["name"] in ("cave_air", "void_air"):
|
||
return False # 空气不是地面
|
||
# ★★ 先问**运行时缓存**（含 mod 方块），再退回本机注册表。
|
||
# 实测依据：本机 blocks.json 只有 1060 个 vanilla 方块，
|
||
# Create 的 643 个里 613 个认不出（覆盖率 4.7%）
|
||
# ⇒ 只能用运行中的官方注册表（M8 的 blockinfo 已在读它）。
|
||
sid = h["state_id"]
|
||
cache = getattr(self.w, "solid_cache", None)
|
||
if cache:
|
||
hit = cache.get(sid)
|
||
if hit is not None:
|
||
v = hit.get("isSolid")
|
||
return None if v is None else bool(v)
|
||
reg = self._reg()
|
||
if reg is None:
|
||
return None
|
||
return reg.is_solid(sid) # None 表示认不出 ⇒ 不知道
|
||
|
||
def is_standable(self, x: int, y: int, z: int) -> bool | None:
|
||
"""玩家的**脚**能否停在 (x,y,z)：下方要实心，本格与头顶要空。
|
||
|
||
★ 这是「脚所在格」，不是「脚下那一格」。
|
||
"""
|
||
below = self.is_ground(x, y - 1, z)
|
||
here = self.is_passable(x, y, z)
|
||
head = self.is_passable(x, y + 1, z)
|
||
if below is None or here is None or head is None:
|
||
return None
|
||
return bool(below and here and head)
|
||
|
||
def is_passable(self, x: int, y: int, z: int) -> bool | None:
|
||
"""身体能否**穿过**这一格。None = 不知道。"""
|
||
h = self.at(x, y, z)
|
||
if h["source"] not in ("registry", "unknown-state"):
|
||
return None
|
||
if h["state_id"] in self._hazards:
|
||
return False
|
||
if h["name"] in ("air", "cave_air", "void_air", "water"):
|
||
return True
|
||
reg = self._reg()
|
||
if reg is None:
|
||
return None
|
||
s = reg.is_solid(h["state_id"])
|
||
return None if s is None else (not s)
|
||
|
||
def supportable(self, x: int, y: int, z: int) -> bool | None:
|
||
"""这一格能否**承重**（用于 place 搭桥时的支撑位判断）。"""
|
||
return self.is_standable(x, y, z)
|
||
|
||
def standable_if_air(self, x: int, y: int, z: int) -> bool:
|
||
"""假设 (x,y,z) 被挖空后，玩家的**脚**能否停在这里。
|
||
|
||
★ 为什么需要这个：2 格高的墙要挖**上方**那格才通得了
|
||
（挖脚下方那格后，头还是顶着石头）。实测抓到：
|
||
只考虑挖脚部位置 ⇒ 2 格墙永远走不通。
|
||
"""
|
||
below = self.is_ground(x, y - 1, z)
|
||
if below is not True:
|
||
return False
|
||
head = self.is_passable(x, y + 1, z)
|
||
return head is True
|
||
|
||
def can_dig(self, x: int, y: int, z: int) -> bool:
|
||
"""能否挖开这一格（需要世界改动的移动才用得到）。
|
||
|
||
★ 判据用**破坏速度**（运行时注册表给的 destroySpeed），
|
||
不靠名字猜「基岩挖不动」：
|
||
mod 方块里可能有各种不可破坏/极硬的方块，靠猜会规划出挖不动的路。
|
||
基岩 destroySpeed = -1（官方约定：负值 = 不可破坏）。
|
||
"""
|
||
if not self.allow_world_change:
|
||
return False
|
||
h = self.at(x, y, z)
|
||
if h["source"] not in ("registry", "unknown-state"):
|
||
return False
|
||
if h["name"] in ("water", "lava", "void_air", "end_portal"):
|
||
return False # 液体/传送门挖了也没用
|
||
sid = h["state_id"]
|
||
cache = getattr(self.w, "solid_cache", None)
|
||
if cache:
|
||
hit = cache.get(sid)
|
||
if hit is not None and hit.get("destroySpeed") is not None:
|
||
return float(hit["destroySpeed"]) > 0
|
||
reg = self._reg()
|
||
if reg is not None:
|
||
b = reg.by_name.get(h["name"] or "")
|
||
if b is not None and b.get("hardness") is not None:
|
||
return float(b["hardness"]) > 0
|
||
if h["name"] in ("bedrock",):
|
||
return False
|
||
return True
|
||
|
||
|
||
# ---------- 邻居生成（移动词表） ----------
|
||
|
||
#: 水平八向。斜向在 MC 里要额外校验对角，代价也更高 ⇒ 单独处理。
|
||
_DIRS: tuple[tuple[int, int], ...] = (
|
||
(1, 0), (-1, 0), (0, 1), (0, -1),
|
||
(1, 1), (1, -1), (-1, 1), (-1, -1),
|
||
)
|
||
|
||
|
||
def neighbors(t: Terrain, pos: tuple[int, int, int],
|
||
allow_world_change: bool = True) -> Iterable[tuple[tuple[int, int, int], Move]]:
|
||
"""从 pos 出发所有合法的一步移动。
|
||
|
||
依次尝试（**先便宜后贵**，A* 才有意义）：
|
||
`forward` → `step_up` → `step_down` → `drop` → `jump` → `dig` → `place`
|
||
"""
|
||
x, y, z = pos
|
||
for dx, dz in _DIRS:
|
||
nx, nz = x + dx, z + dz
|
||
diag = dx != 0 and dz != 0
|
||
base = COSTS["step_up"] if diag else COSTS["forward"]
|
||
if diag:
|
||
# 斜向必须两条正交方向都通（否则会穿墙角）
|
||
if not (t.is_passable(x + dx, y, z) and t.is_passable(x, y, z + dz)):
|
||
continue
|
||
|
||
# ① 平地走 / 上一格（脚位置 = 目标格）
|
||
# ★ 上一格是 MC 里最常见的移动（上一级台阶），所以**先**给 step_up，
|
||
# 代价也比 dig/jump 低 ⇒ A* 会自然选它。
|
||
for up in (0, 1):
|
||
if t.is_standable(nx, y + up, nz) is True:
|
||
act = "step_up" if up else "forward"
|
||
yield (nx, y + up, nz), Move(act, dx, up, dz)
|
||
break
|
||
|
||
# ② 下一格 / 掉落（脚位置下移）
|
||
if t.is_standable(nx, y - 1, nz) is True:
|
||
yield (nx, y - 1, nz), Move("step_down", dx, -1, dz)
|
||
else:
|
||
for drop in range(2, 5): # 最多落 4 格，不至于摔死
|
||
if t.is_standable(nx, y - drop, nz) is True:
|
||
yield (nx, y - drop, nz), Move("drop", dx, -drop, dz)
|
||
break
|
||
|
||
# ③ 跳上 1 格（前方要高一级）
|
||
if t.is_standable(nx, y + 1, nz) is True and t.is_standable(nx, y, nz) is not True:
|
||
yield (nx, y + 1, nz), Move("jump", dx, 1, dz, detail="跳过障碍")
|
||
|
||
# ④ 挖开挡路的（需世界改动）—— 支持「挖一列」
|
||
# ★★★ 三次实测才做对，记录下来免得再错：
|
||
# a) 不能用「更靠上的格子 standable」当「已经能过」的依据
|
||
# —— 那是挖完才成立的 ⇒ dig 永远不生成（2 格墙直接走不通）。
|
||
# b) 必须考虑挖**头部**那格，不只脚部（否则 2 格墙不通）。
|
||
# c) ★ 必须是「挖 k 格直到能站」而不是「挖 1 格」——
|
||
# 4 格高的墙要挖 4 格，只查 1 格就永远走不通（实测）。
|
||
# 代价按挖的格数线性增长 ⇒ 高墙自然比低墙贵，A* 会先试绕。
|
||
if allow_world_change:
|
||
already = t.is_standable(nx, y, nz) is True
|
||
if not already:
|
||
# 只做一件事：**从脚所在高度向上挖穿**，挖到能站为止。
|
||
# ★ 为什么这样就够（而不是先试向下、再试向上）：
|
||
# 脚所在高度就是玩家现在的高度；把这一段挖通即可继续前进。
|
||
# 我先前写的「先向下挖再向上挖」两段式既复杂又产生了
|
||
# 挖空气的荒谬清单（因为「挖掉 y+k」之后脚该站哪算错了）。
|
||
# 简化成「从 y 起往上逐格挖，直到 standable」后：
|
||
# 1 格障碍 ⇒ 挖 1 格；4 格墙 ⇒ 挖 4 格，全部正确。
|
||
carved = 0
|
||
for k in range(0, MAX_DIG_DEPTH):
|
||
if not t.can_dig(nx, y + k, nz):
|
||
break
|
||
carved += 1
|
||
# 挖掉 y..y+k 之后，脚停在 y+k（上面已清空、下面由 y+k-1 承重）
|
||
if t.standable_if_air(nx, y + k, nz) and k + 1 <= carved:
|
||
yield (nx, y + k, nz), Move(
|
||
"dig", dx, k, dz, needs_world_change=True,
|
||
detail=f"挖 {k + 1} 格穿过 {t.at(nx, y, nz)['name']}",
|
||
dig_count=k + 1)
|
||
break
|
||
|
||
|
||
# ⑤ 搭桥：前方本格与头顶是空的，但**脚下没有承重块** ⇒ 放一块垫脚
|
||
if allow_world_change:
|
||
for dx, dz in _DIRS:
|
||
nx, nz = x + dx, z + dz
|
||
if t.is_passable(nx, y, nz) is not True \
|
||
or t.is_passable(nx, y + 1, nz) is not True:
|
||
continue
|
||
if t.is_ground(nx, y - 1, nz) is not False:
|
||
continue # 已有地面（或不知道）⇒ 不需要搭
|
||
yield (nx, y, nz), Move("place", dx, 0, dz, needs_world_change=True,
|
||
detail="在缺口处搭一块")
|
||
|
||
|
||
# ---------- 启发式 ----------
|
||
|
||
|
||
|
||
|
||
def _octile(a, b) -> float:
|
||
"""八向 + 高度的启发式（可采纳：不超过真实代价）。
|
||
|
||
★ 高度权重用 HEIGHT_PENALTY（比 step_up 略高）——
|
||
这样「宁可多走一格」而不是「上下一级」，
|
||
寻路自然优先走平路（实测：修复后 1 格台阶不再被挖）。
|
||
"""
|
||
dx = abs(a[0] - b[0]); dz = abs(a[2] - b[2]); dy = abs(a[1] - b[1])
|
||
flat = max(dx, dz) + (math.sqrt(2.0) - 1.0) * min(dx, dz)
|
||
return flat + dy * HEIGHT_PENALTY
|
||
|
||
|
||
# ---------- A* 核心 ----------
|
||
|
||
|
||
def _astar(
|
||
t: Terrain,
|
||
start: tuple[int, int, int],
|
||
goal: tuple[int, int, int],
|
||
*,
|
||
max_nodes: int = 6000,
|
||
max_radius: int = 96,
|
||
allow_world_change: bool = True,
|
||
goal_test=None,
|
||
) -> dict[str, Any]:
|
||
"""A*。`goal_test` 为真时（而非坐标相等）即算到达 —— 可达性判断用。"""
|
||
c0 = _octile(start, goal)
|
||
open_heap: list[tuple[float, tuple[int, int, int]]] = [(c0, start)]
|
||
came: dict[tuple[int, int, int], tuple[tuple[int, int, int], Move]] = {}
|
||
best: dict[tuple[int, int, int], float] = {start: 0.0}
|
||
nodes = 0
|
||
touched_unloaded = False
|
||
furthest = 0.0
|
||
# ★ 记录离目标最近的已探索点：未到达时用它给「走到边界」的部分路径
|
||
nearest = start
|
||
|
||
def _dist_to_goal(p):
|
||
return _octile(p, goal)
|
||
|
||
while open_heap:
|
||
_, cur = heapq.heappop(open_heap)
|
||
if goal_test is not None:
|
||
if goal_test(cur):
|
||
return _reconstruct(came, cur, start, nodes, False, furthest)
|
||
else:
|
||
if cur == goal:
|
||
return _reconstruct(came, cur, start, nodes, False, furthest)
|
||
if _dist_to_goal(cur) < _dist_to_goal(nearest):
|
||
nearest = cur
|
||
nodes += 1
|
||
if nodes > max_nodes:
|
||
return _reconstruct(came, None, start, nodes, True, furthest,
|
||
reason="node_limit", nearest=nearest)
|
||
|
||
# ★★ 先单独探测「未加载边界」（实测踩到）：
|
||
# `neighbors()` 在未加载格上**不会生成**移动（因为
|
||
# is_standable 返回 None、无法判断），所以下面循环里的
|
||
# 「if not t.known(*nxt)」**永远执行不到** ⇒ 结果报 `exhausted`
|
||
# （听起来像「地形不通」），而真相是「**数据用完了**」。
|
||
# 这两者对 agent 的意义完全不同：
|
||
# 地形不通 ⇒ 换目标；数据用完 ⇒ 走过去加载再问。
|
||
# 所以必须在这里主动看一眼水平邻居。
|
||
for dx, dz in _DIRS:
|
||
if not t.known(cur[0] + dx, cur[1], cur[2] + dz):
|
||
touched_unloaded = True
|
||
break
|
||
|
||
for nxt, mv in neighbors(t, cur, allow_world_change):
|
||
# 未加载 ⇒ 记下来（不是墙，但要如实报告）
|
||
if not t.known(*nxt):
|
||
touched_unloaded = True
|
||
continue
|
||
# 半径剪枝
|
||
d = math.dist((nxt[0], nxt[2]), (start[0], start[2]))
|
||
if d > max_radius:
|
||
continue
|
||
furthest = max(furthest, d)
|
||
# ★ 每一步的真实代价 = 动作基础代价 + **背离目标的额外代价**。
|
||
# 少了后面这一项就会出荒谬路径（实测）：目标 y=0、挡路的是 1 格台阶，
|
||
# 「上 1 级再下 1 级」只要 2.7，比「挖 6 格」便宜 —— 听起来对，
|
||
# 但 1 格台阶明明**能挖 0 次**通过（原地挖一格就行），
|
||
# 于是 A* 选了「挖」，白白改世界。
|
||
# 真正该反映的是：为了绕过一点高度差要付出多少额外时间。
|
||
step_cost = COSTS[mv.action]
|
||
if mv.action == "dig":
|
||
# 挖穿 N 格 ⇒ 代价按 N 线性增长（挖 4 格远贵于挖 1 格）
|
||
step_cost = COSTS["dig"] * max(1, mv.dig_count)
|
||
if goal_test is None and nxt[1] != cur[1]:
|
||
step_cost += HEIGHT_PENALTY * 0.5
|
||
g = best[cur] + step_cost
|
||
if g < best.get(nxt, float("inf")):
|
||
best[nxt] = g
|
||
came[nxt] = (cur, mv)
|
||
f = g + (_dist_to_goal(nxt) if goal_test is None else 0.0)
|
||
heapq.heappush(open_heap, (f, nxt))
|
||
|
||
return _reconstruct(came, None, start, nodes, True, furthest,
|
||
reason="exhausted" if not touched_unloaded else "hit_unloaded",
|
||
touched_unloaded=touched_unloaded, nearest=nearest)
|
||
|
||
|
||
def _reconstruct(came, end, start, nodes, truncated, radius_reached,
|
||
reason: str = "", touched_unloaded: bool = False,
|
||
nearest=None) -> dict[str, Any]:
|
||
"""把 came[] 反向拼成移动序列；未到达时给「走到边界」的部分路径。"""
|
||
if end is not None:
|
||
moves: list[Move] = []
|
||
cur = end
|
||
while cur in came:
|
||
prev, mv = came[cur]
|
||
moves.append(mv)
|
||
cur = prev
|
||
moves.reverse()
|
||
return {
|
||
"reached": True, "moves": [m.as_dict() for m in moves],
|
||
"move_objects": moves,
|
||
"steps": len(moves), "nodes_expanded": nodes,
|
||
"truncated": False,
|
||
"cost": round(sum(COSTS[m.action] for m in moves), 2),
|
||
"needs_world_change": any(m.needs_world_change for m in moves),
|
||
}
|
||
# ★ 没到达：给**走到边界**的部分路径（而不是空手而归），并如实说明为什么停。
|
||
# 为什么给部分路径：agent 可以「先走到这儿」再重新寻路 —— 这比
|
||
# 「完全没路」有用，且与用户裁定的「未加载就走到边界为止」一致。
|
||
best_node = nearest if nearest is not None else start
|
||
moves = []
|
||
cur = best_node
|
||
while cur in came:
|
||
prev, mv = came[cur]
|
||
moves.append(mv)
|
||
cur = prev
|
||
moves.reverse()
|
||
# 用真实的最远距离（而不是 0）
|
||
far = math.dist((best_node[0], best_node[2]), (start[0], start[2])) \
|
||
if best_node != start else 0.0
|
||
return {
|
||
"reached": False, "moves": [m.as_dict() for m in moves],
|
||
"move_objects": moves,
|
||
"steps": len(moves), "nodes_expanded": nodes,
|
||
"truncated": truncated,
|
||
"stopped_because": reason or "exhausted",
|
||
"stopped_at": list(best_node),
|
||
"radius_reached": round(max(radius_reached, far), 1),
|
||
"cost": round(sum(COSTS[m.action] for m in moves), 2),
|
||
"needs_world_change": any(m.needs_world_change for m in moves),
|
||
}
|
||
|
||
|
||
# ---------- 两个独立工具 ----------
|
||
|
||
|
||
def _player_cell(world) -> tuple[int, int, int]:
|
||
p = world.player
|
||
return int(math.floor(p.x)), int(math.floor(p.y)), int(math.floor(p.z))
|
||
|
||
|
||
def reachable(
|
||
world,
|
||
x: int | None = None, y: int | None = None, z: int | None = None,
|
||
*,
|
||
max_radius: int = 48,
|
||
allow_world_change: bool = False,
|
||
) -> dict[str, Any]:
|
||
"""【工具 1】可达性判断：**到得了吗 / 大概多远**。
|
||
|
||
刻意**不返回**具体路线 —— 只要一个是非 + 距离，让 agent 决定值不值得去。
|
||
|
||
参数缺省时用玩家所在格 ⇒ 「我脚下能站吗」这类问题也能直接问。
|
||
"""
|
||
start = _player_cell(world)
|
||
goal = (x, y, z) if x is not None and y is not None and z is not None else start
|
||
t = Terrain(world, allow_world_change=allow_world_change)
|
||
r = _astar(t, start, goal, max_radius=max_radius,
|
||
allow_world_change=allow_world_change,
|
||
goal_test=lambda n: _octile(n, goal) <= 1.5)
|
||
return {
|
||
"reachable": r["reached"],
|
||
"from": list(start),
|
||
"to": list(goal),
|
||
"approximate_steps": r["steps"],
|
||
"cost": r["cost"],
|
||
# ★ 这个字段是**两个工具的分界**：reachable 默认不挖不放，
|
||
# 所以它答的是「**现有地形**能过去吗」。
|
||
"without_changing_world": not allow_world_change,
|
||
"would_need_world_change": r["needs_world_change"] if allow_world_change else None,
|
||
"stopped_because": r.get("stopped_because"),
|
||
"stopped_at": r.get("stopped_at"),
|
||
"truncated": r["truncated"],
|
||
"nodes_expanded": r["nodes_expanded"],
|
||
"note": _reachable_note(r, allow_world_change),
|
||
}
|
||
|
||
|
||
def _reachable_note(r: dict[str, Any], allow_world_change: bool) -> str:
|
||
if r["reached"]:
|
||
base = ("在**现有地形**上能过去（不挖、不放）" if not allow_world_change
|
||
else "能过去")
|
||
return f"{base}；只回能不能与大概步数，要具体路线用 find_path"
|
||
why = r.get("stopped_because")
|
||
if why == "hit_unloaded":
|
||
return ("走到**未加载区块边界**为止：再往前没有数据，无法判断。"
|
||
"让玩家走过去加载后重新问（不乐观推断）")
|
||
if why == "node_limit":
|
||
return "搜索节点用尽（目标可能太远或地形太绕）；可缩小距离或分段多次问"
|
||
if not allow_world_change:
|
||
return ("**现有地形**上不通。若要挖开/搭桥过去，用 find_path"
|
||
"（它允许改世界）")
|
||
return "在已加载范围内没找到路；走过去加载新区块后可再试"
|
||
|
||
|
||
def find_path(
|
||
world,
|
||
x: int | None = None, y: int | None = None, z: int | None = None,
|
||
*,
|
||
max_radius: int = 48,
|
||
max_nodes: int = 6000,
|
||
allow_world_change: bool = True,
|
||
allow_dig: bool = True,
|
||
) -> dict[str, Any]:
|
||
"""【工具 2】完整版寻路：给出**具体怎么走**。
|
||
|
||
完整版 = 除常规移动外，还允许：
|
||
- `dig` 挖开挡路的方块
|
||
- `place` 在缺口处搭桥
|
||
|
||
这些动作 `needs_world_change=True`，返回值里会明确列出
|
||
**要改哪几个方块**（与 M5 的 dig/place 串联）。
|
||
"""
|
||
start = _player_cell(world)
|
||
goal = (x, y, z) if x is not None and y is not None and z is not None else start
|
||
t = Terrain(world, allow_world_change=allow_world_change)
|
||
r = _astar(t, start, goal, max_nodes=max_nodes, max_radius=max_radius,
|
||
allow_world_change=allow_world_change)
|
||
moves: list[Move] = r.pop("move_objects", [])
|
||
digs: list[dict[str, Any]] = []
|
||
places: list[dict[str, Any]] = []
|
||
cx, cy, cz = start
|
||
for mv in moves:
|
||
if mv.action == "dig":
|
||
# 挖一列 ⇒ 列出**每一格**（agent 要按顺序逐格挖）。
|
||
# mv.dy 指向「脚最终停在的高度」，要挖的是它**下方**的 dig_count 格。
|
||
base_y = cy + mv.dy - (mv.dig_count - 1)
|
||
for k in range(mv.dig_count):
|
||
px, py, pz = cx + mv.dx, base_y + k, cz + mv.dz
|
||
digs.append({"position": [px, py, pz],
|
||
"block": t.at(px, py, pz)["name"]})
|
||
elif mv.action == "place":
|
||
places.append({"position": [cx + mv.dx, cy + mv.dy, cz + mv.dz]})
|
||
cx += mv.dx
|
||
cy += mv.dy
|
||
cz += mv.dz
|
||
r["world_changes"] = {
|
||
"dig": digs,
|
||
"place": places,
|
||
"count": len(digs) + len(places),
|
||
"note": ("执行这些要用 dig/place 工具；每一步都可能失败"
|
||
"（权限/放不下），失败需重新寻路"
|
||
if digs or places else "不需要改动世界"),
|
||
}
|
||
r["from"] = list(start)
|
||
r["to"] = list(goal)
|
||
r["allow_dig"] = allow_dig
|
||
r["collision_caveat"] = ("碰撞数据只有 block/empty 两档：半砖与楼梯会被当整块，"
|
||
"所以路径可能多绕路，但不会撞墙")
|
||
return r
|