通常なやり方

　Keras でCNNは非常に便利で、ネットワークのサンプルも多くて割愛します。画像をメモリに読み込み、その後下記のように一発で訓練すれば良い。

cnn_re.py

model.fit(X_train, y_train, batch_size=batch_size, epochs=epochs, verbose=1, validation_data=(X_val, y_val))

問題

画像数が少ない時にはよかったですが、多くなるとメモリに入りきれなくなって、エラーとなりました。
keras ならmodel.fitの代わりにmodel.fit_generatorを使えば、小出し（minibatchごと）に画像データを与えばいいのです。GeneratorならKerasのImageDataGeneratorがあって便利ですね。でも残念なことに今回は分類問題ではなく、回帰となっているので、どうやら目標の数値を与えるにはそのGeneratorが自作するしかないようです（本当はあるが自分が知らないだけかもしれませんが）。それで以下のようにGeneratorを作ってみました。

呼ぶ側：

cnn_re.py

subfolder = Path(args.fromdir)
pathlist = Path(subfolder).glob('**/*.jpg')
image_file_paths = list(pathlist)
image_file_paths = [str(item) for item in image_file_paths]

random.shuffle(image_file_paths)
all_file_len = len(image_file_paths)
test_pos = int(all_file_len * 0.1)  # get 10% of data for test
val_pos = test_pos * 2 # get 10% of data for validation

test_gen = FishImageGenerator(image_file_paths[0:test_pos],batch_size,img_rows,img_cols,channels)
valid_gen = FishImageGenerator(image_file_paths[test_pos:val_pos],batch_size,img_rows,img_cols,channels)
train_gen = FishImageGenerator(image_file_paths[val_pos:],batch_size,img_rows,img_cols,channels)

model.fit_generator(
    generator=train_gen,
    epochs=epochs,
    steps_per_epoch=len(train_gen),
    verbose=1,
    validation_data=valid_gen,
    validation_steps=len(valid_gen),
    callbacks=[es_cb,
              TensorBoard(log_dir=log_dir),
              ModelCheckpoint(model_path, save_best_only=True)]
    )

Generator側：

cnn_re.py

class FishImageGenerator(Sequence):
    def __init__(self,file_paths=None,batch_size = 16,img_rows=256, img_cols=256, channels=3):
        self._file_paths = file_paths
        self._batch_size = batch_size
        self._img_rows = img_rows
        self._img_cols = img_cols
        self._channels = channels

        if file_paths is None:
            self._file_paths = self._get_paths()

        self._array_len = len(self._file_paths)
        self._num_batches_per_epoch = math.ceil(self._array_len / batch_size)
        self.reset()

    def reset(self):
        random.shuffle(self._file_paths)
        self.inner_targets = []
        for path_in_str in self._file_paths:
            path_obj = Path(path_in_str)
            self.inner_targets.append(path_obj.parent._parts[-1])

    def get_targets(self):
        return self.inner_targets

    def _get_paths(self,root_dir=None):
        if root_dir is None:
            root_dir = Path.cwd()
            pathlist = Path(root_dir).glob('**/*.jpg')

        return pathlist

    def __getitem__(self, idx):
        start_pos = self._batch_size * idx
        end_pos = start_pos + self._batch_size
        if end_pos > self._array_len:
            end_pos = self._array_len
        item_paths = self._file_paths[start_pos: end_pos]

        loaded_data = []
        target_label = []

        for path_in_str in item_paths:
            path_obj = Path(path_in_str)
            try:
                img = cv2.imread(path_in_str)
                loaded_data.append(img)
                target_label.append(path_obj.parent._parts[-1])
            except Exception as e:
                logger.error(e)

        imgs = np.array(loaded_data).astype(np.float32)
        imgs = imgs /255.
        imgs = imgs.reshape(-1, self._img_rows, self._img_cols, self._channels)
        targets = np.array(target_label).astype("uint8")
        targets = targets.reshape(-1, 1)

        return imgs, targets

    def __len__(self):
        """Batch length"""
        return self._num_batches_per_epoch

    def on_epoch_end(self):
        """Task when end of epoch"""
        print(read_pic.cache_info())
        logger.info(read_pic.cache_info())
        self.reset()

Generatorの書き方について、Kerasのドキュメントにもあるので、割愛します。書き方もいろいろですが、Sequenceから継承して、getitem、len、on_epoch_endをそれぞれ実装すれば良いです。それらのメソッドの目的はコメントしています。

試して見たところ、確かのメモリアウトのエラーは出なくなりましたが、毎回ディスクから読み込むのもどうもスッキリしなくて直ぐに思い浮かぶのはやはりCacheですよね。Python3ならfunctoolsにlru_cacheがあるので、それを使えば楽々じゃないですか。

generator.py


import functools
@functools.lru_cache(maxsize=30000)
def read_pic(path_to_pic):
    pic_data = cv2.imread(path_to_pic)
    return pic_data

import functools　と　@functools.lru_cache の二行を追加することでCacheを使うことになりました。

で、なんか変ですね、print(read_pic.cache_info())　の一行で出力されたCacheのHit率があまりにも低いのではないか、maxsizeを拡大したら少しは改善しますが、本来もっと活躍してもらいたかったのに失望ですよ。
もう少し考えたら、なんだバカなのはこの自分じゃないですか。本来一Epochごとに全ての訓練データを使うのに、LRUだとすでに使ったデータをCacheに入れるのは逆効果に他にないよ。まぁ、どうせならCacheも自作してしまえば。

generator.py


import functools
from pathlib import Path
import cv2
import numpy as np
import random
from keras.utils import Sequence
import math
from collections import namedtuple
from _thread import RLock

import logging
logger = logging.getLogger(__name__)

_CacheInfo = namedtuple("CacheInfo", ["hits", "misses", "maxsize", "currsize"])

def _make_key(args):
    return hash(args[0])

def simple_cache(maxsize=128):
    if maxsize is not None and not isinstance(maxsize, int):
        raise TypeError('Expected maxsize to be an integer or None')

    def decorating_function(user_function):
        wrapper = _simple_cache_wrapper(user_function, maxsize, _CacheInfo)
        return functools.update_wrapper(wrapper, user_function)

    return decorating_function

def _simple_cache_wrapper(user_function, maxsize, _CacheInfo):
    sentinel = object()          # unique object used to signal cache misses
    make_key = _make_key         # build a key from the function arguments

    cache = {}
    hits = misses = 0
    full = False
    cache_get = cache.get    # bound method to lookup a key or return None
    cache_len = cache.__len__  # get cache size without calling len()
    lock = RLock()           # because linkedlist updates aren't threadsafe

    if maxsize == 0 or maxsize is None:

        def wrapper(*args, **kwds):
            # No caching -- just a statistics update after a successful call
            nonlocal misses
            result = user_function(*args, **kwds)
            misses += 1
            return result

    else:

        def wrapper(*args, **kwds):
            # Size limited caching that tracks accesses by recency
            nonlocal hits, misses, full
            key = make_key(args)
            with lock:  # Python's built-in structures are thread-safe for single operations but
                # a lock here will add almost no overhead, and will give me peace of mind.
                result = cache_get(key)
                if result is not None:
                    hits += 1
                    return result
            result = user_function(*args, **kwds)
            with lock:
                if key in cache:
                    # Getting here means that this same key was added to the
                    # cache while the lock was released.  Since the link
                    # update is already done, we need only return the
                    # computed result and update the count of misses.
                    pass
                elif full:
                    # our simple is  full,so just ignore the coming one
                    pass
                else:
                    cache[key] = result
                    full = (cache_len() >= maxsize)
                misses += 1
            return result

    def cache_info():
        """Report cache statistics"""
        with lock:
            return _CacheInfo(hits, misses, maxsize, cache_len())

    def cache_clear():
        """Clear the cache and cache statistics"""
        nonlocal hits, misses, full
        with lock:
            cache.clear()
            hits = misses = 0
            full = False

    wrapper.cache_info = cache_info
    wrapper.cache_clear = cache_clear
    return wrapper

@simple_cache(maxsize=2000)
def read_pic(path_to_pic):
    pic_data = cv2.imread(path_to_pic)
    return pic_data

class FishImageGenerator(Sequence):
    def __init__(self,file_paths=None,batch_size = 16,img_rows=256, img_cols=256, channels=3):
        self._file_paths = file_paths
        self._batch_size = batch_size
        self._img_rows = img_rows
        self._img_cols = img_cols
        self._channels = channels

        if file_paths is None:
            self._file_paths = self._get_paths()

        self._array_len = len(self._file_paths)
        self._num_batches_per_epoch = math.ceil(self._array_len / batch_size)
        self.reset()

    def reset(self):
        random.shuffle(self._file_paths)
        self.inner_targets = []
        for path_in_str in self._file_paths:
            path_obj = Path(path_in_str)
            self.inner_targets.append(path_obj.parent._parts[-1])

    def get_targets(self):
        return self.inner_targets

    def _get_paths(self,root_dir=None):
        if root_dir is None:
            root_dir = Path.cwd()
            pathlist = Path(root_dir).glob('**/*.jpg')

        return pathlist

    def __getitem__(self, idx):
        start_pos = self._batch_size * idx
        end_pos = start_pos + self._batch_size
        if end_pos > self._array_len:
            end_pos = self._array_len
        item_paths = self._file_paths[start_pos: end_pos]

        loaded_data = []
        target_label = []

        for path_in_str in item_paths:
            path_obj = Path(path_in_str)
            try:
                img = read_pic(path_in_str)
                #img = cv2.imread(path_in_str)
                loaded_data.append(img)
                target_label.append(path_obj.parent._parts[-1])
            except Exception as e:
                print(e)
                logger.error(e)

        imgs = np.array(loaded_data).astype(np.float32)
        imgs = imgs /255.
        imgs = imgs.reshape(-1, self._img_rows, self._img_cols, self._channels)
        targets = np.array(target_label).astype("uint8")
        targets = targets.reshape(-1, 1)

        #self.inner_imgs,self.inner_targets = img,targets
        return imgs, targets

    def __len__(self):
        """Batch length"""
        return self._num_batches_per_epoch

    def on_epoch_end(self):
        """Task when end of epoch"""
        print(read_pic.cache_info())
        logger.info(read_pic.cache_info())
        self.reset()

やっとこれで少しスッキリしました。でもメモリの消耗が画像サイズの十倍もあるようで、それではCacheの効果もだいぶ落ちますね。よく考えたら、本来圧縮された画像ファイルをNumpyのndarray形式に展開された時点で膨らむよね、それじゃあ当然Cacheできる数も減りますね。

test.py

    pic_data = cv2.imread(path_to_pic)
    with open(path_to_pic, 'rb') as f:
        binary = f.read()
       
    import sys
    print ("array size is {}".format(sys.getsizeof(pic_data)))
    print ("binary size is {}".format(sys.getsizeof(binary)))

測ってみたら、やはり20倍ぐらいの差がありますね。array size is 196736
binary size is 7833

それで、前のソースの中の

generator.py

def read_pic(path_to_pic):
    pic_data = cv2.imread(path_to_pic)
    return pic_data

を下記のように直せば完成

generator.py

@simple_cache(maxsize=150000)
def read_pic(path_to_pic):
    #pic_data = cv2.imread(path_to_pic)
    with open(path_to_pic,'rb') as f:
        binary = f.read()
    return binary

def bin2arr(binary):
    arr = np.asarray(bytearray(binary), dtype=np.uint8)
    pic_data = cv2.imdecode(arr, -1)  

    return pic_data

色々失敗して、やっと形になりました。でもそれで毎回画像ファイルの解凍にもCPU時間を費やさなければならないことになります、その辺の調整は訓練の状況をみてするしかないようです。

あ、言い忘れたところです、自作のCacheのソースの大部分はPythonのfunctoolsからパクったものです。