pytorchでのgradcamの実装が出来ない

Question

前提・実現したいこと

初心者で申し訳ありません。
pytorch用いresnet50で転移学習を行い、gradcamにて特徴領域の可視化を試みています。下記のようなエラーメッセージが繰り返し出力されます。
どなたか解決法等御教示願えませんでしょうか。
よろしくお願いいたします。

発生している問題・エラーメッセージ

RuntimeError: Input type (torch.FloatTensor) and weight type (torch.cuda.FloatTensor) should be the same

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, sampler, random_split
import numpy as np
import pandas as pd
import torchvision
from torchvision import datasets, models, transforms
import matplotlib.pyplot as plt
from PIL import Image
import time
import os 
import copy
import cv2
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

image_transforms = {
    'train': transforms.Compose([
        transforms.RandomResizedCrop(224),
        transforms.RandomHorizontalFlip(),
        transforms.ToTensor(),
        transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
    ]),
    'val': transforms.Compose([
        transforms.Resize(256),
        transforms.CenterCrop(224),
        transforms.ToTensor(),
        transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
    ]),
    'test':
    transforms.Compose([
        transforms.Resize(size=256),
        transforms.CenterCrop(size=224),
        transforms.ToTensor(),
        transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
    ]),
}

batch_size = 32

all_data = datasets.ImageFolder(root='/Users/ora/Desktop/CNNdataset3')
train_data_len = int(len(all_data)*0.8)
valid_data_len = int((len(all_data) - train_data_len)/2)
test_data_len = int(len(all_data) - train_data_len - valid_data_len)
train_data, val_data, test_data = random_split(all_data, [train_data_len, valid_data_len, test_data_len])
train_data.dataset.transform = image_transforms['train']
val_data.dataset.transform = image_transforms['val']
test_data.dataset.transform = image_transforms['test']
print(len(train_data), len(val_data), len(test_data))

train_loader = DataLoader(train_data, batch_size=batch_size, shuffle=True)
val_loader = DataLoader(val_data, batch_size=batch_size, shuffle=True)
test_loader = DataLoader(test_data, batch_size=batch_size, shuffle=True)

model = models.resnet50(pretrained=True)
model = model.to("cuda")
# Freeze early layers
for param in model.parameters():
    param.requires_grad = False
n_classes = 2
# n_futures = model.fc.in_features
num_ftrs = model.fc.in_features
model.fc = nn.Linear(num_ftrs, n_classes)# n_inputs will be 4096 for this case

# model.classifier
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

model.class_to_idx = all_data.class_to_idx
model.idx_to_class = {
    idx: class_
    for class_, idx in model.class_to_idx.items()
}

def train(model,
         criterion,
         optimizer,
         train_loader,
         val_loader,
         save_location,
         early_stop=3,
         n_epochs=50,
         print_every=2):
   
    #Initializing some variables
    valid_loss_min = np.Inf
    stop_count = 0
    valid_max_acc = 0
    history = []
    model.epochs = 0
  
    #Loop starts here
    for epoch in range(n_epochs):
        train_loss = 0
        valid_loss = 0
        train_acc = 0
        valid_acc = 0
        model.train()
        ii = 0
    
        for data, label in train_loader:
            ii += 1
            data, label = data.cuda(), label.cuda()
            optimizer.zero_grad()
            output = model(data)
      
            loss = criterion(output, label)
            loss.backward()
            optimizer.step()
      
            # Track train loss by multiplying average loss by number of examples in batch
            train_loss += loss.item() * data.size(0)
      
            # Calculate accuracy by finding max log probability
            _, pred = torch.max(output, dim=1) # first output gives the max value in the row(not what we want), second output gives index of the highest val
            correct_tensor = pred.eq(label.data.view_as(pred)) # using the index of the predicted outcome above, torch.eq() will check prediction index against label index to see if prediction is correct(returns 1 if correct, 0 if not)
            accuracy = torch.mean(correct_tensor.type(torch.FloatTensor)) #tensor must be float to calc average
            train_acc += accuracy.item() * data.size(0)
            if ii%15 == 0:
                print(f'Epoch: {epoch}\t{100 * (ii + 1) / len(train_loader):.2f}% complete.')
      
        model.epochs += 1
        with torch.no_grad():
            model.eval()
            for data, label in val_loader:
                data, label = data.cuda(), label.cuda()
                output = model(data)
                loss = criterion(output, label)
                valid_loss += loss.item() * data.size(0)
        
                _, pred = torch.max(output, dim=1)
                correct_tensor = pred.eq(label.data.view_as(pred))
                accuracy = torch.mean(correct_tensor.type(torch.FloatTensor))
                valid_acc += accuracy.item() * data.size(0)
        
        train_loss = train_loss / len(train_loader.dataset)
        valid_loss = valid_loss / len(val_loader.dataset)
      
        train_acc = train_acc / len(train_loader.dataset)
        valid_acc = valid_acc / len(val_loader.dataset)
      
        history.append([train_loss, valid_loss, train_acc, valid_acc])
      
        if (epoch + 1) % print_every == 0:
            print(f'\nEpoch: {epoch} \tTraining Loss: {train_loss:.4f} \tValidation Loss: {valid_loss:.4f}')
            print(f'\t\tTraining Accuracy: {100 * train_acc:.2f}%\t Validation Accuracy: {100 * valid_acc:.2f}%')
        
        if valid_loss < valid_loss_min:
            torch.save(model.state_dict(), save_location)
            stop_count = 0
            valid_loss_min = valid_loss
            valid_best_acc = valid_acc
            best_epoch = epoch
        
        else:
            stop_count += 1

class GradCAM:
    def __init__(self, model, feature_layer):
        self.model = model
        self.feature_layer = feature_layer
        self.model.eval()
        self.feature_grad = None
        self.feature_map = None
        self.hooks = []

        # 最終層逆伝播時の勾配を記録する
        def save_feature_grad(module, in_grad, out_grad):
            self.feature_grad = out_grad[0]
        self.hooks.append(self.feature_layer.register_backward_hook(save_feature_grad))
        # 最終層の出力 Feature Map を記録する
        def save_feature_map(module, inp, outp):
            self.feature_map = outp[0]
        self.hooks.append(self.feature_layer.register_forward_hook(save_feature_map))

    def forward(self, x):
        return self.model(x)

    def backward_on_target(self, output, target):
        self.model.zero_grad()
        one_hot_output = torch.zeros([1, output.size()[-1]])
        one_hot_output[0][target] = 1
        output.backward(gradient=one_hot_output, retain_graph=True)

    def clear_hook(self):
        for hook in self.hooks:
            hook.remove()

path = "/Users/ora/Desktop/2019_33 ①.jpg"
VISUALIZE_SIZE = (300, 200)
image = Image.open(path)
image.thumbnail(VISUALIZE_SIZE, Image.ANTIALIAS)

plt.imshow(image)

image_orig_size = image.size # (W, H)

test_image_tensor = image_transforms['test'](image)
test_image_tensor = test_image_tensor.unsqueeze(0)
device = torch.device("cuda")
model = models.resnet50(pretrained=False)
n_classes = 2
num_ftrs = model.fc.in_features
model.fc = nn.Linear(num_ftrs, n_classes)
model.to(device)
model.eval()

from collections import OrderedDict
import torch
checkpoint=torch.load('./natural_images_resnet.pt')

state_dict=checkpoint
new_state_dict=OrderedDict()

model.load_state_dict(state_dict)

#############GradCamによる可視化
grad_cam = GradCAM(model, feature_layer=list(model.layer4.modules())[26])
#画像をGradcamに入力
model_output = grad_cam.forward(test_image_tensor)

if len(model_output) == 1:
    target = model_output.argmax(1).item()
    grad_cam.backward_on_target(model_output, target)

# Get feature gradient
feature_grad = grad_cam.feature_grad.data.numpy()[0]

# Get weights from gradient
weights = np.mean(feature_grad, axis=(1, 2))  # Take averages for each gradient
# Get features outputs
feature_map = grad_cam.feature_map.data.numpy()
grad_cam.clear_hook()

cam = np.sum((weights * feature_map.T), axis=2).T
cam = np.maximum(cam, 0)  # apply ReLU to cam

cam = cv2.resize(cam, VISUALIZE_SIZE)
cam = (cam - np.min(cam)) / (np.max(cam) - np.min(cam))  # Normalize between 0-1
cam = np.uint8(cam * 255)  # Scale between 0-255 to visualize
activation_heatmap = np.uint8(cv2.applyColorMap(cam, cv2.COLORMAP_JET))
activation_heatmap = cv2.cvtColor(activation_heatmap, cv2.COLOR_BGR2RGB)  #色反転

plt.imshow(activation_heatmap)

org_img = np.asarray(image.resize(VISUALIZE_SIZE))
intensity = 0.4   
img_with_heatmap = cv2.addWeighted(activation_heatmap, intensity, org_img, 1, 0)
org_img = cv2.resize(org_img, image_orig_size)
img_with_heatmap = cv2.resize(img_with_heatmap, image_orig_size)

plt.figure(figsize=(10,5))

plt.subplot(1,2,1)
plt.imshow(org_img)
plt.xticks(color="None")
plt.yticks(color="None")
plt.tick_params(length=0)
plt.subplot(1,2,2)
plt.imshow(img_with_heatmap)
plt.xticks(color="None")
plt.yticks(color="None")
plt.tick_params(length=0)
plt.show()

Comments

[toast-uz]

エラーメッセージは、スタックトレース部分が重要なので、メッセージ部分だけを切り出さず、全体を掲載ください

[shujiu]

RuntimeError Traceback (most recent call last)
<ipython-input-57-a8dff80bae72> in <module>
30 grad_cam = GradCAM(model, feature_layer=list(model.layer4.modules())[26])
31 #画像をGradcamに入力
---> 32 model_output = grad_cam.forward(test_image_tensor)
33
34 if len(model_output) == 1:

<ipython-input-56-5b19a4a57393> in forward(self, x)
22
23 def forward(self, x):
---> 24 return self.model(x)
25
26 def backward_on_target(self, output, target):

~\anaconda3\envs\envtest2\lib\site-packages\torch\nn\modules\module.py in __call__(self, *input, **kwargs)
548 result = self._slow_forward(*input, **kwargs)
549 else:
--> 550 result = self.forward(*input, **kwargs)
551 for hook in self._forward_hooks.values():
552 hook_result = hook(self, input, result)

~\anaconda3\envs\envtest2\lib\site-packages\torchvision\models\resnet.py in forward(self, x)
218
219 def forward(self, x):
--> 220 return self._forward_impl(x)
221
222

~\anaconda3\envs\envtest2\lib\site-packages\torchvision\models\resnet.py in _forward_impl(self, x)
201 def _forward_impl(self, x):
202 # See note [TorchScript super()]
--> 203 x = self.conv1(x)
204 x = self.bn1(x)
205 x = self.relu(x)

~\anaconda3\envs\envtest2\lib\site-packages\torch\nn\modules\module.py in __call__(self, *input, **kwargs)
548 result = self._slow_forward(*input, **kwargs)
549 else:
--> 550 result = self.forward(*input, **kwargs)
551 for hook in self._forward_hooks.values():
552 hook_result = hook(self, input, result)

~\anaconda3\envs\envtest2\lib\site-packages\torch\nn\modules\conv.py in forward(self, input)
347
348 def forward(self, input):
--> 349 return self._conv_forward(input, self.weight)
350
351 class Conv3d(_ConvNd):

~\anaconda3\envs\envtest2\lib\site-packages\torch\nn\modules\conv.py in _conv_forward(self, input, weight)
344 _pair(0), self.dilation, self.groups)
345 return F.conv2d(input, weight, self.bias, self.stride,
--> 346 self.padding, self.dilation, self.groups)
347
348 def forward(self, input):
RuntimeError: Input type (torch.FloatTensor) and weight type (torch.cuda.FloatTensor) should be the same

[shujiu]

申し訳ありません。Error messageについて追記しました.
よろしくお願いいたします。

Answer 1

手元で再現試験などしていないため、ソースコードとエラーメッセージのみからの判断になりますが、test_image_tensor がCPU上に置かれている一方で grad_cam (より正確には、初期化時に渡されているResNet50のモデル) がGPU上に転送済みになっている (=GradCAM() へ渡す前に model.to(device) している) ことによる、型の不一致が原因と思われます。

forward() の実行前に test_image_tensor を test_image_tensor.to(device) として、GPUへデータを転送しておくと動くのではないかと思います。

ちなみに最初に添付いただいたエラーメッセージをよーく見ていただくと、

• Input type (torch.FloatTensor)
• weight type (torch. cuda .FloatTensor)

と、入力のデータ型と重みのデータ型が違うことが示されているので、注意深く見ていただくと良いかと思います。(といいつつ、見落として深みにハマりがちなので難しいところですが……)

Comments

[shujiu]

上記の通り修正いたしましたところ、問題なく稼働いたしました。
ありがとうございました