"""区块解析：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