"""把世界数据渲染成画面 —— mcfold 的对外渲染入口。 分工（与设计文档附录 D 一致）： 本模块（Python） : 世界数据 —— 区块、方块、光照、相机 客户端 jar（Java）: 方块名与 state id 的权威映射 + 画面 为什么 state id 必须由渲染器侧解析： 存档 palette 里是**方块名**（`minecraft:oak_log[axis=y]`）， 而网络载荷要 **state id**。名字→id 的映射随版本/加载器/mod 变化， 只有客户端自带的注册表是权威的（1.20.1 不由网络下发，附录 D）。 用法： from mcfold.render import WorldRenderer with WorldRenderer.open(instance) as wr: wr.load_save("/home/mc-test/mc-test", center=(-1, -1), radius=3) wr.camera(x=8, y=80, z=8, yaw=135, pitch=35) img = wr.screenshot("/tmp/out.png") """ from __future__ import annotations import base64 import os from dataclasses import dataclass, field from pathlib import Path from typing import Any, Iterable from . import anvil from .backends import backend_class, detect_generation from .backends.base import BackendError, RenderBackend, RenderResult # ── 协议层常量（与版本无关，协议本身定义）── SECTION_VOLUME = 16 * 16 * 16 # 一个 section 的方块数 BIOME_VOLUME = 4 * 4 * 4 # 一个 biome section 的值数 # ── 世界尺寸：★ 不硬编码，运行时问 Java ── # 这些量随版本/世界类型变（1.20 起 worldHeight 384、minSection -4； # 更早的版本是 256/-0）。Python 只做路由与协议编解码， # 版本相关的量由后端回答（cmd="capacity"）。 # 下面三个仅在没有后端时的降级默认值。 FALLBACK_SECTION_COUNT = 24 FALLBACK_MIN_SECTION_Y = -4 FALLBACK_DEFAULT_BIOME = "minecraft:plains" def _decode_storage(longs: Iterable[int], volume: int, bits: int) -> list[int]: """int[] 位数组 → 索引列表（不跨 long）。""" longs = list(longs) if bits == 0 or not longs: return [0] * volume vpl = 64 // bits mask = (1 << bits) - 1 out = [] for i in range(volume): li = i // vpl if li >= len(longs): out.append(0); continue off = (i - li * vpl) * bits out.append((longs[li] >> off) & mask) return out @dataclass class SaveChunk: """存档里一个区块的、已解析好的方块与生物群系。""" cx: int cz: int status: str = "" sections: dict[int, dict] = field(default_factory=dict) # y_idx -> {"palette": [名字], "states": [4096 名字], "biomes": [64 名字]} palette_specs: list[str] = field(default_factory=list) # 全部去重后的方块名(+属性)规格 def _props_to_spec(name: str, props: dict | None) -> str: """{'Name':'minecraft:oak_log','Properties':{'axis':'y'}} → 'minecraft:oak_log[axis=y]'""" if not props: return name parts = ",".join(f"{k}={v}" for k, v in sorted(props.items())) return f"{name}[{parts}]" def parse_save_chunk(cx: int, cz: int, nbt: dict) -> SaveChunk: """存档区块 NBT → SaveChunk（方块用**名字规格**表示，state id 留给渲染器解析）。""" ch = SaveChunk(cx=cx, cz=cz, status=str(nbt.get("Status", ""))) specs_seen: set[str] = set() by_y: dict[int, dict] = {} for s in nbt.get("sections", []) or []: sy = s.get("Y") if sy is None: continue bs = s.get("block_states") if not isinstance(bs, dict): continue palette = bs.get("palette") or [] spec_list = [ _props_to_spec(str(p.get("Name", "minecraft:air")), p.get("Properties")) for p in palette ] for sp in spec_list: specs_seen.add(sp) # 存档 data 是「按 long 打包的调色板索引」（无 varint 前缀） storage = bs.get("data") or [] n = len(spec_list) if n <= 1: states = [spec_list[0] if spec_list else "minecraft:air"] * SECTION_VOLUME else: bits = max(4, (n - 1).bit_length()) idxs = _decode_storage(storage, SECTION_VOLUME, bits) states = [spec_list[i] if i < n else "minecraft:air" for i in idxs] # 生物群系：同样结构，4x4x4 bio = s.get("biomes") or {} bio_pal = bio.get("palette") or [] bio_specs = [ (p if isinstance(p, str) else str(p.get("Name", "minecraft:plains"))) for p in bio_pal ] if not bio_specs: biomes = ["minecraft:plains"] * BIOME_VOLUME elif len(bio_specs) == 1: biomes = [bio_specs[0]] * BIOME_VOLUME else: bb = max(1, (len(bio_specs) - 1).bit_length()) bidx = _decode_storage(bio.get("data") or [], BIOME_VOLUME, bb) biomes = [bio_specs[i] if i < len(bio_specs) else "minecraft:plains" for i in bidx] by_y[int(sy)] = {"palette": spec_list, "states": states, "biomes": biomes} ch.sections = by_y ch.palette_specs = sorted(specs_seen) return ch class WorldRenderer: """世界 → 画面。持有一个常驻 Java 渲染器进程。""" def __init__(self, bridge: RenderBackend, instance: Any | None = None): self.bridge = bridge self.instance = instance self._spec_cache: dict[str, int] = {} self._loaded: set[tuple[int, int]] = set() self.chunks: dict[tuple[int, int], SaveChunk] = {} self._capacity: dict[str, Any] | None = None self._biome_id: int | None = None # ---------------- 构造 ---------------- @classmethod def open(cls, instance: Any = None, *, store: str | None = None, classpath: str | None = None, start: bool = True, java: str = "java", client_jar: str | None = None, generation: str | None = None, launcher: str = "", mods_dir: str | None = None) -> "WorldRenderer": """从 instance（mcfold.instance.Instance）或 store 目录打开渲染器。 classpath 缺省取环境变量 MCFOLD_RENDER_CLASSPATH； 可先跑 mcfold/scaffold/build.sh 生成。 """ game_dir = assets = asset_index = None # ★ 版本隔离策略：产物在**内容根**的 .mcfold/ 下（可只读共享）， # 不是 /tmp（tmpfs，重启即丢）。见 scaffold/build.sh。 content_root = None if instance is not None: cr = getattr(instance, "content_root", None) if cr: content_root = Path(cr) elif store: content_root = Path(store) if Path(store).name.startswith(("vanilla-", "forge-", "fabric-", "neoforge-", "quilt-")) else None if content_root is not None and (content_root / ".mcfold").is_dir(): mf = content_root / ".mcfold" if (mf / "classes").is_dir(): # 内容根产物优先（除非调用方显式给了 classpath） if classpath is None: cp_file = mf / "classpath.txt" parts = [] if cp_file.is_file(): parts = [ln.strip() for ln in cp_file.read_text().splitlines() if ln.strip()] parts += [str(mf / "client-named.jar"), str(mf / "classes")] classpath = os.pathsep.join(parts) if (mf / "natives").is_dir(): os.environ.setdefault("MCFOLD_NATIVES_DIR", str(mf / "natives")) if instance is not None: root = getattr(instance, "content_root", None) or getattr(instance, "path", None) if root: root = Path(root) assets = str(root / "assets") idx = root / "assets" / "indexes" if idx.is_dir(): names = sorted(p.stem for p in idx.glob("*.json")) asset_index = names[-1] if names else None game_dir = str(getattr(instance, "path", root) if root else "") if store: store_p = Path(store) assets = str(store_p / "assets") names = sorted(p.stem for p in (store_p / "assets" / "indexes").glob("*.json")) asset_index = names[-1] if names else None # ★ 不默认写生产实例 default（版本隔离策略）。 # 实例目录与内容根**同级**（都挂在 mcview/ 下）， # 绝不嵌套进内容根 —— 内容根要可只读共享（见 MULTI-INSTANCE-DESIGN §1.1）。 # 缺省按内容根名派生一个独立实例，避免撞上任何真实实例。 # 权威布局（mcview/lib/versions.cjs:25）： # BASE（可由 MCVIEW_BASE 覆盖）= .../mcview # contentRoot = BASE/store/- # instanceDir = BASE/instances/ # 所以 store 参数若是内容根，则 BASE = 内容根的 parent.parent。 base = Path(os.environ.get("MCVIEW_BASE") or store_p.parent.parent) game_dir = game_dir or str(base / "instances" / f"exp-{store_p.name}") # ★ Python 层 switch：按 client jar 的世代选后端 # （1.21+ 的 DeltaTracker 在 1.20.x jar 中不存在 ⇒ 必须独立编译的后端） # ★ 变量名不能叫 cls —— 会覆盖 classmethod 的 cls， # 导致下面 cls(bridge, instance) 构造出后端而非 WorldRenderer（实测踩过） backend_cls = backend_class(client_jar=client_jar, generation=generation, launcher=launcher) bridge = backend_cls( java=java, classpath=classpath, game_dir=game_dir, assets=assets, asset_index=asset_index, mods_dir=mods_dir, ) wr = cls(bridge, instance) if start: bridge.start() wr.refresh_capacity() return wr def refresh_capacity(self) -> dict[str, Any]: """向 Java 后端询问世界尺寸（Python 不硬编码版本相关量）。""" try: self._capacity = self.bridge._request({"cmd": "capacity"}) except Exception: self._capacity = None return self._capacity or {} @property def sections_count(self) -> int: if self._capacity and self._capacity.get("sectionsCount"): return int(self._capacity["sectionsCount"]) return FALLBACK_SECTION_COUNT @property def min_section_y(self) -> int: if self._capacity and "minSection" in self._capacity: return int(self._capacity["minSection"]) return FALLBACK_MIN_SECTION_Y @property def _default_biome(self) -> int: """默认生物群系 id —— 运行时向 Java 查（不硬编码 1）。""" if self._biome_id is None: name = (self._capacity or {}).get("defaultBiome") or FALLBACK_DEFAULT_BIOME self._biome_id = self._resolve_biome_id(name) return self._biome_id def _resolve_biome_id(self, name: str) -> int: key = f"__biome__{name}" if key in self._spec_cache: return self._spec_cache[key] try: ids = self.bridge.resolve_blocks([name]) bid = ids[0] if ids else 1 except Exception: bid = 1 self._spec_cache[key] = bid return bid def close(self) -> None: self.bridge.stop() def __enter__(self) -> "WorldRenderer": return self def __exit__(self, *exc) -> None: self.close() # ---------------- 方块名 → state id ---------------- def get_block_info(self, x: int, y: int, z: int) -> dict: """查询坐标处的方块信息（Java 侧回答，Python 只路由）。 含：方块名（mod 方块亦然）、stateId、属性、光照/硬度、 贴图缺失的程序化判据、是否来自 mod；未加载区块明确区分于空气。 """ return self.bridge.get_block_info(x, y, z) def load_mods(self, mods_dir: str | Path | None = None) -> dict: """加载 mod（Java 侧全权处理：扫 jar / 注册方块 / 挂资源包）。 mods_dir 缺省取实例的 mods/ 目录。 ★ 为什么不需要 Fabric/Forge 加载器（见 scaffold/MOD-BLOCKS.md）： 方块**渲染**只需要「注册表里有名字」+「资源包有模型/贴图」， 与方块行为无关。所以零加载器也能让 mod 方块正确显示。 代价（诚实标注）：方块行为（红石/交互）与第三方光影包（Iris）不支持。 """ if mods_dir is None: inst_path = getattr(self.instance, "path", None) if self.instance else None if inst_path: mods_dir = Path(inst_path) / "mods" if mods_dir is None: return {"ok": False, "error": "未提供 mods_dir 且无法从实例推断"} d = Path(mods_dir) if not d.is_dir(): return {"ok": True, "mods": 0, "note": f"{d} 不存在"} r = self.bridge.load_mods(str(d)) # 加载 mod 后方块变了 ⇒ 规格缓存作废（否则仍按旧表解析） self._spec_cache.clear() self._biome_id = None self.refresh_capacity() return r def resolve_specs(self, specs: list[str]) -> dict[str, int]: """批量把方块名规格解析成 state id（渲染器侧权威注册表）。""" todo = [s for s in dict.fromkeys(specs) if s not in self._spec_cache] if todo: resp = self.bridge._request({"cmd": "resolve", "blocks": todo}, timeout=120.0) ids = resp.get("ids", []) for spec, sid in zip(todo, ids): self._spec_cache[spec] = int(sid) return {s: self._spec_cache[s] for s in specs} # ---------------- 装载 ---------------- def load_save(self, save_dir: str | Path, *, center: tuple[int, int] | None = None, radius: int | None = None, dim: str | None = None, only_status: str = "minecraft:full") -> dict: """从存档装载区块（零网络、零登录）。 radius —— 加载半径（区块）。**缺省跟随视距**（这是关键： 视距设了 12 但只加载 3 圈数据，画面看起来就像视距没生效）。 传 None 时用 renderDistance；显式传值则两者取大。 """ if radius is None: radius = max(3, self.render_distance) else: radius = max(radius, 1) save_dir = Path(save_dir) level = anvil.read_level_dat(save_dir) data = level.get("Data", {}) spawn = data.get("SpawnX"), data.get("SpawnY"), data.get("SpawnZ") if center is None: center = ((spawn[0] or 0) >> 4, (spawn[2] or 0) >> 4) regs: dict[tuple[int, int], anvil.RegionFile] = {} # ★ 先设视野中心：ClientChunkCache 只保留视野内的区块（实测否则注入被丢弃） self.set_viewport(center[0], center[1], radius=radius + 2) parsed: list[SaveChunk] = [] skipped = 0 for cx in range(center[0] - radius, center[0] + radius + 1): for cz in range(center[1] - radius, center[1] + radius + 1): rx, rz = cx >> 5, cz >> 5 key = (rx, rz) if key not in regs: p = anvil.region_path(save_dir, dim, rx, rz) if not p.is_file(): skipped += 1; continue regs[key] = anvil.RegionFile(p) raw = regs[key].chunk_nbt(cx, cz) if not raw: skipped += 1; continue nbt = anvil.nbt_read(raw) if only_status and str(nbt.get("Status", "")) != only_status: skipped += 1; continue parsed.append(parse_save_chunk(cx, cz, nbt)) return self.load_chunks(parsed, meta={ "level_name": data.get("LevelName"), "data_version": data.get("DataVersion"), "spawn": spawn, "center": center, "skipped": skipped, }) def load_chunks(self, chunks: list[SaveChunk], meta: dict | None = None) -> dict: """把存档区块交给渲染器。 ★ Python 只做「方块名 → state id」的路由与 RLE 压缩； 调色板组装、位打包、协议序列化、入 cache 全在 Java 侧用官方 API 完成。 这样历代改过的协议格式不需要在 Python 里复制一份。 """ all_specs: list[str] = [] for c in chunks: all_specs.extend(c.palette_specs) id_map = self.resolve_specs(all_specs) unknown = sum(1 for v in id_map.values() if v < 0) n_sections = self.sections_count min_sy = self.min_section_y injected = 0 for c in chunks: sections: dict[int, list[int]] = {} for idx in range(n_sections): sec = c.sections.get(idx + min_sy) if sec is None: continue # 空 section 由 Java 侧留空 sections[idx] = [max(0, id_map.get(sp, -1)) for sp in sec["states"]] if sections: self.bridge.set_blocks(c.cx, c.cz, sections) injected += 1 self.chunks[(c.cx, c.cz)] = c self._loaded.update((c.cx, c.cz) for c in chunks) out = {"injected": injected, "unknownBlocks": unknown, "uniqueSpecs": len(id_map)} if meta: out.update(meta) return out # ---------------- 视角与截图 ---------------- @property def render_distance(self) -> int: """当前视距（区块）。默认 8，可用 set_view(render_distance=) 改。""" return getattr(self, "_render_distance", 8) def set_render_distance(self, rd: int) -> dict: """设置视距（区块）。内部同时设 options + serverRenderDistance + ClientChunkCache 的 storage 半径 —— 三者缺一则视距不生效（实测）。""" self._render_distance = int(rd) return self.bridge.set_view(render_distance=int(rd)) def set_viewport(self, cx: int, cz: int, radius: int | None = None) -> dict: """设定视野中心 —— 必须在注入前设好，否则区块会被 ClientChunkCache 丢弃。""" req: dict[str, Any] = {"cmd": "viewport", "cx": cx, "cz": cz} if radius is not None: req["radius"] = radius return self.bridge._request(req) def set_view(self, **kw) -> dict: """视觉设置（camera_type / render_distance / fov / hide_gui / …）。 ★ 这个方法原先**缺失**，而桥层（`RenderBackend.set_view`）早就实现了 完整的 `view` 命令（含 `hide_gui` → 官方 `options.hideGui`）。 结果 M8 里我调 `renderer.set_view(hide_gui=True)` 时 撞上 AttributeError，又被 `except: pass` 静默吞掉 ⇒ **出图带着 HUD**，而返回值里还写着 `hide_gui: True`（自相矛盾）。 ⇒ 两个教训： 1. 桥层有能力 ≠ 上层有入口（中间这层要一起补）； 2. `except: pass` 掩盖了真实错误 —— 修 HUD 时必须让它**显式失败**， 否则「以为设了」与「实际没设」无法区分。 """ return self.bridge.set_view(**kw) def debug(self) -> dict: """诊断：玩家坐标/区块/网格编译/视野中心的实际状态。""" return self.bridge._request({"cmd": "debug"}) def camera(self, x: float, y: float, z: float, yaw: float = 0.0, pitch: float = 0.0) -> None: self.bridge.set_camera(x, y, z, yaw, pitch) def camera_above(self, cx: float = 0.0, cz: float = 0.0, height: float = 90.0, yaw: float = 0.0, pitch: float = 70.0) -> None: """俯视整片区域（默认视角）。""" self.bridge.set_camera(cx, height, cz, yaw, pitch) def screenshot(self, out: str | None = None, frames: int = 60) -> RenderResult: return self.bridge.render(out=out, frames=frames)