[kaggle] 🐘Greyscale MobileNet [LB=0.892] - Review 하기

이 글은 친구와 kaggle에 익숙해지기 위해 스터디하는 걸 정리해봤습니다.
목적으로 코드도 중요하지만 데이터를 어떻게 다루고 왜 다루고 전처리는 어떤 방식으로 하는 지 궁금해기 때문에 하나하나 관찰하며 혜안을 얻고자 하는 목적입니다.

CODE

https://www.kaggle.com/code/gaborfodor/greyscale-mobilenet-lb-0-892/notebook

흑백이미지 낙서 분류하기

이 문제는 340개의 클래스로 나눠진 흑백이미지 낙서 사진을 분류한다. 모델이 분류해낸 상위 3개의 데이터를 제출하면 된다.
제출 파일은 다음과 방식으로 이해하면 된다.

key_id	word
9000003627287624	radio stereo stove
9000010688666847	hockey_puck bottlecap steak
9000023642890129	The_Great_Wall_of_China castle fence
9000038588854897	mountain tent The_Eiffel_Tower
9000052667981386	fireplace campfire fire_hydrant

DP_DIR = '../input/shuffle-csvs/'
INPUT_DIR = '../input/quickdraw-doodle-recognition/'

BASE_SIZE = 256
NCSVS = 100 # CSV 파일 개수
NCATS = 340 # 총 클래스 개수
np.random.seed(seed=1987)
tf.set_random_seed(seed=1987)

# 파일 이름에서 카테코리이름을 추출하는 함수 입니다.
def f2cat(filename: str) -> str:
    return filename.split('.')[0]

# 모든 카테고리를 리스트로 반환합니다.
def list_all_categories():
    files = os.listdir(os.path.join(INPUT_DIR, 'train_simplified'))
    return sorted([f2cat(f) for f in files], key=str.lower)

def apk(actual, predicted, k=3):
    """
    Source: https://github.com/benhamner/Metrics/blob/master/Python/ml_metrics/average_precision.py
    """
    if len(predicted) > k:
        predicted = predicted[:k]
    score = 0.0
    num_hits = 0.0
    for i, p in enumerate(predicted):
        if p in actual and p not in predicted[:i]:
            num_hits += 1.0
            score += num_hits / (i + 1.0)
    if not actual:
        return 0.0
    return score / min(len(actual), k)

apk는 Actual Precision at K의 약자로, 예측된 상위 K개의 결과에서 실제 값과의 일치율을 평가하는 함수입니다.
actual은 실제 정답 리스트, predicted는 예측된 값 리스트, k는 평가할 상위 K개의 수입니다.
예측된 값 리스트에서 상위 K개만 사용하며, 각 예측된 값이 실제 값에 포함되는지 확인하고 점수를 계산합니다.
실제 값 리스트가 비어 있으면 0.0을 반환하고, 그렇지 않으면 평균 점수를 반환합니다.

def mapk(actual, predicted, k=3):
    """
    Source: https://github.com/benhamner/Metrics/blob/master/Python/ml_metrics/average_precision.py
    """
    return np.mean([apk(a, p, k) for a, p in zip(actual, predicted)])

mapk는 Mean Actual Precision at K의 약자로, 여러 예측 결과에 대해 평균적인 정확도를 계산합니다.
actual과 predicted는 각각 실제 값 리스트와 예측된 값 리스트입니다.
apk 함수를 사용하여 각 쌍에 대해 정확도를 계산한 후, 그 평균을 반환합니다.

def preds2catids(predictions):
    return pd.DataFrame(np.argsort(-predictions, axis=1)[:, :3], columns=['a', 'b', 'c'])

preds2catids는 예측 확률 배열을 상위 3개의 카테고리 ID로 변환합니다.

def top_3_accuracy(y_true, y_pred):
    return top_k_categorical_accuracy(y_true, y_pred, k=3)

top_3_accuracy는 상위 3개의 예측값 중 하나가 실제 값과 일치하는지 평가하는 함수입니다.

STEPS = 800
EPOCHS = 16
size = 64
batchsize = 680

model = MobileNet(input_shape=(size, size, 1), alpha=1., weights=None, classes=NCATS)
model.compile(optimizer=Adam(lr=0.002), loss='categorical_crossentropy',
              metrics=[categorical_crossentropy, categorical_accuracy, top_3_accuracy])
print(model.summary())

하이퍼 파라미터 설정 및 모델을 정의합니다.

def draw_cv2(raw_strokes, size=256, lw=6, time_color=True):
    img = np.zeros((BASE_SIZE, BASE_SIZE), np.uint8)
    for t, stroke in enumerate(raw_strokes):
        for i in range(len(stroke[0]) - 1):
            color = 255 - min(t, 10) * 13 if time_color else 255
            _ = cv2.line(img, (stroke[0][i], stroke[1][i]),
                         (stroke[0][i + 1], stroke[1][i + 1]), color, lw)
    if size != BASE_SIZE:
        return cv2.resize(img, (size, size))
    else:
        return img

draw_cv2 함수는 주어진 raw strokes (낙서) 데이터를 이미지로 변환합니다.

def image_generator_xd(size, batchsize, ks, lw=6, time_color=True):
    while True:
        for k in np.random.permutation(ks):
            filename = os.path.join(DP_DIR, 'train_k{}.csv.gz'.format(k))
            for df in pd.read_csv(filename, chunksize=batchsize):
                df['drawing'] = df['drawing'].apply(json.loads)
                x = np.zeros((len(df), size, size, 1))
                for i, raw_strokes in enumerate(df.drawing.values):
                    x[i, :, :, 0] = draw_cv2(raw_strokes, size=size, lw=lw,
                                             time_color=time_color)
                x = preprocess_input(x).astype(np.float32)
                y = keras.utils.to_categorical(df.y, num_classes=NCATS)
                yield x, y

CSV 파일에서 데이터를 읽어와 이미지를 생성하고, 전처리한 후 배치 단위로 반환합니다.

def image_generator_xd(size, batchsize, ks, lw=6, time_color=True):
    while True:
        for k in np.random.permutation(ks):
            filename = os.path.join(DP_DIR, 'train_k{}.csv.gz'.format(k))
            for df in pd.read_csv(filename, chunksize=batchsize):
                df['drawing'] = df['drawing'].apply(json.loads)
                x = np.zeros((len(df), size, size, 1))
                for i, raw_strokes in enumerate(df.drawing.values):
                    x[i, :, :, 0] = draw_cv2(raw_strokes, size=size, lw=lw,
                                             time_color=time_color)
                x = preprocess_input(x).astype(np.float32)
                y = keras.utils.to_categorical(df.y, num_classes=NCATS)
                yield x, y

df_to_image_array_xd 함수는 데이터프레임을 이미지 배열로 변환합니다.

train_datagen = image_generator_xd(size=size, batchsize=batchsize, ks=range(NCSVS - 1))
x, y = next(train_datagen)
n = 8
fig, axs = plt.subplots(nrows=n, ncols=n, sharex=True, sharey=True, figsize=(12, 12))
for i in range(n**2):
    ax = axs[i // n, i % n]
    (-x[i]+1)/2
    ax.imshow((-x[i, :, :, 0] + 1)/2, cmap=plt.cm.gray)
    ax.axis('off')
plt.tight_layout()
fig.savefig('gs.png', dpi=300)
plt.show()

callbacks = [
    ReduceLROnPlateau(monitor='val_top_3_accuracy', factor=0.75, patience=3, min_delta=0.001, mode='max', min_lr=1e-5, verbose=1),
    ModelCheckpoint('model.h5', monitor='val_top_3_accuracy', mode='max', save_best_only=True, save_weights_only=True),
]

ReduceLROnPlateau: 학습이 정체되면 학습률을 줄여주는 콜백입니다.
monitor: val_top_3_accuracy를 모니터링.
factor: 학습률 감소 비율.
patience: 개선되지 않는 에포크 수.
min_lr: 최소 학습률.

ModelCheckpoint: 최적 모델을 저장하는 콜백입니다.
monitor: val_top_3_accuracy를 모니터링.
mode: 최고값 저장.
save_best_only: 최고 성능 모델만 저장.
save_weights_only: 가중치만 저장.

hists = []
hist = model.fit_generator(
    train_datagen, steps_per_epoch=STEPS, epochs=70, verbose=1,
    validation_data=(x_valid, y_valid),
    callbacks=callbacks
)
hists.append(hist)

fit_generator: 생성기를 사용하여 모델을 학습합니다.
train_datagen: 학습 데이터 생성기.
steps_per_epoch: 에포크당 스텝 수.
epochs: 총 에포크 수.
validation_data: 검증 데이터.
callbacks: 콜백 리스트.
hists.append(hist): 학습 히스토리를 저장합니다.

test = pd.read_csv(os.path.join(INPUT_DIR, 'test_simplified.csv'))
test.head()
x_test = df_to_image_array_xd(test, size)
print(test.shape, x_test.shape)
print('Test array memory {:.2f} GB'.format(x_test.nbytes / 1024.**3))

테스트 데이터 로드 및 전처리를 합니다.

top3 = preds2catids(test_predictions)
top3.head()
top3.shape

예측 결과의 상위 3개를 반환합니다.

cats = list_all_categories()
id2cat = {k: cat.replace(' ', '_') for k, cat in enumerate(cats)}
top3cats = top3.replace(id2cat)
top3cats.head()
top3cats.shape

카테고리 이름으로 변환합니다.

test = pd.read_csv(os.path.join(INPUT_DIR, 'test_simplified.csv'))
test.head()
x_test = df_to_image_array_xd(test, size)
print(test.shape, x_test.shape)
print('Test array memory {:.2f} GB'.format(x_test.nbytes / 1024.**3 ))