[Python] Streaming audio to Amazon Connect through Kinesis Video Streams

What I am trying to achieve is the following.

In Amazon Connect, when I create a Flow with Start Streaming Data block, it starts streaming to a Kinesis Video Stream, and I can call a lambda function with the streams information.

Next, I want this lambda function to stream some audio into the Kinesis Video Stream, so the customer on the phone can hear it.

I have created the following python script, but I cannot hear the audio on the phone call. I have looked at the error logs of connect, and there are no errors that are occuring. A solution to fix this script, or any examples of a similar pipeline would be awesome.

import boto3
import botocore.exceptions
import types
from functools import lru_cache
import time
import io
import wave

class KinesisVideo(object):
_CONTROL_SERVICES = ('kinesisvideo', )
_DATA_SERVICES = ('kinesis-video-media', 'kinesis-video-archived-media')

def __init__(self, session=None):
self._session = session or boto3.Session()
self._methods = {}
for service in self._CONTROL_SERVICES + self._DATA_SERVICES:
prototype = self._get_client_for_service(service)
for method in prototype.meta.method_to_api_mapping.keys():
self._methods[method] = service

@lru_cache()
def _get_arn_for_stream_name(self, stream_name):
response = self.describe_stream(StreamName=stream_name)
return response['StreamInfo']['StreamARN']

@lru_cache()
def _get_endpoint_for_stream_method(self, stream_arn, method):
response = self.get_data_endpoint(StreamARN=stream_arn, APIName=method.upper())
return response['DataEndpoint']

@lru_cache()
def _get_client_for_service(self, service, endpoint_url=None):
client = self._session.client(service, endpoint_url=endpoint_url)
if service == 'kinesis-video-media':
client = self._patch_kinesis_video_media(client)
return client

@lru_cache()
def _get_client_by_arguments(self, method, stream_name=None, stream_arn=None):
service = self._methods[method]
if service not in self._DATA_SERVICES:
return self._get_client_for_service(service)
if not (bool(stream_name) ^ bool(stream_arn)):
raise botocore.exceptions.ParamValidationError(report=
'One of StreamName or StreamARN must be defined ' + \
'to determine service endpoint'
)
stream_arn = self._get_arn_for_stream_name(stream_name) if stream_name else stream_arn
endpoint_url = self._get_endpoint_for_stream_method(stream_arn, method)
return self._get_client_for_service(service, endpoint_url)

def __getattr__(self, method):
if method not in self._methods:
return getattr(super(), method)
kwarg_map = {'StreamName': 'stream_name', 'StreamARN': 'stream_arn'}
def _api_call(**kwargs):
filtered_kwargs = {kwarg_map[k]: v for k, v in kwargs.items() if k in kwarg_map}
client = self._get_client_by_arguments(method, **filtered_kwargs)
return getattr(client, method)(**kwargs)
return _api_call

@staticmethod
def _patch_kinesis_video_media(client):
client.meta.service_model._service_description['operations']['PutMedia'] = {
'name': 'PutMedia',
'http': {'method': 'POST', 'requestUri': '/putMedia'},
'input': {'shape': 'PutMediaInput'},
'output': {'shape': 'PutMediaOutput'},
'errors': [
{'shape': 'ResourceNotFoundException'},
{'shape': 'NotAuthorizedException'},
{'shape': 'InvalidEndpointException'},
{'shape': 'ClientLimitExceededException'},
{'shape': 'ConnectionLimitExceededException'},
{'shape': 'InvalidArgumentException'}
],
'authtype': 'v4-unsigned-body',
}
client.meta.service_model._shape_resolver._shape_map['PutMediaInput'] = {
'type': 'structure',
'required': ['FragmentTimecodeType', 'ProducerStartTimestamp'],
'members': {
'FragmentTimecodeType': {
'shape': 'FragmentTimecodeType',
'location': 'header',
'locationName': 'x-amzn-fragment-timecode-type',
},
'ProducerStartTimestamp': {
'shape': 'Timestamp',
'location': 'header',
'locationName': 'x-amzn-producer-start-timestamp',
},
'StreamARN': {
'shape': 'ResourceARN',
'location': 'header',
'locationName': 'x-amzn-stream-arn',
},
'StreamName': {
'shape': 'StreamName',
'location': 'header',
'locationName': 'x-amzn-stream-name',
},
'Payload': {
'shape': 'Payload',
},
},
'payload': 'Payload',
}
client.meta.service_model._shape_resolver._shape_map['PutMediaOutput'] = {
'type': 'structure',
'members': {'Payload': {'shape': 'Payload'}},
'payload': 'Payload',
}
client.meta.service_model._shape_resolver._shape_map['FragmentTimecodeType'] = {
'type': 'string',
'enum': ['ABSOLUTE', 'RELATIVE'],
}
client.put_media = types.MethodType(
lambda self, **kwargs: self._make_api_call('PutMedia', kwargs),
client,
)
client.meta.method_to_api_mapping['put_media'] = 'PutMedia'
return client

def main():
session = boto3.Session(
aws_access_key_id='MY_ACCESS_KEY_ID',
aws_secret_access_key='MY_SECRET_ACCESS_KEY',
region_name='MY_REGION'
)

video = KinesisVideo(session=session)

print(video.list_streams())

start_tmstp = repr(time.time()).split('.')[0]

print(start_tmstp)

# Ensure the audio data is correctly handled for Amazon Connect
with open('packages/python/output.mkv', 'rb') as payload_file:
payload_data = payload_file.read()

# Create a PCM file from the audio data
pcm_data = io.BytesIO(payload_data)
with wave.open(pcm_data, 'wb') as wav_file:
wav_file.setnchannels(1) # Mono
wav_file.setsampwidth(2) # 2 bytes per sample
wav_file.setframerate(8000) # Sample rate
wav_file.writeframes(payload_data)

response = video.put_media(
StreamName='MY_STREAM_NAME',
Payload=pcm_data,
FragmentTimecodeType='RELATIVE',
ProducerStartTimestamp=start_tmstp,
)

import json
for event in map(json.loads, response['Payload'].read().decode('utf-8').splitlines()):
print(event)

if __name__ == '__main__':
main()

Continue reading...