120 lines
4.6 KiB
Python
120 lines
4.6 KiB
Python
"""世界模型：聚合区块，按世界坐标查方块。
|
||
|
||
对应 mineflayer 侧的 prismarine-world（811 行）。
|
||
职责：
|
||
* 以 (chunkX, chunkZ) 为键存区块列
|
||
* 世界坐标 (x, y, z) → 方块名（跨区块自动定位）
|
||
* 未加载区域**显式**返回 unloaded，而不是假装是空气
|
||
|
||
★ 设计要点（与 mineflayer 的关键区别）：
|
||
mineflayer 对越界 state id 会**静默**映射成别的方块；
|
||
本实现返回 `unknown(<id>)`，并把该 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
|