diff --git a/README.md b/README.md index a58aed7..fa14526 100644 --- a/README.md +++ b/README.md @@ -55,6 +55,11 @@ 可提取:mod id / 名称 / 版本 / **方块名清单** / 物品名清单 **诚实边界**:拿不到 state id(已在摘要里明说,见附录 D) +- [x] **P6** 世界模型(`mcfold/world.py`,7 项回归全绿) + 多区块聚合 / 负坐标定位 / 世界高度边界 + **关键语义**:`unloaded`(没收到)与 `air`(真是空气)**必须可区分** + mod 方块计入 `unknown_state_ids`,世界摘要带告警 + ## 累计测试 | 套件 | 项数 | @@ -63,5 +68,6 @@ | protodef | 6 | | 区块解析 | 11 | | mod jar | 8 | +| 世界模型 | 7 | | 跨实现互操作 | 2 | -| **合计** | **33** | +| **合计** | **40** | diff --git a/mcfold/__pycache__/world.cpython-312.pyc b/mcfold/__pycache__/world.cpython-312.pyc new file mode 100644 index 0000000..370920b Binary files /dev/null and b/mcfold/__pycache__/world.cpython-312.pyc differ diff --git a/mcfold/world.py b/mcfold/world.py new file mode 100644 index 0000000..2aaf2b9 --- /dev/null +++ b/mcfold/world.py @@ -0,0 +1,119 @@ +"""世界模型:聚合区块,按世界坐标查方块。 + +对应 mineflayer 侧的 prismarine-world(811 行)。 +职责: + * 以 (chunkX, chunkZ) 为键存区块列 + * 世界坐标 (x, y, z) → 方块名(跨区块自动定位) + * 未加载区域**显式**返回 unloaded,而不是假装是空气 + +★ 设计要点(与 mineflayer 的关键区别): + mineflayer 对越界 state id 会**静默**映射成别的方块; + 本实现返回 `unknown()`,并把该 id 归因到「可能是 mod 方块」。 + 这符合文档附录 D 的结论:让降级可见,而不是无声错误。 +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Iterable + +from .chunk import Chunk, BlockRegistry, MIN_Y, WORLD_HEIGHT, NUM_SECTIONS, decode_chunk + + +@dataclass +class BlockHit: + """一次方块查询的结果(含来源说明,便于 agent 理解可信度)。""" + name: str | None # 方块名;None 表示未知/未加载 + state_id: int + loaded: bool # 该区块是否已加载 + source: str # "registry" | "unknown-state" | "unloaded" | "out-of-world" + + def describe(self) -> str: + if self.source == "unloaded": + return "未加载(该区块还没收到)" + if self.source == "out-of-world": + return f"超出世界高度(y 必须在 {MIN_Y}..{MIN_Y + WORLD_HEIGHT - 1})" + if self.source == "unknown-state": + return f"unknown({self.state_id}) —— 本机注册表无此 state id,可能来自 mod 方块" + return self.name or "?" + + +class World: + """区块集合 + 方块查询。""" + + def __init__( + self, + registry: BlockRegistry | None = None, + num_sections: int = NUM_SECTIONS, + min_y: int = MIN_Y, + ): + self.registry = registry + self.chunks: dict[tuple[int, int], Chunk] = {} + self.num_sections = num_sections + self.min_y = min_y + # 统计:用于给 agent 报告「我看到了什么」 + self.unknown_state_ids: dict[int, int] = {} # state_id -> 出现次数 + + # ---------- 装载 ---------- + def load_chunk_data(self, cx: int, cz: int, data: bytes) -> Chunk: + ch = decode_chunk(cx, cz, data, self.num_sections) + self.chunks[(cx, cz)] = ch + return ch + + def unload_chunk(self, cx: int, cz: int) -> None: + self.chunks.pop((cx, cz), None) + + @property + def loaded_count(self) -> int: + return len(self.chunks) + + # ---------- 查询 ---------- + def block_at(self, x: int, y: int, z: int) -> BlockHit: + """世界坐标 → 方块。坐标可为负(MC 坐标就是这么用的)。""" + if y < self.min_y or y >= self.min_y + WORLD_HEIGHT: + return BlockHit(None, 0, False, "out-of-world") + + cx, cz = x >> 4, z >> 4 # Python 的 >> 对负数即 floor 除 16 ✓ + ch = self.chunks.get((cx, cz)) + if ch is None: + return BlockHit(None, 0, False, "unloaded") + + sid = ch.state_id_at(x & 15, y, z & 15) + return self._resolve(sid) + + def _resolve(self, sid: int) -> BlockHit: + if self.registry is None: + return BlockHit(None, sid, True, "unknown-state") + name = self.registry.name_of(sid) + if name is None: + self.unknown_state_ids[sid] = self.unknown_state_ids.get(sid, 0) + 1 + return BlockHit(None, sid, True, "unknown-state") + return BlockHit(name, sid, True, "registry") + + def columns(self) -> Iterable[tuple[int, int]]: + return self.chunks.keys() + + # ---------- 摘要(给 agent 用) ---------- + def summary(self) -> dict[str, Any]: + """世界现状摘要。**含降级提示**,让 agent 知道数据可信度。""" + out: dict[str, Any] = { + "loaded_chunks": self.loaded_count, + "min_y": self.min_y, + "world_height": WORLD_HEIGHT, + } + if self.unknown_state_ids: + out["unknown_state_count"] = len(self.unknown_state_ids) + out["unknown_state_sample"] = dict( + sorted(self.unknown_state_ids.items(), key=lambda kv: -kv[1])[:5]) + out["warning"] = ( + "检测到本机注册表没有的 state id ⇒ 本实例可能装了 mod。" + "这些方块会显示为 unknown(id),**不是**某个原版方块。" + ) + return out + + def reset_unknown_stats(self) -> None: + self.unknown_state_ids.clear() + + +def in_world(y: int, min_y: int = MIN_Y, height: int = WORLD_HEIGHT) -> bool: + return min_y <= y < min_y + height diff --git a/tests/test_world.py b/tests/test_world.py new file mode 100644 index 0000000..2121f8d --- /dev/null +++ b/tests/test_world.py @@ -0,0 +1,120 @@ +"""世界模型回归:区块聚合 + 世界坐标查询 + 显式降级。""" +import sys, os, json, subprocess +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from mcfold.world import World, in_world +from mcfold.chunk import BlockRegistry, MIN_Y, WORLD_HEIGHT +from mcfold.protodef import Writer + +RUNTIME = "/home/newqqagent/plugin_data/mc/runtime" +BLOCKS = f"{RUNTIME}/node_modules/minecraft-data/minecraft-data/data/pc/1.20/blocks.json" + +def reg(): return BlockRegistry.from_json_file(BLOCKS) + +def test_negative_coords(): + """★ 负坐标定位:MC 世界坐标可为负,这是最容易错的地方。""" + w = World(reg()) + # 在 (-1,-1) 区块放一个方块:(x=-1,z=-1) ⇒ chunk (-1,-1), 内部 (15,15) + assert (-1) >> 4 == -1 and (-1) & 15 == 15 + # 先确认未加载时报 unloaded(而不是假装空气) + hit = w.block_at(-1, 64, -1) + assert not hit.loaded and hit.source == "unloaded", hit + print(" ✓ 负坐标:未加载区块显式返回 unloaded(非假装空气)") + +def test_unloaded_vs_air_distinction(): + """★ 关键语义:「未加载」与「空气」必须可区分。 + + mineflayer 的 bot.blockAt 在未加载时返回 null,但很多调用方当空气处理; + 我们要求显式区分,避免 agent 误判。 + """ + w = World(reg()) + unloaded = w.block_at(0, 64, 0) + assert unloaded.source == "unloaded", unloaded + # 加载一个全空气区块后,同一坐标应变成 air(registry) + wr = Writer() + for _ in range(24): + wr.write((0).to_bytes(2,"big")); wr.write(bytes([0])) + wr.varint(0); wr.varint(0) + wr.write(bytes([0])); wr.varint(0); wr.varint(0) + w.load_chunk_data(0, 0, wr.getvalue()) + air = w.block_at(0, 64, 0) + assert air.loaded and air.name == "air", (air.loaded, air.name) + print(" ✓ 未加载(unloaded) 与 空气(air) 语义分明") + +def test_real_chunk_via_world(): + """用 prismarine 产出的真实区块走完 World 全路径。""" + r = subprocess.run(["node", f"{RUNTIME}/chunkgen.js"], capture_output=True, + text=True, cwd=RUNTIME, timeout=120) + if r.returncode != 0: + print(f" - 跳过({r.stderr[:80]})"); return + d = json.loads(r.stdout) + w = World(reg()) + w.load_chunk_data(0, 0, bytes.fromhex(d["hex"])) + checks = [((0,-60,0),"stone"), ((1,-59,2),"gold_block"), + ((5,-59,7),"diamond_block"), ((3,20,4),"diamond_block")] + for (x,y,z), want in checks: + hit = w.block_at(x,y,z) + assert hit.loaded and hit.name == want, (x,y,z,hit) + print(f" ✓ World 全路径: 4/4 方块命中(解真实区块数据)") + +def test_world_height_bounds(): + w = World(reg()) + assert w.block_at(0, MIN_Y-1, 0).source == "out-of-world" + assert w.block_at(0, MIN_Y+WORLD_HEIGHT, 0).source == "out-of-world" + assert w.block_at(0, MIN_Y, 0).source == "unloaded" # 合法高度但未加载 + assert in_world(MIN_Y) and not in_world(MIN_Y-1) + print(f" ✓ 世界高度边界: [{MIN_Y}, {MIN_Y+WORLD_HEIGHT}) 之外显式 out-of-world") + +def test_unknown_state_reported(): + """★ mod 方块的可见降级:unknown(id) 且计入统计。""" + r = reg() + w = World(r) + wr = Writer() + # section0: palette=[99999] + wr.write((1).to_bytes(2,"big")); wr.write(bytes([4])) + wr.varint(1); wr.varint(99999); wr.varint(256) + wr.write(b"\x00"*(256*8)) + wr.write(bytes([0])); wr.varint(0); wr.varint(0) + for _ in range(23): + wr.write((0).to_bytes(2,"big")); wr.write(bytes([0])) + wr.varint(0); wr.varint(0) + wr.write(bytes([0])); wr.varint(0); wr.varint(0) + w.load_chunk_data(0, 0, wr.getvalue()) + hit = w.block_at(0, MIN_Y, 0) # section 0 的首个方块 + assert hit.source == "unknown-state", hit + assert "mod" in hit.describe(), hit.describe() + s = w.summary() + assert s.get("unknown_state_count") == 1, s + assert "mod" in s.get("warning",""), s + print(f" ✓ ★ mod 方块可见降级: {hit.describe()[:40]}… + 世界摘要含告警") + +def test_multiple_chunks(): + """多区块聚合:跨区块查询各自命中自己的列。""" + w = World(reg()) + wr = Writer() + for _ in range(24): + wr.write((0).to_bytes(2,"big")); wr.write(bytes([0])) + wr.varint(0); wr.varint(0) + wr.write(bytes([0])); wr.varint(0); wr.varint(0) + data = wr.getvalue() + for cx, cz in [(0,0), (1,0), (-1,-1), (2,-3)]: + w.load_chunk_data(cx, cz, data) + assert w.loaded_count == 4, w.loaded_count + # (16,64,0) 属于区块 (1,0) + assert w.block_at(16,64,0).loaded + # (-1,64,-1) 属于区块 (-1,-1) + assert w.block_at(-1,64,-1).loaded + # (32,64,-48) 属于区块 (2,-3) + assert w.block_at(32,64,-48).loaded + # (16,64,-1) 属于区块 (1,-1) —— 未加载 + assert w.block_at(16,64,-1).source == "unloaded" + print(" ✓ 多区块聚合: 4 列,跨区块/负坐标定位均正确,未加载区可辨") + # 卸载 + w.unload_chunk(1, 0) + assert w.block_at(16,64,0).source == "unloaded" + print(" ✓ 区块卸载后查询回到 unloaded") + +if __name__ == "__main__": + for fn in [test_negative_coords, test_unloaded_vs_air_distinction, test_real_chunk_via_world, + test_world_height_bounds, test_unknown_state_reported, test_multiple_chunks]: + fn() + print("\n世界模型:全部通过")