<jupyter_start><jupyter_text>colab<jupyter_code>!nvidia-smi
# ドライブをマウント
import sys
if 'google.colab' in sys.modules:
    from google.colab import drive
    drive.mount('/content/drive')
import os, sys

if "google.colab" in sys.modules:
    CP_DIR = f"/content/drive/MyDrive/Work/probspace_religious_art/notebook/{NAME}_colab/output"
    INPUT_DIR = "./eda_output/output"
    sys.path.append("/content/drive/MyDrive/Work/probspace_religious_art/code")
elif "kaggle_web_client" in sys.modules:
    pass
elif "/kqi/output" in os.getcwd():
    pass
else:
    # local
    CP_DIR = "output"
    INPUT_DIR = "../../eda/output"
    sys.path.append("../../../code")
    sys.path.append('../../../Git/Ranger-Deep-Learning-Optimizer')
    sys.path.append('../../../Git/pytorch-optimizer')

from mix_aug import cutmix, fmix, snapmix, SnapMixLoss, resizemix
# driveからzipコピーしてくる
if os.getcwd() == "/content" and os.path.exists(INPUT_DIR) == False:
    !mkdir -p "./eda_output"
    !cp -r "/content/drive/MyDrive/Work/probspace_religious_art/notebook/eda/output.zip" "./eda_output"
    !unzip -qq "./eda_output/output.zip" -d "./eda_output"
    pass
# colabで足りないライブラリinstall
import os, sys
if ("google.colab" in sys.modules) or ("kaggle_web_client" in sys.modules) or ("/kqi/output" in os.getcwd()):
    !pip install --upgrade albumentations
    !pip install --upgrade timm
    !pip install torch-optimizer
    pass<jupyter_output><empty_output><jupyter_text>data load<jupyter_code>import pandas as pd

# ====================================================
# Data Load
# ====================================================
def get_test_file_path(image_id):
    return f"{INPUT_DIR}/test/{str(image_id)}.jpg"

test = pd.read_csv(INPUT_DIR + "/test.csv")
test["file_path"] = test["image_id"].apply(get_test_file_path)<jupyter_output><empty_output><jupyter_text>influence<jupyter_code>CFGs = []
NUM_CLASSES = 13
TTA_ROUND = 2
import yaml

# ====================================================
# Param
# ====================================================
cfg_yml = CP_DIR + "/cfg.yaml"

with open(cfg_yml, "r") as f:
    cfg_dict = yaml.safe_load(f)
cfg_dict["batch_size"] = cfg_dict["batch_size"] * 2
cfg_dict["num_workers"] = os.cpu_count() if ("google.colab" in sys.modules) or ("kaggle_web_client" in sys.modules) or ("/kqi/output" in os.getcwd()) else 0
cfg_dict["name"] = NAME

class Cfg:
    def __init__(self, **entries):
        self.__dict__.update(entries)
CFG = Cfg(**cfg_dict)
print("CFG:", CFG.__dict__)

CFGs.append(CFG)
import glob
from sklearn.metrics import accuracy_score

for seed in CFG.seeds:
    oof_df = pd.read_csv(CP_DIR + f"/oof_seed{seed}.csv")
    score = accuracy_score(oof_df['label'].values, oof_df['pred'].values)
    print("seed:", seed, ", oof score:", score)
import os
import sys
import cv2
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

import torch
from torch.utils.data import DataLoader, Dataset

import albumentations as A
from albumentations import Compose
from albumentations.pytorch import ToTensorV2

# ====================================================
# Dataset
# ====================================================
class TestDataset(Dataset):
    def __init__(self, df, transform=None):
        super().__init__()
        self.df = df
        self.file_paths = df["file_path"].values
        self.transform = transform

    def __len__(self):
        return len(self.df)

    def __getitem__(self, idx):
        file_path = self.file_paths[idx]
        image = cv2.imread(file_path)
        image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB).astype(np.float32)
        if self.transform:
            augmented = self.transform(image=image)
            image = augmented["image"]
        return image


def get_transforms(*, data):

    if data == "train":
        pass

    elif data == "valid":
        return Compose(
            [
                A.Resize(CFG.size, CFG.size),
                A.Normalize(
                    mean=[0.485, 0.456, 0.406],
                    std=[0.229, 0.224, 0.225],
                ),
                ToTensorV2(),
            ]
        )


def collate(batch):
    """DataLoaderに追加可能なbatchを加工する関数"""
    images, labels = list(zip(*batch))
    images = torch.stack(images)
    labels = torch.stack(labels)
    return images, labels.long()


# ====================================================
# Library
# ====================================================
import sys
import os
import gc
import re
import math
import time
import random
import yaml
import shutil
import glob
import pickle
import pathlib
from pathlib import Path
from contextlib import contextmanager
from collections import defaultdict, Counter

import scipy as sp
import numpy as np
import pandas as pd
from tqdm.auto import tqdm

from functools import partial

import cv2
from PIL import Image

import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.nn.parameter import Parameter
from torch.utils.data import DataLoader, Dataset
from torch.cuda.amp import autocast, GradScaler
sys.path.append(r'C:\Users\81908\Git\Ranger-Deep-Learning-Optimizer')
sys.path.append(r'C:\Users\81908\Git\pytorch-optimizer')
from torch_optimizer import RAdam, Lookahead

import timm
print("timm version:", timm.__version__)

import warnings
warnings.filterwarnings("ignore")

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(device)


# ====================================================
# Helper functions
# ====================================================
def tta(img, ops):
    # input: NxCxHxW
    if ops == 0:
        pass
    elif ops == 1:
        img = torch.flip(img, [-1])
    elif ops == 2:
        img = torch.flip(img, [-2])
    elif ops == 3:
        img = torch.flip(img, [-1, -2])
    elif ops == 4:
        img = torch.rot90(img, 1, [2, 3])
    elif ops == 5:
        img = torch.rot90(img, 3, [2, 3])
    else:
        pass
    return img


def test_fn(test_loader, model, device, tta_round=1):
    # switch to evaluation mode
    model.eval()
    preds = []
    tk0 = tqdm(test_loader, total=len(test_loader))
    for step, (images) in enumerate(tk0):
        images = images.to(device)
        batch_size = images.size(0)
        with torch.no_grad():
            with autocast():
                if tta_round == 1:
                    predictions = model.forward_softmax(images)  # 確信度
                else:
                    predictions = torch.zeros([images.shape[0], NUM_CLASSES], device=device)
                    for i, x in enumerate(images):
                        for ops in range(tta_round):
                            xi = torch.unsqueeze(tta(x, ops), 0)
                            predictions[i] += model.forward_softmax(xi)[0] / tta_round  # 確信度
        pred = predictions.detach().cpu().numpy()
        preds.append(pred)
    preds = np.concatenate(preds)
    return preds


# ====================================================
# Utils
# ====================================================
def seed_torch(seed=42):
    random.seed(seed)
    os.environ["PYTHONHASHSEED"] = str(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    torch.cuda.manual_seed(seed)
    torch.cuda.manual_seed_all(seed)
    torch.backends.cudnn.deterministic = True


# ====================================================
# Model
# ====================================================
class TimmModel(nn.Module):
    def __init__(self, n_classes, model_name="resnet18", pretrained=True):
        super().__init__()
        self.cnn = timm.create_model(model_name, pretrained=pretrained)
        if "efficient" in model_name:
            self.cnn.classifier = nn.Linear(self.cnn.classifier.in_features, n_classes)
        elif "nfnet" in model_name:
            self.cnn.head.fc = nn.Linear(self.cnn.head.fc.in_features, n_classes)
        elif "vit" in model_name:
            self.cnn.head = nn.Linear(self.cnn.head.in_features, n_classes)
        elif "tnt" in model_name:
            self.cnn.head = nn.Linear(self.cnn.head.in_features, n_classes)
        elif "swin" in model_name:
            self.cnn.head = nn.Linear(self.cnn.head.in_features, n_classes)
        elif "cait" in model_name:
            self.cnn.head = nn.Linear(self.cnn.head.in_features, n_classes)
        elif "mixer" in model_name:
            self.cnn.head = nn.Linear(self.cnn.head.in_features, n_classes)
        else:
            self.cnn.fc = nn.Linear(self.cnn.fc.in_features, n_classes)

    def forward(self, x):
        return self.cnn(x)
    
    def forward_softmax(self, x):
        return torch.softmax(self.cnn(x), 1)
    
    def forward_argmax(self, x):
        return self.cnn(x).argmax(1)


# ====================================================
# main
# ====================================================
def main(test, CFGs):

    Y_pred_ens = np.zeros((len(test), NUM_CLASSES))
    
    for CFG in CFGs:
        
        print(f"========== {CFG.name} testing ==========")
        
        Y_pred_seed = np.zeros((len(test), CFG.n_classes))
    
        for seed in CFG.seeds:
            seed_torch(seed=seed)
            
            print(f"========== seed: {seed} testing ==========")

            Y_pred = np.zeros((len(test), CFG.n_classes))
            #Y_pred = pd.DataFrame(Y_pred, index=test.index)  # , columns=Y.columns
            if CFG.debug:
                test = test.sample(n=1000, random_state=seed).reset_index(drop=True)
                Y_pred = np.zeros((len(test), CFG.n_classes))

            for fold in CFG.trn_fold:

                print(f"========== fold: {fold} testing ==========")

                #states = torch.load(f'fold{CFG.trn_fold[fold]}_seed{seed}_best.pth', map_location=torch.device('cpu'))
                states = torch.load(CP_DIR + f'/fold{CFG.trn_fold[fold]}_seed{seed}_best.pth', map_location=torch.device('cpu'))
                model = TimmModel(CFG.n_classes, model_name=CFG.model_name, pretrained=False)
                model.to(device)
                model.load_state_dict(states['model'])
                del states; gc.collect()

                test_dataset = TestDataset(test, transform=get_transforms(data='valid'))
                test_loader = DataLoader(test_dataset, 
                                         batch_size=CFG.batch_size, 
                                         shuffle=False, 
                                         num_workers=CFG.num_workers)
                Y_pred += test_fn(test_loader, model, device, tta_round=TTA_ROUND) / len(CFG.trn_fold)
                del test_loader, model; gc.collect()
    
            _test = test.copy()
            _test["label"] = Y_pred.argmax(1)  # soft avgしたlogitが最大のidをラベルとする
            _test = pd.concat([_test, pd.DataFrame(Y_pred)], axis=1)  # 確信度も残す
            _test.to_csv(CP_DIR+f"/{NAME}_inf_test_seed{seed}.csv", index=False)
            print(f"OUTPUT: {CP_DIR}/{NAME}_inf_test_seed{seed}.csv")
            
            Y_pred_seed += Y_pred / len(CFG.seeds)

        Y_pred_ens += Y_pred_seed / len(CFGs)
    
    test["label"] = Y_pred_ens.argmax(1)  # soft avgしたlogitが最大のidをラベルとする
    test = pd.concat([test, pd.DataFrame(Y_pred_ens)], axis=1)  # 確信度も残す
    test.to_csv(CP_DIR+f"/{NAME}_inf_test_seed_mean.csv", index=False)
    print(f"OUTPUT: {CP_DIR}/{NAME}_inf_test_seed_mean.csv")

    return test


if __name__ == "__main__":
    # ====================================================
    # make submission.csv
    # ====================================================
    test = main(test, CFGs)
    print("test.shape: " + str(test.shape))
    #display(test.head())
    
    print("\ntest finish!!!")<jupyter_output>timm version: 0.4.9
cuda
========== ex15_pseudo3 testing ==========
========== seed: 0 testing ==========
========== fold: 0 testing ==========<jupyter_text>make submission.csv<jupyter_code>sub = pd.read_csv(f"{CP_DIR}/{NAME}_inf_test_seed_mean.csv")[['image_id', 'label']]
sub = sub.rename(columns={'image_id':"id", 'label':"y"})
sub.to_csv(f"{CP_DIR}/{NAME}_seed_mean_submission.csv", index=False)
sub<jupyter_output><empty_output>
StarcoderdataJupyterStructuredCleanDedup