#!/usr/bin/python3 """Build-tree-only Rime proof of concept, isolated from the host desktop.""" import argparse import json import os from pathlib import Path import shutil import subprocess import time ROOT = Path(__file__).resolve().parents[1] DATA = ROOT / ".artifacts/rime-poc-20261028" RUNTIME = Path(f"/run/user/{os.getuid()}/ferese-rime-poc") MAIN = "ferese-rime-poc-20261028.service" IME = "ferese-rime-native-20261028.service" APP = "ferese-rime-entry-20261028.service" STRATA = "ferese-rime-strata-20261028.service" UNITS = [APP, STRATA, IME, MAIN] def run(args, *, env=None, check=True): p = subprocess.run(args, env=env, text=True, capture_output=True) if check and p.returncode: raise RuntimeError(p.stderr.strip() or p.stdout.strip() or f"command failed: {args[0]}") return p.stdout.strip() def prop(unit, field): return run(["systemctl", "--user", "show", unit, f"--property={field}", "--value"], check=False) def start_unit(unit, command, env, limit): args = ["systemd-run", "--user", "--collect", f"--unit={unit}", f"--service-type={'notify' if unit == IME else 'exec'}", f"--working-directory={ROOT}", f"--property=MemoryMax={limit}", "--property=TasksMax=512"] if unit != MAIN: args += [f"--property=BindsTo={MAIN}", f"--property=After={MAIN}"] if unit == IME: args += ["--property=Restart=on-failure", "--property=RestartSec=1", "--property=StartLimitIntervalSec=60", "--property=StartLimitBurst=5", "--property=WatchdogSec=10", "--property=WatchdogSignal=SIGKILL", "--property=TimeoutStopSec=3", "--property=TimeoutStartSec=45"] run(args + [f"--setenv={key}={value}" for key, value in env.items()] + list(command)) def wait_for(fn, message, timeout=20): until = time.monotonic() + timeout while time.monotonic() < until: value = fn() if value: return value time.sleep(.1) raise RuntimeError(message) def context(): group = prop(MAIN, "ControlGroup") if not group: return None try: pids = (Path("/sys/fs/cgroup") / group.lstrip("/") / "cgroup.procs").read_text().split() except OSError: return None for pid in pids: try: proc = Path("/proc") / pid command = (proc / "cmdline").read_bytes().split(b"\0")[0].decode() if Path(command).name != "ferese-shell": continue env = dict(s.split("=", 1) for s in (proc / "environ").read_text().split("\0") if "=" in s) if env.get("FERESE_SOCKET") and env.get("FERESE_PUBLIC_WAYLAND_DISPLAY"): return {"socket": env["FERESE_SOCKET"], "wayland": env["FERESE_PUBLIC_WAYLAND_DISPLAY"], "bus": env["DBUS_SESSION_BUS_ADDRESS"], "shell_pid": int(pid)} except (OSError, ValueError): pass return None def environment(): values = {"XDG_RUNTIME_DIR": str(RUNTIME), "XDG_CURRENT_DESKTOP": "Ferese", "XDG_SESSION_DESKTOP": "ferese", "XDG_SESSION_TYPE": "wayland", "GDK_BACKEND": "wayland", "GTK_IM_MODULE": "wayland", "QT_IM_MODULE": "wayland", "QT_QPA_PLATFORM": "wayland", "NO_AT_BRIDGE": "1", "GTK_A11Y": "none", "GSETTINGS_BACKEND": "keyfile", "IBUS_ADDRESS": "", "FERESE_SESSION_IMPORT_ENV": "0", "FERESE_ENABLE_INPUT_METHOD": "1", "FERESE_RIME_CONTROL": str(RUNTIME / "rime-control.sock"), "FERESE_ENABLE_TEST_POINTER": "1", "FERESE_ENABLE_SCREENCOPY": "1", "XDG_DATA_DIRS": f"{Path.home()}/.local/share:/usr/local/share:/usr/share"} values.update({f"XDG_{kind.upper()}_HOME": str(DATA / "profile" / kind) for kind in ("config", "data", "cache", "state")}) return values def child_environment(): c = context() if not c: raise RuntimeError("Rime development preview is not running") return {**environment(), "WAYLAND_DISPLAY": c["wayland"], "DBUS_SESSION_BUS_ADDRESS": c["bus"], "FERESE_SOCKET": c["socket"]} def windows(): c = context() return json.loads(run([str(ROOT / "target/debug/feresectl"), "--socket", c["socket"], "--json", "windows"])) def focus_entry(): wait_for(lambda: any(w["app_id"] == "dev.ferese.RimeValidation" for w in windows()), "Validation window did not map") if not any(w["focused"] and w["app_id"] == "dev.ferese.RimeValidation" for w in windows()): c = context() entry=next(w for w in windows() if w["app_id"]=="dev.ferese.RimeValidation") run([str(ROOT / "target/debug/feresectl"), "--socket", c["socket"], "focus-window", str(entry["id"])]) wait_for(lambda: any(w["focused"] and w["app_id"] == "dev.ferese.RimeValidation" for w in windows()), "Validation window did not focus") def start_ime(): if prop(IME, "ActiveState") == "active": return start_unit(IME, [str(ROOT / "target/debug/ferese-rime"), "--user-data", str(DATA / "rime-user")], child_environment(), "768M") pid = prop(IME, "MainPID") wait_for(lambda: '"event":"ready"' in run( ["journalctl", "--user", f"_PID={pid}", "--no-pager", "-o", "cat"], check=False), "Native Rime did not become ready", timeout=40) def start(): if prop(MAIN, "ActiveState") == "active": print("开发预览已在运行。") return DATA.mkdir(mode=0o700,parents=True,exist_ok=True) DATA.chmod(0o700) if not (ROOT / "target/debug/ferese-rime").exists(): raise RuntimeError("Build ferese, feresectl and ferese-rime first") for kind in ("config/ferese", "data", "cache", "state"): (DATA / "profile" / kind).mkdir(parents=True, exist_ok=True) RUNTIME.mkdir(mode=0o700, exist_ok=True) if RUNTIME.stat().st_uid != os.getuid() or RUNTIME.stat().st_mode & 0o077: raise RuntimeError("Unexpected private runtime owner or permissions") (DATA / "profile/config/ferese/config.kdl").write_text(''' xwayland { enabled #false; } status { keybinding-guide #false; } window-rule app-id="dev.ferese.RimeValidation" floating=#true width=880 height=560 window-rule app-id="io.github.lgse.Strata" floating=#true width=1050 height=720 window-rule app-id="QQ" floating=#true width=1000 height=720 window-rule app-id="teams-for-linux" floating=#true width=1000 height=720 theme { typography { font-family "MiSans"; }; } ''') host_runtime = Path(os.environ.get("XDG_RUNTIME_DIR", f"/run/user/{os.getuid()}")) host_display = Path(os.environ.get("WAYLAND_DISPLAY", "wayland-0")) if not host_display.is_absolute(): host_display = host_runtime / host_display if not host_display.exists(): raise RuntimeError("Parent Wayland display is missing") env = {**environment(), "WAYLAND_DISPLAY": str(host_display), "RUST_LOG": "info"} start_unit(MAIN, ["/usr/bin/dbus-run-session", "--", str(ROOT / "target/debug/ferese"), "--backend", "nested", "--supervise-client", "--grant-effects", "--grant-shell-control", "--", "/usr/bin/ferese-shell"], env, "3G") c = wait_for(context, "Development compositor did not publish its private connection") (DATA / "context.json").write_text(json.dumps(c, indent=2) + "\n") start_ime() fixture = DATA / "fixtures" fixture.mkdir(exist_ok=True) (fixture / "中文测试.txt").write_text("Ferese Rime 独立测试文件。\n") strata = shutil.which("strata") or str(Path.home() / ".local/bin/strata") if Path(strata).exists(): start_unit(STRATA, [strata, str(fixture)], child_environment(), "2G") wait_for(lambda: any(w["app_id"] == "io.github.lgse.Strata" for w in windows()), "Strata did not map") start_unit(APP, ["/usr/bin/gjs", str(ROOT / "scripts/tests/fixtures/rime-entry.js"), str(DATA / "entry-state.json")], child_environment(), "512M") focus_entry() print("Ferese 原生 Rime 开发预览已打开。") def stop(): loaded = [u for u in UNITS if prop(u, "LoadState") not in ("", "not-found")] if loaded: run(["systemctl", "--user", "stop", *loaded]) run(["systemctl", "--user", "reset-failed", *loaded], check=False) print("Rime 开发预览已停止。") def action(name): focus_entry() env = child_environment() return run(["gdbus", "call", "--address", env["DBUS_SESSION_BUS_ADDRESS"], "--dest", "dev.ferese.RimeValidation", "--object-path", "/dev/ferese/RimeValidation", "--method", "org.gtk.Actions.Activate", name, "[]", "{}"]) def probe(arguments): env = {**os.environ, **child_environment(), "FERESE_TEST_INPUT": "1"} print(run([str(ROOT / "target/debug/ferese-rime-probe"), *arguments], env=env)) if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("action", choices=["start", "stop", "restart", "status", "ime", "probe", "entry"]) args, rest = parser.parse_known_args() if args.action == "restart": stop(); start() elif args.action == "status": print(json.dumps({u: prop(u, "ActiveState") for u in UNITS}, indent=2)) elif args.action == "entry": print(action(rest[0])) elif args.action == "probe": probe(rest) elif args.action == "ime": start_ime() else: globals()[args.action]()