mieki256's diary

パイプを使って画像データを渡すPythonスクリプトをAI(Google Gemini)に作ってもらった。一発では動くスクリプトが出てこなかったけど、数回やり取りしたら動くようになった。

• 親スクリプトは Pillow を使って画像を開いてから、子スクリプトを実行する。
• 親スクリプトは、子スクリプトの標準入力(STDIN)に画像データを送る。画像データの先頭には画像の横幅と縦幅を付加しておく。
• 子スクリプトは標準入力から得られた画像データに対して何らかの処理をする。今回はグレースケール化のみをしてみた。
• 子スクリプトの処理の進捗具合は STDERR に出力される。親スクリプトは STDERR を読み取ってコンソールに出力する。
• 親スクリプトは、子スクリプトの標準出力(STDOUT)から処理後の画像データを受け取って表示する。

_親スクリプト parent.py

import subprocess
import struct
import threading
from PIL import Image


def log_reader(pipe):
    """
    標準エラー出力をリアルタイムで読み取って表示する関数。
    メインスレッドとは別に動かすことで、子プロセスが吐き出すログを
    メインの読み書き処理を止めずに処理できます。
    """
    try:
        with pipe:
            # pipe.readline はデータが来るまで待機し、b''（空）が来たら終了
            for line in iter(pipe.readline, b""):
                # decode()でbytesを文字列に変換。strip()で末尾の改行を除去
                print(f"[Child Log]: {line.decode().strip()}")
    except Exception as e:
        print(f"Log error: {e}")


def run_example(image_path):
    # 画像を開き、RGBA形式（1ピクセル4バイト）に統一して読み込み
    img = Image.open(image_path).convert("RGBA")
    width, height = img.size
    img_bytes = img.tobytes()

    # 子プロセス（child.py）を起動
    # すべての入出力を PIPE で接続し、親からコントロール可能にする
    process = subprocess.Popen(
        ["python", "child.py"],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        bufsize=0,  # バイナリデータのやり取りで遅延を防ぐためバッファリングを無効化
    )

    # 1. 進捗（stderr）を読み取る専用スレッドを開始
    # ※重要：ここを別スレッドにしないと、子プロセスがstderrに大量のログを出した際、
    # 　OSのパイプバッファがいっぱいになり、子プロセスが停止（デッドロック）してしまいます。
    t = threading.Thread(target=log_reader, args=(process.stderr,))
    t.daemon = True  # 親スクリプトが異常終了してもこのスレッドが残らないように設定
    t.start()

    # 2. 子プロセスへデータを送信
    # child.py 側が待っている 8byte のヘッダー（幅・高さ）をパック
    header = struct.pack("ii", width, height)

    print(f"Sending image data ({width}x{height})...")

    # ヘッダーと画像本体を結合して stdin へ書き込み
    process.stdin.write(header + img_bytes)

    # 書き込み終了を明示的に伝える（EOFを送る）。
    # これにより、child.py 側の stdin.read() が「これ以上データが来ない」と判断し、処理が進みます。
    process.stdin.close()

    # 3. 処理済みデータ（stdout）を受信
    # stderr の読み取りは別スレッドに任せているため、メインスレッドは stdout の受信に専念できます
    print("Waiting for processed image...")
    result_bytes = process.stdout.read()

    # 子プロセスの終了を待ち、リソースをクリーンアップ
    process.wait()

    # ログ出力スレッドが残っている場合、最大1秒待機して同期をとる
    t.join(timeout=1)

    # 受信したバイナリを Pillow で画像オブジェクトに戻す
    if result_bytes:
        res_img = Image.frombytes("RGBA", (width, height), result_bytes)

        # プレビュー表示（Windowsのフォトアプリ等が起動します）
        res_img.show()

        print("Success: Image received and displayed.")
    else:
        print("エラー: データを受信できませんでした。")


if __name__ == "__main__":
    # カレントディレクトリに input.png がある前提
    run_example("input.png")

_子スクリプト child.py

import sys
import struct


def process_image_real_progress():
    # Windows環境などで改行コードの自動変換（\n -> \r\n）を防ぐため、
    # .buffer を使用してバイナリモードでストリームを扱います。
    stdin = sys.stdin.buffer
    stdout = sys.stdout.buffer
    stderr = sys.stderr

    # 1. ヘッダー読み込み。画像の幅と高さを取得
    # 親側で struct.pack('ii', w, h) して送られてくることを想定（4byte * 2 = 8byte）
    header = stdin.read(8)
    if not header:
        return

    # 'i'はsigned int (4byte)。リトルエンディアンで展開されます
    width, height = struct.unpack("ii", header)

    # 2. 全データ読み込み（一気に読み込んでメモリ上で保持）
    # RGBA 32bit (4byte) 前提の計算: width * height * 4
    # 大容量画像の場合はメモリ負荷に注意が必要ですが、1枚単位ならこの手法が高速です
    raw_data = stdin.read(width * height * 4)

    # 処理後のデータを格納するためのバッファを先に確保（不変なbytesではなく可変なbytearrayを使用）
    processed_data = bytearray(len(raw_data))

    # 3. 行単位でループ処理を行い、計算と進捗の書き出しを両立させる
    for y in range(height):
        # 処理対象の行のバイト配列上の開始・終了インデックスを計算
        row_start = y * width * 4
        row_end = row_start + (width * 4)

        # 1行分のピクセルを走査。4バイト刻み（R, G, B, A）で処理
        for i in range(row_start, row_end, 4):
            # RGBAの各要素を抽出
            r = raw_data[i]
            g = raw_data[i + 1]
            b = raw_data[i + 2]
            a = raw_data[i + 3]

            # 輝度計算 (ITU-R BT.601) に基づくグレースケール変換
            # 重み付け: R=0.299, G=0.587, B=0.114
            v = int(0.299 * r + 0.587 * g + 0.114 * b)

            # 処理結果を格納。RGBすべてに同じ輝度値を代入し、Alphaは元の値を維持
            processed_data[i] = v
            processed_data[i + 1] = v
            processed_data[i + 2] = v
            processed_data[i + 3] = a

        # 標準エラー出力(stderr)を使って進捗を表示
        # stdoutは画像データ用に使用しているため、混ざらないようstderrを使います。
        # 20行ごともしくは最終行で進捗を出力（頻繁なIOによる速度低下を防止）
        if y % 20 == 0 or y == height - 1:
            progress = int((y + 1) / height * 100)
            # \n を含めて書き出すことで、親プロセス側で readline() して取得可能
            stderr.write(f"Processing: {progress}%\n")
            stderr.flush()

    # 4. 変換済みの画像バイナリを一括して標準出力へ書き出し
    stdout.write(processed_data)

    # 明示的にフラッシュし、パイプの先（親プロセス）へデータを送り出す
    stdout.flush()


if __name__ == "__main__":
    process_image_real_progress()

対象画像は以下。


_input.png

以下で実行。

> python parent.py

Sending image data (512x512)...
Waiting for processed image...
[Child Log]: Processing: 0%
[Child Log]: Processing: 4%
[Child Log]: Processing: 8%
...
[Child Log]: Processing: 97%
[Child Log]: Processing: 100%
Success: Image received and displayed.

以下の結果になった。


たしかに、パイプを利用することで、親スクリプトから子スクリプトに画像データを渡して、フィルタ処理の結果を得ることができた。

Python 3.8 から、共有メモリを簡単に利用できる multiprocessing.shared_memory というライブラリが標準添付されるようになったらしい。その機能を使って親スクリプトから子スクリプトに画像データを渡してみる。これも AI(Google Gemini)に作ってもらった。

• 親スクリプトは Pillow で画像を開いてから、子スクリプトを実行する。
• 親スクリプトは、標準入力を利用して、子スクリプトに「共有メモリの名前」「画像の横幅」「縦幅」を渡す。
• 子スクリプトは標準入力から「共有メモリの名前」「画像の横幅」「縦幅」を取得して、共有メモリ上の画像データにアクセスする。
• 子スクリプト側の処理の進捗具合は STDERR を介して親スクリプトに伝える。
• 処理結果は共有メモリを介して親スクリプト側が取得する。

_parent_shm.py

import subprocess
import struct
import threading
from multiprocessing import shared_memory
from PIL import Image


def log_reader(pipe):
    """子の進捗ログを読み取る"""
    with pipe:
        for line in iter(pipe.readline, b""):
            print(f"[Child Log]: {line.decode().strip()}")


def run_simple_shm(image_path):
    # 画像準備
    img = Image.open(image_path).convert("RGBA")
    width, height = img.size
    img_bytes = img.tobytes()
    size = len(img_bytes)

    # 1. 共有メモリの作成とデータ書き込み
    shm = shared_memory.SharedMemory(create=True, size=size)
    try:
        shm.buf[:size] = img_bytes

        # 2. 子プロセス起動
        # セマフォを使わないため、単純な Popen でOK
        process = subprocess.Popen(
            ["python", "child_shm.py"],
            stdin=subprocess.PIPE,
            stderr=subprocess.PIPE,
            bufsize=0,
        )

        # 進捗表示用スレッド
        t = threading.Thread(target=log_reader, args=(process.stderr,))
        t.start()

        # 3. メモリ情報を子に送信
        # 固定長 64byte の名前と、幅・高さの int を送る
        header = struct.pack("64sii", shm.name.encode(), width, height)
        process.stdin.write(header)
        process.stdin.close()

        # 4. 子の終了を待つ
        # これにより、子がメモリを書き換え終わるまで親は待機します（＝同期）
        print("Child process is working...")
        process.wait()
        t.join()

        # 5. 処理結果をメモリから直接読み出して表示
        print("Displaying result from Shared Memory.")
        res_img = Image.frombytes("RGBA", (width, height), shm.buf[:size])
        res_img.show()

    finally:
        # 6. リソース解放
        shm.close()
        shm.unlink()


if __name__ == "__main__":
    run_simple_shm("input.png")

_child_shm.py

import sys
import struct
from multiprocessing import shared_memory


def process_shm_simple():
    # 1. 親からメモリ名と画像サイズを受信 (計72byte)
    # メモリ名(64byte) + 幅(4byte) + 高さ(4byte)
    header = sys.stdin.buffer.read(72)
    if len(header) < 72:
        return

    shm_name_raw, width, height = struct.unpack("64sii", header)
    shm_name = shm_name_raw.decode().strip("\x00")

    # 2. 既存の共有メモリに接続
    try:
        shm = shared_memory.SharedMemory(name=shm_name)
        data = shm.buf

        # 3. 画像処理（グレースケール化）
        # メモリを直接書き換えるため、戻り値を stdout で送る必要がありません
        for y in range(height):
            row_offset = y * width * 4
            for x in range(width):
                offset = row_offset + (x * 4)

                # RGBAを取得して輝度計算
                r, g, b, a = data[offset: offset + 4]
                v = int(0.299 * r + 0.587 * g + 0.114 * b)

                # 直接上書き
                data[offset] = v
                data[offset + 1] = v
                data[offset + 2] = v
                # Alphaは維持

            # 進捗報告（stderrを使用）
            if y % 20 == 0 or y == height - 1:
                progress = int((y + 1) / height * 100)
                sys.stderr.write(f"Progress: {progress}%\n")
                sys.stderr.flush()

        # 4. クローズ（アンリンクは親の役目）
        shm.close()

    except Exception as e:
        sys.stderr.write(f"Child Error: {e}\n")


if __name__ == "__main__":
    process_shm_simple()

実行は以下。

> python parent_shm.py

Child process is working...
[Child Log]: Progress: 0%
[Child Log]: Progress: 4%
[Child Log]: Progress: 8%
[Child Log]: Progress: 11%
...
[Child Log]: Progress: 97%
[Child Log]: Progress: 100%
Displaying result from Shared Memory.

パイプを使った場合と同じ結果が得られた。

共有メモリとやらを使うことでも、別の実行バイナリに画像データを渡して処理させることが可能と分かった。ただ、標準入力からアレコレ渡したり、STDERR で進捗状態を取得するのはどうなんだろうとは思うけど…。

とりあえずこういうことができるのであれば、GIMPのフィルタプラグインと同様に、実行バイナリの形でフィルタプラグインを公開配布することもできそうだなと…。いやまあ、そういう共通規格を作るとしたら、という話だけど。後は、どういうパラメータをどういうフォーマットで渡すかを考えれば形になるのかもしれない。ヘッダー部分に色々入れるとして、ヘッダーサイズを内包したり、画像データサイズを内包したり…。おそらくレイヤー情報を数枚渡す場合もあるだろうし…。

余談。今回の実験で、別の実行バイナリ(フィルタプラグイン)に画像データを渡す方法はいくつかあると分かったけれど。仮に、こういった方法を利用してフィルタプラグインの規格を作ってみたとしても、おそらく誰も使わないだろうなと思えてきた。

と言うのも、既存の画像編集ソフトは既に自前で各フィルタを実装しちゃってるものが大半なので…。「今更そんな規格を提唱されてもねえ」「当の昔に自分で実装しちゃったからそんな規格は要らねえ」みたいな。

8bitPC時代のMSXみたいなものかも。MSX規格に沿ってなかった、それ以前からあったPC製品シリーズのほうがスペックは高いしやれることも多いから、今更MSX規格と言われてもウチはそんなの作りませんよ…みたいな。

これから画像編集ソフトを新規に開発する場合は、そういった規格があれば助かるのかもしれないけれど、「仕様を把握して対応するのが面倒臭いから独自実装でいいや」ということになりそうでもある…。

DTM/DAWソフトのVSTi/VSTeのようには行かないのだろうな…。

そう考えると、既にあるGIMPのプラグインを利用できるようにするラッパーを作るとか、G'MICを自作アプリに組み込むためのチュートリアル文書でも作ったほうが良さそうかもしれないなと…。

「そういうのはPhotoshopプラグインでいいじゃん」と言われそうな気もするけれど8bfはバージョン毎の互換性の問題が…。