FERMAT_TEST_REST_SERVER/GLASSES_AI_SERVER
4
0
Fork 0
Code Issues Pull Requests Projects Releases 1 Wiki Activity

edit: 안광학 서버 입고 세팅

Browse Source
This commit is contained in:
김재웅
2025-09-22 14:43:31 +09:00
parent 4ef360448a
commit 605a7660c4
4 changed files with 211 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
main_rest/app/common/consts.py
View File

@@ -12,11 +12,11 @@
# SUPPORT PROJECT # SUPPORT PROJECT
SUPPORT_PROJECT_BASIC = 'PROJECT_BASIC' SUPPORT_PROJECT_BASIC = 'PROJECT_BASIC'
PROJECT_NAME = 'FERMAT-TEST' PROJECT_NAME = 'K_EYEWEAR'
SW_TITLE= f'{PROJECT_NAME} - REST API' SW_TITLE= f'{PROJECT_NAME} - REST API'
SW_VERSION = '0.1.0' SW_VERSION = '0.1.0'
SW_DESCRIPTION = f''' SW_DESCRIPTION = f'''
### FERMAT-TEST REST API ### K_EYEWEAR REST API
## API 이용법 ## API 이용법
- 개별 API 설명과 Request/Response schema 참조 - 개별 API 설명과 Request/Response schema 참조
@@ -87,7 +87,7 @@ NUM_RETRY_UUID_GEN = 3
DB_ADDRESS = "localhost" DB_ADDRESS = "localhost"
DB_PORT = 53306 DB_PORT = 53306
DB_USER_ID = 'root' DB_USER_ID = 'root'
DB_USER_PW = '1234' DB_USER_PW = 'Fermat3514!'
DB_NAME = 'FM_TEST' DB_NAME = 'FM_TEST'
DB_CHARSET = 'utf8mb4' DB_CHARSET = 'utf8mb4'

390
main_rest/app/routes/services.py
View File

@@ -123,88 +123,88 @@ router = APIRouter(prefix="/services")
# LOG.error(traceback.format_exc()) # LOG.error(traceback.format_exc())
# return response.set_error(e) # return response.set_error(e)
@router.post("/imageGenerate/imagen", summary="이미지 생성(AI) - imagen", response_model=M.ResponseBase) # @router.post("/imageGenerate/imagen", summary="이미지 생성(AI) - imagen", response_model=M.ResponseBase)
async def imagen(request: Request, request_body_info: M.ImageGenerateReq): # async def imagen(request: Request, request_body_info: M.ImageGenerateReq):
""" # """
## 이미지 생성(AI) - imagen # ## 이미지 생성(AI) - imagen
> imagen AI를 이용하여 이미지 생성 # > imagen AI를 이용하여 이미지 생성
""" # """
# imagen 사용중단 gemini로 변경 # # imagen 사용중단 gemini로 변경
response = M.ResponseBase() # response = M.ResponseBase()
try: # try:
# NOTE(JWKIM) : imagen 사용 중단 # # NOTE(JWKIM) : imagen 사용 중단
# img_length = imagen_generate_image(prompt=request_body_info.prompt, # # img_length = imagen_generate_image(prompt=request_body_info.prompt,
# download_count=request_body_info.downloadCount # # download_count=request_body_info.downloadCount
# ) # # )
temp_image_path = gemini_image(request_body_info.prompt) # temp_image_path = gemini_image(request_body_info.prompt)
if rest_config.config != 'release': # if rest_config.config != 'release':
_remote_folder = os.path.join(rest_config.remote_folder,"imagen") # _remote_folder = os.path.join(rest_config.remote_folder,"imagen")
# remote save # # remote save
sftp_client.remote_copy_data( # sftp_client.remote_copy_data(
temp_image_path, # temp_image_path,
os.path.join(_remote_folder,f"imagen_{request_body_info.prompt}_{1}_{D.date_file_name()}.png")) # os.path.join(_remote_folder,f"imagen_{request_body_info.prompt}_{1}_{D.date_file_name()}.png"))
else: # else:
_local_forder = os.path.join(rest_config.local_folder,"image_generate","imagen") # _local_forder = os.path.join(rest_config.local_folder,"image_generate","imagen")
os.makedirs(_local_forder,exist_ok=True) # os.makedirs(_local_forder,exist_ok=True)
shutil.copy(temp_image_path, os.path.join(_local_forder,f"imagen_{request_body_info.prompt}_{1}_{D.date_file_name()}.png")) # shutil.copy(temp_image_path, os.path.join(_local_forder,f"imagen_{request_body_info.prompt}_{1}_{D.date_file_name()}.png"))
# Clean up temporary files # # Clean up temporary files
if 'temp_image_path' in locals(): # if 'temp_image_path' in locals():
if os.path.exists(temp_image_path): # if os.path.exists(temp_image_path):
os.remove(temp_image_path) # os.remove(temp_image_path)
del temp_image_path # del temp_image_path
return response.set_message() # return response.set_message()
except Exception as e: # except Exception as e:
LOG.error(traceback.format_exc()) # LOG.error(traceback.format_exc())
# Clean up temporary files # # Clean up temporary files
if 'query_image_path' in locals(): # if 'query_image_path' in locals():
if os.path.exists(query_image_path): # if os.path.exists(query_image_path):
os.remove(query_image_path) # os.remove(query_image_path)
del query_image_path # del query_image_path
return response.set_error(error=e) # return response.set_error(error=e)
@router.post("/imageGenerate/imagen/data", summary="이미지 생성(AI) - imagen(data)", response_model=M.ImageGenerateDataRes) # @router.post("/imageGenerate/imagen/data", summary="이미지 생성(AI) - imagen(data)", response_model=M.ImageGenerateDataRes)
async def imagen_data(request: Request, request_body_info: M.ImageGenerateReq): # async def imagen_data(request: Request, request_body_info: M.ImageGenerateReq):
""" # """
## 이미지 생성(AI) - imagen # ## 이미지 생성(AI) - imagen
> imagen AI를 이용하여 이미지 데이터생성 # > imagen AI를 이용하여 이미지 데이터생성
""" # """
# imagen 사용중단 gemini로 변경 # # imagen 사용중단 gemini로 변경
response = M.ImageGenerateDataRes() # response = M.ImageGenerateDataRes()
try: # try:
# NOTE(JWKIM) : imagen 사용 중단 # # NOTE(JWKIM) : imagen 사용 중단
# img_length = imagen_generate_image(prompt=request_body_info.prompt, # # img_length = imagen_generate_image(prompt=request_body_info.prompt,
# download_count=request_body_info.downloadCount # # download_count=request_body_info.downloadCount
# ) # # )
temp_image_path = gemini_image(request_body_info.prompt) # temp_image_path = gemini_image(request_body_info.prompt)
b64data = image_to_base64_string(temp_image_path) # b64data = image_to_base64_string(temp_image_path)
# Clean up temporary files # # Clean up temporary files
if 'temp_image_path' in locals(): # if 'temp_image_path' in locals():
if os.path.exists(temp_image_path): # if os.path.exists(temp_image_path):
os.remove(temp_image_path) # os.remove(temp_image_path)
del temp_image_path # del temp_image_path
return response.set_message(b64data) # return response.set_message(b64data)
except Exception as e: # except Exception as e:
LOG.error(traceback.format_exc()) # LOG.error(traceback.format_exc())
# Clean up temporary files # # Clean up temporary files
if 'query_image_path' in locals(): # if 'query_image_path' in locals():
if os.path.exists(query_image_path): # if os.path.exists(query_image_path):
os.remove(query_image_path) # os.remove(query_image_path)
del query_image_path # del query_image_path
return response.set_error(error=e) # return response.set_error(error=e)
# @router.post("/vectorImageSearch/imagenet/imageGenerate/imagen", summary="벡터 이미지 검색(imagenet) - imagen", response_model=M.ResponseBase) # @router.post("/vectorImageSearch/imagenet/imageGenerate/imagen", summary="벡터 이미지 검색(imagenet) - imagen", response_model=M.ResponseBase)
# async def vactor_imagenet(request: Request, request_body_info: M.vectorImageSearchReq): # async def vactor_imagenet(request: Request, request_body_info: M.vectorImageSearchReq):
@@ -256,164 +256,164 @@ async def imagen_data(request: Request, request_body_info: M.ImageGenerateReq):
# LOG.error(traceback.format_exc()) # LOG.error(traceback.format_exc())
# return response.set_error(error=e) # return response.set_error(error=e)
@router.post("/vectorImageSearch/vit/imageGenerate/imagen", summary="벡터 이미지 검색(clip-vit) - imagen", response_model=M.ResponseBase) # @router.post("/vectorImageSearch/vit/imageGenerate/imagen", summary="벡터 이미지 검색(clip-vit) - imagen", response_model=M.ResponseBase)
async def vactor_vit(request: Request, request_body_info: M.VectorImageSearchVitReq): # async def vactor_vit(request: Request, request_body_info: M.VectorImageSearchVitReq):
""" # """
## 벡터 이미지 검색(clip-vit) - imagen # ## 벡터 이미지 검색(clip-vit) - imagen
> imagen AI를 이용하여 이미지 생성 후 vector 검색 그후 결과 이미지 생성 # > imagen AI를 이용하여 이미지 생성 후 vector 검색 그후 결과 이미지 생성
### options # ### options
> - modelType -> b32,b16,l14,l14_336 # > - modelType -> b32,b16,l14,l14_336
> - indexType -> l2,cos # > - indexType -> l2,cos
""" # """
response = M.ResponseBase() # response = M.ResponseBase()
try: # try:
if not download_range(request_body_info.searchNum, max=10): # if not download_range(request_body_info.searchNum, max=10):
raise Exception(f"downloadCound is invalid (current value = {request_body_info.searchNum})") # raise Exception(f"downloadCound is invalid (current value = {request_body_info.searchNum})")
if request_body_info.modelType not in [M.VitModelType.b32, M.VitModelType.b16, M.VitModelType.l14, M.VitModelType.l14_336]: # if request_body_info.modelType not in [M.VitModelType.b32, M.VitModelType.b16, M.VitModelType.l14, M.VitModelType.l14_336]:
raise Exception(f"modelType is invalid (current value = {request_body_info.modelType})") # raise Exception(f"modelType is invalid (current value = {request_body_info.modelType})")
if request_body_info.indexType not in [M.VitIndexType.cos, M.VitIndexType.l2]: # if request_body_info.indexType not in [M.VitIndexType.cos, M.VitIndexType.l2]:
raise Exception(f"indexType is invalid (current value = {request_body_info.indexType})") # raise Exception(f"indexType is invalid (current value = {request_body_info.indexType})")
# query_image_path = imagen_generate_temp_image_path(image_prompt=request_body_info.prompt) #imagen # # query_image_path = imagen_generate_temp_image_path(image_prompt=request_body_info.prompt) #imagen
query_image_path = gemini_image(request_body_info.prompt) #gemini # query_image_path = gemini_image(request_body_info.prompt) #gemini
vector_request_data = {'query_image_path' : query_image_path, # vector_request_data = {'query_image_path' : query_image_path,
'index_type' : request_body_info.indexType, # 'index_type' : request_body_info.indexType,
'model_type' : request_body_info.modelType, # 'model_type' : request_body_info.modelType,
'search_num' : request_body_info.searchNum} # 'search_num' : request_body_info.searchNum}
vector_response = requests.post('http://localhost:51002/api/services/faiss/vector/search/vit', data=json.dumps(vector_request_data)) # vector_response = requests.post('http://localhost:51002/api/services/faiss/vector/search/vit', data=json.dumps(vector_request_data))
vector_response_dict = json.loads(vector_response.text) # vector_response_dict = json.loads(vector_response.text)
if vector_response.status_code != 200: # if vector_response.status_code != 200:
raise Exception(f"response error: {vector_response_dict['error']}") # raise Exception(f"response error: {vector_response_dict['error']}")
if vector_response_dict["error"] != None: # if vector_response_dict["error"] != None:
raise Exception(f"vector error: {vector_response_dict['error']}") # raise Exception(f"vector error: {vector_response_dict['error']}")
result_image_paths = vector_response_dict.get('img_list').get('result_image_paths') # result_image_paths = vector_response_dict.get('img_list').get('result_image_paths')
result_percents = vector_response_dict.get('img_list').get('result_percents') # result_percents = vector_response_dict.get('img_list').get('result_percents')
if rest_config.config != 'release': # if rest_config.config != 'release':
# 원격지 폴더 생성 # # 원격지 폴더 생성
remote_directory = os.path.join(rest_config.remote_folder, f"imagen_query_{request_body_info.modelType}_{request_body_info.indexType}_{request_body_info.prompt}_{D.date_file_name()}") # remote_directory = os.path.join(rest_config.remote_folder, f"imagen_query_{request_body_info.modelType}_{request_body_info.indexType}_{request_body_info.prompt}_{D.date_file_name()}")
sftp_client.remote_mkdir(remote_directory) # sftp_client.remote_mkdir(remote_directory)
# 원격지에 이미지 저장 # # 원격지에 이미지 저장
sftp_client.remote_copy_data(local_path=query_image_path, remote_path=os.path.join(remote_directory,"query.png")) # sftp_client.remote_copy_data(local_path=query_image_path, remote_path=os.path.join(remote_directory,"query.png"))
for img_path, img_percent in zip(result_image_paths,result_percents): # for img_path, img_percent in zip(result_image_paths,result_percents):
sftp_client.remote_copy_data(local_path=img_path, remote_path=os.path.join(remote_directory,f"search_{img_percent}.png")) # sftp_client.remote_copy_data(local_path=img_path, remote_path=os.path.join(remote_directory,f"search_{img_percent}.png"))
else: # else:
local_directory = os.path.join(rest_config.local_folder, f"imagen_query_{request_body_info.modelType}_{request_body_info.indexType}_{request_body_info.prompt}_{D.date_file_name()}") # local_directory = os.path.join(rest_config.local_folder, f"imagen_query_{request_body_info.modelType}_{request_body_info.indexType}_{request_body_info.prompt}_{D.date_file_name()}")
os.makedirs(local_directory, exist_ok=True) # os.makedirs(local_directory, exist_ok=True)
shutil.copy(query_image_path, os.path.join(local_directory,"query.png")) # shutil.copy(query_image_path, os.path.join(local_directory,"query.png"))
for img_path, img_percent in zip(result_image_paths,result_percents): # for img_path, img_percent in zip(result_image_paths,result_percents):
shutil.copy(img_path, os.path.join(local_directory,f"search_{img_percent}.png")) # shutil.copy(img_path, os.path.join(local_directory,f"search_{img_percent}.png"))
# Clean up temporary files # # Clean up temporary files
if 'query_image_path' in locals(): # if 'query_image_path' in locals():
if os.path.exists(query_image_path): # if os.path.exists(query_image_path):
os.remove(query_image_path) # os.remove(query_image_path)
del query_image_path # del query_image_path
return response.set_message() # return response.set_message()
except Exception as e: # except Exception as e:
LOG.error(traceback.format_exc()) # LOG.error(traceback.format_exc())
# Clean up temporary files # # Clean up temporary files
if 'query_image_path' in locals(): # if 'query_image_path' in locals():
if os.path.exists(query_image_path): # if os.path.exists(query_image_path):
os.remove(query_image_path) # os.remove(query_image_path)
del query_image_path # del query_image_path
return response.set_error(error=e) # return response.set_error(error=e)
@router.post("/vectorImageSearch/vit/imageGenerate/imagen/data", summary="벡터 이미지 검색(clip-vit) - imagen(data)", response_model=M.VectorImageSerachDataRes) # @router.post("/vectorImageSearch/vit/imageGenerate/imagen/data", summary="벡터 이미지 검색(clip-vit) - imagen(data)", response_model=M.VectorImageSerachDataRes)
async def vactor_vit_data(request: Request, request_body_info: M.VectorImageSearchVitDataReq): # async def vactor_vit_data(request: Request, request_body_info: M.VectorImageSearchVitDataReq):
""" # """
## 벡터 이미지 검색(clip-vit) - imagen # ## 벡터 이미지 검색(clip-vit) - imagen
> imagen AI를 이용하여 이미지 생성 후 vector 검색 그후 결과 이미지데이터 return # > imagen AI를 이용하여 이미지 생성 후 vector 검색 그후 결과 이미지데이터 return
### options # ### options
> - modelType -> b32,b16,l14,l14_336 # > - modelType -> b32,b16,l14,l14_336
> - indexType -> l2,cos # > - indexType -> l2,cos
""" # """
response = M.VectorImageSerachDataRes() # response = M.VectorImageSerachDataRes()
query_image_data = '' # query_image_data = ''
try: # try:
if not download_range(request_body_info.searchNum, max=10): # if not download_range(request_body_info.searchNum, max=10):
raise Exception(f"downloadCound is invalid (current value = {request_body_info.searchNum})") # raise Exception(f"downloadCound is invalid (current value = {request_body_info.searchNum})")
if request_body_info.modelType not in [M.VitModelType.b32, M.VitModelType.b16, M.VitModelType.l14, M.VitModelType.l14_336]: # if request_body_info.modelType not in [M.VitModelType.b32, M.VitModelType.b16, M.VitModelType.l14, M.VitModelType.l14_336]:
raise Exception(f"modelType is invalid (current value = {request_body_info.modelType})") # raise Exception(f"modelType is invalid (current value = {request_body_info.modelType})")
if request_body_info.indexType not in [M.VitIndexType.cos, M.VitIndexType.l2]: # if request_body_info.indexType not in [M.VitIndexType.cos, M.VitIndexType.l2]:
raise Exception(f"indexType is invalid (current value = {request_body_info.indexType})") # raise Exception(f"indexType is invalid (current value = {request_body_info.indexType})")
# query_image_path = imagen_generate_temp_image_path(image_prompt=request_body_info.prompt) #imagen # # query_image_path = imagen_generate_temp_image_path(image_prompt=request_body_info.prompt) #imagen
query_image_path = gemini_image(request_body_info.prompt) #gemini # query_image_path = gemini_image(request_body_info.prompt) #gemini
vector_request_data = {'query_image_path' : query_image_path, # vector_request_data = {'query_image_path' : query_image_path,
'index_type' : request_body_info.indexType, # 'index_type' : request_body_info.indexType,
'model_type' : request_body_info.modelType, # 'model_type' : request_body_info.modelType,
'search_num' : request_body_info.searchNum} # 'search_num' : request_body_info.searchNum}
vector_response = requests.post('http://localhost:51002/api/services/faiss/vector/search/vit', data=json.dumps(vector_request_data)) # vector_response = requests.post('http://localhost:51002/api/services/faiss/vector/search/vit', data=json.dumps(vector_request_data))
vector_response_dict = json.loads(vector_response.text) # vector_response_dict = json.loads(vector_response.text)
if vector_response.status_code != 200: # if vector_response.status_code != 200:
raise Exception(f"response error: {vector_response_dict['error']}") # raise Exception(f"response error: {vector_response_dict['error']}")
if vector_response_dict["error"] != None: # if vector_response_dict["error"] != None:
raise Exception(f"vector error: {vector_response_dict['error']}") # raise Exception(f"vector error: {vector_response_dict['error']}")
result_image_paths = vector_response_dict.get('img_list').get('result_image_paths') # result_image_paths = vector_response_dict.get('img_list').get('result_image_paths')
result_percents = vector_response_dict.get('img_list').get('result_percents') # result_percents = vector_response_dict.get('img_list').get('result_percents')
# 이미지 데이터 생성 # # 이미지 데이터 생성
vector_image_results = [] # vector_image_results = []
for img, percents in zip(result_image_paths, result_percents): # for img, percents in zip(result_image_paths, result_percents):
b64_data = image_to_base64_string(img) # b64_data = image_to_base64_string(img)
float_percent = float(percents) # float_percent = float(percents)
info = M.VectorImageResult(image=b64_data,percents=float_percent) # info = M.VectorImageResult(image=b64_data,percents=float_percent)
vector_image_results.append(info) # vector_image_results.append(info)
if request_body_info.querySend: # if request_body_info.querySend:
query_image_data = image_to_base64_string(query_image_path) # query_image_data = image_to_base64_string(query_image_path)
# Clean up temporary files # # Clean up temporary files
if 'query_image_path' in locals(): # if 'query_image_path' in locals():
if os.path.exists(query_image_path): # if os.path.exists(query_image_path):
os.remove(query_image_path) # os.remove(query_image_path)
del query_image_path # del query_image_path
return response.set_message(vector_result=vector_image_results,query_img=query_image_data) # return response.set_message(vector_result=vector_image_results,query_img=query_image_data)
except Exception as e: # except Exception as e:
LOG.error(traceback.format_exc()) # LOG.error(traceback.format_exc())
# Clean up temporary files # # Clean up temporary files
if 'query_image_path' in locals(): # if 'query_image_path' in locals():
if os.path.exists(query_image_path): # if os.path.exists(query_image_path):
os.remove(query_image_path) # os.remove(query_image_path)
del query_image_path # del query_image_path
return response.set_error(error=e) # return response.set_error(error=e)
@router.post("/vectorImageSearch/vit/inputImage/data", summary="벡터 이미지 검색(clip-vit) - input image(data)", response_model=M.VectorImageSerachDataRes) @router.post("/vectorImageSearch/vit/inputImage/data", summary="벡터 이미지 검색(clip-vit) - input image(data)", response_model=M.VectorImageSerachDataRes)
@@ -422,12 +422,21 @@ async def vactor_vit_input_img_data(request: Request, request_body_info: M.Vecto
## 벡터 이미지 검색(clip-vit) - inputimage ## 벡터 이미지 검색(clip-vit) - inputimage
> 입력된 이미지로 vector 검색 그후 결과 이미지데이터 return > 입력된 이미지로 vector 검색 그후 결과 이미지데이터 return
### Requriements ### Input
> - googlecli 설치(https://cloud.google.com/sdk/docs/install?hl=ko#linux) > 입력이미지(inputImage)는 base64로 변환된 데이터 입력
### options ### Output
> - modelType -> b32,b16,l14,l14_336 > - 결과(vectorResult)는 base64로 변환된 데이터(image), 유사도(percents)가 쌍으로 나오며, 요청한 searchNum 갯수에 맞춰서 결과가 나옴
> - indexType -> l2,cos > - queryImage는 Input시 입력한 inputImage 이미지 데이터
### Options
> - modelType -> b32, b16, l14, l14_336 (기본값: l14)
> - indexType -> l2, cos (기본값: l2)
> - searchNum -> 결과이미지 갯수 (기본값: 4)
### Notice
> - 일반검색시 indexType , modelType 은 기본값으로 사용
> - indexType , modelType - 새로운 조합 (ex l14, cos)으로 처음 요청할경우 모델을 빌드하는 과정이 추가됨
""" """
response = M.VectorImageSerachDataRes() response = M.VectorImageSerachDataRes()
@@ -441,6 +450,7 @@ async def vactor_vit_input_img_data(request: Request, request_body_info: M.Vecto
raise Exception(f"indexType is invalid (current value = {request_body_info.indexType})") raise Exception(f"indexType is invalid (current value = {request_body_info.indexType})")
query_image_path = os.path.join(TEMP_FOLDER, f'input_{D.date_file_name()}_query.png') query_image_path = os.path.join(TEMP_FOLDER, f'input_{D.date_file_name()}_query.png')
os.makedirs(TEMP_FOLDER, exist_ok=True)
save_base64_as_image_file(request_body_info.inputImage ,query_image_path) save_base64_as_image_file(request_body_info.inputImage ,query_image_path)
vector_request_data = {'query_image_path' : query_image_path, vector_request_data = {'query_image_path' : query_image_path,
@@ -453,10 +463,10 @@ async def vactor_vit_input_img_data(request: Request, request_body_info: M.Vecto
vector_response_dict = json.loads(vector_response.text) vector_response_dict = json.loads(vector_response.text)
if vector_response.status_code != 200: if vector_response.status_code != 200:
raise Exception(f"response error: {vector_response_dict['error']}") raise Exception(f"search server error: {vector_response_dict['error']}")
if vector_response_dict["error"] != None: if vector_response_dict["error"] != None:
raise Exception(f"vector error: {vector_response_dict['error']}") raise Exception(f"search result error: {vector_response_dict['error']}")
result_image_paths = vector_response_dict.get('img_list').get('result_image_paths') result_image_paths = vector_response_dict.get('img_list').get('result_image_paths')
result_percents = vector_response_dict.get('img_list').get('result_percents') result_percents = vector_response_dict.get('img_list').get('result_percents')
@@ -491,8 +501,8 @@ async def vactor_vit_input_img_data(request: Request, request_body_info: M.Vecto
return response.set_error(error=e) return response.set_error(error=e)
@router.post("/vectorImageSearch/vit/imageGenerate/imagen/report", summary="벡터 이미지 검색(clip-vit) - imagen, report 생성", response_model=M.ResponseBase) # @router.post("/vectorImageSearch/vit/imageGenerate/imagen/report", summary="벡터 이미지 검색(clip-vit) - imagen, report 생성", response_model=M.ResponseBase)
async def vactor_vit_report(request: Request, request_body_info: M.VectorImageSearchVitReportReq): # async def vactor_vit_report(request: Request, request_body_info: M.VectorImageSearchVitReportReq):
""" """
## 벡터 이미지 검색(clip-vit) - imagen, report 생성 ## 벡터 이미지 검색(clip-vit) - imagen, report 생성
> imagen AI를 이용하여 이미지 생성 후 vector 검색 그후 종합결과 이미지 생성 > imagen AI를 이용하여 이미지 생성 후 vector 검색 그후 종합결과 이미지 생성

4
main_rest/app/utils/parsing_utils.py
View File

@@ -65,8 +65,10 @@ def save_base64_as_image_file(base64_data: str, output_path: str):
""" """
Base64 문자열을 디코딩하여 이미지 파일로 저장합니다. Base64 문자열을 디코딩하여 이미지 파일로 저장합니다.
""" """
try:
# Base64 문자열을 디코딩하여 이진 데이터로 변환합니다. # Base64 문자열을 디코딩하여 이진 데이터로 변환합니다.
decoded_data = base64.b64decode(base64_data) decoded_data = base64.b64decode(base64_data)
with open(output_path, "wb") as image_file: with open(output_path, "wb") as image_file:
image_file.write(decoded_data) image_file.write(decoded_data)
except Exception as e:
raise Exception(f"input image data error : {e}")

2
vector_rest/app/common/consts.py
View File

@@ -87,7 +87,7 @@ NUM_RETRY_UUID_GEN = 3
DB_ADDRESS = "localhost" DB_ADDRESS = "localhost"
DB_PORT = 53306 DB_PORT = 53306
DB_USER_ID = 'root' DB_USER_ID = 'root'
DB_USER_PW = '1234' DB_USER_PW = 'Fermat3514!'
DB_NAME = 'FM_TEST' DB_NAME = 'FM_TEST'
DB_CHARSET = 'utf8mb4' DB_CHARSET = 'utf8mb4'