🧵

【keyhac】JobQueueジョブキューについてわかっていることまとめ

2024/07/28に公開

Python

keyhac

tech

https://crftwr.github.io/keyhac/doc/ja/

基本機能

JobItem() に2つの関数を渡すと、

第1引数の関数をサブスレッドで実行してから、
第2引数の関数をメインスレッドで実行する

というジョブを作れる（各関数には引数として JobItem オブジェクトが渡される）。

このジョブを JobQueue.defaultQueue().enqueue() でデフォルトキューに投入することで順次関数を実行できる。

from keyhac import *
from ckit import JobItem, JobQueue

def configure(keymap):
    def do_as_job() -> None:

        def _subthread_func(job_item: JobItem) -> None:
            print("これはサブスレッド内で実行される")

        def _finished_func(job_item: JobItem) -> None:
            print("これはメインスレッド内で実行される")

        job = JobItem(_subthread_func, _finished_func)
        JobQueue.defaultQueue().enqueue(job)

    keymap_global = keymap.defineWindowKeymap()
    keymap_global["C-0"] = do_as_job # Ctrl + 0 で実行

出力：

これはサブスレッド内で実行される
これはメインスレッド内で実行される

keyhacは、キーが押されて離されるまでの間に処理が終わっていないとフック処理がうまくいかずにキーのすり抜けが起きたり固まったりしてしまう。
ジョブ処理はそうした制約のなかで時間のかかる処理を行うのに便利な機能。

公式によると、300ミリ秒以上かかるかどうかが1つの基準とのこと。

実行順としては、

キー入力にバインドされた関数
サブスレッド内関数
メインスレッド内関数

の順で実行される。

from keyhac import *
from ckit import JobItem, JobQueue

def configure(keymap):
    def do_as_job() -> None:
        print("あああ")
        def _subthread_func(job_item: JobItem) -> None:
            print("これはサブスレッド内で実行される")

        def _finished_func(job_item: JobItem) -> None:
            print("これはメインスレッド内で実行される")

        print("いいい")
        job = JobItem(_subthread_func, _finished_func)
        JobQueue.defaultQueue().enqueue(job)
        print("ううう")

    keymap_global = keymap.defineWindowKeymap()
    keymap_global["C-0"] = do_as_job

出力：

あああ
いいい
ううう
これはサブスレッド内で実行される
これはメインスレッド内で実行される

スレッド内で値をやり取りする

実行順とスコープの制約から、サブスレッド内関数の実行結果をメインスレッドに渡すには工夫が必要。

単純に変数代入しようとすると、各関数の実行時点では変数が定義されていないことになるのでエラーになる。

from keyhac import *
from ckit import JobItem, JobQueue

def configure(keymap):
    def do_as_job() -> None:
        hoge = ""
        def _subthread_func(job_item: JobItem) -> None:
            hoge = hoge + "aa"

        def _finished_func(job_item: JobItem) -> None:
            hoge = hoge + "bb"

        job = JobItem(_subthread_func, _finished_func)
        JobQueue.defaultQueue().enqueue(job)

    keymap_global = keymap.defineWindowKeymap()
    keymap_global["C-0"] = do_as_job

出力：

UnboundLocalError: local variable 'hoge' referenced before assignment

リストを使用すればスコープの問題は回避可能。
とはいえPythonの罠ともよく言われる部分なのでコードが複雑になってくるとバグの元になるかもしれない。

from keyhac import *
from ckit import JobItem, JobQueue

def configure(keymap):
    def do_as_job() -> None:
        hoge = []
        def _subthread_func(job_item: JobItem) -> None:
            hoge.append("aa")

        def _finished_func(job_item: JobItem) -> None:
            hoge.append("bb")
            print(hoge)

        print(hoge)
        job = JobItem(_subthread_func, _finished_func)
        JobQueue.defaultQueue().enqueue(job)

    keymap_global = keymap.defineWindowKeymap()
    keymap_global["C-0"] = do_as_job

出力：

[]
['aa', 'bb']

↑do_as_job 実行時点では hoge が空のままであるのに注意。あくまで関数の実行後に _subthread_func → _finished_func の順に実行される。

別の回避策として公式のソースには、各関数に引数として渡される job_item オブジェクトのプロパティを利用するものが紹介されている。

ここでは job_item に .hoge として自作のプロパティをぶら下げて値をやり取りしてみる。

from keyhac import *
from ckit import JobItem, JobQueue

def configure(keymap):
    def do_as_job() -> None:
        counter = 0
        print("関数の開始時点", counter)

        def _subthread_func(job_item: JobItem) -> None:
            job_item.hoge = counter + 1
            print("サブスレッド内", job_item.hoge)

        def _finished_func(job_item: JobItem) -> None:
            print("メインスレッド内", job_item.hoge)

        job = JobItem(_subthread_func, _finished_func)
        JobQueue.defaultQueue().enqueue(job)
        print("関数の終了時点", counter)

    keymap_global = keymap.defineWindowKeymap()
    keymap_global["C-0"] = do_as_job

出力：

関数の開始時点 0
関数の終了時点 0
サブスレッド内 1
メインスレッド内 1

実用例：文字列を確実にコピーしてから次の処理を行う

仮想的に Ctrl+C を押すことで選択している文字をシステムのクリップボードに格納できる。
しかしクリップボードが更新されるまでは僅かに時間がかかり、たまにコピーできないまま次の処理が走ってしまうことがある。
そうしたケースが生じないよう、クリップボードが更新されたのを待ってから次の処理を行ってみる。

import time

from keyhac import *
from ckit import JobItem, JobQueue

def configure(keymap):

    def delay(msec: int = 50) -> None:
        time.sleep(msec / 1000)

    def copy_and_process() -> None:
        origin = getClipboardText()
        keymap.InputKeyCommand("C-C")() # サブスレッドに入る前にキー入力をしておくのがポイント

        def _watch_clipboard(job_item: JobItem) -> None:
            # 10ミリ秒ずつ、20回クリップボードを確認する
            interval = 10
            timeout = interval * 20
            while timeout > 0:
                delay(interval)
                if (s := getClipboardText()) != origin:
                    job_item.copied = s # 更新されたら copied プロパティに格納
                    return
                timeout -= interval
            job_item.copied = origin # 更新できなかったら元のまま

        def _finished(job_item: JobItem) -> None:
            # コピーした文字列でGoogle検索する
            url = "https://www.google.com/search?q=" + job_item.copied
            keymap.ShellExecuteCommand(None, url, "", "")()

        job = JobItem(_watch_clipboard, _finished)
        JobQueue.defaultQueue().enqueue(job)

    keymap_global["C-0"] = copy_and_process

NGパターン

keyhacがハングして固まったり、そうでなくても極端に処理が遅くなるアンチパターンがいくつかある様子。

サブスレッド内からのサブスレッド呼び出し

~~サブスレッド内関数でさらにもう一段階サブスレッド処理を行うと固まる。~~

↑再現できないこともあるので様子見中。

文字入力をサブスレッドから実行しようとする

「ABCDEFG…」と文字入力する処理をサブスレッド内で実行したいとする。

from keyhac import *
from ckit import JobItem, JobQueue

def configure(keymap):
    def do_as_job() -> None:

        def _subthread_func(job_item: JobItem) -> None:
            keymap.InputTextCommand("ABCDEFGHIJKLMN")()

        def _finished_func(job_item: JobItem) -> None:
            pass

        job = JobItem(_subthread_func, _finished_func)
        JobQueue.defaultQueue().enqueue(job)

    keymap_global = keymap.defineWindowKeymap()
    keymap_global["C-0"] = do_as_job

上記のようにサブスレッド内で keymap.InputTextCommand() を実行すると、マウスを始めとした各種入力系が操作を受け付けなくなり、タスクトレイのアイコンも右クリックできなくなる（タスクマネージャからkeyhacのプロセス自体を終了させればOK）。