Spaces:
Running
Running
File size: 2,321 Bytes
5291ba9 34046e2 5291ba9 34046e2 5291ba9 34046e2 5291ba9 34046e2 5291ba9 34046e2 5291ba9 34046e2 |
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 |
import base64
import json
import os
import time
from pathlib import Path
import boto3
from dotenv import load_dotenv
from api.flux import FluxAPI
class AWSBedrockAPI(FluxAPI):
def __init__(self):
load_dotenv()
# AWS credentials should be set via environment variables
# AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, and AWS_SESSION_TOKEN
os.environ["AWS_ACCESS_KEY_ID"] = ""
os.environ["AWS_SECRET_ACCESS_KEY"] = ""
os.environ["AWS_SESSION_TOKEN"] = ""
os.environ["AWS_REGION"] = "us-east-1"
self._client = boto3.client("bedrock-runtime")
self._model_id = "amazon.nova-canvas-v1:0"
@property
def name(self) -> str:
return "aws_nova_canvas"
def generate_image(self, prompt: str, save_path: Path) -> float:
start_time = time.time()
# Format the request payload
native_request = {
"taskType": "TEXT_IMAGE",
"textToImageParams": {"text": prompt},
"imageGenerationConfig": {
"seed": 0,
"quality": "standard",
"height": 1024,
"width": 1024,
"numberOfImages": 1,
},
}
try:
# Convert request to JSON and invoke the model
request = json.dumps(native_request)
response = self._client.invoke_model(modelId=self._model_id, body=request)
# Process the response
model_response = json.loads(response["body"].read())
if not model_response.get("images"):
raise Exception("No images returned from AWS Bedrock API")
# Save the image
base64_image_data = model_response["images"][0]
self._save_image_from_base64(base64_image_data, save_path)
except Exception as e:
raise Exception(f"Error generating image with AWS Bedrock: {str(e)}")
end_time = time.time()
return end_time - start_time
def _save_image_from_base64(self, base64_data: str, save_path: Path):
"""Save a base64 encoded image to the specified path."""
save_path.parent.mkdir(parents=True, exist_ok=True)
image_data = base64.b64decode(base64_data)
with open(save_path, "wb") as f:
f.write(image_data)
|