api/aws.py

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)