画像の処理をMixup→Random Erasing→ImageDataGeneratorの変換の順で処理したい

Question

前提

kerasのImageDataGeneratorをカスタマイズしてRandom ErasingとMixupの実装ができたのですが、画像を変換する際の処理順をMixup→Random Erasing→ImageDataGeneratorの変換の順に処理することは可能でしょうか？

また、できればMixupの発生させる確立を指定したいです。

実現したいこと

・画像の処理順をMixup→Random Erasing→ImageDataGeneratorの変換の順で処理
・Mixupの発生させる確立を指定

発生している問題

・apply_transform内でmixupがうまくできていない

該当のソースコード

from random_eraser import get_random_eraser

class MyImageDataGenerator(ImageDataGenerator):
    def  __init__(self,mix_up_alpha = 0.0, *args, **kwargs):
        self.random_eraser = get_random_eraser()
        super().__init__(*args, **kwargs)
        assert mix_up_alpha >= 0.0
        self.mix_up_alpha = mix_up_alpha  
        
    ##mixup用    
    def img(gen: ImageDataGenerator):
        for x, y in gen:
            yield x, y
    
    def mix_up(self, X1, y1, X2, y2):
            assert X1.shape[0] == y1.shape[0] == X2.shape[0] == y2.shape[0]
            batch_size = X1.shape[0]
            l = np.random.beta(self.mix_up_alpha, self.mix_up_alpha, batch_size)
            X_l = l.reshape(batch_size, 1, 1, 1)
            y_l = l.reshape(batch_size, 1)
            X = X1 * X_l + X2 * (1-X_l)
            y = y1 * y_l + y2 * (1-y_l)
            return X, y
    
    def apply_transform(self, x, transform_parameters):
       
        x = self.mix_up_alpha(x)   # 先に処理する
        x = self.random_eraser(x)            
        return super().apply_transform(x, transform_parameters)

試したこと

回答を参考にmixupを実装したかったのですが、どのようにdef imgの画像ジェネレーターとmixupするのか詰まっている状況です。

補足情報（FW/ツールのバージョンなど）

・https://dev.classmethod.jp/articles/tensorflow-image-generator-custom/
・https://dev.classmethod.jp/articles/tensorflow-image-generator-custom-mixup/
上記二つのサイトを参考に実装しました。

使用ツール
anacondaを使用
python　3.8.13
TensorFlow　2.3
Spyder 5.3.3

Comments

[ps_aux_grep]

super().flow()からのnext(batch_gen)の時点でImageDataGeneratorの変換結果が生成されているので，これを後に持っていくよう記述するべきです．

[semisemi]

コメントありがとうございます。
def flow内のwhile True下にある
batch_x, batch_y = next(batch_gen)
batch_x_2, batch_y_2 = next(batch_gen2)
をif self.random_erasing_probabilityとself.mixup_alphaより下に書くということでしょうか？

[ps_aux_grep]

それをしてしまうと，random_erasingやmixup_alphaをするデータはどこから持ってくるか論になってしまいます．flowはOverrideせずに，apply_transformをOverrideしてかつapply_transform()の最後でsuper().apply_transform()するべきです．

[semisemi]

まだ、私自身の理解が出来ていないですが質問のソースコードに、apply_transformメソッドを追記しておりますので確認していただけると幸いです。

[ps_aux_grep]

返信を失念していました．確認お願いいたします．

Answer 1

簡単のためgithubにあったrandom_eraserを利用して次のように実装します．

コマンドで

wget https://raw.githubusercontent.com/yu4u/cutout-random-erasing/master/random_eraser.py -O random_eraser.py

と実行すればカレントディレクトリ配下にrandom_eraser.pyがダウンロードされるので，これを利用して

from random_eraser import get_random_eraser
import csv
import numpy as np
import tensorflow as tf
from keras.preprocessing.image import ImageDataGenerator
from matplotlib import pyplot as plt

def mix_up(X1, y1, X2, y2, mix_up_alpha):
    assert X1.shape[0] == y1.shape[0] == X2.shape[0] == y2.shape[0]
    batch_size = X1.shape[0]
    l = np.random.beta(mix_up_alpha, mix_up_alpha, batch_size)
    X_l = l.reshape(batch_size, 1, 1, 1)
    y_l = l.reshape(batch_size, 1)
    X = X1 * X_l + X2 * (1 - X_l)
    y = y1 * y_l + y2 * (1 - y_l)
    return X, y

class MyImageDataGenerator(ImageDataGenerator):
    def  __init__(self, *args, **kwargs):
        self.random_eraser = get_random_eraser()
        super().__init__(*args, **kwargs)
  
    def apply_transform(self, x, transform_parameters):
        x = self.random_eraser(x) # 先に処理する
        return super().apply_transform(x, transform_parameters)

from keras.datasets import mnist
(X, y), (_, _) = mnist.load_data()
X = X[:, :, :, np.newaxis]

mix_up_gen = ImageDataGenerator(
    rescale = 1/255,
).flow(X, y)
train_gen = MyImageDataGenerator(
    rescale = 1/255,
    rotation_range = 90
).flow(X, y)

def my_flow(mix_up_gen, gen):
    for (X1, y1), (X2, y2) in zip(mix_up_gen, gen):
        yield mix_up(X1, y1, X2, y2, 0.5)

fig = plt.figure(figsize = (10, 7))
rows, cols = 3, 3
one_batch = next(my_flow(mix_up_gen, train_gen))
for i, (img, label) in enumerate(zip(one_batch[0], one_batch[1])):
    if i >= rows * cols:
        break
    fig.add_subplot(rows, cols, i + 1)
    img = img.reshape(28, 28)
    plt.imshow(img, vmin = 0, vmax = 1)
    plt.axis("off")

apply_transform()に要求されている引数は，GitHubのコードからも分かるとおり，単一画像xとその変換方法transform_parameterなので，これが処理されてしまう前に，単一画像xを独自の方法で変換します．

また，MixUpをImageDataGeneratorによる変換前に挿入するのはIteratorの書き換えが必要であることがわかり現実的ではないので，従来通りジェネレータを定義するしかありません．

Comments

[ps_aux_grep]

mixupに関しては，画像変換において複数画像を持てるシーンが存在しなかったので，mixup用の画像ジェネレータを別でbatch_size = 1で生成してMyImageDataGeneratorに渡しておき，apply_transformで逐次mixupする方法を取るのが良さそうに見えます．

とにかく，super().flow()では既にNumpyArrayIteratorを介して変換済み画像が返されており，これがジェネレータとして機能することから，これより前に画像を変換しようとすると，書き換えることになるのはNumpyArrayIterator._get_batches_of_transformed_samplesかapply_transformの二択になります．
前者であれば，変換対象のindex_arrayに対して，mixupに必要なもう一つの画像のindex_arrayを用意してやって，mixupをすることが可能に見えます．

[semisemi]

返信遅れて申し訳ないです。
回答ありがとうございます。
前の質問に回答していただいた際の下記コードでMyImageDataGeneratorに組み込むことでbatch_size = 1を実装できていますか？
def img(gen: ImageDataGenerator):
for x, y in gen:
yield x, y

[ps_aux_grep]

そうですね，それで組み合わせることが可能です

[semisemi]

質問に追記いたしましたので確認いただけると幸いです。

[ps_aux_grep]

回答に追記しました．ImageDataGeneratorにMixUpを組み込むのは諦めた方が良さそうです．

[semisemi]

ありがとうございます。
回答いただいた手法で学習を行ってみようと思います。

[semisemi]

この質問に関連する質問を投稿いたしましたのでできればアドバイスをいただけると幸いです
よろしくお願いします。