Python: Nova Reel Text-to-Video example

.doc_gen/metadata/bedrock-runtime_metadata.yaml

@@ -1403,6 +1403,24 @@ bedrock-runtime_InvokeModel_StableDiffusion:
   services:
     bedrock-runtime: {InvokeModel}
 
+# Video Generation Models
+bedrock-runtime_Scenario_AmazonNova_TextToVideo:
+  title: Use Amazon Nova Reel to generate a video from a text prompt
+  title_abbrev: "Text-to-video"
+  synopsis: use Amazon Nova Reel to generate a video from a text prompt.
+  category: Amazon Nova Reel
+  languages:
+    Python:
+      versions:
+        - sdk_version: 3
+          github: python/example_code/bedrock-runtime
+          excerpts:
+            - description: Use Amazon Nova Reel to generate a video from a text prompt.
+              snippet_tags:
+                - python.example_code.bedrock-runtime.Scenario_AmazonNova_TextToVideo
+  services:
+    bedrock-runtime: {GetAsyncInvoke, StartAsyncInvoke}
+
 # Embedding Models
 bedrock-runtime_InvokeModelWithResponseStream_TitanTextEmbeddings:
   title: Invoke Amazon Titan Text Embeddings on &BR;

python/example_code/bedrock-runtime/README.md

@@ -64,6 +64,10 @@ functions within the same service.
 
 - [InvokeModel](models/amazon_nova/amazon_nova_canvas/invoke_model.py#L4)
 
+### Amazon Nova Reel
+
+- [Text-to-video](models/amazon_nova/amazon_nova_reel/text_to_video.py#L4)
+
 ### Amazon Titan Image Generator
 
 - [InvokeModel](models/amazon_titan_image_generator/invoke_model.py#L4)

python/example_code/bedrock-runtime/models/amazon_nova/amazon_nova_reel/text_to_video.py

@@ -0,0 +1,126 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+# snippet-start:[python.example_code.bedrock-runtime.Scenario_AmazonNova_TextToVideo]
+"""
+This example demonstrates how to use Amazon Nova Reel to generate a video from a text prompt.
+
+It shows how to:
+- Set up the Amazon Bedrock runtime client
+- Configure a text-to-video request
+- Submit an asynchronous job for video generation
+- Poll for job completion status
+- Access the generated video from S3
+"""
+
+import random
+import time
+
+import boto3
+
+# Replace with your own S3 bucket to store the generated video
+# Format: s3://your-bucket-name
+OUTPUT_S3_URI = "s3://REPLACE-WITH-YOUR-S3-BUCKET-NAME"
+
+
+def start_text_to_video_generation_job(bedrock_runtime, prompt, output_s3_uri):
+    """
+    Starts an asynchronous text-to-video generation job using Amazon Nova Reel.
+
+    :param bedrock_runtime: The Bedrock runtime client
+    :param prompt: The text description of the video to generate
+    :param output_s3_uri: S3 URI where the generated video will be stored
+
+    :return: The invocation ARN of the async job
+    """
+    # Specify the model ID for text-to-video generation
+    model_id = "amazon.nova-reel-v1:0"
+
+    # Generate a random seed between 0 and 2,147,483,646
+    # This helps ensure unique video generation results
+    seed = random.randint(0, 2147483646)
+
+    # Configure the video generation request with additional parameters
+    model_input = {
+        "taskType": "TEXT_VIDEO",
+        "textToVideoParams": {"text": prompt},
+        "videoGenerationConfig": {
+            "fps": 24,
+            "durationSeconds": 6,
+            "dimension": "1280x720",
+            "seed": seed,
+        },
+    }
+
+    # Specify the S3 location for the output video
+    output_config = {"s3OutputDataConfig": {"s3Uri": output_s3_uri}}
+
+    # Invoke the model asynchronously
+    response = bedrock_runtime.start_async_invoke(
+        modelId=model_id, modelInput=model_input, outputDataConfig=output_config
+    )
+
+    invocation_arn = response["invocationArn"]
+
+    return invocation_arn
+
+
+def query_job_status(bedrock_runtime, invocation_arn):
+    """
+    Queries the status of an asynchronous video generation job.
+
+    :param bedrock_runtime: The Bedrock runtime client
+    :param invocation_arn: The ARN of the async invocation to check
+
+    :return: The runtime response containing the job status and details
+    """
+    return bedrock_runtime.get_async_invoke(invocationArn=invocation_arn)
+
+
+def main():
+    """
+    Main function that demonstrates the complete workflow for generating
+    a video from a text prompt using Amazon Nova Reel.
+    """
+    # Create a Bedrock Runtime client
+    # Note: Credentials will be loaded from the environment or AWS CLI config
+    bedrock_runtime = boto3.client("bedrock-runtime", region_name="us-east-1")
+
+    # Configure the text prompt and output location
+    prompt = "Closeup of a cute old steampunk robot. Camera zoom in."
+
+    # Verify the S3 URI has been set to a valid bucket
+    if "REPLACE-WITH-YOUR-S3-BUCKET-NAME" in OUTPUT_S3_URI:
+        print("ERROR: You must replace the OUTPUT_S3_URI with your own S3 bucket URI")
+        return
+
+    print("Submitting video generation job...")
+    invocation_arn = start_text_to_video_generation_job(
+        bedrock_runtime, prompt, OUTPUT_S3_URI
+    )
+    print(f"Job started with invocation ARN: {invocation_arn}")
+
+    # Poll for job completion
+    while True:
+        print("\nPolling job status...")
+        job = query_job_status(bedrock_runtime, invocation_arn)
+        status = job["status"]
+
+        if status == "Completed":
+            bucket_uri = job["outputDataConfig"]["s3OutputDataConfig"]["s3Uri"]
+            print(f"\nSuccess! The video is available at: {bucket_uri}/output.mp4")
+            break
+        elif status == "Failed":
+            print(
+                f"\nVideo generation failed: {job.get('failureMessage', 'Unknown error')}"
+            )
+            break
+        else:
+            print("In progress. Waiting 15 seconds...")
+            time.sleep(15)
+
+
+if __name__ == "__main__":
+    main()
+
+# snippet-end:[python.example_code.bedrock-runtime.Scenario_AmazonNova_TextToVideo]

python/example_code/bedrock-runtime/requirements.txt

@@ -1,5 +1,5 @@
-boto3==1.36.22
-botocore==1.36.22
+boto3==1.37.9
+botocore==1.37.9
 colorama==0.4.6
 iniconfig==2.0.0
 jmespath==1.0.1

python/example_code/bedrock-runtime/test/test_nova_reel.py

@@ -0,0 +1,182 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""
+Unit tests for the text-to-video generation example using Amazon Nova Reel.
+Uses mocking to test the code without making actual API calls to AWS.
+"""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+from models.amazon_nova.amazon_nova_reel import text_to_video
+
+
+@pytest.fixture
+def mock_bedrock_runtime():
+    """A mocked bedrock runtime client."""
+    with patch("boto3.client") as mock_client:
+        mock_bedrock = MagicMock()
+        mock_client.return_value = mock_bedrock
+        yield mock_bedrock
+
+
+@pytest.fixture
+def successful_job_responses(mock_bedrock_runtime):
+    """Mock responses for a successful job."""
+    mock_bedrock_runtime.start_async_invoke.return_value = {
+        "invocationArn": "arn:aws:bedrock:us-east-1:123456789012:invocation/abcdef123456"
+    }
+    mock_bedrock_runtime.get_async_invoke.side_effect = [
+        {"status": "InProgress"},
+        {
+            "status": "Completed",
+            "outputDataConfig": {"s3OutputDataConfig": {"s3Uri": "s3://test-bucket"}},
+        },
+    ]
+
+    return mock_bedrock_runtime
+
+
+@pytest.fixture
+def failed_job_responses(mock_bedrock_runtime):
+    """Mock responses for a failed job."""
+    mock_bedrock_runtime.start_async_invoke.return_value = {
+        "invocationArn": "arn:aws:bedrock:us-east-1:123456789012:invocation/abcdef123456"
+    }
+    mock_bedrock_runtime.get_async_invoke.return_value = {
+        "status": "Failed",
+        "failureMessage": "Test failure message",
+    }
+
+    return mock_bedrock_runtime
+
+
+def test_start_text_to_video_generation_job(mock_bedrock_runtime):
+    # Set up mock return value
+    mock_bedrock_runtime.start_async_invoke.return_value = {
+        "invocationArn": "arn:aws:bedrock:us-east-1:123456789012:invocation/abcdef123456"
+    }
+
+    # Test parameters
+    prompt = "Test prompt"
+    output_s3_uri = "s3://test-bucket"
+
+    # Call the function
+    result = text_to_video.start_text_to_video_generation_job(
+        mock_bedrock_runtime, prompt, output_s3_uri
+    )
+
+    # Verify the client was called correctly
+    mock_bedrock_runtime.start_async_invoke.assert_called_once()
+
+    # Check the parameters
+    call_args = mock_bedrock_runtime.start_async_invoke.call_args[1]
+    assert call_args["modelId"] == "amazon.nova-reel-v1:0"
+    assert call_args["modelInput"]["textToVideoParams"]["text"] == prompt
+    assert call_args["outputDataConfig"]["s3OutputDataConfig"]["s3Uri"] == output_s3_uri
+
+    # Verify the return value
+    assert result == "arn:aws:bedrock:us-east-1:123456789012:invocation/abcdef123456"
+
+
+def test_query_job_status(mock_bedrock_runtime):
+    """Test the job status query function with pytest."""
+    # Set up mock return value
+    mock_job_status = {"status": "InProgress"}
+    mock_bedrock_runtime.get_async_invoke.return_value = mock_job_status
+
+    # Test parameter
+    invocation_arn = "arn:aws:bedrock:us-east-1:123456789012:invocation/abcdef123456"
+
+    # Call the function
+    result = text_to_video.query_job_status(mock_bedrock_runtime, invocation_arn)
+
+    # Verify the client was called correctly
+    mock_bedrock_runtime.get_async_invoke.assert_called_once_with(
+        invocationArn=invocation_arn
+    )
+
+    # Verify the return value
+    assert result == mock_job_status
+
+
+@patch("time.sleep", return_value=None)  # Prevent actual sleep in tests
+def test_main_success(mock_sleep, successful_job_responses, monkeypatch):
+    # Patch boto3.client to use our fixture
+    monkeypatch.setattr(
+        "boto3.client", lambda service_name, region_name: successful_job_responses
+    )
+
+    # Patch the OUTPUT_S3_URI constant to use a valid test value
+    monkeypatch.setattr(text_to_video, "OUTPUT_S3_URI", "s3://test-bucket")
+
+    # Capture print output
+    printed_messages = []
+    monkeypatch.setattr(
+        "builtins.print", lambda *args: printed_messages.append(args[0])
+    )
+
+    # Call the main function
+    text_to_video.main()
+
+    # Verify the printed messages
+    assert "Submitting video generation job..." in printed_messages
+    assert (
+        "\nSuccess! The video is available at: s3://test-bucket/output.mp4"
+        in printed_messages
+    )
+
+
+@patch("time.sleep", return_value=None)  # Prevent actual sleep in tests
+def test_main_failure(mock_sleep, failed_job_responses, monkeypatch):
+    # Patch boto3.client to use our fixture
+    monkeypatch.setattr(
+        "boto3.client", lambda service_name, region_name: failed_job_responses
+    )
+
+    # Patch the OUTPUT_S3_URI constant to use a valid test value
+    monkeypatch.setattr(text_to_video, "OUTPUT_S3_URI", "s3://test-bucket")
+
+    # Capture print output
+    printed_messages = []
+    monkeypatch.setattr(
+        "builtins.print", lambda *args: printed_messages.append(args[0])
+    )
+
+    # Call the main function
+    text_to_video.main()
+
+    # Verify the printed messages
+    assert "Submitting video generation job..." in printed_messages
+    assert "\nVideo generation failed: Test failure message" in printed_messages
+
+
+def test_main_invalid_s3_uri(monkeypatch):
+    """Test the main function with an invalid S3 URI."""
+    # Mock boto3.client to ensure no real AWS calls are made
+    mock_client = MagicMock()
+    monkeypatch.setattr("boto3.client", lambda service_name, region_name: mock_client)
+
+    # Ensure the OUTPUT_S3_URI is the default placeholder value
+    monkeypatch.setattr(
+        text_to_video, "OUTPUT_S3_URI", "s3://REPLACE-WITH-YOUR-S3-BUCKET-NAME"
+    )
+
+    # Capture print output
+    printed_messages = []
+    monkeypatch.setattr(
+        "builtins.print", lambda *args: printed_messages.append(args[0])
+    )
+
+    # Call the main function
+    text_to_video.main()
+
+    # Verify the error message is printed
+    assert (
+        "ERROR: You must replace the OUTPUT_S3_URI with your own S3 bucket URI"
+        in printed_messages
+    )
+
+    # Verify that no AWS API calls were made
+    mock_client.start_async_invoke.assert_not_called()