Spaces:
Sleeping
Sleeping
File size: 6,157 Bytes
c14816f 20779fb c14816f 823b070 c14816f 823b070 c14816f c7b8eef c14816f 843f3c8 c14816f 20779fb c14816f 20779fb c14816f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 |
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
import api_module as mj
import random
from io import BytesIO
import openai
#from transparent_background import Remover
import requests
from PIL import Image
from qiniu import Auth, put_data, etag, Zone
import replicate
import os
mj_api_key = os.environ['MJ_API_KEY']
openai.api_key = os.environ['OPENAI_API_KEY']
access_key = os.environ['ACCESS_KEY']
secret_key = os.environ['SECRET_KEY']
bucket_name = 'mystory-games'
def translate_to_english(text):
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[
{
"role": "system",
"content": "accurately translating Chinese text to English, while keeping the context and idiomatically correct in English, ensure that only translated content is replied to \n"
},
{
"role": "user",
"content": text
},
],
temperature=0.2,
max_tokens=256,
top_p=1,
frequency_penalty=0,
presence_penalty=0
)
translated_text = response['choices'][0]['message']['content'].strip()
return translated_text
def translate_to_chinese(text):
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[
{
"role": "system",
"content": "accurately translating English text to Chinese, while keeping the context and idiomatically correct in Chinese, ensure that only translated content is replied to \n"
},
{
"role": "user",
"content": text
},
],
temperature=0.2,
max_tokens=256,
top_p=1,
frequency_penalty=0,
presence_penalty=0
)
translated_text = response['choices'][0]['message']['content'].strip()
return translated_text
def avatar_create(text):
#translate to english
text = translate_to_english(text)
#add text to prompt
prompt = f'{text}, concept art by Pixar, cgsociety, character, furry art, full body shot, white background, Realistic 3d render --v 5.2'
#make imagine request
max_retries = 5
retries = 0
while retries < max_retries:
task_id = mj.make_imagine_request(mj_api_key, prompt)
if task_id is not None:
break
else:
retries += 1
print(f'Failed to get task id, retrying {retries}/{max_retries}...')
if retries == max_retries:
print('Failed to get task id, aborting')
return None
#fetch request
# make sure the imagine process has finished
while True:
image_url = mj.fetch_request(mj_api_key, task_id)
if image_url is not None:
break
elif image_url is None:
print('Imagine process failed, exiting...')
raise SystemExit('Imagine process failed')
return task_id
# def avatar_remove_background(image_url):
# remover = Remover() # default settings
# # downlaod image
# response = requests.get(image_url)
# # convert to PIL image
# img = Image.open(BytesIO(response.content)).convert('RGB')
# # remove background
# out = remover.process(img, type='rgba')
# image_io = BytesIO()
# Image.fromarray(out).save(image_io, format='PNG')
# image_data = image_io.getvalue()
# # Upload the image data
# q = Auth(access_key, secret_key)
# token = q.upload_token(bucket_name)
# ret, info = put_data(token, None, image_data, mime_type="image/png")
# if info.status_code == 200:
# cdn_url = 'http://image1.juramaia.com/' + ret['key']
# return cdn_url
# else:
# print('Failed to upload image to CDN')
# return None
def avatar_upscale(task_id, index):
task_id = mj.make_upscale_request(mj_api_key, task_id, index)
# make sure the upscale task sucessfully
if task_id is not None:
pass
elif task_id is None:
print('Upscale process failed, exiting...')
raise SystemExit('Upscale process failed')
# make sure the upscale process has finished
while True:
upscale_image_url = mj.fetch_request(mj_api_key, task_id)
if upscale_image_url is not None:
break
elif upscale_image_url is None:
print('Upscale process failed, exiting...')
raise SystemExit('Upscale process failed')
return upscale_image_url
def main_process(text):
#create avatar
task_id = avatar_create(text)
print(f'\navatar create successful task_id is {task_id}\n')
#upscale avatar and remove background
index = random.randint(1, 4)
while True:
print(f'upsclae index is {index} and task_id is {task_id}')
upscale_image_url = avatar_upscale(task_id, index)
if upscale_image_url is not None:
break
elif upscale_image_url is None:
print('Upscale process failed, exiting...')
raise SystemExit('Upscale process failed')
print(f'upscale_image_url is {upscale_image_url}')
return upscale_image_url
def analyze_process(url):
print(f'\nanalyzing image: {url}')
eng_output = replicate.run(
"andreasjansson/blip-2:4b32258c42e9efd4288bb9910bc532a69727f9acd26aa08e175713a0a857a608",
input={
"image": url,
"caption": True,
"temperature": 0.5
}
)
output = translate_to_chinese(eng_output)
print(f'analyzed caption: {output}\n')
return output
class Item(BaseModel):
text: str
api = FastAPI()
@api.post("/main_process/")
async def main_api(item: Item):
try:
image_url = main_process(item.text)
return {"image_url": image_url}
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
@api.post("/analyze_process/")
async def analyze_api(item: Item):
try:
output = analyze_process(item.text)
return {"output": output}
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
|