csapi 프로젝트 (3) - celery 함수화

https://not-to-be-reset.tistory.com/343 글에 작성한 대로, 모델 다운로드 링크가 만료되고 코드가 동작하지 않아서 다시 포스팅 합니다.
단일 인스턴스 부터 쿠버네티스 클러스터까지 확장해 나가려고 합니다.

개발환경

2023. 7. 22 기준 wsl 윈도우에서 진행

github 주소

https://github.com/kyeongjun-dev/csapi

커밋 : https://github.com/kyeongjun-dev/csapi/tree/1f31bd5716c217bf741375102b20e457c56d43c2

---

tasks.py 코드 변경

현재는 test_input.png 라는 이름의 파일만 이미지처리를 하고, out.png 라는 이름으로만 결과물을 출력합니다. 이를 변경하기 위해 아래의 원래 코드와 비교하여 tasks.py 코드를 변경합니다.

- 변경 전

import cv2
import numpy as np
from tensorflow.keras.models import load_model
import tensorflow as tf

class fashion_tools(object):
    def __init__(self,imageid,model,version=1.1):
        self.imageid = imageid
        self.model   = model
        self.version = version
        
    def get_dress(self,stack=False):
        """limited to top wear and full body dresses (wild and studio working)"""
        """takes input rgb----> return PNG"""
        name =  self.imageid
        file = cv2.imread(name)
        file = tf.image.resize_with_pad(file,target_height=512,target_width=512)
        rgb  = file.numpy()
        file = np.expand_dims(file,axis=0)/ 255.
        seq = self.model.predict(file)
        seq = seq[3][0,:,:,0]
        seq = np.expand_dims(seq,axis=-1)
        c1x = rgb*seq
        c2x = rgb*(1-seq)
        cfx = c1x+c2x
        dummy = np.ones((rgb.shape[0],rgb.shape[1],1))
        rgbx = np.concatenate((rgb,dummy*255),axis=-1)
        rgbs = np.concatenate((cfx,seq*255.),axis=-1)
        if stack:
            stacked = np.hstack((rgbx,rgbs))
            return stacked
        else:
            return rgbs

    def get_patch(self):
        return None

from celery import Celery, current_task

app = Celery('tasks', broker="pyamqp://guest:guest@rabbit:5672//")

@app.task
def api():
    saved = load_model("./model.h5")
    api = fashion_tools("test_input.png", saved)
    image_ = api.get_dress(False)
    print('task ID :', current_task.request.id)
    cv2.imwrite("out.png", image_)
    return current_task.request.id

 

- 변경 후

import cv2
import numpy as np
from tensorflow.keras.models import load_model
import tensorflow as tf

class fashion_tools(object):
    def __init__(self,imageid,model,version=1.1):
        self.imageid = imageid
        self.model   = model
        self.version = version
        
    def get_dress(self,stack=False):
        """limited to top wear and full body dresses (wild and studio working)"""
        """takes input rgb----> return PNG"""
        name =  self.imageid
        file = cv2.imread(name)
        file = tf.image.resize_with_pad(file,target_height=512,target_width=512)
        rgb  = file.numpy()
        file = np.expand_dims(file,axis=0)/ 255.
        seq = self.model.predict(file)
        seq = seq[3][0,:,:,0]
        seq = np.expand_dims(seq,axis=-1)
        c1x = rgb*seq
        c2x = rgb*(1-seq)
        cfx = c1x+c2x
        dummy = np.ones((rgb.shape[0],rgb.shape[1],1))
        rgbx = np.concatenate((rgb,dummy*255),axis=-1)
        rgbs = np.concatenate((cfx,seq*255.),axis=-1)
        if stack:
            stacked = np.hstack((rgbx,rgbs))
            return stacked
        else:
            return rgbs

    def get_patch(self):
        return None

from celery import Celery, current_task

app = Celery('tasks', broker="pyamqp://guest:guest@rabbit:5672//")

import uuid

@app.task
def api(input_file_name):
    saved = load_model("./model.h5")
    api = fashion_tools(input_file_name, saved)
    image_ = api.get_dress(False)
    print('task ID :', current_task.request.id)
    cv2.imwrite(str(uuid.uuid4()), image_)
    return current_task.request.id

 

- 변경사항

 app = Celery('tasks', broker="pyamqp://guest:guest@rabbit:5672//")
 
+import uuid
+
 @app.task
-def api():
+def api(input_file_name):
     saved = load_model("./model.h5")
-    api = fashion_tools("test_input.png", saved)
+    api = fashion_tools(input_file_name, saved)
     image_ = api.get_dress(False)
     print('task ID :', current_task.request.id)
-    cv2.imwrite("out.png", image_)
+    cv2.imwrite(str(uuid.uuid4()), image_)
     return current_task.request.id

 

테스트

먼저 동작을 잘 하는지 확인하기 위해 test_input.png 파일을 이름을 다르게 하여 복사합니다.

tree csapi/
csapi/
├── celery
│   ├── Dockerfile
│   ├── model.h5
│   ├── requirements.txt
│   ├── run.py
│   ├── tasks.py
│   ├── test_input.png
│   ├── test_input_1.png
│   ├── test_input_2.png
│   └── test_input_3.png
└── docker-compose.yml

 

다시 docker-compose build를 통해 복사한 이미지 파일들이 컨테이너 안에 저장되도록 합니다

docker-compose up --build

 

컨테이너 안에 복사한 이미지들이 있는지 확인합니다.

docker exec csapi-worker-1 ls
Dockerfile
__pycache__
model.h5
requirements.txt
run.py
tasks.py
test_input.png
test_input_1.png
test_input_2.png
test_input_3.png

 

아래 순서로 celery 동작을 확인합니다.

docker exec -it csapi-worker-1 bash
root@0d0a90f6082d:/csapi#

root@0d0a90f6082d:/csapi# python
Python 3.7.16 (default, May 23 2023, 14:46:23) 
[GCC 10.2.1 20210110] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>>

>>> from tasks import api
2023-07-22 14:56:04.626862: I tensorflow/core/platform/cpu_feature_guard.cc:193] This TensorFlow binary is optimized with oneAPI Deep Neural Network Library (oneDNN) to use the following CPU instructions in performance-critical operations:  AVX2 FMA
To enable them in other operations, rebuild TensorFlow with the appropriate compiler flags.
2023-07-22 14:56:04.790861: W tensorflow/compiler/xla/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libcudart.so.11.0'; dlerror: libcudart.so.11.0: cannot open shared object file: No such file or directory
2023-07-22 14:56:04.790943: I tensorflow/compiler/xla/stream_executor/cuda/cudart_stub.cc:29] Ignore above cudart dlerror if you do not have a GPU set up on your machine.
2023-07-22 14:56:05.606489: W tensorflow/compiler/xla/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libnvinfer.so.7'; dlerror: libnvinfer.so.7: cannot open shared object file: No such file or directory
2023-07-22 14:56:05.606648: W tensorflow/compiler/xla/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libnvinfer_plugin.so.7'; dlerror: libnvinfer_plugin.so.7: cannot open shared object file: No such file or directory
2023-07-22 14:56:05.606669: W tensorflow/compiler/tf2tensorrt/utils/py_utils.cc:38] TF-TRT Warning: Cannot dlopen some TensorRT libraries. If you would like to use Nvidia GPU with TensorRT, please make sure the missing libraries mentioned above are installed properly.
>>>

>>> api.delay("test_input.png")
<AsyncResult: ab6eb916-61bf-42c0-9195-000539044159>
>>>

 

docker compose의 로그를 확인하면 아래와 같이 에러가 발생합니다.

csapi-worker-1  | [2023-07-22 14:57:15,486: ERROR/ForkPoolWorker-1] Task tasks.api[ab6eb916-61bf-42c0-9195-000539044159] raised unexpected: error("OpenCV(4.5.1) /tmp/pip-req-build-7m_g9lbm/opencv/modules/imgcodecs/src/loadsave.cpp:682: error: (-2:Unspecified error) could not find a writer for the specified extension in function 'imwrite_'\n")
csapi-worker-1  | Traceback (most recent call last):
csapi-worker-1  |   File "/usr/local/lib/python3.7/site-packages/celery/app/trace.py", line 450, in trace_task
csapi-worker-1  |     R = retval = fun(*args, **kwargs)
csapi-worker-1  |   File "/usr/local/lib/python3.7/site-packages/celery/app/trace.py", line 731, in __protected_call__
csapi-worker-1  |     return self.run(*args, **kwargs)
csapi-worker-1  |   File "/csapi/tasks.py", line 50, in api
csapi-worker-1  |     cv2.imwrite(str(uuid.uuid4()), image_)
csapi-worker-1  | cv2.error: OpenCV(4.5.1) /tmp/pip-req-build-7m_g9lbm/opencv/modules/imgcodecs/src/loadsave.cpp:682: error: (-2:Unspecified error) could not find a writer for the specified extension in function 'imwrite_'

 

cv.imwrite 함수에서 파일의 확장자를 지정하지 않아서 입니다. (감사합니다 - 링크)
tasks.py 코드를 수정합니다. (확장자 부분을 파싱해서, uuid 문자열에 붙입니다.)

import cv2
import numpy as np
from tensorflow.keras.models import load_model
import tensorflow as tf

class fashion_tools(object):
    def __init__(self,imageid,model,version=1.1):
        self.imageid = imageid
        self.model   = model
        self.version = version
        
    def get_dress(self,stack=False):
        """limited to top wear and full body dresses (wild and studio working)"""
        """takes input rgb----> return PNG"""
        name =  self.imageid
        file = cv2.imread(name)
        file = tf.image.resize_with_pad(file,target_height=512,target_width=512)
        rgb  = file.numpy()
        file = np.expand_dims(file,axis=0)/ 255.
        seq = self.model.predict(file)
        seq = seq[3][0,:,:,0]
        seq = np.expand_dims(seq,axis=-1)
        c1x = rgb*seq
        c2x = rgb*(1-seq)
        cfx = c1x+c2x
        dummy = np.ones((rgb.shape[0],rgb.shape[1],1))
        rgbx = np.concatenate((rgb,dummy*255),axis=-1)
        rgbs = np.concatenate((cfx,seq*255.),axis=-1)
        if stack:
            stacked = np.hstack((rgbx,rgbs))
            return stacked
        else:
            return rgbs

    def get_patch(self):
        return None

from celery import Celery, current_task

app = Celery('tasks', broker="pyamqp://guest:guest@rabbit:5672//")

import uuid

@app.task
def api(input_file_name):
    saved = load_model("./model.h5")
    api = fashion_tools(input_file_name, saved)
    image_ = api.get_dress(False)
    file_type = input_file_name.split('.')[1]
    cv2.imwrite(str(uuid.uuid4()) + '.' + file_type, image_)
    return current_task.request.id

 

다시 docker-compose up --build 부터 진행합니다

docker exec -it csapi-worker-1 bash
root@4c6739e4e39e:/csapi#

root@4c6739e4e39e:/csapi# python
Python 3.7.16 (default, May 23 2023, 14:46:23) 
[GCC 10.2.1 20210110] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>>

>>> from tasks import api
2023-07-22 16:03:27.345265: I tensorflow/core/platform/cpu_feature_guard.cc:193] This TensorFlow binary is optimized with oneAPI Deep Neural Network Library (oneDNN) to use the following CPU instructions in performance-critical operations:  AVX2 FMA
To enable them in other operations, rebuild TensorFlow with the appropriate compiler flags.
2023-07-22 16:03:27.767133: W tensorflow/compiler/xla/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libcudart.so.11.0'; dlerror: libcudart.so.11.0: cannot open shared object file: No such file or directory
2023-07-22 16:03:27.767211: I tensorflow/compiler/xla/stream_executor/cuda/cudart_stub.cc:29] Ignore above cudart dlerror if you do not have a GPU set up on your machine.
2023-07-22 16:03:28.843788: W tensorflow/compiler/xla/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libnvinfer.so.7'; dlerror: libnvinfer.so.7: cannot open shared object file: No such file or directory
2023-07-22 16:03:28.843989: W tensorflow/compiler/xla/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libnvinfer_plugin.so.7'; dlerror: libnvinfer_plugin.so.7: cannot open shared object file: No such file or directory
2023-07-22 16:03:28.844555: W tensorflow/compiler/tf2tensorrt/utils/py_utils.cc:38] TF-TRT Warning: Cannot dlopen some TensorRT libraries. If you would like to use Nvidia GPU with TensorRT, please make sure the missing libraries mentioned above are installed properly.
>>>

>>> api.delay("test_input.png")
<AsyncResult: 9cc70caf-8b28-4a5f-87c1-9d0ea32bd585>
>>>

 

docker compose 로그를 확인하면 성공한 것을 확인할수 있습니다

csapi-worker-1  | 
csapi-worker-1  | 
csapi-worker-1  | [2023-07-22 16:04:12,326: INFO/ForkPoolWorker-1] Task tasks.api[9cc70caf-8b28-4a5f-87c1-9d0ea32bd585] succeeded in 10.483806963999996s: '9cc70caf-8b28-4a5f-87c1-9d0ea32bd585'
csapi-rabbit-1  | 2023-07-22 16:04:12.450570+00:00 [warning] <0.770.0> closing AMQP connection <0.770.0> (172.18.0.2:46302 -> 172.18.0.3:5672, vhost: '/', user: 'guest'):
csapi-rabbit-1  | 2023-07-22 16:04:12.450570+00:00 [warning] <0.770.0> client unexpectedly closed TCP connection

 

파이썬 쉘을 실행한 컨테이너에서 파일 목록을 확인하면 아래와 같이 <uuid>.png 파일이 생성된 것을 확인할 수 있습니다

>>> exit()

root@4c6739e4e39e:/csapi# ls
5fd3d852-1d90-4093-b2a6-d1484805bf8e.png  tasks.py
Dockerfile                                test_input.png
__pycache__                               test_input_1.png
model.h5                                  test_input_2.png
requirements.txt                          test_input_3.png
run.py

 

<uuid>.png 파일을 확인해보면 아래와 같이 정상적으로 이미지 처리가 된것을 확인할 수 있습니다.

 

이어서 다른 input 이미지들도 진행합니다

root@4c6739e4e39e:/csapi# python
Python 3.7.16 (default, May 23 2023, 14:46:23) 
[GCC 10.2.1 20210110] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>>

>>> from tasks import api
2023-07-22 16:07:45.579900: I tensorflow/core/platform/cpu_feature_guard.cc:193] This TensorFlow binary is optimized with oneAPI Deep Neural Network Library (oneDNN) to use the following CPU instructions in performance-critical operations:  AVX2 FMA
To enable them in other operations, rebuild TensorFlow with the appropriate compiler flags.
2023-07-22 16:07:45.734173: W tensorflow/compiler/xla/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libcudart.so.11.0'; dlerror: libcudart.so.11.0: cannot open shared object file: No such file or directory
2023-07-22 16:07:45.734230: I tensorflow/compiler/xla/stream_executor/cuda/cudart_stub.cc:29] Ignore above cudart dlerror if you do not have a GPU set up on your machine.
2023-07-22 16:07:46.580093: W tensorflow/compiler/xla/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libnvinfer.so.7'; dlerror: libnvinfer.so.7: cannot open shared object file: No such file or directory
2023-07-22 16:07:46.580159: W tensorflow/compiler/xla/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libnvinfer_plugin.so.7'; dlerror: libnvinfer_plugin.so.7: cannot open shared object file: No such file or directory
2023-07-22 16:07:46.580179: W tensorflow/compiler/tf2tensorrt/utils/py_utils.cc:38] TF-TRT Warning: Cannot dlopen some TensorRT libraries. If you would like to use Nvidia GPU with TensorRT, please make sure the missing libraries mentioned above are installed properly.
>>>

>>> api.delay("test_input_1.png")
<AsyncResult: 1d593d42-baaf-4d83-80ae-ac163a897146>
>>> api.delay("test_input_2.png")
<AsyncResult: 28374bab-79e2-47d4-9077-47dde472575f>
>>> api.delay("test_input_3.png")
<AsyncResult: d75173c0-ff04-442a-b986-a5d49b44c9a3>
>>> exit()

root@4c6739e4e39e:/csapi# ls
5fd3d852-1d90-4093-b2a6-d1484805bf8e.png  requirements.txt
Dockerfile                                run.py
__pycache__                               tasks.py
b194c9f7-202e-476f-8d99-05a1b5cb15b0.png  test_input.png
d19de34b-f1d0-4235-ba98-54846d7dee12.png  test_input_1.png
d6ec2046-c633-407d-bb0e-830361e60726.png  test_input_2.png
model.h5                                  test_input_3.png