Files
agentmc/mcfold/backends/base.py
JianFeeeee 51ee57e556 实例创建接口(Python 层)+ 后端透传新参数
## ① mcfold/instances.py —— Python 提供「建实例」接口

用户定向(2026-09-26):
> 这个文件夹内容也是 python 层创建的,而不是标准游戏目录。
> 即 python 提供创建实例的接口(含 mod/插件的实例,内部有对应目录,
> 插件 mod 放在目录中就会被加载)。

所以实例是**我们定义的目录**,不是一份 MC 客户端安装:

    create_instance(name, version=, loader=, mods=[], force=False)  → Path
    list_instances()                                    → [实例摘要]
    find_content_root(version, loader)                  → 共享只读那层

- 实例内部:`instance.json`(自述)+ **mods/ shaderpacks/ resourcepacks/
  datapacks/**(★ agent 的接口面,放进去即生效)+ config/ saves/ logs/
  options.txt(MC/加载器自己写的运行时产物)
- **不在实例里建 versions/libraries/assets** —— 那些属于内容根(只读共享)。
  放进实例会让「同一版本多实例」变成多份副本,也把「实例私有/版本公有」
  的边界糊掉。
- 实例名白名单 `^[A-Za-z0-9._-]{1,64}$`(它要进路径)
- 已存在时**报错而非默默复用**,并提示「换名字 = 装不同 mod 的方式」

★ 这与已有的三层口径一致(mcview/lib/versions.cjs:59/61):
  store/<loader>-<version>/ · instances/<name>/ · runtime/<name>/

## ② base.py 三个修正

**(a) cwd 缺省改为 game_dir** —— 版本隔离漏洞:
    渲染进程的 cwd 原是仓库根,而 Sodium 等会往 cwd 写 config/
    ⇒ 实测 config/sodium-mixins.properties 落进了 mcfold 仓库根。
    这些产物属于**实例**。已改为 `cwd=self.cwd or self.game_dir`。

**(b) 命名空间按 client jar 形态自动判** —— 原先硬编码 `named`:
    官方 obf jar + 加载器 ⇒ intermediary(mod 的 mixin 能解析)
    client-named.jar      ⇒ named(仅裸启动可用)
    实测:named 下 42 个 mod 的 464 处 mixin 全部找不到目标。

**(c) 新增 names_file 参数** —— 把 names.json 路径透给后端
    (B 方案的 extern 参数在 Python 侧的可入口)。
2026-09-26 10:18:26 +08:00

511 lines
20 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""渲染后端基类 —— 世代无关的协议与生命周期。
设计(用户 2026-09-25 定向):
Python 层做 switch 选择后端;每个世代一个**独立编译**的 Java 服务端。
原因:Java 是编译期绑定的,世代间的签名差异(DeltaTracker 等)无法写在同一份源码里。
本基类承载所有世代共用的部分:
- NDJSON 通道(子进程 + 行分隔 JSON)
- 命令集(chunk / camera / resolve / render / registry / debug / view)
- base64 载荷编解码
子类只需声明世代特征(Java 目录、判据类、API 差异说明)。
"""
from __future__ import annotations
import base64
import json
import os
import subprocess
import threading
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
def rle_encode(values) -> list[list[int]]:
"""行程编码:[[值, 重复次数], ...]。
★ 这是**格式无关**的压缩 —— 只是把连续相同的值合并成一段,
不含任何 MC 协议知识(调色板/位宽/字节序都不在这里)。
真正的协议组装由 Java 侧用官方 API 完成。
"""
out: list[list[int]] = []
for v in values:
iv = int(v)
if out and out[-1][0] == iv:
out[-1][1] += 1
else:
out.append([iv, 1])
return out
def rle_decode(rle, expected: int | None = None) -> list[int]:
"""rle_encode 的逆(测试与诊断用)。"""
out: list[int] = []
for v, n in rle:
out.extend([int(v)] * int(n))
if expected is not None and len(out) != expected:
raise ValueError(f"RLE 长度 {len(out)} != 期望 {expected}")
return out
class BackendError(RuntimeError):
"""后端不可用或渲染失败。"""
@dataclass
class RenderResult:
"""一次渲染的结果。
distinct_colors 是本项目实测最锐利的判据:
空屏/纯色 1-2 色 / 加载屏 ~199 / 简单地形 ~1450 / 真实存档 >10000
(non-black 会被红底加载屏骗过 —— 实测踩过)
"""
image: bytes | None = None
path: str | None = None
width: int = 0
height: int = 0
chunks_rendered: int = 0
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:
return self.distinct_colors > 400 and self.chunks_rendered > 0
class RenderBackend:
"""所有世代后端共用的实现。子类覆写类属性即可。"""
generation: str = "base"
min_mc_version: str = ""
max_mc_version: str = ""
java_dir: str = ""
markers: tuple[str, ...] = ()
main_class: str = "McfoldRenderServer"
# 启动架构(与世代正交):"" = 裸启动(我们自己的 main)
# "fabric" = 经 Fabric 加载器启动(用 mod 与 Iris,但不写我们的 mod)
launcher: str = ""
def __init__(
self,
java: str = "java",
classpath: str | list[str] | None = None,
game_dir: str | None = None,
assets: str | None = None,
asset_index: str | None = None,
java_args: list[str] | None = None,
startup_timeout: float = 240.0,
cwd: str | None = None,
mods_dir: str | None = None,
game_mapping_namespace: str = "",
names_file: str | None = None,
):
self.java = java
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.game_dir = game_dir
self.assets = assets
self.asset_index = asset_index
self.startup_timeout = startup_timeout
self.cwd = cwd
# 加载器后端的 mod 目录(launcher="fabric" 时经 -Dfabric.modsFolder 传入)
self.mods_dir = mods_dir
# 命名空间:"" = 按 client jar 自动判;"named" / "intermediary" 可显式指定
self.game_mapping_namespace = game_mapping_namespace
# extern 参数:反射名的次表(McfoldNames 读的 names.json)
self.names_file = names_file
self.java_args = java_args or [
"-Xmx1200M", "-Xms512M", "-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._stderr_tail: list[str] = []
# ---------------- 启动架构(加载器) ----------------
def _client_jar_is_official(self) -> bool:
"""classpath 里的是**官方 obf jar**(而非 client-named.jar)吗。
它决定运行时命名空间:
官方 obf jar + Fabric ⇒ intermediary(mod 的 mixin 能命中)
client-named.jar ⇒ named(仅裸启动可用)
"""
if not self.classpath:
return False
for p in self.classpath.split(os.pathsep):
if "client-named" in p:
return False
return True
def _fabric_classpath(self, cp: str) -> str:
"""把普通 classpath 变成 **Fabric 加载器** 可用的那条。
★ 两个变换都是实测必需,且都是**用户不该自己拼**的机械操作:
① 剔除 1.20.1 自带的 asm(9.2 与 9.6 **两版并存**)
否则 Knot.<clinit> 抛
"duplicate ASM classes found on classpath"
(Fabric 自带 asm 9.10.1,必须只留它一份)
② 追加 Fabric 的 8 个库(asm ×5 / sponge-mixin / intermediary / fabric-loader)
库来源:与 client jar 同级的 `.mcfold/fabric/`(内容根,随版本隔离)。
找不到就**原样返回**(让用户自己拼,并会在 Knot 报错里看到原因)。
"""
parts = [p for p in cp.split(os.pathsep) if p]
# ① 去 asm
kept = [p for p in parts if "/org/ow2/asm/" not in p.replace("\\", "/")]
# ② 找 Fabric 库目录
fab_dir = None
for p in kept:
cand = Path(p).parent / "fabric"
if cand.is_dir():
fab_dir = cand
break
if fab_dir is None:
return os.pathsep.join(kept)
fab = sorted(str(j) for j in fab_dir.glob("*.jar"))
return os.pathsep.join(kept + fab)
# ---------------- 生命周期 ----------------
@property
def running(self) -> bool:
return self._proc is not None and self._proc.poll() is None
def start(self) -> "RenderBackend":
if self.running:
return self
if not self.classpath:
raise BackendError(
f"未提供 classpath。设置 MCFOLD_RENDER_CLASSPATH,"
f"或先跑 scaffold/build.sh 编译 {self.generation} 后端。"
)
java_args = list(self.java_args)
# ★ 启动架构:走 Fabric 加载器时,主类是 launcher_fabric.LoaderBoot,
# 并需两个系统属性(实测缺一不可):
# fabric.gameMappingNamespace=named → 游戏类名保持 official 名
# fabric.modsFolder=<dir> → mod 目录
main_class = self.main_class
classpath = self.classpath
if self.launcher == "fabric":
main_class = "LoaderBoot"
# ★ classpath 组装由后端负责(已实测必需的两步:去 asm + 加 Fabric 8 库)
classpath = self._fabric_classpath(classpath)
# ★ 命名空间按 client jar 形态定:
# intermediary(官方 obf jar + 加载器)⇒ mod 的 mixin 能解析
# named (client-named.jar) ⇒ 仅裸启动可用
# 实测:named 下 42 个 mod 的 mixin 全部找不到目标(464 处)。
if not self.game_mapping_namespace:
self.game_mapping_namespace = (
"intermediary" if self._client_jar_is_official() else "named")
java_args.append(f"-Dfabric.gameMappingNamespace={self.game_mapping_namespace}")
# ★ extern 参数:反射名次表路径(B 方案)
if self.names_file:
java_args.append(f"-Dmcfold.names={self.names_file}")
if self.mods_dir:
java_args.append(f"-Dfabric.modsFolder={self.mods_dir}")
cmd = [self.java, *java_args, "-cp", classpath, 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)
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 or self.game_dir, env=env,
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
text=False, bufsize=0,
)
except FileNotFoundError as e:
raise BackendError(f"无法启动 java: {e}") from e
threading.Thread(target=self._read_loop, daemon=True).start()
threading.Thread(target=self._stderr_loop, daemon=True).start()
ready = self._await_event("ready", self.startup_timeout)
if not ready.get("_ok"):
self.stop()
tail = "".join(self._stderr_tail[-6:])
raise BackendError(
f"{self.generation} 后端启动失败:{ready.get('error') or '超时'}"
+ (f"\n{tail}" if 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:
try:
self._send_raw({"cmd": "shutdown"})
p.wait(timeout=timeout)
except subprocess.TimeoutExpired:
p.terminate()
try:
p.wait(timeout=5)
except subprocess.TimeoutExpired:
p.kill()
except Exception:
try:
p.kill()
except Exception:
pass
finally:
self._proc = None
def __enter__(self):
return self.start()
def __exit__(self, *exc):
self.stop()
# ---------------- 数据注入 ----------------
def feed_chunk(self, cx: int, cz: int, chunk_data: bytes) -> dict:
return self._request({"cmd": "chunk", "cx": cx, "cz": cz,
"data": self._b64(chunk_data)})
def feed_chunks(self, chunks: list[tuple[int, int, bytes]]) -> dict:
return self._request({
"cmd": "chunks",
"items": [{"cx": cx, "cz": cz, "data": self._b64(d)} for cx, cz, d in chunks],
}, timeout=300.0)
def set_blocks(self, cx: int, cz: int, sections: dict[int, list[int]]) -> dict:
"""直接写方块(Python 只做路由,不碰协议格式)。
sections: {sectionIndex: [4096 个 state id]},索引顺序 (y,z,x) 线性化
(与 MC 的 PalettedContainer 索引一致:x 最快、y 最慢)。
为什么走这条路(用户 2026-09-25 定向):
方块写入、调色板组装、位打包、入 cache 全在 Java 侧用官方 API 完成。
Python 不复制历代变过的协议格式 —— 版本差异留在 Java。
"""
items = [{"index": int(idx), "rle": rle_encode(ids)}
for idx, ids in sorted(sections.items())]
return self._request({"cmd": "setblocks", "cx": cx, "cz": cz,
"sections": items}, timeout=300.0)
def load_mods(self, mods_dir: str) -> dict:
"""加载 mod:Java 侧自己扫 jar、注册占位方块、挂资源包。
★ Python 只传目录路径 —— 版本/格式相关的处理全在 Java 侧
(用户 2026-09-25 定向:Python 相当于路由)。
原理(见 scaffold/MOD-BLOCKS.md):方块**渲染**只需要
(a) 注册表里有这个名字(拿 state id)
(b) 资源包里有 blockstates/models/textures
与方块行为无关 ⇒ 不需要 Fabric/Forge 加载器。
代价(诚实标注):方块**行为**(红石/交互)与第三方光影包(Iris)不支持。
"""
return self._request({"cmd": "mods", "dir": str(mods_dir)}, timeout=600.0)
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._request(req)
# ---------------- 视角与视觉 ----------------
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, **kw) -> dict:
"""视觉设置。各世代支持的项见子类 api_notes()。
世代无关的项:camera_type / render_distance / fov / hide_gui
原版画质(所有世代都有):graphics_mode / ambient_occlusion /
mipmap_levels / gamma / entity_shadows / clouds / particles
后处理链:post_effect(原版 PostChain,非第三方光影包)
★ hide_gui:关掉 HUD(血量/热键栏/教学提示…)。
截图场景下**应该关**:HUD 是噪声,而且它给 distinctColors
掺一大块固定像素(实测污染判据)。
走的是官方选项 `options.hideGui`(= 游戏里 F1),不改渲染代码。
"""
req: dict[str, Any] = {"cmd": "view"}
mapping = {
"camera_type": "cameraType", "render_distance": "renderDistance",
"fov": "fov", "graphics_mode": "graphicsMode",
"ambient_occlusion": "ambientOcclusion", "mipmap_levels": "mipmapLevels",
"gamma": "gamma", "entity_shadows": "entityShadows",
"clouds": "clouds", "particles": "particles",
"post_effect": "postEffect",
"hide_gui": "hideGui",
}
for py, js in mapping.items():
if py in kw and kw[py] is not None:
req[js] = kw[py]
return self._request(req)
# ---------------- 查询 ----------------
def registry_info(self) -> dict:
return self._request({"cmd": "registry"})
def lookup_state(self, state_id: int) -> dict:
return self._request({"cmd": "lookup", "stateId": state_id})
def resolve_blocks(self, specs: list[str]) -> list[int]:
"""方块名(+属性) → state id。名字→id 的权威映射在客户端注册表。"""
if not specs:
return []
return self._request({"cmd": "resolve", "blocks": specs})["ids"]
def get_block_info(self, x: int, y: int, z: int) -> dict:
"""查询某个坐标的方块信息。
★ 由 Java 侧回答(它才有 level + 注册表 + 模型状态),Python 只路由。
返回:block 名(含 mod 方块)、stateId、属性(朝向等)、isAir/isSolid、
光照/硬度、贴图是否缺失(程序化判据)、是否来自 mod。
未加载的区块会明确回 known=false + reason="chunk_not_loaded"
—— 与「空气」语义区分(设计文档附录 D)。
"""
return self._request({"cmd": "blockinfo", "x": int(x), "y": int(y), "z": int(z)})
def debug(self) -> dict:
return self._request({"cmd": "debug"})
# ---------------- 渲染 ----------------
def render(self, out: str | None = None, frames: int = 60,
width: int | None = None, height: int | None = None) -> RenderResult:
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=600.0)
img = resp.pop("image", None)
cam = resp.get("camera") or (0.0, 0.0, 0.0)
if not isinstance(cam, (list, tuple)):
cam = (0.0, 0.0, 0.0)
return RenderResult(
image=base64.b64decode(img) if img 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(float(c) for c in cam[:3]),
warnings=resp.get("warnings", []),
raw=resp,
)
# ---------------- 内部 ----------------
@staticmethod
def _b64(b: bytes) -> str:
return base64.b64encode(b).decode("ascii")
def _request(self, req: dict, timeout: float = 180.0) -> dict:
if not self.running:
raise BackendError("后端未运行(先 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 BackendError(f"后端无响应({timeout}s):cmd={req.get('cmd')}")
_, resp = self._pending.pop(rid)
if not resp.get("_ok", True):
raise BackendError(resp.get("error", "未知错误"))
return resp
def _await_event(self, kind: str, timeout: float) -> dict:
deadline = time.time() + timeout
while time.time() < deadline:
if not self.running:
code = self._proc.poll() if self._proc else "?"
return {"_ok": False, "error": f"进程退出(code={code})"}
box = self._pending.pop(-1, None)
if box is not None:
ev, payload = box
if payload.get("_event") == kind:
return {"_ok": True, **payload}
self._pending[-1] = (ev, payload) # 不是目标事件,放回
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 BackendError("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
if msg.get("event") == "ready":
ev = threading.Event()
box = dict(msg); box["_event"] = "ready"
self._pending[-1] = (ev, box)
ev.set()
continue
rid = msg.get("id")
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) > 300:
del self._stderr_tail[:150]
# ---------------- 元信息 ----------------
@staticmethod
def api_notes() -> dict:
"""本世代的 API 特征(子类覆写)。
★ 静态信息(不依赖实例)⇒ 定义成 staticmethod,
这样才能在**不启动后端**的前提下做世代对照(测试用)。
"""
return {}
def describe(self) -> str:
return (f"{self.generation} [{self.min_mc_version}..{self.max_mc_version}] "
f"java_dir={self.java_dir}")