"""渲染桥：把 Python 侧的世界数据交给 Java 客户端渲染器出图。 设计要点 -------- 两侧格式**同源**，不需要转换： Python client.run_smoke()/read_packet() 收到的 packet_map_chunk 载荷， 其 chunkData 部分与 Java 侧 LevelChunkSection.read(FriendlyByteBuf) 期待的是同一套字节（nonEmptyBlockCount + PalettedContainer×N + Biomes）。 所以桥的协议是「把 chunk 原始字节交给 Java」，而不是「把方块逐个传过去」—— 既避免重建世界的语义损失（mod 方块、NBT、biome、block entity）， 又天然支持任意版本（格式由同一个 protocol.json 描述）。 通信 ---- 子进程 + 行分隔 JSON（NDJSON）： Java 侧 stdout 每行一个 JSON 对象；Python 写 stdin 用同样格式。 选 NDJSON 而非 stdin 二进制：可读、可日志、便于调试； chunk 字节量不大（单区块约 10-100KB），base64 开销可接受。 """ from __future__ import annotations import base64 import json import os import shutil import subprocess import threading import time from dataclasses import dataclass, field from pathlib import Path from typing import Any class RenderBridgeError(RuntimeError): """桥不可用或渲染失败。""" @dataclass class RenderResult: """一次渲染的结果。""" image: bytes | None = None # PNG 字节 path: str | None = None width: int = 0 height: int = 0 chunks_rendered: int = 0 # countRenderedChunks distinct_colors: int = 0 # 判据：区分「世界」与「加载屏/空屏」 camera: tuple[float, float, float] = (0.0, 0.0, 0.0) warnings: list[str] = field(default_factory=list) raw: dict[str, Any] = field(default_factory=dict) @property def has_world_content(self) -> bool: """distinct_colors 是实测有效的判据： 空屏/纯色 1-2 色，加载屏 ~199（logo 抗锯齿），真实世界 >1000。""" return self.distinct_colors > 400 and self.chunks_rendered > 0 class RenderBridge: """驱动 Java 渲染器子进程。 生命周期：start() → feed_chunk()×N → set_camera() → render() → stop()。 渲染器是**常驻进程**：起一次约 3-5s（GL + 资源重载）， 之后每次渲染只需几十到几百毫秒。 """ def __init__( self, java: str = "java", classpath: str | list[str] | None = None, main_class: str = "McfoldRenderServer", game_dir: str | None = None, assets: str | None = None, asset_index: str | None = None, java_args: list[str] | None = None, startup_timeout: float = 180.0, cwd: str | None = None, ): self.java = java self.main_class = main_class self.game_dir = game_dir self.assets = assets self.asset_index = asset_index self.startup_timeout = startup_timeout self.cwd = cwd if classpath is None: classpath = os.environ.get("MCFOLD_RENDER_CLASSPATH") if isinstance(classpath, (list, tuple)): classpath = os.pathsep.join(classpath) self.classpath = classpath self.java_args = java_args or [ "-Xmx1200M", "-Xms512M", # headless：不要 AWT 窗口；GL 仍走 GLFW + Xvfb "-Djava.awt.headless=true", ] self._proc: subprocess.Popen | None = None self._next_id = 1 self._pending: dict[int, tuple[threading.Event, dict]] = {} self._lock = threading.Lock() self._reader: threading.Thread | None = None self._stderr_tail: list[str] = [] # ---------------- 生命周期 ---------------- @property def running(self) -> bool: return self._proc is not None and self._proc.poll() is None def start(self) -> "RenderBridge": if self.running: return self if not self.classpath: raise RenderBridgeError( "未提供 classpath。设置 MCFOLD_RENDER_CLASSPATH 或传 classpath=。" "构建：mcfold/scaffold/build.sh " ) cmd = [self.java, *self.java_args, "-cp", self.classpath, self.main_class] if self.game_dir: cmd += ["--gameDir", self.game_dir] if self.assets: cmd += ["--assets", self.assets] if self.asset_index: cmd += ["--assetIndex", self.asset_index] env = dict(os.environ) # natives 目录（lwjgl .so）：构建脚本解到 MCFOLD_NATIVES_DIR nat = env.get("MCFOLD_NATIVES_DIR") if nat: env["LD_LIBRARY_PATH"] = nat + os.pathsep + env.get("LD_LIBRARY_PATH", "") try: self._proc = subprocess.Popen( cmd, cwd=self.cwd, env=env, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=False, bufsize=0, ) except FileNotFoundError as e: raise RenderBridgeError(f"无法启动 java: {e}") from e self._reader = threading.Thread(target=self._read_loop, daemon=True) self._reader.start() threading.Thread(target=self._stderr_loop, daemon=True).start() ready = self._await_event("ready", self.startup_timeout) if not ready.pop("_ok", False): self.stop() raise RenderBridgeError( "渲染器启动失败：" + (ready.get("error") or "超时") + ("\n" + "".join(self._stderr_tail[-5:]) if self._stderr_tail else "") ) return self def stop(self, timeout: float = 10.0) -> None: p = self._proc if p is None: return try: if p.poll() is None: self._send_raw({"cmd": "shutdown"}) try: p.wait(timeout=timeout) except subprocess.TimeoutExpired: p.terminate() try: p.wait(timeout=5) except subprocess.TimeoutExpired: p.kill() finally: self._proc = None def __enter__(self) -> "RenderBridge": return self.start() def __exit__(self, *exc) -> None: self.stop() # ---------------- 数据注入 ---------------- def feed_chunk(self, cx: int, cz: int, chunk_data: bytes, heightmaps_nbt: bytes | None = None) -> dict: """注入一个区块。 chunk_data —— packet_map_chunk 载荷里「长度前缀之后」的那段， 与 Java LevelChunkSection.read() 直接对应。 调 client 侧：decode 后的原始 payload 切片。 heightmaps_nbt —— 可选 NBT（Java 侧允许空）。 """ return self._request({ "cmd": "chunk", "cx": cx, "cz": cz, "data": base64.b64encode(chunk_data).decode("ascii"), "heightmaps": base64.b64encode(heightmaps_nbt).decode("ascii") if heightmaps_nbt else None, }) def feed_chunks(self, chunks: list[tuple[int, int, bytes]]) -> dict: """批量注入（减少往返）。""" return self._request({ "cmd": "chunks", "items": [ {"cx": cx, "cz": cz, "data": base64.b64encode(d).decode("ascii")} for cx, cz, d in chunks ], }) def set_camera(self, x: float, y: float, z: float, yaw: float = 0.0, pitch: float = 0.0) -> dict: """设置观察者位置与朝向（相机 = 我们想看哪里，不一定是玩家）。""" return self._request({"cmd": "camera", "x": x, "y": y, "z": z, "yaw": yaw, "pitch": pitch}) def set_view(self, *, camera_type: str = "THIRD_PERSON_BACK", render_distance: int | None = None, fov: float | None = None) -> dict: req: dict[str, Any] = {"cmd": "view", "cameraType": camera_type} if render_distance is not None: req["renderDistance"] = render_distance if fov is not None: req["fov"] = fov return self._request(req) # ---------------- 渲染 ---------------- def render(self, out: str | None = None, frames: int = 60, width: int | None = None, height: int | None = None) -> RenderResult: """渲若干帧并取回 PNG。 frames 不能太小：区块网格重建队列靠 tick 推进， 实测 20 帧时 countRenderedChunks 已到位，60 帧更稳。 """ req: dict[str, Any] = {"cmd": "render", "frames": frames} if out: req["out"] = out if width: req["width"] = width if height: req["height"] = height resp = self._request(req, timeout=300.0) img_b64 = resp.pop("image", None) res = RenderResult( image=base64.b64decode(img_b64) if img_b64 else None, path=resp.get("path"), width=resp.get("width", 0), height=resp.get("height", 0), chunks_rendered=resp.get("chunksRendered", 0), distinct_colors=resp.get("distinctColors", 0), camera=tuple(resp.get("camera", (0.0, 0.0, 0.0))), warnings=resp.get("warnings", []), raw=resp, ) return res def registry_info(self) -> dict: """客户端方块注册表：state id ↔ 方块名（含 mod 方块的完整状态）。""" return self._request({"cmd": "registry"}) def lookup_state(self, state_id: int) -> dict: """查单个 state id（返回方块名 + 完整属性，如 instrument/note/powered）。""" return self._request({"cmd": "lookup", "stateId": state_id}) # ---------------- 内部：NDJSON 通道 ---------------- def _request(self, req: dict, timeout: float = 120.0) -> dict: if not self.running: raise RenderBridgeError("渲染器未运行（先 start()）") with self._lock: rid = self._next_id self._next_id += 1 req["id"] = rid ev = threading.Event() self._pending[rid] = (ev, {}) self._send_raw(req) if not ev.wait(timeout): self._pending.pop(rid, None) raise RenderBridgeError(f"渲染器无响应（{timeout}s）：cmd={req.get('cmd')}") _, resp = self._pending.pop(rid) if not resp.get("_ok", True): raise RenderBridgeError(resp.get("error", "未知错误")) return resp def _await_event(self, kind: str, timeout: float) -> dict: """等待一个主动事件（如 ready）。""" deadline = time.time() + timeout while time.time() < deadline: if not self.running: return {"_ok": False, "error": f"进程退出（code={self._proc.poll() if self._proc else '?'}）"} for rid, (ev, box) in list(self._pending.items()): if box.get("_event") == kind and ev.is_set(): self._pending.pop(rid, None) box["_ok"] = True return box time.sleep(0.05) return {"_ok": False, "error": f"等待 {kind} 超时"} def _send_raw(self, obj: dict) -> None: if not self._proc or self._proc.stdin is None: raise RenderBridgeError("stdin 不可用") self._proc.stdin.write((json.dumps(obj) + "\n").encode("utf-8")) self._proc.stdin.flush() def _read_loop(self) -> None: assert self._proc and self._proc.stdout for raw in self._proc.stdout: line = raw.decode("utf-8", "replace").strip() if not line.startswith("{"): continue try: msg = json.loads(line) except json.JSONDecodeError: continue rid = msg.get("id") if rid is None and msg.get("event") == "ready": # ready 事件：塞进 _pending 供 _await_event 取 ev = threading.Event() box = dict(msg); box["_event"] = "ready" self._pending[-1] = (ev, box) ev.set() continue if rid in self._pending: ev, box = self._pending[rid] box.update(msg) ev.set() def _stderr_loop(self) -> None: assert self._proc and self._proc.stderr for raw in self._proc.stderr: line = raw.decode("utf-8", "replace").rstrip() self._stderr_tail.append(line) if len(self._stderr_tail) > 200: del self._stderr_tail[:100] # ---------------- 便利 ---------------- @staticmethod def discover_defaults(store: str | Path) -> dict: """从 store 目录推出 assets/assetIndex/gameDir 的默认值（照官方启动器布局）。""" store = Path(store) assets = store / "assets" idx_dir = assets / "indexes" asset_index = None if idx_dir.is_dir(): names = sorted(p.stem for p in idx_dir.glob("*.json")) asset_index = names[-1] if names else None return { "assets": str(assets), "asset_index": asset_index, "game_dir": str(store / "instances" / "default"), }