[2024-05-25] 딥러닝을 이용하여 그림체 분류하기
최근 기계학습, 딥러닝 등을 공부하며 뭔가를 해보고 싶어졌다. 그래서 그림의 그림체를 분류해 보기로 했다.
…
나도 뜬금없는거 안다. 하지만 나름대로 이걸 하는 이유는 있다.
사실은 그냥 그림 AI를 사용하기 위해 그림체를 학습하려고 했다. 하지만 그림체 학습을 위해서는, 다들 알다시피 많은 그림들이 필요하다.
물론 현재 AI가 많은 논란이 있는 건 사실이다. 하지만 나는 지금 AI로 생성된 그림을 학습하려고 하고 있고… 어느정도 정상참작이 되지 않을까?
AI로 생성된 그림을 재학습한다는건 사실 권장되지 않는 방법이지만, 정말 멋진 그림들이 하필 AI로 생성된 걸 어쩌겠는가. 그래도 이것저것 세팅만 잘 해 놓으면 큰 문제는 발생하지 않는다.
아무튼, 이를 위해 인터넷에서 그림들을 가져왔다. 하지만 큰 문제가 생겼으니…
그림의 그림체가 전부 다르다!!!
그림을 학습할 때는 일관된 그림체가 권장된다. 그렇지 않으면 이도저도 아닌 이상한 그림만 나온다. 하지만 내가 모은 그림들은 전부 미묘하게 그림체가 다르고, 또 어떤 프롬포트로 생성되었는지도 모른다. 즉, 개고생의 시작이었다.
이 그림이 AI그림이라니...
이걸 일일이 손으로 분류한다는건… 할 수는 있지만 하고싶진 않다. 그래서 어떻게 할까 생각한 것이, ‘기계학습으로 분류하면 되지 않을까?’ 였다.
바로 실행에 옮겨서, 이미지들을 넘파이 배열로 변환하고 DBSCAN과 Hierarchical clustering을 이용해 이미지를 분류했다. 하지만 이게 쉽게 될리가… 당연히 나온 결과는 쓰레기값이었다. 생각해 보면 단순히 RGB값만 전달해 놨는데, 당연히 분류가 될 리가 없다.
그래서 생각한 것이, 딥러닝을 이용하는 것이었다. 딥러닝은 데이터만 때려 넣으면 왠진 모르겠지만 돌아가기 때문에 시도해볼만 했다. 하지만 또다시 문제가 생겼으니, 그림체를 어떻게 분류하냐는 것이다.
그림체는 정말 미묘한 차이로 완성된다. 선을 그리는 방식이라든지, 아니면 특유의 색감이나 신체를 표현하는 방식 등 수많은 요소를 고려해야 한다.
당연히 그림은 전혀 모르는 나는 구분할 수 없다. 아니 구분할 순 있지만, 하다보면 이게 당최 어디로 분류해야 할지 고민되는 애매한 그림들이 너무 많다.
내가 알고 있는 딥러닝에서의 분류는 label의 갯수를 정해놓고 예측하는 것이다. 하지만 나는 label을 분류할 수 없다. 어떻게 해야 할까…
뭐든 방법은 있다. 나는 다음과 같은 아이디어를 떠올렸다.
-
확실히 다른 그림과 차별화되는 그림체를 선별한다.
-
해당 그림체를 분류할 수 있는 모델을 학습한다.
-
이미지를 예측한다.
-
예측값은 확률값으로 나오므로, 무슨 그림체 몇%, 무슨 그림체 몇%… 이런 식으로 출력된다. 그러면 이 예측값으로 클러스터링을 하면 되지 않을까?
생각해 보면 비슷한 그림체면 나오는 확률값도 비슷하지 않을까? 그러면 이 확률값으로 클러스터링을 하면 분명 의미있는 결과가 나올 것이다.
실로 싱크빅한 생각이다. 나는 바로 이미지 샘플 몇 개를 분류했고, 학습에 들어갔다. 하지만 꼭 한번에 해결되는 일이 없다. 또 무슨 문제가 발생했냐 하면…
- 이미지 갯수가 부족하다.
사실 이건 큰 문제는 아니다. 데이터는 불리면 되는 것이고, 추가로 이미지 파일을 집어넣으면 되니까.
하지만 두번째 문제가 가장 문제였으니…
- GPU 메모리가 부족하다.
어… 이건 돈으로 해결해야 하는 문제다. 하지만 GTX1080… 아직 현역이야…
아무튼 이런 일 때문에 GPU를 산다는 건 배보다 배꼽이 더 크다. 나는 어떻게든 해결을 봐야 한다. 그래서 어떻게든 방법을 찾았고, 다음과 같은 방법으로 해결했다.
이미지를 배열로 넘기지 말고 제너레이터로 넘겨버리자!
무슨 뜻이냐 하면, 기존에는 이미지를 불러와서, 배열로 변환하고, 또 증강시켜서 model.fit 함수에 넘겼다. 하지만 이러면 모든 이미지 배열을 한 번에 GPU에 올리게 되므로 당연히 메모리가 부족해진다.
이때 할 수 있는 것이 제너레이터다. 제너레이터를 이용해서 한 번에 올리는 것이 아니라, 필요한 만큼만 넘기면 되는 것이다. 하지만 기본으로 제공되는 제너레이터는 내가 원하는 기능이 들어있지 않다. 나는 데이터를 불러오면서 동시에 증강시키기를 원했고, 이를 만족하는 함수는 없었다.
없으면 만들어야지… 그래서 만들었다.
datagen = ImageDataGenerator(rescale=1./255)
if augmentation:
datagen_aug = ImageDataGenerator(
rescale=1./255,
rotation_range=30,
zoom_range=0.2,
horizontal_flip=True,
fill_mode='constant',
cval=0
)
else:
datagen_aug = None
label = int(path.split(os.sep)[-2])
img = image.load_img(path, target_size=(512, 512))
img_array = image.img_to_array(img)
img_array = np.expand_dims(img_array, axis=0)
if augmentation:
for _ in range(num_augmentation):
img_array = next(datagen_aug.flow(img_array, batch_size=1))
yield image_array, label
else:
img_array = next(datagen.flow(img_array, batch_size=1))
yield image_array, label
어… 정말 단순한 함수다. 하지만 이것도 문제가 없는 것이 아니었다. 이게 한 번에 이미지를 하나씩 불러오니, 병목현상이 생긴다. 그러면 한 번에 많이 불러와서 내보내야 하는데…
그래서 멀티 스레드를 이용해서 한 번에 여러 이미지를 불러왔다. 하지만 이것도 문제가 있으니, 한 번 읽고 내보내면 바로 다음 이미지를 내보내야 하는데, 기존 코드로는 한 번 읽고 내보내고, 다시 읽고 내보내고에서 딜레이가 발생한다.
그래서 내가 알고 있는 지식을 이용했다. 그래픽 파이프라인을 배울 때 GPU는 여러 장의 이미지를 미리 생성하고, 교대로 내보낸다. 이것도 마찬가지로 교차로 내보낸다면 병목현상을 해결할 수 있을 것이다.
결론적으로, 멋지게 해결했다.
class PreloadedBatchGenerator:
def __init__(self, paths, batch_size=32, max_batch=3, augmentation=False, num_classes=23, max_queue_size=10, max_workers=None):
self.paths = paths
self.batch_size = batch_size
self.augmentation = augmentation
self.num_classes = num_classes
self.queue = queue.Queue(maxsize=max_queue_size)
self.stop_thread = False
self.max_workers = max_workers
self.max_batch = max_batch
self.batches = [[]] * self.max_batch
self.datagen = ImageDataGenerator(rescale=1./255)
if self.augmentation:
self.datagen_aug = ImageDataGenerator(
rescale=1./255,
rotation_range=30,
zoom_range=0.3,
horizontal_flip=True,
vertical_flip=True,
fill_mode='constant',
cval=0,
brightness_range=(0.1, 1.5),
shear_range=0.3,
width_shift_range=0.2,
height_shift_range=0.2
)
else:
self.datagen_aug = None
self.thread = threading.Thread(target=self._generate_batches)
self.thread.start()
def _process_image(self, path):
label = int(path.split(os.sep)[-2])
img = image.load_img(path, target_size=(512, 512))
img_array = image.img_to_array(img)
img_array = np.expand_dims(img_array, axis=0)
if self.augmentation:
img_array = next(self.datagen_aug.flow(img_array, batch_size=1, seed=np.random.randint(low=1, high=np.iinfo(np.int32).max)))
else:
img_array = next(self.datagen.flow(img_array, batch_size=1))
return img_array[0], tf.keras.utils.to_categorical(label, num_classes=self.num_classes)
def _generate_batches(self):
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
idx = min(self.batch_size * self.max_batch, len(self.paths))
batch_idx = 0
futures = queue.Queue()
for path in self.paths[:idx]:
futures.put(executor.submit(self._process_image, path))
try:
while True:
if self.stop_thread:
break
future = futures.get()
img, label = future.result()
while len(self.batches[batch_idx]) == self.batch_size:
time.sleep(0.1)
self.batches[batch_idx].append((img, label))
if len(self.batches[batch_idx]) == self.batch_size or idx + 1 == len(self.paths):
batch_images = np.array([item[0] for item in self.batches[batch_idx]])
batch_labels = np.array([item[1] for item in self.batches[batch_idx]])
self.queue.put((batch_images, batch_labels), block=True)
self.batches[batch_idx].clear()
batch_idx = (batch_idx + 1) % self.max_batch
idx %= len(self.paths)
futures.put(executor.submit(self._process_image, self.paths[idx]))
idx += 1
for batch_idx in range(self.max_batch):
if not self.batches[batch_idx]:
continue
batch_images = [item[0] for item in self.batches[batch_idx]]
batch_labels = [item[1] for item in self.batches[batch_idx]]
batch_images = np.array(batch_images)
batch_labels = np.array(batch_labels)
batch_labels = tf.keras.utils.to_categorical(batch_labels, num_classes=self.num_classes)
self.queue.put((batch_images, batch_labels))
except:
pass
finally:
executor.shutdown(wait=False, cancel_futures=True)
self.queue.put(None)
def __iter__(self):
return self
def __next__(self):
item = self.queue.get()
if item is None:
self.stop()
raise StopIteration
return item
def stop(self):
self.stop_thread = True
while not self.queue.empty():
self.queue.get_nowait()
self.thread.join()
메모리 한 번 절약해 보겠다고 별 짓을 다했다. 결국 나는 해냈고, 기존에는 batch size가 32면 터질락 말락 했지만 이것을 적용하고 난 후 64까지도 거뜬하게 가능했다. 혹시 다른 사람이 이미지 처리할 때 메모리가 부족하다면 이 방법을 시도해 보라고 말하고 싶어서 오늘 글을 올렸다.
split = int(len(paths) * 0.2)
train_path = paths[split:] * 10
val_path = paths[:split]
...
atch_size = 64
train_gen = PreloadedBatchGenerator(train_path, augmentation=True, batch_size=batch_size)
val_gen = PreloadedBatchGenerator(val_path, batch_size=batch_size)
model.fit(train_gen, epochs=50, batch_size=batch_size, steps_per_epoch=math.ceil(len(train_path) / batch_size), validation_data=val_gen, validation_batch_size=batch_size, validation_steps=math.ceil(len(val_path) / batch_size), callbacks=[checkpoint, early_stopping])
train_gen.stop()
val_gen.stop()
이를 통해 성공적으로 이미지를 증강시키고, 또 메모리를 절약해서 학습을 할 수 있었다.
그래서 결과가 어떻게 됐냐 하면…
아직 학습중이다. 힘내라, GTX 1080!
toss Donation