Generate and store images in S3 using Titan Image Generator G1

zhalbert@6c7e67d927df work % sam init

You can preselect a particular runtime or package type when using the `sam init` experience.
Call `sam init --help` to learn more.

Which template source would you like to use?
1 - AWS Quick Start Templates
2 - Custom Template Location
Choice: 1

Choose an AWS Quick Start application template
1 - Hello World Example
2 - Data processing
3 - Hello World Example with Powertools for AWS Lambda
4 - Multi-step workflow
5 - Scheduled task
6 - Standalone function
7 - Serverless API
8 - Infrastructure event management
9 - Lambda Response Streaming
10 - Serverless Connector Hello World Example
11 - Multi-step workflow with Connectors
12 - GraphQLApi Hello World Example
13 - Full Stack
14 - Lambda EFS example
15 - Hello World Example With Powertools for AWS Lambda
16 - DynamoDB Example
17 - Machine Learning
Template: 1

Use the most popular runtime and package type? (Python and zip) [y/N]: y

Would you like to set Structured Logging in JSON format on your Lambda functions? [y/N]: y
Structured Logging in JSON format might incur an additional cost. View https://docs.aws.amazon.com/lambda/latest/dg/monitoring-cloudwatchlogs.html#monitoring-cloudwatchlogs-pricing for more details

Project name [sam-app]: s3imagegenerator

import json
import logging
import boto3
from botocore.exceptions import ClientError
import base64

# This creates a logger instance
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# This initializes the clients Bedrock Runtime and S3
bedrock_runtime_client = boto3.client('bedrock-runtime', region_name='us-east-1')
s3 = boto3.client('s3')

bucket_name = 's3bucketname' # Add the s3 bucket you want to upload photos to

def lambda_handler(event, context):
# We need to extract 'text' and 'seed' from the event, provide defaults if not present
prompt = event.get('text', 'default prompt')
seed = event.get('seed', 0) # Default seed value if not provided

try:
base64_image_data = invoke_titan_image(prompt, seed)

# The image data is a base64-encoded string, so we need to decode it to get the actual image data
image_data = base64.b64decode(base64_image_data)
object_key = f"generated_images/image_{prompt.replace(' ', '_')}_{seed}.jpg"

# Now we upload the image data to S3
s3.put_object(
Bucket=bucket_name,
Key=object_key,
Body=image_data,
ContentType='image/jpeg' # This is adjustable!
)
logger.info(f"Uploaded image to s3://{bucket_name}/{object_key}")

return {
'statusCode': 200,
'body': json.dumps({'message': f"Image uploaded successfully to {bucket_name}/{object_key}."})
}

except Exception as e:
logger.error("Error: %s", str(e))
return {
'statusCode': 500,
'body': json.dumps({'error': 'Failed to invoke model or upload image', 'detail': str(e)})
}

def invoke_titan_image(prompt, seed):
try:
request = json.dumps({
"taskType": "TEXT_IMAGE",
"textToImageParams": {"text": prompt},
"imageGenerationConfig": {
"numberOfImages": 1,
"quality": "standard",
"cfgScale": 8.0,
"height": 640, # Permissible sizes: https://docs.aws.amazon.com/bedrock/latest/userguide/model-parameters-titan-image.html#:~:text=The%20following%20sizes%20are%20permissible.
"width": 1408,
"seed": seed,
},
})

response = bedrock_runtime_client.invoke_model(
modelId="amazon.titan-image-generator-v1", body=request
)

response_body = json.loads(response["body"].read())
base64_image_data = response_body["images"][0]

return base64_image_data

except ClientError as e:
logger.error(f"Couldn't invoke Titan Image generator: {e}")
raise

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: >
A SAM application to invoke a Lambda that calls the Bedrock Titan Image Generator then stores the output in S3 and logs in CloudWatch, and a Hello World function

Globals:
Function:
Timeout: 60
MemorySize: 128

Resources:
HelloWorldFunction:
Type: AWS::Serverless::Function
Properties:
CodeUri: hello_world/
Handler: app.lambda_handler
Runtime: python3.9
Architectures:
- x86_64
Events:
HelloWorld:
Type: Api
Properties:
Path: /hello
Method: get

S3ImageGeneratorFunction:
Type: AWS::Serverless::Function
Properties:
CodeUri: s3imagegenerator/
Handler: app.lambda_handler
Runtime: python3.8 # Ensure this matches the Python version you're using in s3imagegenerator
Policies:
- S3CrudPolicy:
BucketName: "s3imagegenerator" # Replace with your actual S3 bucket name
- Version: "2012-10-17"
Statement:
- Effect: Allow
Action:
- "bedrock-runtime:InvokeModel"
Resource: "*"
- Effect: Allow
Action:
- "logs:CreateLogGroup"
- "logs:CreateLogStream"
- "logs:PutLogEvents"
Resource: "arn:aws:logs:*:*:*"
Events:
TitanImageApi:
Type: Api
Properties:
Path: /generate-image
Method: post

Outputs:
HelloWorldApi:
Description: "API Gateway endpoint URL for Prod stage for Hello World function"
Value: !Sub "https://${ServerlessRestApi}.execute-api.${AWS::Region}.amazonaws.com/Prod/hello/"
HelloWorldFunction:
Description: "Hello World Lambda Function ARN"
Value: !GetAtt HelloWorldFunction.Arn
HelloWorldFunctionIamRole:
Description: "Implicit IAM Role created for Hello World function"
Value: !GetAtt HelloWorldFunctionRole.Arn

S3ImageGeneratorApi:
Description: "API Gateway endpoint URL for Prod stage for S3 Image Generator function"
Value: !Sub "https://${ServerlessRestApi}.execute-api.${AWS::Region}.amazonaws.com/Prod/generate-image/"
S3ImageGeneratorFunction:
Description: "S3 Image Generator Function ARN"
Value: !GetAtt S3ImageGeneratorFunction.Arn
S3ImageGeneratorFunctionIamRole:
Description: "Implicit IAM Role created for S3 Image Generator function"
Value: !GetAtt S3ImageGeneratorFunction.Arn

zhalbert@6c7e67d927df s3imagegenerator % aws lambda invoke \
--function-name s3imagegenerator-<function name> \
--payload '{"text":"dog with cowboy hat","seed":2000}' \
--cli-binary-format raw-in-base64-out \
response.json