Unreal Python 子进程快捷键监听

前言

  这个常识其实早在去年就已经弄过了，然而当时的实现方案不好。
  虽然我合并到 PyToolkit 里面了，最后还是设置成了不开启状态。
  最近我研究出新方案，可以解决我之前提到的问题。

Python 键盘监听

  以前我有做过一个基于 Python 的启动器，需要使用 C++ 在 Unreal 注册快捷键来启动。 链接
  这么做一方面是需要 C++ 进行编译，另一方面注册的快捷键并不是全局的，只能允许在注册的特定窗口触发 (可能是没有使用合适的 C++ API 导致的)
  然而既然 Unreal 支持 Python ，可以依赖 Python 生态完成键盘的监听。

  最初我是使用了比较出名的 keyboard 库来实现键盘监听。 Github链接
  这个库是纯 Python 实现，而且可以跨平台使用，非常强大。
  使用上也很简单，可以查找官方提供的案例。

import keyboard
keyboard.add_hotkey('ctrl+shift+a', print, args=('triggered', 'hotkey'))
# Block forever, like `while True`.
keyboard.wait()

  上面的代码可以实现全局快捷键触发监听。
  那么似乎直接使用这个 库 到 Unreal Python 下使用即可。
  的确直接在 Unreal 下运行这个库可以用快捷键触发相关的 函数。
  然而，当我不关注 Unreal ，转而在别的窗口打字的时候，会变得非常的卡顿。
  以前不知道为什么，只想着在不关注窗口的时候将 hook 关掉的。

  最近我才领悟到是什么原因导致的。

  主要原因是勾选了这个选项，在 Unreal 切换到后台之后减少 CPU 的消耗。
  导致 Python 运行效率极大降低。
  我查了 keyboard 的源码，背后是通过多线程进行监听，需要持续运行的。
  然而这个由于 Python 的处理效率降低，导致键盘事件的 hook 一直处在等待处理的状态。
  以至于切换到代码编辑器写代码就会异常地卡顿。

  我去年也深入剖析源码研究了这个问题，发现使用 keyboard 库可能不好进行修改。
  于是我改用另一个库 pynput 进行快捷键监听，并且对这个库的函数进行重载，通过 tick 间隔时间来判断是否在卡顿。
  如果卡顿就把 监听 给去掉。

pynput 监听魔改

  pynput 是另一个比较知名的键盘监听库 Github链接
  这个库也有 GlobalHotKeys 的功能实现全局键盘监听。
  其实实现原理和 keyboard 大差不差。

  经过我的研究，这些库别后都是多线程然后有个函数会进入到 while True 死循环进行持续监听。
  我需要在 Unreal 卡顿 的时候去掉监听，然后再 Unreal 不卡顿的时候重新启用监听。
  经过我对源码的研究我找到了 while True 循环的位置，并且进行一些魔改，可以减轻卡顿问题。

def message_itr(self):
    global listener
    assert self._threadid is not None
    try:
        # Pump messages until WM_STOP
        while True:
            if delta_queue.empty():
                continue
            elasped = delta_queue.get()
            # NOTE 如果虚幻阻塞 去掉 键盘监听
            if elasped > 0.1:
                listener and listener.stop()
                listener = None
                break
            
            # print('start capture key ...')
            msg = wintypes.MSG()
            lpmsg = byref(msg)
            # NOTE _PeekMessage 队列不阻塞
            r = self._PeekMessage(lpmsg, None, 0, 0 , 0x0001)
            if r <= 0:
                continue
            elif msg.message == self.WM_STOP:
                break
            else:
                yield msg
    finally:
        self._threadid = None
        self.thread = None


def get_hotkey():
    json_path = posixpath.join(DIR, "hotkey.json")
    key_data = read_json(json_path)
    key_map = {}
    for k,v in key_data.items():
        v = v if isinstance(v, dict) else {"command":v,"type":"COMMAND"}
        command = v.get("command")
        typ = v.get("type","").upper()
        func = HOTKEY_TYPE.get(typ)
        if not func:
            continue

        key_map[k] = func(command)
    return key_map

def __key_listener__(delta_seconds):
    global listener,key_map,hotkey_enabled
    if not hotkey_enabled:
        return
    if delta_queue.empty():
        delta_queue.put(delta_seconds)
    elif listener is None:
        # NOTE 重新构建 键盘 监听
        listener = keyboard.GlobalHotKeys(key_map)
    elif listener not in listener_list:
        del listener_list[:]
        listener_list.append(listener)
        if not listener.is_alive():
            listener.start()
            
tick_handle = unreal.register_slate_pre_tick_callback(__key_listener__)
unreal_app.aboutToQuit.connect(partial(unreal.unregister_slate_post_tick_callback, tick_handle))

delta_queue = Queue(1)
from pynput._util import win32
win32.MessageLoop.__iter__ = message_itr
key_map = get_hotkey()
listener = None
listener_list = []           

  在 Windows 下 win32.MessageLoop 是死循环的元凶。
  于是我魔改获取 tick 的变化时间，如果处于卡顿状态就打断 while 循环。
  这样可以使监听减速，随后在需要开启的状态重新启动新的进程。
  上面的方案只能减轻卡顿，不能从根本上解决这个问题，所以当时研究完了也没有写博客记录，不算是特别好的解决方案。

Python 子进程监听

  直到最近，我发现是 CPU 问题导致卡顿之后，就自然而然地想到使用 Python 的子进程进行监听。
  子进程不受 Unreal 的 CPU 限制影响，就不会导致卡顿了。
  于是我又开始研究进程通信的实现，最后还真的给我捣鼓了一个比较可行的方案。

  关于子进程，在 Python 下可以用过 subprocess.Popen 开启子进程，然后使用 PIPE 或 queue 实现进程通信。
  当然也可以使用 python 插件自带的 remote_execution 通过远程端口进行通信。
  理论上使用第一套方案效率更好，只是局限在单台电脑，使用远程端口的话甚至可以跨电脑调用。

  那么如何实现不阻塞的进程通信呢？ 这里我直接参考了 Stack Overflow 的代码实现的功能 回答链接
  基于上面的回答我编写自己的监听脚本。
  首先是 init_unreal.py 脚本

FORMAT_ARGS = {"Content": DIR}

# NOTE 初始化键盘事件
hotkey_enabled = setting.get("hotkey")
if hotkey_enabled:
    # NOTE 根据系统获取 Python exe 路径
    os_config = {
        "Windows": "Win64",
        "Linux": "Linux",
        "Darwin": "Mac",
    }
    os_platform = os_config.get(platform.system())
    if not os_platform:
        raise OSError("Unsupported platform '{}'".format(platform.system()))

    PYTHON = "Python" if six.PY2 else "Python3"
    ThirdParty = os.path.join(sys.executable, "..", "..", "ThirdParty")
    interpreter = os.path.join(ThirdParty, PYTHON, os_platform, "python.exe")
    interpreter = os.path.abspath(interpreter)

    exec_file = os.path.join(DIR, "_key_listener", "__main__.py")
    msg = "lost path \n%s\n%s" % (interpreter, exec_file)
    assert os.path.exists(interpreter) and os.path.exists(exec_file), msg

    # NOTE 开一个 Python 子进程进行键盘监听
    # NOTE https://stackoverflow.com/a/4896288
    ON_POSIX = "posix" in sys.builtin_module_names
    p = Popen(
        [interpreter, exec_file],
        shell=True,
        stdout=PIPE,
        bufsize=1,
        close_fds=ON_POSIX,
    )
    # NOTE 关闭 Unreal 的时候同步关闭进程
    unreal_app.aboutToQuit.connect(p.terminate)

    def enqueue_output(out, queue):
        for line in iter(out.readline, b""):
            queue.put(line)
        out.close()

    q = Queue()
    t = Thread(target=enqueue_output, args=(p.stdout, q))
    t.daemon = True  # thread dies with the program
    t.start()

    # NOTE 读取注册快捷键的配置
    hotkey_path = posixpath.join(DIR, "hotkey.json")
    hotkey_config = read_json(hotkey_path)

    def __red_key_tick__(delta_seconds):
        try:
            line = q.get_nowait()
        except:
            return

        # NOTE https://stackoverflow.com/a/26970249
        # NOTE 获取 windows 下当前关注窗口的名称
        title = get_foreground_window_title()
        title = title if title else ""
        ue_active = "Unreal Editor" not in title
        if delta_seconds > 0.1 or ue_active:
            return

        # NOTE python3 将获取的数据转换为 unicode
        line = str(line.strip(), "utf-8")
        config = hotkey_config.get(line)
        if not config:
            return

        callbacks = {
            "COMMAND": lambda command: sys_lib.execute_console_command(
                None, command
            ),
            "PYTHON": lambda command: eval(command),
        }

        typ = config.get("type", "").upper()
        callback = callbacks.get(typ)
        if not callback:
            return

        command = config.get("command", "").format(**FORMAT_ARGS)
        callback(command)

    tick_handle = unreal.register_slate_post_tick_callback(__red_key_tick__)
    __QtAppQuit__ = partial(unreal.unregister_slate_post_tick_callback, tick_handle)
    unreal_app.aboutToQuit.connect(__QtAppQuit__)

  监听用的脚本如下

import os
import sys
import time
import json
import keyboard

def main():
    hotkey_path = os.path.join(Content, "hotkey.json")
    if not os.path.exists(hotkey_path):
        print("{} not exists".format(hotkey_path))
        return

    with open(hotkey_path, "r", encoding="utf-8") as f:
        hotkey_config = json.load(f)

    # NOTE 注册的快捷键会打印快捷键信息 传输到主进程
    for hotkey in hotkey_config:
        keyboard.add_hotkey(hotkey, print, args=(hotkey,))

    while True:
        # NOTE 降低 CPU 占用
        time.sleep(0.1)
        # NOTE 保持 stdout 刷新确保主进程可以持续获取数据
        sys.stdout.flush()

if __name__ == "__main__":
    main()

  使用 while True 循环持续监听，由于是另一个进程不受 Unreal 的 CPU 响应限制。
  加入 time.sleep 可以降低 Python 对 CPU 的占用。
  经过上面的配置，我只需要配置好 hotkey.json 文件，就可以调用类似之前做的菜单一样进行调用。

{
    "ctrl+f":{
        "type":"COMMAND",
        "command": "py \"{Content}/Msic/ue_launcher/launcher.py\""
    },
    "ctrl+shift+c":{
        "type":"COMMAND",
        "command": "py \"{Content}/_key_listener/copy_reference.py\""
    }
}

  比如我自定义添加 ctrl+shift+c 复制 Unreal 里面选中资源的路径。

总结

  总算是把去年的文章给补上了。
  借助 Python 生态，这个方案甚至也可以作为 DCC 的功能扩展。
  趁着这个契机，我打算做一个 Unreal Launcher