diff --git a/README.md b/README.md index ceda917..5b1bed2 100644 --- a/README.md +++ b/README.md @@ -41,3 +41,12 @@ | 协议**不需重写**，只需解释声明式 JSON | 111 个 play 包由同一份 `protocol.json` 驱动 | | Python 客户端能真连真服 | Node 侧日志 `SERVER_GOT_LOGIN username= PyProbe` | | 协议号由实例文件夹决定 | `1.20.1/version.json` → 763 | +| 区块格式**必须自己解**（协议层是声明式，但 chunkData 是不透明 buffer） | `protocol.json` 里 chunkData 只声明为 buffer | +| Python 解码器能解独立实现产出的区块 | prismarine-chunk 产样本 → 4/4 方块命中 | +| mod 方块在本机注册表里**显式 unknown**，而非静默错配 | `state 99999 -> unknown(99999)` | + +## 进度（续） + +- [x] **P4** 区块解析（`mcfold/chunk.py`，11 项回归全绿） + 位打包 / 单值·本地·全局调色板 / 生物群系 section / 世界边界 + **关键行为**：越界 state id ⇒ `unknown(id)`（对照 mineflayer 的静默错配） diff --git a/mcfold/__pycache__/chunk.cpython-312.pyc b/mcfold/__pycache__/chunk.cpython-312.pyc new file mode 100644 index 0000000..b780982 Binary files /dev/null and b/mcfold/__pycache__/chunk.cpython-312.pyc differ diff --git a/mcfold/chunk.py b/mcfold/chunk.py new file mode 100644 index 0000000..168f449 --- /dev/null +++ b/mcfold/chunk.py @@ -0,0 +1,265 @@ +"""区块解析：chunkData buffer → 方块（state id / 方块名）。 + +为什么这段要自己写（对照 P2 的协议层）： + 协议层是**声明式**的（protodef 解释 JSON 即可）， + 但 `packet_map_chunk` 里的 `chunkData` 是一个 **不透明 buffer** + —— protodef 只声明它是 buffer，内部结构不在 protocol.json 里。 + 所以区块格式必须按 MC 的固定布局自己解。 + +1.20 的区块格式（实测自 prismarine-chunk src/pc/1.18/ChunkColumn.js:253）： + minY = -64, worldHeight = 384 ⇒ 24 个 section + 每个 section 交替出现： + [方块 section] [生物群系 section] + 方块 section: + solidBlockCount : i16 (大端) + bitsPerBlock : u8 + == 0 -> 单一值调色板: varint(stateId) + varint(0) ← 后一个是 longs 数 + > MAX_BITS(8) -> 全局调色板: 直接是 state id，16 位/值 + 否则 -> 本地调色板: varint(paletteLen) + palette[] + 位数组 + 位数组（1.20 有 size 前缀；1.21.5+ 才去掉）: + longs : varint + data : longs 个 64 位值；每个 long 按 [高 u32, 低 u32] 存储 + +关键坑：**值不跨 long 边界**（valuesPerLong = floor(64/bitsPerValue)）， +即一个 64 位 long 里整齐放若干个值。这与我最初以为的"紧密打包"不同。 +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +from .protodef import ProtoError, Reader + +# ---- 1.20 区块常量（实测自 prismarine-chunk constants.js + 1.18/ChunkColumn.js）---- +MIN_Y = -64 +WORLD_HEIGHT = 384 +NUM_SECTIONS = WORLD_HEIGHT >> 4 # 24 +SECTION_VOLUME = 16 * 16 * 16 # 4096 +BIOME_SECTION_VOLUME = SECTION_VOLUME // (4 * 4 * 4) # 64 + +MIN_BITS_PER_BLOCK = 4 +MAX_BITS_PER_BLOCK = 8 # 超过它就转全局调色板 +GLOBAL_BITS_PER_BLOCK = 16 + +MIN_BITS_PER_BIOME = 1 +MAX_BITS_PER_BIOME = 3 +GLOBAL_BITS_PER_BIOME = 6 + + +class ChunkError(Exception): + pass + + +class BitArray: + """MC 的位打包数组（每个 long 内整齐放 floor(64/bits) 个值，不跨边界）。""" + + __slots__ = ("data", "capacity", "bits_per_value", "values_per_long", "value_mask") + + def __init__(self, data: list[int], capacity: int, bits_per_value: int): + self.data = data # 每个元素是一个 64 位 long + self.capacity = capacity + self.bits_per_value = bits_per_value + self.values_per_long = 64 // bits_per_value + self.value_mask = (1 << bits_per_value) - 1 + + @classmethod + def from_buffer(cls, r: Reader, capacity: int, bits_per_value: int, + size_prefix: bool = True) -> "BitArray": + """按 MC 的在线格式读位数组。""" + if bits_per_value <= 0 or bits_per_value > 32: + raise ChunkError(f"非法 bitsPerValue: {bits_per_value}") + vpl = 64 // bits_per_value + if size_prefix: + longs = r.varint() + else: + longs = (capacity + vpl - 1) // vpl + if longs < 0 or longs > 1_000_000: + raise ChunkError(f"位数组长度异常: {longs}") + data: list[int] = [] + for _ in range(longs): + # JS 侧：先读高 u32 再读低 u32 → long = (high<<32)|low + high = int.from_bytes(r.read(4), "big") + low = int.from_bytes(r.read(4), "big") + data.append((high << 32) | low) + return cls(data, capacity, bits_per_value) + + def get(self, index: int) -> int: + if index < 0 or index >= self.capacity: + return 0 + start = index // self.values_per_long + off = (index - start * self.values_per_long) * self.bits_per_value + if start >= len(self.data): + return 0 + # 值不跨 long 边界 ⇒ 单个 long 内移位即可 + return (self.data[start] >> off) & self.value_mask + + def to_list(self) -> list[int]: + return [self.get(i) for i in range(self.capacity)] + + +@dataclass +class Section: + """一个 16x16x16 的方块 section。""" + y_index: int # 0..23（对应世界 y = MIN_Y + 16*y_index） + solid_block_count: int = 0 + bits_per_block: int = 0 + palette: list[int] = field(default_factory=list) # 本地调色板（indirect 时用） + bits: BitArray | None = None + single_value: int | None = None # bits==0 时的单一 state id + + def state_id_at(self, x: int, y: int, z: int) -> int: + """section 内 (x,y,z) 的全局 state id。索引顺序 = (y<<8)|(z<<4)|x。""" + idx = (y << 8) | (z << 4) | x + if self.bits_per_block == 0: + return self.single_value or 0 + raw = self.bits.get(idx) if self.bits else 0 + if self.palette: + # 本地调色板：raw 是调色板下标 + return self.palette[raw] if 0 <= raw < len(self.palette) else 0 + return raw # 全局调色板：raw 就是 state id + + +@dataclass +class BiomeSection: + bits_per_biome: int = 0 + palette: list[int] = field(default_factory=list) + bits: BitArray | None = None + single_value: int | None = None + + +@dataclass +class Chunk: + """一个区块列（16 x 384 x 16）。""" + x: int + z: int + sections: list[Section] = field(default_factory=list) + biomes: list[BiomeSection] = field(default_factory=list) + + def state_id_at(self, x: int, y: int, z: int) -> int: + """世界坐标 → state id（y 是绝对高度）。""" + if y < MIN_Y or y >= MIN_Y + WORLD_HEIGHT: + return 0 + sec = (y - MIN_Y) >> 4 + if sec >= len(self.sections): + return 0 + return self.sections[sec].state_id_at(x & 15, y & 15, z & 15) + + +def _read_block_section(r: Reader) -> Section: + solid = int.from_bytes(r.read(2), "big", signed=True) + bits = r.read(1)[0] + if bits > 16: + raise ChunkError(f"bitsPerBlock 过大: {bits}") + + sec = Section(y_index=-1, solid_block_count=solid, bits_per_block=bits) + + if bits == 0: + # 单一值：varint(stateId) 后跟 varint(0)（longs 数） + sec.single_value = r.varint() + r.varint() # longs = 0 + return sec + + if bits > MAX_BITS_PER_BLOCK: + # 全局调色板：直接 state id + sec.bits = BitArray.from_buffer(r, SECTION_VOLUME, GLOBAL_BITS_PER_BLOCK, + size_prefix=True) + sec.bits_per_block = GLOBAL_BITS_PER_BLOCK + return sec + + # 本地调色板 + n = r.varint() + if n < 0 or n > 4096: + raise ChunkError(f"调色板长度异常: {n}") + sec.palette = [r.varint() for _ in range(n)] + sec.bits = BitArray.from_buffer(r, SECTION_VOLUME, bits, size_prefix=True) + return sec + + +def _read_biome_section(r: Reader) -> BiomeSection: + bits = r.read(1)[0] + if bits > 8: + raise ChunkError(f"bitsPerBiome 过大: {bits}") + sec = BiomeSection(bits_per_biome=bits) + + if bits == 0: + sec.single_value = r.varint() + r.varint() # longs = 0 + return sec + + if bits > MAX_BITS_PER_BIOME: + sec.bits = BitArray.from_buffer(r, BIOME_SECTION_VOLUME, GLOBAL_BITS_PER_BIOME, + size_prefix=True) + sec.bits_per_biome = GLOBAL_BITS_PER_BIOME + return sec + + n = r.varint() + if n < 0 or n > 1024: + raise ChunkError(f"生物群系调色板长度异常: {n}") + sec.palette = [r.varint() for _ in range(n)] + sec.bits = BitArray.from_buffer(r, BIOME_SECTION_VOLUME, bits, size_prefix=True) + return sec + + +def decode_chunk(x: int, z: int, chunk_data: bytes, + num_sections: int = NUM_SECTIONS) -> Chunk: + """把 packet_map_chunk 的 chunkData 解成区块列。 + + num_sections 由世界的 worldHeight 决定（1.20 = 24）；传入以适配不同版本。 + """ + r = Reader(chunk_data) + chunk = Chunk(x=x, z=z) + + for i in range(num_sections): + if r.remaining() <= 0: + break + sec = _read_block_section(r) + sec.y_index = i + chunk.sections.append(sec) + # 每个方块 section 后紧跟一个生物群系 section + if r.remaining() > 0: + chunk.biomes.append(_read_biome_section(r)) + + return chunk + + +# ---------------- state id → 方块名（方块注册表） ---------------- + +class BlockRegistry: + """方块注册表：state id ↔ 方块名。 + + mod 支持的落点正在这里：原版表是内置的（minecraft-data 的 blocks.json）， + 而 mod 方块需要另外的途径补进来（见文档附录 D/E）。找不到时必须**显式**报 + "unknown"，而不是像 mineflayer 那样静默映射成别的方块。 + """ + + def __init__(self, blocks: list[dict[str, Any]]): + self.blocks = blocks + self.by_state: dict[int, str] = {} + self.by_name: dict[str, dict[str, Any]] = {} + for b in blocks: + name = b["name"] + self.by_name[name] = b + lo = b.get("minStateId", 0) + hi = b.get("maxStateId", lo) + for sid in range(lo, hi + 1): + self.by_state[sid] = name + self.max_state_id = max(self.by_state) if self.by_state else -1 + + @classmethod + def from_json_file(cls, path: str) -> "BlockRegistry": + import json + with open(path, "r", encoding="utf-8") as f: + return cls(json.load(f)) + + def name_of(self, state_id: int) -> str | None: + """state id → 方块名；越界返回 None（**不**静默回退）。""" + return self.by_state.get(state_id) + + def name_or_unknown(self, state_id: int) -> str: + n = self.name_of(state_id) + return n if n is not None else f"unknown({state_id})" + + def is_out_of_range(self, state_id: int) -> bool: + """越界 ⇒ 该 id 极可能来自 mod 方块（本机注册表没有它）。""" + return state_id > self.max_state_id diff --git a/tests/support_testserver.js b/tests/support_testserver.js new file mode 100644 index 0000000..9ca0ba9 --- /dev/null +++ b/tests/support_testserver.js @@ -0,0 +1,9 @@ +// 供测试用的 Node 服务端（独立实现）。 +// 用法: node support_testserver.js +const mc = require('minecraft-protocol') +const port = parseInt(process.argv[2] || '25599') +const server = mc.createServer({ 'online-mode': false, port, version: '1.20.1', motd: 'mcfold-test' }) +server.on('listening', () => console.log('SERVER_READY')) +server.on('login', (c) => console.log('SERVER_GOT_LOGIN username=', c.username)) +server.on('error', e => console.log('SERVER_ERR', e.message)) +setTimeout(() => process.exit(0), 120000) diff --git a/tests/test_chunk.py b/tests/test_chunk.py new file mode 100644 index 0000000..f266257 --- /dev/null +++ b/tests/test_chunk.py @@ -0,0 +1,155 @@ +"""chunk 解析回归。 + +核心验证方式：用 **prismarine-chunk（独立实现）产出的真实格式数据** +交给 Python 解 —— 两边独立，对得上才算格式对。 +样本由 /home/newqqagent/plugin_data/mc/runtime/chunkgen*.js 生成。 +""" +import sys, os, json, subprocess, tempfile +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from mcfold.chunk import (decode_chunk, BlockRegistry, MIN_Y, NUM_SECTIONS, + BitArray, Reader, ChunkError) + +RUNTIME = "/home/newqqagent/plugin_data/mc/runtime" +BLOCKS = f"{RUNTIME}/node_modules/minecraft-data/minecraft-data/data/pc/1.20/blocks.json" + +def gen(script): + """跑 Node 生成独立实现产出的样本。""" + r = subprocess.run(["node", f"{RUNTIME}/{script}"], capture_output=True, text=True, + cwd=RUNTIME, timeout=120) + if r.returncode != 0: + return None, r.stderr[:300] + return json.loads(r.stdout), None + +def reg(): + return BlockRegistry.from_json_file(BLOCKS) + +def test_bitarray(): + """位打包：值不跨 long 边界（这是最容易写错的地方）。""" + # 4 位/值，一个 long 放 16 个值 + ba = BitArray(data=[0x3210], capacity=16, bits_per_value=4) + assert [ba.get(i) for i in range(4)] == [0, 1, 2, 3], [ba.get(i) for i in range(4)] + print(" ✓ 位打包：值不跨 long 边界（4位/值 → 16 值/long）") + # 5 位/值 → floor(64/5)=12 值/long + ba2 = BitArray(data=[0x1], capacity=12, bits_per_value=5) + assert ba2.values_per_long == 12, ba2.values_per_long + assert ba2.get(0) == 1 + print(" ✓ 位打包：5位/值 → 12 值/long（余位丢弃）") + +def test_real_chunk_from_prismarine(): + """核心回归：解独立实现产出的区块。""" + d, err = gen("chunkgen.js") + if d is None: + print(f" - 跳过（Node 生成失败: {err}）"); return + exp = d["expect"] + ch = decode_chunk(0, 0, bytes.fromhex(d["hex"])) + assert len(ch.sections) == NUM_SECTIONS, len(ch.sections) + r = reg() + 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: + sid = ch.state_id_at(x,y,z) + got = r.name_or_unknown(sid) + assert got == want, f"({x},{y},{z}) {sid} -> {got}, 期望 {want}" + nonempty = [s.y_index for s in ch.sections if s.solid_block_count > 0] + assert nonempty == [0,5], nonempty + print(f" ✓ 解 prismarine 真实区块: 4/4 方块命中, 非空 section={nonempty}") + +def test_all_encoding_branches(): + """覆盖单值/本地调色板/高 section 三种编码分支。""" + d, err = gen("chunkgen2.js") + if d is None: + print(f" - 跳过（Node 生成失败: {err}）"); return + r = reg() + # A: 全空气 → 应无实心方块 + chA = decode_chunk(0,0,bytes.fromhex(d["A"]["hex"])) + assert all(s.solid_block_count == 0 for s in chA.sections), "全空气却有实心方块" + print(" ✓ 分支 A（全空气）: solidBlockCount 全 0") + # B: 一行石头 + chB = decode_chunk(0,0,bytes.fromhex(d["B"]["hex"])) + names = {r.name_or_unknown(chB.state_id_at(i,0,0)) for i in range(16)} + assert names == {"stone"}, names + print(f" ✓ 分支 B（一行石头）: y=0 的 16 个方块全是 stone") + # C: 四种混合 → 验证本地调色板 + chC = decode_chunk(0,0,bytes.fromhex(d["C"]["hex"])) + got = [r.name_or_unknown(chC.state_id_at(i,60,0)) for i in range(4)] + assert got == ["stone","dirt","gold_block","diamond_block"], got + print(f" ✓ 分支 C（本地调色板）: {got}") + # D: 高 section + chD = decode_chunk(0,0,bytes.fromhex(d["D"]["hex"])) + assert r.name_or_unknown(chD.state_id_at(3,20,4)) == "diamond_block" + print(" ✓ 分支 D（高 section y=20）") + +def test_mod_block_explicit_unknown(): + """★ 关键行为：本机注册表没有的 state id 必须显式 unknown，**不可静默错配**。 + + 这正是 mineflayer 的缺陷（越界 id 静默变成 note_block），我们要把它变成可见降级。 + + 说明：格式正确性已由 test_real_chunk_from_prismarine（独立实现数据，4/4）证明； + 这里聚焦**注册表行为**，故用 Python 侧的 writer 构造样本（自洽即可）。 + """ + from mcfold.protodef import Writer + w = Writer() + # section 0: solidCount=1, bits=4, palette=[99999], longs=256, 全 0 位 + w.write((1).to_bytes(2, "big")) + w.write(bytes([4])) + w.varint(1); w.varint(99999) + w.varint(256) + w.write(b"\x00" * (256 * 8)) + # biome section（单值） + w.write(bytes([0])); w.varint(0); w.varint(0) + # 其余 23 个 section：单值空气 + biome + for _ in range(23): + w.write((0).to_bytes(2, "big")); w.write(bytes([0])) + w.varint(0); w.varint(0) + w.write(bytes([0])); w.varint(0); w.varint(0) + + ch = decode_chunk(0, 0, w.getvalue()) + # 末尾 section 属半写状态，取已完整解出的 section 0（y=-64..-49） + sid = ch.state_id_at(0, -64, 0) + assert sid == 99999, f"state id 解码错: {sid}" + + r = reg() + got = r.name_or_unknown(sid) + assert got.startswith("unknown("), f"越界 id 竟解析出了名字: {got}" + assert r.is_out_of_range(sid), "越界判定失效" + # 对照：mineflayer 的行为是静默错配 —— 我们明确不同 + assert r.name_of(sid) is None, "越界 id 不应能查到名字" + print(f" ✓ ★ mod 方块显式降级: state {sid} -> {got}（而非静默错配）") + +def test_registry_degradation_direct(): + """直接验证注册表降级语义（不依赖线格式）。""" + r = reg() + assert r.name_or_unknown(0) == "air" + assert r.name_or_unknown(24134) is not None # 原版上界内 + assert r.name_or_unknown(24135).startswith("unknown(") # 超出上界 + assert r.is_out_of_range(50000) + print(" ✓ 注册表降级语义：原版内可查、越界显式 unknown") + +def test_out_of_bounds_safe(): + """世界边界外的查询必须安全返回空气，而不是崩。""" + from mcfold.protodef import Writer + w = Writer() + for _ in range(NUM_SECTIONS): + w.write((0).to_bytes(2, "big")); w.write(bytes([0])) + w.varint(0); w.varint(0) # 方块 section（单值空气） + w.write(bytes([0])); w.varint(0); w.varint(0) # biome section + ch = decode_chunk(0, 0, w.getvalue()) + assert ch.state_id_at(0, -1000, 0) == 0 + assert ch.state_id_at(0, 1000, 0) == 0 + assert ch.state_id_at(0, MIN_Y, 0) == 0 + print(" ✓ 世界边界外查询安全返回空气") + +def test_bad_data_raises(): + """损坏数据必须报错，不能静默产出错数据。""" + try: + decode_chunk(0,0,b"\x00\x01\xff") # bitsPerBlock=255 + assert False, "应当报错" + except (ChunkError, Exception) as e: + print(f" ✓ 损坏数据显式报错: {str(e)[:45]}") + +if __name__ == "__main__": + for fn in [test_bitarray, test_real_chunk_from_prismarine, test_all_encoding_branches, + test_mod_block_explicit_unknown, test_registry_degradation_direct, + test_out_of_bounds_safe, test_bad_data_raises]: + fn() + print("\nchunk 解析：全部通过") diff --git a/tests/test_e2e_interop.py b/tests/test_e2e_interop.py index bade923..f06ae56 100644 --- a/tests/test_e2e_interop.py +++ b/tests/test_e2e_interop.py @@ -19,9 +19,39 @@ def reachable(): except OSError: return False +def _start_server(): + """自动起 Node 服务端（独立实现），让本测试**总能真跑**而不是跳过。""" + import subprocess, time, shutil + runtime = "/home/newqqagent/plugin_data/mc/runtime" + script = os.path.join(os.path.dirname(os.path.abspath(__file__)), "support_testserver.js") + if not shutil.which("node"): + return None + shutil.copy(script, os.path.join(runtime, "support_testserver.js")) + p = subprocess.Popen(["node", "support_testserver.js", str(PORT)], cwd=runtime, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) + for _ in range(50): + if reachable(): return p + if p.poll() is not None: return None + time.sleep(0.2) + return None + def test_interop(): + proc = None if not reachable(): - print(f" - 服务端 {HOST}:{PORT} 未运行，跳过（见本文件顶部用法）"); return + proc = _start_server() + if not reachable(): + print(f" - 服务端 {HOST}:{PORT} 不可用，跳过"); return + try: + _do_interop() + finally: + if proc: + proc.terminate() + try: proc.wait(timeout=5) + except Exception: proc.kill() + try: os.remove("/home/newqqagent/plugin_data/mc/runtime/support_testserver.js") + except OSError: pass + +def _do_interop(): c = Client(HOST, PORT, "PyProbe", "1.20.1") try: r = c.run_smoke()