keras gradcam 예제

keras-gradcam-p1-re.py

#!/usr/bin/env python
# coding: utf-8

# In[1]:


## gist: https://gist.github.com/rhee-elten/ca633edcf039069015e8ccd7faaa099a
## keras 모델로 p1 데이터셋을 학습하고, 테스트 데이터셋에 대한 gradcam 을 추출
## 93.87% 94.86% VAL AUC 최고치 였던 저장 파일 사용
_backbone = 'DenseNet169'
_pretrained = 'save/p1-keras-9/checkpoints_/best_model.053-0.114.hdf5'
_image_files_dir = 'data/p1/650x900'
_image_size = 900, 650


# In[2]:


# %env TF_FORCE_GPU_ALLOW_GROWTH=true


# In[3]:


get_ipython().run_line_magic('matplotlib', 'inline')


# In[4]:


import matplotlib.pyplot as plt
plt.rcParams['figure.figsize'] = (4.5, 3.25)
plt.rcParams['figure.dpi'] = 100
plt.rcParams['image.cmap'] = 'gray'


# In[5]:


## 미지정 파라메터의 디폴트 값 설정

def def_global(name, default_value, type=None, comment=None):
    if name not in globals():
        globals()[name] = default_value
    print('{:>22s}: {:64.64s}'.format(name, repr(globals()[name])), flush=True)


# _save_dir 또는 checkpoint .hdf5 파일 경로를 지정
def_global('_pretrained', None)
def_global('_image_files_dir', None)
def_global('_image_size', None) # (900, 650))
def_global('_num_classes', 2)

# weight 만 저장된 파일을 읽어오는 경우 아래 규칙으로 모델을 먼저 빌드함.
def_global('_backbone', None)
def_global('_input_layer_name','input_1')
def_global('_inner_model_name', None) # 'densenet169')
def_global('_last_conv_layer_name', None) # 'conv5_block32_concat')

def_global('_gradcam_examples', 20)

def_global('_seed', 4223585)

## _inner_model_name 를 지정하지 않으면, _backbone 으로 부터 추측
if _inner_model_name is None:
    _inner_model_name = _backbone.lower()
print('_inner_model_name:',_inner_model_name)

## _inner_model_name 를 지정하지 않으면, _inner_model_name 으로 부터 추측
if _last_conv_layer_name is None:
    _last_conv_layer_name = {
        'densenet169': 'conv5_block32_concat',
        'resnet101v2': 'conv5_block3_out',
        'resnet50v2': 'conv5_block3_out',
    }[_inner_model_name]
print('_last_conv_layer_name:', _last_conv_layer_name)


# In[ ]:





# In[ ]:





# In[ ]:





# In[6]:


"""
keras 로 학습하고 저장한 모델 파일을 로딩하는 코드
"""
import sys
import numpy as np
import pandas as pd
from glob import glob
import tensorflow as tf
K = tf.compat.v1.keras.backend

try:
    from tensorflow.python.framework.ops import disable_eager_execution
    disable_eager_execution()
    tf.compat.v1.disable_v2_behavior()
except:
    _, ex_value, _ = sys.exc_info()
    print(repr(ex_value), file=sys.stderr)

tf.compat.v1.disable_eager_execution()



from importlib import import_module
from inspect import getmembers

from tensorflow.keras import Sequential
from tensorflow.keras.initializers import TruncatedNormal
from tensorflow.keras.layers import Activation, Conv2D, Dense, Input, Layer
from tensorflow.keras.models import Model
from tensorflow.keras.regularizers import l2 as l2_regularizer



##################################################

class Scaler(Layer):
    @tf.function
    def call(self, inp):
        return tf.cast(inp, tf.float32) / 255.0 * 2.0 - 1.0


## 구 버젼으로 저장한 weights 파일 지원을 위해 생성
class Resizer(Layer):
    @tf.function
    def call(self, inp):
        resize_area = tf.compat.v1.image.resize_area
        return resize_area(inp, (_image_size[1], _image_size[0]))


def build_model(backbone, input_shape, num_classes, _exclude_top=0):

    input_ = Input(shape=input_shape)
    layer = Scaler()(input_)

    if _exclude_top:
        keras_apps = import_module("tensorflow.keras.applications")
        network_fn = dict(getmembers(keras_apps))[backbone]
        mdl_ = network_fn(
            input_shape=input_shape,
            include_top=False,
            pooling="avg",
        )
        layer = mdl_(layer)
        layer = Dense(units=num_classes, activation=None, name="logits")(layer)
        output = Activation("softmax")(layer)
    else:
        keras_apps = import_module("tensorflow.keras.applications")
        network_fn = dict(getmembers(keras_apps))[backbone]
        mdl_ = network_fn(
            input_shape=input_shape,
            weights=None,
            pooling="avg",
            classes=num_classes,
        )
        output = mdl_(layer)

    mdl = Model(input_, output)

    mdl.compile(tf.keras.optimizers.SGD(), "sparse_categorical_crossentropy", ["acc"])

    return mdl

custom_objects = {
    "Scaler": Scaler,
    "Resizer": Resizer,
}

def load_model(_pretrained, _image_size=_image_size, custom_objects=custom_objects):

    ## 1차 시도 - load_model() 을 이용하여 모델과 weights 를 로딩
    try:
        mdl = tf.keras.models.load_model(_pretrained, custom_objects=custom_objects)
        print("\n")
        print("============================================================")
        print("== model loaded using tf.keras.models.load_model()")
        print("============================================================")
        print("\n", flush=True)
        return mdl
    except ValueError:  # ValueError: Unknown layer: Functional
        _, ex_value, _ = sys.exc_info()
        print("***** load_model() failed by:", ex_value)

    ## 2차 시도 - exclude_top=0 으로 하여 모델을 생성하고, weights 를 로딩
    try:
        mdl = build_model(_backbone, (_image_size[1], _image_size[0], 3), _num_classes)
        mdl.load_weights(_pretrained)
        print("\n")
        print("============================================================")
        print("== model loaded using model.load_weights() (exclude_top=0)")
        print("============================================================")
        print("\n", flush=True)
        return mdl
    except (
        ValueError
    ):  ## "ValueError: You are trying to load a weight file containing 2 layers into a model with 1 layers."
        _, ex_value, _ = sys.exc_info()
        print("***** load_model(exclude_top=0) failed by:", ex_value)

    ## 3차 시도 - exclude_top=1 로 하여 모델을 생성하고, weights 를 로딩
    try:
        mdl = build_model(
            _backbone, (_image_size[1], _image_size[0], 3), _num_classes, _exclude_top=1
        )
        mdl.load_weights(_pretrained)
        print("\n")
        print("============================================================")
        print("== model loaded using model.load_weights() (exclude_top=1)")
        print("============================================================")
        print("\n", flush=True)
        return mdl
    except ValueError:
        _, ex_value, _ = sys.exc_info()
        print("***** load_model(exclude_top=1) failed by:", ex_value)

    assert 0, "model load failed"

    return None


# In[7]:


"""
지정된 테스트 데이터셋 이미지를 읽어오고,
파일명 규칙에 따라 레이블을 설정하는 코드
"""
import re
from glob import glob
from os.path import basename, splitext
import cv2

def read_dataset(test_files_dir, image_size=None, label_fn=None):

    if label_fn is None:
        def label_fn(x):
            b = splitext(basename(x))[0]
            m = re.match(
                r'^(\d{9})_([abcd])_(right|left)_([012])(-(.*))?$', b)
            assert m, b
            pid, compo, side, mal, aug = m.group(1, 2, 3, 4, 6)
            mal = int(mal)
            yval = int(mal == 2)
            return yval

    test_file_names = glob(test_files_dir+'/*.jpg')
    test_file_names += glob(test_files_dir+'/*.png')
    test_file_names = sorted(test_file_names)

    assert(len(test_file_names) > 0)

    test_files = []
    for x in test_file_names:
        if x.strip():
            yval = label_fn(x)
            test_files.append((x, yval))
    print('found:', len(test_files))

    num_files = len(test_files)
    X = None
    if image_size is not None:
        X = np.zeros((num_files, image_size[1], image_size[0], 3), 'uint8')
    y = np.zeros((num_files,), 'int')

    for i, itm in enumerate(test_files):
        pth, yval = itm
        im = cv2.imread(pth, flags=cv2.IMREAD_COLOR)
        h, w = im.shape[:2]
        if X is None:
            X = np.zeros((num_files, h, w, 3), 'uint8')
            image_size = X.shape[2], X.shape[1]
        if h != image_size[1] or w != image_size[0]:
            im = cv2.resize(im, image_size, interpolation=cv2.INTER_AREA)
        X[i, ...] = im
        y[i] = yval

    print('X:', X.shape, 'y:', y.shape)

    return X, y, test_file_names


# In[ ]:





# In[ ]:





# In[ ]:





# In[8]:


"""
로딩된 모델에서,
지정된 레이어 (모델 마다 다름) 의 출력을 추출하여
gradcam heatmap 을 만드는 함수를 생성하는 코드
"""


def build_gradcam_fn(
    model,
    last_conv_layer_name=_last_conv_layer_name,
    inner_model_name=_inner_model_name,
    input_layer_name=_input_layer_name,
):

    # First, we create a model that maps the input image to the activations
    # of the last conv layer as well as the output predictions
    try:
        ## inner_model 이 지정된 경우 inner 모델 내부의 last_conv_layer 선택
        inner_model = model.get_layer(inner_model_name)
        last_conv_layer = inner_model.get_layer(last_conv_layer_name).get_output_at(1)
    except:
        _, ex_value, _ = sys.exc_info()
        print("***** get innner_model failed:", ex_value)
        ## inner_model 이 없는 경우 바깥쪽 모델의 last_conv_layer 선택
        last_conv_layer = model.get_layer(last_conv_layer_name).output
    print("last_conv_layer.shape:", last_conv_layer.get_shape())

    ## 특정 클래스 (OK,NG 등등) 를 선택할 수 있는 
    ## placeholder 를 네트워크에 추가

    print("model.output.shape", model.output.get_shape())
    
    pred_index_ph = tf.compat.v1.placeholder(tf.int64)
    class_channel = model.output[:, pred_index_ph]

    ## last_conv_layer를 구성하는 각각의 값들이
    ## 최종 출력값에 미치는 영향을
    ## 계산하기 위해서 grdients 값 추출
    print(
        "getting tf.gradients...:",
        class_channel.get_shape(),
        last_conv_layer.get_shape(),
    )
    grads = tf.gradients(class_channel, last_conv_layer)
    if grads is None:
        print("tf.gradients: grads is None")
    if grads[0] is None:
        print("tf.gradients: grads[0] is None")

    print("get_shape(grads)", tuple(g.get_shape() for g in grads))

    # This is a vector where each entry is the mean intensity of the gradient
    # over a specific feature map channel
    pooled_grads = tf.reduce_mean(grads[0], axis=(0, 1, 2))
    print("reduce_mean(grads[0], (0,1,2)).shape:", pooled_grads.get_shape())

    # We multiply each channel in the feature map array
    # by "how important this channel is" with regard to the top predicted class
    # then sum all the channels to obtain the heatmap class activation
    print("last_conv_layer[0].shape:", last_conv_layer[0].get_shape())
    print(
        "pooled_grads[..., tf.newaxis].shape:",
        pooled_grads[..., tf.newaxis].get_shape(),
    )
    heatmap = last_conv_layer[0] @ pooled_grads[..., tf.newaxis]
    print("heatmap.shape:", heatmap.get_shape())
    heatmap = tf.squeeze(heatmap)
    print("heatmap.shape squeezed:", heatmap.get_shape())

    # For visualization purpose, we will also normalize the heatmap between 0 & 1
    heatmap = tf.maximum(heatmap, 0) / tf.math.reduce_max(heatmap)

    def gradcam_fn(img_array, pred_index):
        return heatmap.eval(
            feed_dict={
                model.get_layer(input_layer_name).input: img_array,
                pred_index_ph: pred_index,
            },
            session=K.get_session(),
        )

    return gradcam_fn


# In[9]:


"""
원본 이미지와 생성된 cam heatmap 을
시각화 하여 그리는 함수
"""
def visualize_cam(im, cam, clss, title=None, heatmap_alpha=0.35, figsize=(19.2, 9.6), dpi=100):
    print('visualize_cam: im.shape:', im.shape, 'cam.shape:', cam.shape)
    
    h, w = im.shape[:2]
    cam_ = cv2.resize(cam, (w, h), interpolation=cv2.INTER_LINEAR_EXACT)
    
    _, (ax1, ax2, ax3) = plt.subplots(1, 3, figsize=figsize, dpi=dpi)
    ax1.imshow(im)
    ax1.axis('off')
    if title is not None:
        ax1.set_title(title)
    
    ax2.matshow(cam,cmap='jet')
    ax2.axis('off')

    ax3.imshow(im)
    ax3.imshow(cam_, cmap='jet', alpha=heatmap_alpha)
    ax3.axis('off')
    ax3.set_title('prediction: {}'.format(clss))
    
    plt.tight_layout()
    plt.show()


# In[ ]:





# In[ ]:





# In[ ]:





# In[10]:


"""
지정된 모델 로딩
"""

mdl = load_model(_pretrained)

## 적합한 모델인지 검증
layer_names = [lyr.name for lyr in mdl.layers]
if not (_last_conv_layer_name in layer_names or _inner_model_name in layer_names):
    mdl.summary()
    assert _last_conv_layer_name in layer_names or _inner_model_name in layer_names, (
        _last_conv_layer_name,
        _inner_model_name,
        layer_names,
    )

## 모델 구조 출력
mdl.summary()
try:
    mdl.layers[-1].summary()
except:
    _, ex_value, _ = sys.exc_info()

# for GRADCAM, remove softmax
# tf.identity or tf.keras.activations.linear
try:
    print("old activation:", mdl.layers[-1].activation)
    mdl.layers[-1].activation = tf.keras.activations.linear
    print("patched activation:", mdl.layers[-1].activation)
    print("mdl.layers[-1].activation patched")
except:
    _, ex_value, _ = sys.exc_info()
    print("mdl.layers[-1].activation patch failed:", ex_value)
    print("old activation:", mdl.layers[-1].layers[-1].activation)
    mdl.layers[-1].layers[-1].activation = tf.keras.activations.linear
    print("patched activation:", mdl.layers[-1].layers[-1].activation)
    print("mdl.layers[-1].layers[-1].activation patched")


# In[11]:


## gradcam 추출하는 함수 생성
gradcam = build_gradcam_fn(mdl)


# In[ ]:





# In[ ]:





# In[ ]:





# In[12]:


## 데이터셋을 X, y 에 로딩
X, y, _image_files = read_dataset(_image_files_dir, _image_size)
if _image_size is None:
    _image_size = X.shape[2], X.shape[1]

## 이상치 데이터만 선택:
## NG (2) 만 선택 (파일명에서 0, 1: OK, 2: MAL)
is_mal = (y == 1)
X_mal = X[is_mal]
y_mal = y[is_mal]
files_mal = np.asarray(_image_files)[is_mal].tolist()


# In[13]:


## X_mal 에서 임의로 몇개만 선택
rng = np.random.default_rng(_seed)
choices = sorted(rng.choice(np.arange(len(y_mal)), size=(_gradcam_examples,), replace=False))
print("choices:", choices)


# In[14]:


from os.path import basename

for n in choices:
    img = X_mal[n]
    img_clss = y_mal[n]
    img_filename = basename(files_mal[n])
    title = 'file: {:s}{}, clss: {:d}'.format(
        img_filename, img.shape, img_clss)
    print(title)

    preds = mdl.predict(np.asarray([img]))
    preds = preds[0]
    pred_index = np.argmax(preds)
    with np.printoptions(precision=3, suppress=True):
        print('preds:', preds, 'pred:', pred_index)

    heatmap = gradcam(np.asarray([img]), pred_index)
    visualize_cam(img, heatmap, pred_index, title=title)


# In[ ]:





# In[ ]:





# In[ ]: