import os
import pickle
import numpy as np
import torch
import cv2
import logging
logger = globals().get('JOB_LOGGER', logging.getLogger())
logger.setLevel(logging.INFO)
class ModelHandler:
def __init__(self, data, context):
self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
self.base_path = os.path.dirname(os.path.abspath(__file__))
model_path = os.path.join(self.base_path, "SampleModel.pth")
self.model = torch.load(model_path, map_location=self.device)
self.model.to(self.device)
self.model.eval()
logger.info(f"[LOG] 모델 로드 완료")
def __call__(self, data, context):
data = pickle.loads(data)
images = np.array(data)
images = images.astype(np.float32) / 255.0
images = torch.from_numpy(images).permute(2, 0, 1).unsqueeze(0).to(self.device)
with torch.no_grad():
predict_result = self.model(images)
predict_result = predict_result.cpu().numpy()
result = {}
result['result_code'] = int(np.argmax(predict_result))
result['score'] = float(np.max(predict_result))
logger.info(f"[RESULT] 예측: {result['result_code']}, 점수: {result['score']:.4f}")
return pickle.dumps(result), context
if __name__ == "__main__":
img = np.random.randint(0, 255, (299, 299, 3), dtype=np.uint8)
data = pickle.dumps(img)
handler = ModelHandler(None, None)
result, context = handler(data=data, context={})
result_dict = pickle.loads(result)
print(f"결과: {result_dict['result_code']}, 점수: {result_dict['score']:.4f}")