![images](../img/imgtop_sub_b.png)
ComfyUI を画像編集アプリ(Krita)のバックエンドにする、シンプルなプラグインを一から作ります。[※1]
![images](../img/imgcmm_202404092230354875683281.png)
- ※1
- 生成画像は、次のモデルを使用しています:
- ・
- 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
- ◯
- 問題
ただ、ワークフローの切り替えには対応していません。自分が作ったワークフローを、このプラグインに適用させるのは難しいわけです。[※1]
- ◯
- 対応
最近 ComfyUI に、生成画像をダイレクトに WebSocket で出力するノードが追加されました。その使用例のスクリプトも公開されています:[※2][※3]
- ・
- https://github.com/comfyanonymous/ComfyUI/blob/master/script_examples/...
ここでは、ComfyUI の WebSocket と Krita のプラグイン(Python / PyQt)を使って、参照画像と生成画像をやり取りする、シンプルなプラグインを作ってみます。
- ◯
- 効果
これで、自分が作ったワークフローから、リアルタイムな画像生成などができるようになります。
- ※1
- 独自のワークフローを適用できないわけではないのですが、専用の関数を使ったりと、いろいろ面倒ですーーこのプラグインでは、機能を増やすほど要素間の関係が密になり、ワークフローも固定するしかなかったようです:
- ・
- https://github.com/Acly/krita-ai-diffusion/discussions/385
- ※2
- これまでも生成画像を WebSocket で出力することはできましたが、ファイルを介したものでした。いっぽう Acly 氏が、生成画像をダイレクトに出力する機能を、カスタムノードとして公開していました(krita-ai-diffusion では、このノードを使ってプラグインを作成しています)ーーこのノードが、ComfyUI のネィテブな機能として取り込まれたかっこうです:
- ・
- https://github.com/Acly/comfyui-tooling-nodes
- ※3
- なら、ノードへのリアルタイムな画像の入力はどうするかというとーーこれはたんに、画像を文字列にエンコードしたものを、それを解釈できるノードの入力にし、ワークフローをサーバに送るだけです(ここでは、この入力ノードは上記の Acly 氏のものを使います)。
前提:ネットワークを介したやり取り
- ◯
- WebSocket
Krita と ComfyUI の参照画像〜生成画像のやり取りは、WebSocket を使います。
- ◯
- コネクション
WebSocket も、ウェブの通常のやり取りと同じポート(80/433)を使います。ただ、その取り決め(プロトコル)が違います:
port | Web | WebSocket |
80 | HTTP (http://...) | WS (ws://...) |
443 | SSL (https://...) | WSS (wss://...) |
HTTP プロトコルによる接続は、1回のやり取りで終わります(クライアントが要求し〜サーバが応答する)。いっぽう WS プロトコルによる接続は、やり取りを何回も繰り返せるよう、その接続(TCP コネクション)を維持します。[※1][※2][※3]
- ※1
- なので WebSocket を使う通信は、次のどれとも違いますーーHTTP による接続をクッキーなどで維持する(HTTP & session)、サーバの応答を遅延させる(XMLHttpRequest & long polling)、サーバがクライアントに送りつける(SSE)。
- ※2
- 同じポートを使うのに、HTTP と WS を分けることができるのは、HTTP にプロトコルを上書きする仕様がふくまれているからです(Upgrade 要求〜応答)。
- ※3
- じっさい ComfyUI は、WS のコネクションを確立している間、要求されたワークフロー(参照画像をふくむ)を実行するのに HTTP を使いますーーこのとき専用のノードが、WS に実行結果の構造体(生成画像をふくむ)を渡すので、そこから画像を読み出す、という流れになります(WS と HTTP は、ワークフローの ID の合致で連携させています)。
前提:スレッドどうしのやり取り
- ◯
- Qt - PyQt
Krita は、その UI の開発環境に Qt を採用しています。[※1][※2]
この 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)、というものです。
これで、ワーカとスレッドの機能が分離され、ワーカはほんらいの処理の記述に専念できます。またフロントで記述していた処理を、けっこうかんたんにスレッドに対応させられる、という利点もあります。
- ※1
- とりあえずスクリプトの動作を確かめたいなら、アプリの次のメニューから、Python / PyQt のインタプリタが使えます:
> tools > script > scripter ... スクリプトのインタプリタを表示
- ※2
- Qt は、(GTK とともに)UNIX の GUI である X Window のツールキットとして知られていますーー当初は Qt のライセンスが GPL と相容れないことから GTK を生むきっかけにもなりましたが、現在は、そのデュアルライセンスのひとつが GPL になっています。
作成
以下、コードです:[※1]
- ◯
- 構成
プラグインの構成を記述します:
- ・
- ${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[※2][※3][※4][※5]
# 参照:ネット import time import uuid import json import urllib.request import websocket # 参照:GUI 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()) # 設定:GUI g_nmlsrc = "L01" # レイヤの名前(参照照像用) g_nmldst = "L02" # レイヤの名前(生成画像用) g_nambgn = "BGN" # ボタンの表記(開始) g_namend = "END" # ボタンの表記(終了) g_objapp = Krita.instance() # クラス:ネット/GUIの各種メソッド 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() # クラス:GUI 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))
- ※1
- このコードは、多様な状況に対応したものではありませんーーエラー処理は省いてあり、操作画面も必要最小限に抑えていますーーなので、最悪の場合、アプリ(Krita)自体がフリーズする/ダウンする可能性があります。
- ※2
- コード中の @pyqtSlot は必須ではありませんーースロットであることを明示するために付けています(ただしコンパイル時の挙動には違いが出るようです)。
- ※3
- 生成を1枚だけで終わらせたいなら、次の1行をコメントアウトしますーーこれで、参照画像を読み込むシグナルが繰り返し出なくなります。結果的に、最初の1回のやり取りで、生成は終わりになります:
@pyqtSlot(str) def rcvstr(self, strimg): ... #self.sigget.emit() # 参照画像の読み込みをコメントアウト
- ※4
- 生成画像を、重ねたレイヤに描画するのでなく、べつのドキュメントに描画する場合はーー新しいウィンドウでドキュメントを作っておき、そのドキュメントのレイヤに描画します:
> window > new window ... 新しいウィンドウを作成
@pyqtSlot(object) def putimg(self, datimg): docdst = g_objapp.documents()[1] # 2番目のドキュメントを選択 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 側では、ワークフローに次のノードを参照画像向け、生成画像向けとして配置します:[※1]
- ・
- 参照画像向け:Load Image (Base64)
- ・
- 生成画像向け:Save Image Websocket
このワークフローを開発者向けとしてダウンロードし、コードで指定した所定のフォルダに置きます。
- ◯
- Krita
Krita 側では、次のように使います:
> view > detach canvas ... キャンバスをドッカーから分離(必要なら) > settings > dockers > layers ... 参照画像向けと生成画像向けのレイヤ群に、コードに記述したレイヤ名を付与 > settings > dockers > WIN001 ... プラグインを起動 > WIN001 > BGN ... 連続した生成を開始 > END ... 連続した生成を停止
- ※1
- たとえば、次のようなワークフローになります(LCM - LoRA と ControlNet - scribble を組み合わせ、落描きから画像を高速に生成):
- ・
- workflow_api.json
![images](../img/imgcmm_202404092230354875683281.png)