Amazon Bedrock AI Karaoke is an interactive demonstration of Amazon Bedrock. Users complete the prompt with the microphone and choose the best response. Powered by Amazon Bedrock, Amazon Transcribe and Amazon ECS Anywhere.
import RPi.GPIO as GPIO
import time
import subprocess

# Constants
FAN_PINS = [14, 26]
PWM_FREQUENCY = 100
TEMP_THRESHOLDS = [75, 70, 60, 50, 40, 30]
FAN_SPEEDS = [100, 85, 70, 50, 25, 15, 0]
SLEEP_TIME = 5

def get_temp():
    """
    Retrieves the CPU temperature of the Raspberry Pi.
    
    :return: CPU temperature as float
    :raises RuntimeError: if temperature retrieval fails
    """
    output = subprocess.run(['vcgencmd', 'measure_temp'], capture_output=True)
    temp_str = output.stdout.decode()
    try:
        return float(temp_str.split('=')[1].split("'")[0])
    except (IndexError, ValueError):
        raise RuntimeError('Could not parse temperature. Output: {}'.format(temp_str))

def set_fan_speed(fans, speed):
    """
    Sets the duty cycle for all fan PWM channels.
    
    :param fans: List of fan PWM channels
    :param speed: Duty cycle to set for the fans
    """
    for fan in fans:
        fan.ChangeDutyCycle(speed)

def main():
    GPIO.setwarnings(False)
    GPIO.setmode(GPIO.BCM)
    
    fans = []
    for pin in FAN_PINS:
        GPIO.setup(pin, GPIO.OUT)
        fans.append(GPIO.PWM(pin, PWM_FREQUENCY))
    
    for fan in fans:
        fan.start(100)

    try:
        print("Starting fan control loop")
        while True:
            temp = get_temp()
            for i, threshold in enumerate(TEMP_THRESHOLDS):
                if temp > threshold:
                    set_fan_speed(fans, FAN_SPEEDS[i])
                    break
            time.sleep(SLEEP_TIME)
    except KeyboardInterrupt:
        pass
    finally:
        for fan in fans:
            fan.stop()
        GPIO.cleanup()

if __name__ == "__main__":
    main()

lib/fans/fans.py
aws-samples/amazon-bedrock-ai-karaoke

# Copy all files in dir to s3 bucket and delete them from local dir.

import boto3
import os
from apscheduler.schedulers.blocking import BlockingScheduler


def main():
    print("Starting scheduled run of main function")
    s3=boto3.resource('s3')

    bucket_name=os.getenv('BUCKET_NAME')

    for subdir, dirs, files in os.walk("/results"):

        for file in files:
            full_path=os.path.join(subdir, file)
            print(full_path)
            s3.Bucket(bucket_name).upload_file(full_path, full_path)
            print("Uploaded " + full_path)
            os.remove(full_path)

    print("Done")

print("Process starting")
scheduler=BlockingScheduler()
scheduler.add_job(main, 'interval', hours = 1)
scheduler.start()

lib/s3-transfer/run.py
aws-samples/amazon-bedrock-ai-karaoke

import time


while True:
    print("Developing")
    time.sleep(10)
lib/server/dev.py
aws-samples/amazon-bedrock-ai-karaoke
#!/usr/bin/env python

import asyncio
from functools import partial

import json
import datetime
from websockets import serve, exceptions, connect

import os
import sounddevice

from amazon_transcribe.client import TranscribeStreamingClient
from amazon_transcribe.handlers import TranscriptResultStreamHandler
from amazon_transcribe.model import TranscriptEvent

from states import State
from server_state import ServerState

try:
    import RPi.GPIO as GPIO
except ImportError:
    print("RPi.GPIO module is not available. Using mouse click instead.")
    # You can use a mock or a dummy module as a fallback
    GPIO = None  # or some mock object

## Uncomment this to see all websocket messages
# import logging
# logger = logging.getLogger('websockets')
# logger.setLevel(logging.DEBUG)
# logger.addHandler(logging.StreamHandler())

WEBSOCKET_PORT = 8765           # Standard websocket port
WEBSOCKET_IP = '127.0.0.1'      # Interface for websocket server to listen on
POLL_DELAY = 0.5                # How long to pause between sending websocket data.
MIC_SAMPLE_RATE_HZ = 48000      # This may change depending on your microphone

aws_region = os.getenv("AWS_DEFAULT_REGION", "us-east-1")
language_code = os.getenv("LANGUAGE_CODE", "en-US")
vocab_filter_name = os.getenv("VOCAB_FILTER_NAME", "BadWords-AiKaraokeStack")
vocab_filter_method = os.getenv("VOCAB_FILTER_METHOD", "mask")

class MyEventHandler(TranscriptResultStreamHandler):

    def __init__(self, stream, server_state):
        super().__init__(stream)
        self.server_state = server_state

    async def modify_string(self, s):
        """Remove the capitals and full stop from the punctuated output from Amazon Transcribe"""
        #TODO - Handle proper nouns so they remain capitalised
        first_cap_to_lower = ''
        full_stop_removed = ''
        
        # Convert the first capital letter to lowercase
        for i, c in enumerate(s):
            if c.isupper():
                first_cap_to_lower = s[:i] + c.lower() + s[i+1:]
                break
        else:  # No capital letter found
            first_cap_to_lower = s
        
        # Remove last full stop
        if first_cap_to_lower and first_cap_to_lower[-1] == '.':
            full_stop_removed = first_cap_to_lower[:-1]
        else:  # No full stop found
            full_stop_removed = first_cap_to_lower
        
        return full_stop_removed

    async def handle_transcript_event(self, transcript_event: TranscriptEvent):
        """Handle transcript events asyncronously."""
        results = transcript_event.transcript.results
        
        for result in results:
            if not result.is_partial:
                if '_' in self.server_state.my_prompt:
                    snippet = await self.modify_string(result.alternatives[0].transcript)
                    self.server_state.my_prompt = self.server_state.my_prompt.replace('_', snippet, 1)
                    print("Prompt:", self.server_state.my_prompt)

                if '_' in self.server_state.my_prompt:
                    break
                else:
                    await asyncio.to_thread(self.server_state.handle_generation)
                    break
    
async def mic_stream(server_state):
    """This function wraps the raw input stream from the microphone 
    forwarding the blocks to an asyncio.Queue."""
    loop = asyncio.get_event_loop()
    # loop.set_debug(True)  # Enable debug
    input_queue = asyncio.Queue()

    def callback(indata, frame_count, time_info, status):
        # Only put audio on queue if state is transcribing - to avoid queuing up chit chat while looking at the image
        if (server_state.my_state == State.TRANSCRIBING):
            loop.call_soon_threadsafe(
                input_queue.put_nowait, (bytes(indata), status))

    # Be sure to use the correct parameters for the audio stream that matches
    # the audio formats described for the source language you'll be using:
    # https://docs.aws.amazon.com/transcribe/latest/dg/streaming.html
    sd_stream = sounddevice.RawInputStream(
        channels=1,
        samplerate=MIC_SAMPLE_RATE_HZ,
        callback=callback,
        blocksize=1024 * 2,
        dtype="int16"
    )
    print(f"Transcribing with sample rate {sd_stream.samplerate} device {sd_stream.device}")
    # Initiate the audio stream and asynchronously yield the audio chunks as they become available.
    with sd_stream:
        while True:
            indata, status = await input_queue.get()
            yield indata, status

async def write_chunks(stream, server_state):
    """This connects the raw audio chunks generator coming from the microphone 
    and passes them along to the transcription stream."""
    async for chunk, status in mic_stream(server_state):
        await stream.input_stream.send_audio_event(audio_chunk=chunk)
    await stream.input_stream.end_stream()


async def basic_transcribe(server_state):
    """Setup transcription client."""
    # Setup up our client with our chosen AWS region
    client = TranscribeStreamingClient(region=aws_region)

    # Start transcription to generate our async stream
    transcribe_stream = await client.start_stream_transcription(
        language_code=language_code,
        media_sample_rate_hz=MIC_SAMPLE_RATE_HZ,
        media_encoding="pcm",
        vocab_filter_name=vocab_filter_name,
        vocab_filter_method=vocab_filter_method
    )

    # Instantiate our handler and start processing events
    handler = MyEventHandler(transcribe_stream.output_stream, server_state)
    try:
        await asyncio.gather(write_chunks(transcribe_stream, server_state),handler.handle_events())
    except Exception as error:
        await transcribe_stream.input_stream.end_stream()
        server_state.my_state = State.ERROR
        server_state.my_error = str(error)
        server_state.my_error_time = datetime.datetime.now().timestamp()
        print(f"ERROR: {server_state.my_error}")

async def consumer_handler(websocket, server_state):
    async for message in websocket:
        if message == "A":
            server_state.red_button_callback(None)
        elif message == "B":
            server_state.blue_button_callback(None)
        else:
            print(f'Unknown message {message}')

async def producer_handler(websocket, server_state):
    while True:        
        try:
            data = {
                "event_time": str(datetime.datetime.now()),
                "prompt": server_state.my_prompt,
                "model": server_state.my_model,
                "result_a": server_state.my_result_a,
                "result_b": server_state.my_result_b,
                "state": str(server_state.my_state),
                "error": server_state.my_error
            }
            await websocket.send(json.dumps(data))

        except exceptions.ConnectionClosed \
                or exceptions.ConnectionClosedOK \
                or exceptions.ConnectionClosedError \
                or KeyboardInterrupt as e:
            pass
        await asyncio.sleep(0.1)

async def handler(websocket, server_state):
    await asyncio.gather(
        consumer_handler(websocket, server_state),
        producer_handler(websocket, server_state),
    )

async def poll_handler(server_state):
    """Handle the GPIO button presses and restarting transcription after errors"""
    server_state.my_task = asyncio.create_task(basic_transcribe(server_state))

    while True:
        try:
            time_now = datetime.datetime.now().timestamp()

            if server_state.my_state == State.ERROR and (time_now - server_state.my_error_time) > 20:
                print("Starting again after 30 seconds after an error.")
                server_state.get_next_prompt()
                if server_state.my_task:
                    server_state.my_task.cancel()
                    server_state.my_task = asyncio.create_task(basic_transcribe(server_state))
                    
            if server_state.my_state == State.REVIEW_TXT or server_state.my_state == State.REVIEW_IMG:
                try:
                    # Initialize GPIO
                    if GPIO is not None:
                        GPIO.setmode(GPIO.BCM)  # Use BCM numbering scheme
                        GPIO.setwarnings(False) # Disable warnings

                        # Pin configuration
                        red_button_pin = 16
                        blue_button_pin = 17

                        # Set up the button pins as input pins with pull-up resistors
                        GPIO.setup(red_button_pin, GPIO.IN, pull_up_down=GPIO.PUD_UP)
                        GPIO.setup(blue_button_pin, GPIO.IN, pull_up_down=GPIO.PUD_UP)

                        # Register the button press events
                        GPIO.add_event_detect(red_button_pin, GPIO.FALLING, callback=server_state.red_button_callback, bouncetime=600)
                        GPIO.add_event_detect(blue_button_pin, GPIO.FALLING, callback=server_state.blue_button_callback, bouncetime=600)
                    else:
                        print("GPIO is not available. Skipping setup.")
                    print("Waiting for button press.")
                    while True:
                        if server_state.button_pressed:
                            await asyncio.sleep(1)
                            server_state.save_results()
                            if server_state.my_task:
                                server_state.my_task.cancel()
                            server_state.my_task = asyncio.create_task(basic_transcribe(server_state))
                            break
                        else: 
                            await asyncio.sleep(1)

                except KeyboardInterrupt:
                    print("Exiting program.")

                finally:
                    if GPIO is not None:
                        GPIO.cleanup()  # Clean up GPIO settings

            await asyncio.sleep(POLL_DELAY)
        
        except Exception as error:
            print(error)
            server_state.my_error = str(error)
            server_state.my_error_time = datetime.datetime.now().timestamp()
            server_state.my_state = State.ERROR
            continue

async def main():
    """Main method to initialize and run the server."""
    server_state = ServerState()
    # Create a partial function for producer_handler with server_state
    handler_with_state = partial(handler, server_state=server_state)
    
    # Schedule these calls *concurrently*:
    tasks = await asyncio.gather(
        poll_handler(server_state),
        serve(handler_with_state, WEBSOCKET_IP, WEBSOCKET_PORT)
    )

if __name__ == "__main__":
    asyncio.run(main())

lib/server/server.py
aws-samples/amazon-bedrock-ai-karaoke
#!/usr/bin/env python

import json
import datetime
import uuid
from random import randint
import boto3
import os

from states import State

aws_region = os.getenv("AWS_DEFAULT_REGION", "us-east-1")

# Encapsulate global state and functionality in a class
class ServerState:
    def __init__(self):
        """Initialize the state variables."""
        self.my_state = State.INITIALIZING
        self.prompts = []
        self.current_prompt_index = 0
        self.my_prompt = None
        self.my_model = None
        self.my_result_a = ""
        self.my_result_b = ""
        self.my_human_preference = None
        self.button_pressed = False
        self.my_task = None
        self.my_error = None
        self.my_error_time = datetime.datetime.now().timestamp()
        self.my_last_interaction_time = datetime.datetime.now().timestamp()
        self.my_uuid = uuid.uuid4()

        self.bedrock_runtime = boto3.client(
            service_name="bedrock-runtime",
            region_name=aws_region
        )

        self.load_prompts("prompts.json")
        self.get_next_prompt()

    def load_prompts(self, file_path):
        """Load prompts from a given JSON file."""
        try:
            with open(file_path, 'r') as f:
                self.prompts = json.load(f)
            print(f"Found {len(self.prompts)} prompts.")
        except Exception as e:
            print(f"An error occurred while loading prompts: {e}")
            self.my_state = State.ERROR
    
    def get_next_prompt(self):
        """Called at the end of a human interaction to get a new prompt and 
        reset the state variables."""
        self.my_result_a = ""
        self.my_result_b = ""
        self.my_human_preference = None
        self.button_pressed = False
        self.my_state = State.TRANSCRIBING
        self.my_uuid = uuid.uuid4()
        if self.prompts:
            if self.current_prompt_index >= len(self.prompts):
                self.current_prompt_index = 0  # Loop back to the start
            self.my_prompt = self.prompts[self.current_prompt_index]["prompt"]
            self.my_model = self.prompts[self.current_prompt_index]["model"]
            self.current_prompt_index += 1  # Increment the index for the next call

        else:
            print("No prompts are loaded.")
            return None
    
    def red_button_callback(self, channel):
        """Logic for handling a red button press."""
        self.my_human_preference = 'a'
        self.my_last_interaction_time = datetime.datetime.now().timestamp()
        print("Red button pressed!") 
        self.button_pressed = True
        if self.my_model == "claude":
            self.my_state = State.SELECT_A_TXT
        elif self.my_model == "sdxl":
            self.my_state = State.SELECT_A_IMG
        else:
            raise ValueError(f"Unknown model specified: {self.my_model}")
        
    def blue_button_callback(self, channel):
        """Logic for handling a blue button press."""
        self.my_human_preference = 'b'
        self.my_last_interaction_time = datetime.datetime.now().timestamp()
        print("Blue button pressed!")
        self.button_pressed = True
        if self.my_model == "claude":
            self.my_state = State.SELECT_B_TXT
        elif self.my_model == "sdxl":
            self.my_state = State.SELECT_B_IMG
        else:
            raise ValueError(f"Unknown model specified: {self.my_model}")
            
    def call_claude2(self):
        # Construct the body dictionary
        body_dict = {
            "prompt": self.my_prompt,
            "max_tokens_to_sample": 150,
            "temperature": 1,
            "top_k": 250,
            "top_p": 0.999,
            "stop_sequences": ["\n\nHuman:"],
            "anthropic_version": "bedrock-2023-05-31"
        }

        # Serialize the body dictionary to a JSON string
        body_str = json.dumps(body_dict)

        # Construct the kwargs dictionary
        kwargs = {
            "modelId": "anthropic.claude-v2",
            "contentType": "application/json",
            "accept": "*/*",
            "body": body_str
        }

        # Invoke the model
        response = self.bedrock_runtime.invoke_model_with_response_stream(**kwargs)
        stream = response.get('body')
        if stream:
            for event in stream:
                chunk = event.get('chunk')
                if chunk:
                    completion = json.loads(chunk.get('bytes')).get('completion')
                    self.my_result_a += completion

        print(f"Completion A: {self.my_result_a}")
        self.my_state = State.INFERENCE_TXT_B

        # Invoke the model
        response = self.bedrock_runtime.invoke_model_with_response_stream(**kwargs)
        stream = response.get('body')
        if stream:
            for event in stream:
                chunk = event.get('chunk')
                if chunk:
                    completion = json.loads(chunk.get('bytes')).get('completion')
                    self.my_result_b += completion

        print(f"Completion B: {self.my_result_b}")

    def invoke_sdxl(self):
        """Specific logic for making a sdxl prediction."""
        body_dict = {
            "text_prompts": [{"text": self.my_prompt}],
            "cfg_scale": 10,
            "seed": randint(0, 1000),
            "steps": 50
        }
        
        # Serialize the dictionary to a JSON string
        body_str = json.dumps(body_dict)

        # Construct the outer dictionary
        kwargs = {
            "modelId": "stability.stable-diffusion-xl-v0",
            "contentType": "application/json",
            "accept": "application/json",
            "body": body_str
        }

        # Invoke the model
        response = self.bedrock_runtime.invoke_model(**kwargs)
        response_body = json.loads(response.get("body").read())

        results = response_body.get("artifacts")[0].get("base64")
        return results

    def handle_image_gen(self):
        image = self.invoke_sdxl()
        self.my_result_a = image
        self.my_state = State.INFERENCE_IMG_B
        image = self.invoke_sdxl()
        self.my_result_b = image        
        
    def handle_generation(self):
        """General method to make a prediction based on the model type."""        
        if self.my_model == "claude":
            self.my_state = State.INFERENCE_TXT_A
            self.call_claude2()
            self.my_state = State.REVIEW_TXT
        elif self.my_model == "sdxl":
            self.my_state = State.INFERENCE_IMG_A
            self.handle_image_gen()
            self.my_state = State.REVIEW_IMG
        else:
            raise ValueError(f"Unknown model specified: {self.my_model}")
        
        self.my_task.cancel()

    def save_results(self):
        """Write to disk so human preferences can be uploaded to S3 later."""
        if self.my_human_preference:
            data = {
                'timestamp': str(datetime.datetime.now()),
                'instruction': self.my_prompt,
                'model': self.my_model,
                'prompt': self.my_prompt,
                'result_a': self.my_result_a,
                'result_b': self.my_result_b,
                'human_preference': self.my_human_preference,
            }

            try:
                with open(f"/results/{self.my_uuid}.json", "w") as f:
                    json.dump(data, f, indent=4)
            except Exception as e:
                print(f"An error occurred while saving results: {e}")
        self.get_next_prompt()
    
lib/server/server_state.py
aws-samples/amazon-bedrock-ai-karaoke
from enum import Enum, unique

@unique
class State(Enum):
    ERROR = 0
    INITIALIZING = 1
    TRANSCRIBING = 2
    INFERENCE_TXT_A = 3
    INFERENCE_TXT_B = 4
    INFERENCE_IMG_A = 5
    INFERENCE_IMG_B = 6
    REVIEW_TXT = 7
    REVIEW_IMG = 8
    SELECT_A_TXT = 9
    SELECT_A_IMG = 10
    SELECT_B_TXT = 11
    SELECT_B_IMG = 12
lib/server/states.py
aws-samples/amazon-bedrock-ai-karaoke