forked from cgvr/DeltaVR
combined pipeline running function
This commit is contained in:
75
3d-generation-pipeline/start_pipeline.py
Normal file
75
3d-generation-pipeline/start_pipeline.py
Normal file
@@ -0,0 +1,75 @@
|
||||
import os
|
||||
import base64
|
||||
import requests
|
||||
from datetime import datetime
|
||||
from dotenv import load_dotenv
|
||||
|
||||
load_dotenv()
|
||||
|
||||
ACCOUNT_ID = os.environ["CF_ACCOUNT_ID"]
|
||||
API_TOKEN = os.environ["CF_API_TOKEN"]
|
||||
|
||||
|
||||
def get_timestamp():
|
||||
return datetime.now().strftime("%Y-%m-%d-%H-%M-%S")
|
||||
|
||||
|
||||
def text_to_image(prompt, output_name):
|
||||
MODEL = "@cf/black-forest-labs/flux-1-schnell"
|
||||
URL = f"https://api.cloudflare.com/client/v4/accounts/{ACCOUNT_ID}/ai/run/{MODEL}"
|
||||
|
||||
payload = {
|
||||
"prompt": prompt,
|
||||
}
|
||||
|
||||
headers = {
|
||||
"Authorization": f"Bearer {API_TOKEN}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
resp = requests.post(URL, json=payload, headers=headers, timeout=60)
|
||||
resp.raise_for_status()
|
||||
|
||||
data = resp.json()
|
||||
b64 = data["result"]["image"]
|
||||
if not b64:
|
||||
raise RuntimeError(f"Unexpected response structure: {data}")
|
||||
|
||||
img_bytes = base64.b64decode(b64)
|
||||
|
||||
out_path = f"images/{output_name}.jpg"
|
||||
with open(out_path, "wb") as f:
|
||||
f.write(img_bytes)
|
||||
|
||||
return out_path
|
||||
|
||||
def refine_text_prompt(prompt):
|
||||
MODEL = "@cf/meta/llama-3.2-3b-instruct"
|
||||
URL = f"https://api.cloudflare.com/client/v4/accounts/{ACCOUNT_ID}/ai/run/{MODEL}"
|
||||
|
||||
instructions = """
|
||||
User is talking about some object. Your task is to generate a short and concise description of it. Use only user's own words, keep it as short as possible.
|
||||
Example:
|
||||
User: 'Umm, okay, I would like a really cool sword, with for example a bright orange crossguard. And also it should be slightly curved.'
|
||||
You: 'a slightly curved sword with bright orange crossguard'
|
||||
"""
|
||||
|
||||
response = requests.post(URL,
|
||||
headers={"Authorization": f"Bearer {API_TOKEN}"},
|
||||
json={
|
||||
"messages": [
|
||||
{"role": "system", "content": instructions},
|
||||
{"role": "user", "content": prompt}
|
||||
]
|
||||
}
|
||||
)
|
||||
data = response.json()
|
||||
return data["result"]["response"]
|
||||
|
||||
def main():
|
||||
user_prompt = "Give epic sword"
|
||||
print(f"User prompt: {user_prompt}")
|
||||
refined_prompt = refine_text_prompt(user_prompt)
|
||||
print(f"Refined prompt: {refined_prompt}")
|
||||
image_path = text_to_image(refined_prompt, get_timestamp())
|
||||
print(f"Generated image file: {image_path}")
|
||||
Reference in New Issue
Block a user