266 lines
9.7 KiB
Python
266 lines
9.7 KiB
Python
"""区块解析：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
|