ComfyUI を画像編集アプリ（Krita）のバックエンドにする、シンプルなプラグインを一から作ります。［※１］
- ※１
- 生成画像は、次のモデルを使用しています：
- ・
- https://civitai.com/models/4468/counterfeit-v30
関連
- ◯
- ComfyUI で画像生成 〜 なぜそこにつなぐのか：ComfyUI, Stable Diffusion
- ◯
- ComfyUI を、クラウドのコンテナに設置する：ComfyUI, Docker, GCP
- ◯
- ComfyUI の API を使う：ComfyUI, API, JSON, Python, Docker, GCP
検証
- ◯
- サーバ
- ・
- クラウド：GCP
- ・
- コンテナ：Docker
- ・
- ホスト：Ubuntu 22.04
- ・
- ゲスト：Ubuntu 22.04
- ◯
- クライアント
- ・
- コンテナ：Docker Desktop
- ・
- ホスト：macOS
- ・
- ゲスト：Ubuntu 22.04
概要
- ◯
- 背景
ComfyU をバックエンドにした画像編集アプリでは、Krita とそのプラグイン krita-ai-diffusion が知られてますよねーーリアルタイムに生成画像を編集する動画が、話題になりました：
- ・
- https://github.com/Acly/krita-ai-diffusion
- ◯
- 問題
ただ、ワークフローの切り替えには対応していません。自分が作ったワークフローを、このプラグインに適用させるのは難しいわけです。［※１］
- ◯
- 対応
最近 ComfyUI に、生成画像をダイレクトに WebSocket で出力するノードが追加されました。その使用例のスクリプトも公開されています：［※２］［※３］
- ・
- https://github.com/comfyanonymous/ComfyUI/blob/master/script_examples/...
ここでは、ComfyUI の WebSocket と Krita のプラグイン（Python / PyQt）を使って、参照画像と生成画像をやり取りする、シンプルなプラグインを作ってみます。
- ◯
- 効果
これで、自分が作ったワークフローから、リアルタイムな画像生成などができるようになります。
- ※１
- 独自のワークフローを適用できないわけではないのですが、専用の関数を使ったりと、いろいろ面倒ですーーこのプラグインでは、機能を増やすほど要素間の関係が密になり、ワークフローも固定するしかなかったようです：
- ・
- https://github.com/Acly/krita-ai-diffusion/discussions/385
- ※２
- これまでも生成画像を WebSocket で出力することはできましたが、ファイルを介したものでした。いっぽう Acly 氏が、生成画像をダイレクトに出力する機能を、カスタムノードとして公開していました（krita-ai-diffusion では、このノードを使ってプラグインを作成しています）ーーこのノードが、ComfyUI のネィテブな機能として取り込まれたかっこうです：
- ・
- https://github.com/Acly/comfyui-tooling-nodes
- ※３
- なら、ノードへのリアルタイムな画像の入力はどうするかというとーーこれはたんに、画像を文字列にエンコードしたものを、それを解釈できるノードの入力にし、ワークフローをサーバに送るだけです（ここでは、この入力ノードは上記の Acly 氏のものを使います）。
前提：ネットワークを介したやり取り
- ◯
- WebSocket
Krita と ComfyUI の参照画像〜生成画像のやり取りは、WebSocket を使います。
- ◯
- コネクション
WebSocket も、ウェブの通常のやり取りと同じポート（80/433）を使います。ただ、その取り決め（プロトコル）が違います：
| port | Web | WebSocket |
| 80 | HTTP (http://...) | WS (ws://...) |
| 443 | SSL (https://...) | WSS (wss://...) |
HTTP プロトコルによる接続は、１回のやり取りで終わります（クライアントが要求し〜サーバが応答する）。いっぽう WS プロトコルによる接続は、やり取りを何回も繰り返せるよう、その接続（TCP コネクション）を維持します。［※１］［※２］［※３］
- ※１
- なので WebSocket を使う通信は、次のどれとも違いますーーHTTP による接続をクッキーなどで維持する（HTTP & session）、サーバの応答を遅延させる（XMLHttpRequest & long polling）、サーバがクライアントに送りつける（SSE）。
- ※２
- 同じポートを使うのに、HTTP と WS を分けることができるのは、HTTP にプロトコルを上書きする仕様がふくまれているからです（Upgrade 要求〜応答）。
- ※３
- じっさい ComfyUI は、WS のコネクションを確立している間、要求されたワークフロー（参照画像をふくむ）を実行するのに HTTP を使いますーーこのとき専用のノードが、WS に実行結果の構造体（生成画像をふくむ）を渡すので、そこから画像を読み出す、という流れになります（WS と HTTP は、ワークフローの ID の合致で連携させています）。
前提：スレッドどうしのやり取り
- ◯
- Qt - PyQt
Krita は、その UI の開発環境に Qt を採用しています。［※１］［※２］
この Qt の Python 向けのラッパが、PyQt ですーーKrita のプラグインでは、この PyQt のライブラリを使って、UI の動作を記述します：
- ・
- https://docs.krita.org/en/user_manual/python_scripting/introduction_to_python_scripting.html
- ・
- https://docs.krita.org/en/user_manual/python_scripting/krita_python_plugin_howto.html
- ◯
- シグナル／スロット
PyQt（というよりその大元の C++ の Qt）では、UI のそれぞれの要素のやり取り（非同期の通信）に、シグナルとスロットという考え方を導入しています。
とはいえ、使う分には難しいところはなく、たんにシグナルを送る先の関数／メソッドがスロットになる、ということです。
- ◯
- ワーカ／スレッド
ネットワーク側とのやり取りは通常、UI をあつかうフォアグラウンドではなく、バックグラウンド（別のスレッド）で行いますーーこのスレッドの生成にも、Qt では独自の作法がありますーーほんらいの処理を記述したクラス（ワーカ）のインスタンスを、スレッドのインスタンスに移行させる（moveToThread）、というものです。
これで、ワーカとスレッドの機能が分離され、ワーカはほんらいの処理の記述に専念できます。またフロントで記述していた処理を、けっこうかんたんにスレッドに対応させられる、という利点もあります。
- ※１
- とりあえずスクリプトの動作を確かめたいなら、アプリの次のメニューから、Python / PyQt のインタプリタが使えます：
> tools > script > scripter ... スクリプトのインタプリタを表示
- ※２
- Qt は、（GTK とともに）UNIX の GUI である X Window のツールキットとして知られていますーー当初は Qt のライセンスが GPL と相容れないことから GTK を生むきっかけにもなりましたが、現在は、そのデュアルライセンスのひとつが GPL になっています。
作成
以下、コードです：［※１］
- ◯
- 構成
プラグインの構成を記述します：
- ・
- ${directory_development}/try001.desktop
[Desktop Entry] Type=Service ServiceTypes=Krita/PythonPlugin X-KDE-Library=try001 X-Python-2-Compatible=false X-Krita-Manual=try001.html Name=TRY001 Comment=
- ◯
- 参照
参照するパッケージ群を指定します：
- ・
- ${directory_development}/try001/__init__.py
from .bin.try001 import *
- ◯
- 処理
パッケージの本体を記述します：
- ・
- ${directory_development}/try001/bin/try001.py［※２］［※３］［※４］［※５］
# 参照：ネット
import time
import uuid
import json
import urllib.request
import websocket
# 参照：ＧＵＩ
from base64 import b64encode
from PyQt5.QtGui import QImage
from PyQt5.QtCore import QByteArray, QBuffer, QIODevice, pyqtSignal, pyqtSlot
from PyQt5.QtWidgets import *
from krita import *
# 設定
g_namdoc = "DOC001" # ドッカーの名前
g_namwin = "WIN001" # ドッカーウィンドウの表題
# 設定：ネット
g_adrsrv = "<address_server>:<port>" # 接続先サーバのアドレス／ポート
g_pthflw = "<directory_plugin_krita>/cnf/workflow_api.json" # 作ったワークフロー（開発者モード）のパス
g_nodput = '<id_node_put>' # ノードの識別子：画像の入力用
g_nodget = '<id_node_get>' # ノードの識別子：画像の出力用
g_secwdf = 0 # 待ち時間（秒）：参照画像の更新があるとき
g_secwdm = 1 # 待ち時間（秒）：参照画像の更新がないとき
g_idtcli = str(uuid.uuid4())
# 設定：ＧＵＩ
g_nmlsrc = "L01" # レイヤの名前（参照照像用）
g_nmldst = "L02" # レイヤの名前（生成画像用）
g_nambgn = "BGN" # ボタンの表記（開始）
g_namend = "END" # ボタンの表記（終了）
g_objapp = Krita.instance()
# クラス：ネット／ＧＵＩの各種メソッド
class Cnncmf():
# 指定したワークフローの実行結果（JSON形式）を返す（※例示のコードを、クラスのメソッド化のため若干変更）
def queue_prompt(self, server_address, client_id, prompt):
p = {"prompt": prompt, "client_id": client_id}
data = json.dumps(p).encode('utf-8')
req = urllib.request.Request("http://{}/prompt".format(server_address), data=data)
return json.loads(urllib.request.urlopen(req).read())
# 生成画像のリストを返す（※例示のコードを、クラスのメソッド化のため若干変更）
def get_images(self, server_address, client_id, prompt, save_image_websocket_node, ws):
prompt_id = self.queue_prompt(server_address, client_id, prompt)['prompt_id']
output_images = {}
current_node = ""
while True:
out = ws.recv()
if isinstance(out, str):
message = json.loads(out)
if message['type'] == 'executing':
data = message['data']
if data['prompt_id'] == prompt_id:
if data['node'] is None:
break #Execution is done
else:
current_node = data['node']
else:
if current_node == save_image_websocket_node:
images_output = output_images.get(current_node, [])
images_output.append(out[8:])
output_images[current_node] = images_output
return output_images
# レイヤに描かれた参照画像を、文字列（Base64/UTF-8形式）に変換
def getimg(self, docsrc, lyrsrc):
datimg = lyrsrc.projectionPixelData(0, 0, docsrc.width(), docsrc.height()).data()
qimage = QImage(datimg, docsrc.width(), docsrc.height(), QImage.Format_ARGB32)
bytimg = QByteArray()
buffer = QBuffer(bytimg)
buffer.open(QIODevice.WriteOnly)
qimage.save(buffer, 'PNG')
bseimg = b64encode(bytimg)
strimg = bseimg.decode('utf-8')
return strimg
# ノードからの生成画像（PNG 形式）を、レイヤで表示できる画像に変換
def putimg(self, docdst, lyrdst, datimg):
qimage = QImage()
qimage.loadFromData(datimg)
ptrimg = qimage.bits()
ptrimg.setsize(qimage.byteCount())
lyrdst.setPixelData(QByteArray(ptrimg.asstring()), 0, 0, docdst.width(), docdst.height())
lyrdst.projectionPixelData(0, 0, docdst.width(), docdst.height())
docdst.refreshProjection()
# ワークフローのファイルを読み込み、ディクショナリ（JSON形式）に変換
def getflw(self, pthflw):
with open(pthflw) as f:
strflw = f.read()
dctflw = json.loads(strflw)
return dctflw
# 文字列化した画像（Base64/UTF-8形式）を、ワークフローの入力ノードの箇所に挿入
def cnnnet(self, cnnweb, adrsrv, idtcli, dctflw, nodput, nodget, strimg):
dctflw[nodput]['inputs']['image'] = strimg
lstimg = self.get_images(adrsrv, idtcli, dctflw, nodget, cnnweb)
datimg = lstimg[nodget][0]
return datimg
# クラス：ネット（ワーカ）
class Wrkcmf(QObject):
# シグナル群を定義
sigget = pyqtSignal()
sigput = pyqtSignal(object)
def __init__(self):
super().__init__()
#
self.flgrun = False
self.strimg = None
self.cnncmf = Cnncmf()
# 連続したやり取りを開始：ネットワークのコネクションを確立し、最初の参照画像の読み込みを要求
def prcrun(self):
self.flgrun = True
self.strimp = ''
self.cnnweb = websocket.WebSocket()
self.cnnweb.connect("ws://{}/ws?clientId={}".format(g_adrsrv, g_idtcli))
self.dctflw = self.cnncmf.getflw(g_pthflw)
#
self.sigget.emit()
# 連続したやり取りを終了
def prcend(self):
self.flgrun = False
self.cnnweb.close()
# スロット：返された参照画像をネットワークに送り、生成画像を取得。指定された待ち時間の経過後、参照画像の読み込みを要求
@pyqtSlot(str)
def rcvstr(self, strimg):
self.strimg = strimg
if self.strimg != self.strimp:
datimg = self.cnncmf.cnnnet(self.cnnweb, g_adrsrv, g_idtcli, self.dctflw, g_nodput, g_nodget, self.strimg)
self.sigput.emit(datimg)
self.strimp = self.strimg
time.sleep(g_secwdf)
else:
time.sleep(g_secwdm)
self.sigget.emit()
# クラス：ＧＵＩ
class Doccmf(DockWidget):
# シグナル群を定義
sigstr = pyqtSignal(str)
# ドッカーの画面を生成
def __init__(self):
super().__init__()
#
self.cnncmf = Cnncmf()
self.setWindowTitle(g_namwin)
objwgt = QWidget(self)
self.setWidget(objwgt)
# ボタン群を生成
btn001 = QPushButton(g_nambgn, objwgt)
btn002 = QPushButton(g_namend, objwgt)
# ボタン群を専用のレイアウトに適用
objwgt.setLayout(QVBoxLayout())
objwgt.layout().addWidget(btn001)
objwgt.layout().addWidget(btn002)
# ボタン群と処理の対応を設定
btn001.clicked.connect(self.prcrun)
btn002.clicked.connect(self.prcend)
# ワーカをスレッド化
def prcrun(self):
self.wrkcmf = Wrkcmf()
self.thread = QThread()
# ワーカをスレッドに移行
self.wrkcmf.moveToThread(self.thread)
# シグナル〜スロット群の対応を設定（ワーカに対する）
self.sigstr.connect(self.wrkcmf.rcvstr)
self.wrkcmf.sigget.connect(self.getimg)
self.wrkcmf.sigput.connect(self.putimg)
# シグナル〜スロット群の対応を設定（スレッドに対する）
self.thread.started.connect(self.wrkcmf.prcrun)
self.thread.finished.connect(self.wrkcmf.deleteLater)
self.thread.finished.connect(self.thread.deleteLater)
# スレッドを開始（イベントループ）
self.thread.start()
# スレッドを終了
def prcend(self):
self.wrkcmf.prcend()
self.thread.quit()
self.thread.wait()
# スロット：スレッド化したワーカの要求に応じて、指定のレイヤの参照画像を送信
@pyqtSlot()
def getimg(self):
docsrc = g_objapp.activeDocument()
lyrsrc = docsrc.nodeByName(g_nmlsrc)
strimg = self.cnncmf.getimg(docsrc, lyrsrc)
self.sigstr.emit(strimg)
# スロット：スレッド化したワーカから、生成画像を受け取り、指定のレイヤに表示
@pyqtSlot(object)
def putimg(self, datimg):
docdst = g_objapp.activeDocument()
lyrdst = docdst.nodeByName(g_nmldst)
self.cnncmf.putimg(docdst, lyrdst, datimg)
def canvasChanged(self, canvas):
pass
g_objapp.addDockWidgetFactory(DockWidgetFactory(g_namdoc, DockWidgetFactoryBase.DockRight, Doccmf))
- ※１
- このコードは、多様な状況に対応したものではありませんーーエラー処理は省いてあり、操作画面も必要最小限に抑えていますーーなので、最悪の場合、アプリ（Krita）自体がフリーズする／ダウンする可能性があります。
- ※２
- コード中の @pyqtSlot は必須ではありませんーースロットであることを明示するために付けています（ただしコンパイル時の挙動には違いが出るようです）。
- ※３
- 生成を１枚だけで終わらせたいなら、次の１行をコメントアウトしますーーこれで、参照画像を読み込むシグナルが繰り返し出なくなります。結果的に、最初の１回のやり取りで、生成は終わりになります：
@pyqtSlot(str)
def rcvstr(self, strimg):
...
#self.sigget.emit() # 参照画像の読み込みをコメントアウト
- ※４
- 生成画像を、重ねたレイヤに描画するのでなく、べつのドキュメントに描画する場合はーー新しいウィンドウでドキュメントを作っておき、そのドキュメントのレイヤに描画します：
> window > new window ... 新しいウィンドウを作成
@pyqtSlot(object)
def putimg(self, datimg):
docdst = g_objapp.documents()[1] # ２番目のドキュメントを選択
lyrdst = docdst.nodeByName(g_nmldst)
...
設置
- ◯
- ComfyUI
ComryUI は、次のバージョン以降に更新しておきます（このバージョンから、生成画像を WebSocket で出力するノードが実装されています）：
- ・
- https://github.com/comfyanonymous/ComfyUI/commit/...
次のカスタムノードを設置します（このライブラリに、参照画像を Base64 で入力するノードが実装されています）：
- ・
- https://github.com/Acly/comfyui-tooling-nodes
$ cd ${directory_project}
$ git clone --depth=1 https://github.com/Acly/comfyui-tooling-nodes.git
$ mv comfyui-tooling-nodes comfyui-tooling-nodes_ver_${yyyy_mm_dd_nnn}
$ cd ${directory_project}/ComfyUI/custom_nodes
$ ln -s ${directory_project}/comfyui-tooling-nodes_ver_${yyyy_mm_dd_nnn} comfyui-tooling-nodes
- ◯
- Krita
ComfyUI の WebSocket の例では、次のライブラリを使っているので、これを導入します（ネットワークの接続は、遅延や障害への対応など処理が面倒ですが、このライブラリは、その面倒なところを引き受けてくれてるようですし）：
- ・
- https://github.com/websocket-client/websocket-client
$ cd ${directory_project}
$ git clone --depth 1 -https://github.com/websocket-client/websocket-client.git
$ cd ${directory_plugin_krita}
$ mv ${directory_project}/websocket .
作成したコードのファイル群を、プラグイン向けのフォルダに置きます：
$ cd ${directory_plugin_krita}
$ cp ${directory_development}/try001.desktop .
$ cp -r ${directory_development}/try001 .
作成したプラグインを有効にし、アプリを再起動します：
> krita > preferences > python plugin manager > TRY001: <yes>
利用
- ◯
- ComfyUI
ComfyUI 側では、ワークフローに次のノードを参照画像向け、生成画像向けとして配置します：［※１］
- ・
- 参照画像向け：Load Image (Base64)
- ・
- 生成画像向け：Save Image Websocket
このワークフローを開発者向けとしてダウンロードし、コードで指定した所定のフォルダに置きます。
- ◯
- Krita
Krita 側では、次のように使います：
> view > detach canvas ... キャンバスをドッカーから分離（必要なら） > settings > dockers > layers ... 参照画像向けと生成画像向けのレイヤ群に、コードに記述したレイヤ名を付与 > settings > dockers > WIN001 ... プラグインを起動 > WIN001 > BGN ... 連続した生成を開始 > END ... 連続した生成を停止
- ※１
- たとえば、次のようなワークフローになります（LCM - LoRA と ControlNet - scribble を組み合わせ、落描きから画像を高速に生成）：
- ・
- workflow_api.json