
subaud is authored by Court Schuett (he/him). While Court is an 8090 Inc employee, opinions and statements on this blog are his own.


Court Schuett


# ffmpeg streaming to KVS

[GitHub Repo](https://github.com/schuettc/ffmpeg-kvs-streaming)

[Amazon Kinesis Video Streams](https://aws.amazon.com/kinesis/video-streams) can be used to get access to other AWS services with streaming video. In this demo, we will see how to use [Node.js](https://nodejs.org/en) and [ffmpeg](https://ffmpeg.org/) to convert a file into a stream that can be sent to KVS.

![Overview](/images/blog/streaming-to-kvs-with-ffmpeg/ffmpegStreaming.png)

1. An object is created in the Amazon Simple Storage Service (Amazon S3) bucket. When this happens, a notification is sent to the associated AWS Lambda function.
2. This Lambda makes a request to the [Application Load Balancer](https://docs.aws.amazon.com/elasticloadbalancing/latest/application/introduction.html) associated with the [AWS Fargate](https://aws.amazon.com/fargate/) task with the object information.
3. The Fargate application downloads the object from S3 and begins processing (processing details below).
4. The Fargate application streams the contents of the object to [Amazon Kinesis Video Streams](https://aws.amazon.com/kinesis/video-streams/).

## Processing with ffmpeg

Once the file has been downloaded from S3, the application will begin processing the file with ffmpeg.

```typescript
async function processFfmpeg(fileName: string): Promise<{
  videoStream: PassThrough;
}> {
  console.log('Processing with ffmpeg');
  const videoStream = new PassThrough();
  let videStreamCount: number = 0;

  ffmpeg(fileName)
    .native()
    .output(videoStream)
    .outputOptions(['-profile:v baseline'])
    .videoCodec('libx264')
    .size('640x480')
    .audioCodec('aac')
    .format('matroska')
    .on('error', (error) => {
      console.log('Cannot process: ' + error.message);
    })
    .on('stderr', (data) => {
      if (videStreamCount % COUNT_FREQUENCY === 0) {
        console.info(`videoStream: ${data}`);
      }
      videStreamCount++;
    })
    .run();

  return { videoStream };
}
```

To do this, we will create a `PassThrough` stream that will be used as the output of the ffmpeg pipe. By using `.native()` as an input option, the ffmpeg pipe will read the contents of the file at the native frame rate. The output options used will format the output stream so that it can be used by KVS.

## Streaming Media

Because Kinesis Video Streams [PutMedia](https://docs.aws.amazon.com/kinesisvideostreams/latest/dg/API_dataplane_PutMedia.html) does not have a corresponding AWS SDK command, we must use a direct HTTP request. This involves several extra steps:

1. GetDataEndpoint
2. GetCredentials
3. SignRequest
4. PutMedia

### GetDataEndpoint

```typescript
async function getEndpoint(
  streamArn: string,
): Promise<GetDataEndpointCommandOutput['DataEndpoint']> {
  const response = await kvsClient.send(
    new GetDataEndpointCommand({
      APIName: APIName.PUT_MEDIA,
      StreamARN: streamArn,
    }),
  );
  return response.DataEndpoint;
}
```

The first step to sending a stream to KVS is to get an endpoint for the stream we are going to use. We have already created the stream during the deployment of the CDK, so we will use that stream as the target.

### GetCredentials

```typescript
const ECS_ROLE = process.env.ECS_ROLE || '';

async function getCredentials(): Promise<
  AssumeRoleCommandOutput['Credentials']
> {
  console.log('Getting credentials');
  const response = await stsClient.send(
    new AssumeRoleCommand({
      RoleArn: ECS_ROLE,
      RoleSessionName: 'kvs-stream',
    }),
  );
  return response.Credentials;
}
```

In order to stream to the KVS, we will need to assume an IAM role that has permission to do this. More details about doing this using a local Docker are outlined below.

#### IAM Role

```typescript
this.kinesisRole = new Role(this, 'kinesisRole', {
  assumedBy: new CompositePrincipal(
    new ArnPrincipal(kvsProducerRole.roleArn),
    new AccountPrincipal(Stack.of(this).account),
  ),
  inlinePolicies: {
    ['KinesisVideoPolicy']: new PolicyDocument({
      statements: [
        new PolicyStatement({
          resources: ['*'],
          actions: ['kinesisvideo:PutMedia'],
        }),
      ],
    }),
  },
});
```

During the deployment of the CDK, an [IAM Role](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles.html) is created. This is the role that we will assume and use to `PutMedia`. This Role is passed to the application through environment variables.

### SignRequest

```typescript
async function signRequest(
  streamArn: string,
  endpoint: string,
): Promise<{ signedUrl: aws4.Request; reqUrl: string }> {
  const credentials = await getCredentials();
  if (!credentials) {
    throw new Error('Failed to get credentials');
  }
  const signedUrl = aws4.sign(
    {
      host: endpoint,
      path: '/putMedia',
      service: 'kinesisvideo',
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'x-amzn-stream-arn': streamArn,
        'x-amzn-fragment-timecode-type': 'ABSOLUTE',
        'x-amz-content-sha256': 'UNSIGNED-PAYLOAD',
      },
    },
    {
      accessKeyId: credentials.AccessKeyId,
      secretAccessKey: credentials.SecretAccessKey,
      sessionToken: credentials.SessionToken,
    },
  );
  if (!signedUrl || !signedUrl.host || !signedUrl.path) {
    throw new Error('Failed to sign request');
  }
  const reqUrl = signedUrl.host + signedUrl.path;
  return { signedUrl: signedUrl, reqUrl: reqUrl };
}
```

Using the credentials from the assumed role, the Stream ARN, and the DataEndpoint, we will use [aws4](https://github.com/mhart/aws4) to sign the request we will be making and get a URL to use in the `PutMedia`.

### PutMedia

```typescript
const agent = new Agent({
  rejectUnauthorized: false,
});
const axiosParams = {
  method: 'POST',
  timeout: 40 * 1000,
  url: reqUrl,
  headers: signedUrl.headers as AxiosHeaders,
  data: stream,
  responseType: 'stream' as ResponseType,
  maxContentLength: Infinity,
  httpsAgent: agent,
};
```

Next, we will combine all of the pieces together as we prepare to `PutMedia`. Using the stream generated by `ffmpeg` and the url generated by `aws4`, we will create a `POST` request using [axios](https://axios-http.com/docs/intro).

```typescript
let streamCount: number = 0;

try {
  const response = (await axios(axiosParams)) as AxiosResponse;
  const startFragmentNumber = await getStartFragmentNumber(response);

  response.data.on('data', (chunk: Buffer) => {
    if (streamCount % COUNT_FREQUENCY === 0) {
      console.info(`Stream: ${streamArn} - Chunk: ${chunk.toString()}`);
    }
    streamCount++;
  });
  response.data.on('end', async () => {
    console.log(`Stream: ${streamArn} ended`);
  });

  response.data.on('error', (error: Error) => {
    console.error(`Error in ${streamArn} stream: ${error}`);
  });
  if (startFragmentNumber) {
    return { streamArn, startFragmentNumber };
  } else {
    throw new Error('StartFragment not found');
  }
} catch (error) {
  console.error('Error in putMedia:', error);
  throw error;
}
```

Finally, we will make the `POST` request to the `PutMedia` API and process the responses. For every fragment sent, KVS will acknowledge with a response:

```json
{
       Acknowledgement : {
          "EventType": enum
          "FragmentTimecode": Long,
          "FragmentNumber": Long,
          "ErrorId" : String
      }
}
```

## Testing

In order to test this demo, you can upload a movie file to the S3 bucket that is created during the CDK deployment. This will start processing on the Fargate container and streaming to KVS. This stream can be seen in the [KVS Console](https://us-east-1.console.aws.amazon.com/kinesisvideo/home?region=us-east-1#/streams) and finding the stream named `ffmpeg-streaming`.

![KVSConsole](/images/blog/streaming-to-kvs-with-ffmpeg/KVS.png)

This stream can be viewed with `Media playback`.

![MediaPlayback](/images/blog/streaming-to-kvs-with-ffmpeg/MediaPlayback.png)

## Developing Locally with Docker

Because development like this can require extensive trial and error testing, it is useful to test locally with a Docker image before deploying to Fargate. This will involve several steps:

1. Build a local Docker container
2. Pass credentials and environment variables to this container
3. Run the local Docker container
4. Trigger this container

### Build the Docker container

To do this, we will use [Docker Compose](https://docs.docker.com/compose/) and a `.env` file.

```docker-compose
version: '3'
services:
  ffmpeg:
    build:
      context: .
      dockerfile: Dockerfile
    environment:
      - AWS_PROFILE
      - AWS_ACCESS_KEY_ID
      - AWS_SECRET_ACCESS_KEY
      - AWS_SESSION_TOKEN
      - COUNT_FREQUENCY=60
      - ECS_ROLE
      - KVS_STREAM_ARN
    volumes:
      - ~/.aws/:/root/.aws:ro
    ports:
      - '80:80'
    command: ['npm', 'start']
```

### Pass credentials and environment variables

This will use the `Dockerfile` used in the deployment with environment variables and a mounted volume that references `~./aws`. This will allow us to pass credentials from the [`~.aws/credentials`](https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-files.html) file and use the associated `AWS_PROFILE`.

Alternatively, you can export `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, and `AWS_SESSION_TOKEN` as environment variables.

`ECS_ROLE` and `KVS_STREAM_ARN` are created during the deployment of the CDK and passed to the `.env` file through the outputs of the CDK.

### Run the Docker container

To run the Docker container:

```bash
docker-compose up --build
```

This will re-build and run the Docker container

### Trigger the Docker container

[Postman](https://www.postman.com/) is a useful tool for making API request. This is similar to what we will be doing with the Lambda function and serves as a convenient way to test locally. The Docker container will be running on port 80, so we will make a `POST` request to `localhost:80/processObject` passing the `bucketName` and `keyName` of the object to be processed.

![Postman](/images/blog/streaming-to-kvs-with-ffmpeg/Postman.png)

## Using this demo

### Requirements

- yarn installed
- ARM processor
- Docker desktop running

### Deploy

```bash
yarn launch
```

### Cleanup

```bash
yarn cdk destroy
```


This demo will go through the steps to process a file with ffmpeg and stream the contents to KVS using Node.js

Streaming to KVS with ffmpeg and Node.js


In this demo, we will see how to build an application that will accept streaming audio sent via gRPC and have the audio transcribed using [OpenAI Whisper](https://openai.com/research/whisper). This demo builds off of the previous [gRPC Streaming Audio with Node](https://subaud.io/blog/node-grpc-server) blog post. However, in this demo, the ECS uses an [EC2 based deployment that supports GPUs](https://docs.aws.amazon.com/AmazonECS/latest/developerguide/ecs-gpu.html). This requires several changes to the deployment that will be highlighted.

## Overview

![Overview](/images/blog/streaming-whisper-ecs-hosted/StreamingWhisper.png)

The basic concepts of the previous [gRPC Streaming Audio with Node](https://subaud.io/blog/node-grpc-server) remain largely the same. An audio stream will be generated from a local gRPC client and sent to a gRPC server hosted in [Amazon Elastic Container Service (ECS)](https://aws.amazon.com/ecs/). The server will transcribe the audio and return the results to the gRPC client. While the previous model used a Fargate hosted container, this demo uses an EC2 deployed container in order to use a GPU. Additionally, while the previous model used [Amazon Transcribe](https://aws.amazon.com/transcribe/) to provide the Speech To Text processing, this uses Whisper as the ASR and is included in the container that is built and deployed. Because Whisper does not normally support streaming audio, we will have to introduce a chunking mechanism that Whisper can use.

## GitHub Repo

https://github.com/schuettc/hosted-whisper-streaming

## ECS Deployment

```typescript
const cluster = new Cluster(this, 'cluster', {
  vpc: props.vpc,
});

const launchTemplate = new LaunchTemplate(this, 'LaunchTemplate', {
  machineImage: EcsOptimizedImage.amazonLinux2(AmiHardwareType.GPU),
  instanceType: new InstanceType('g4dn.2xlarge'),
  requireImdsv2: true,
  userData: UserData.forLinux(),
  securityGroup: ecsServiceSecurityGroup,
  role: ec2Role,
  blockDevices: [
    {
      deviceName: '/dev/xvda',
      volume: BlockDeviceVolume.ebs(300),
    },
  ],
});

const autoScalingGroup = new AutoScalingGroup(this, 'AutoScalingGroup', {
  vpc: props.vpc,
  launchTemplate: launchTemplate,
  desiredCapacity: 1,
  minCapacity: 1,
  maxCapacity: 2,
  vpcSubnets: { subnetType: SubnetType.PUBLIC },
});

const capacityProvider = new AsgCapacityProvider(this, 'AsgCapacityProvider', {
  autoScalingGroup,
});

cluster.addAsgCapacityProvider(capacityProvider);
```

Some changes are required in the CDK to use an EC2 based deployment of ECS. Here we see the [Launch Template](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ec2-launch-templates.html) that defines the EC2 instance parameters. An autoscaling group is also configured to allow this demo to scale up if needed.

```typescript
const taskDefinition = new Ec2TaskDefinition(this, 'taskDefinition', {
  taskRole: hostedWhisperStreamingRole,
  networkMode: NetworkMode.BRIDGE,
});

taskDefinition.addContainer('HostedWhisperStreaming', {
  image: ContainerImage.fromAsset('src/resources/whisperServer'),
  environment: {
    ECS_LOGLEVEL: props.logLevel,
  },
  gpuCount: 1,
  memoryLimitMiB: 31690,
  cpu: 8192,
  portMappings: [{ containerPort: 50051, hostPort: 50051 }],
  logging: new AwsLogDriver({ streamPrefix: 'HostedWhisperStreaming' }),
});

this.ecsService = new Ec2Service(this, 'ECSService', {
  cluster: cluster,
  taskDefinition: taskDefinition,
  capacityProviderStrategies: [
    { capacityProvider: capacityProvider.capacityProviderName, weight: 1 },
  ],
});
```

Next we configure the [Task Definition](https://docs.aws.amazon.com/AmazonECS/latest/developerguide/task_definitions.html) that will use the included Docker container image.

## Docker Image

```Dockerfile
FROM --platform=linux/amd64 nvidia/cuda:12.2.2-cudnn8-runtime-ubuntu22.04 AS base

RUN apt-get update && apt-get install -y --no-install-recommends \
    python3 python3.10-dev python3-pip python3.10-venv libsndfile1 build-essential curl git  && \
    rm -rf /var/lib/apt/lists/*

FROM base AS builder

COPY src/requirements.txt ./

RUN pip3 install --upgrade pip setuptools wheel && \
    pip3 install --user -r requirements.txt && \
    pip3 install --user git+https://github.com/openai/whisper.git

FROM base
ENV MODEL=${MODEL}
ENV LOG_LEVEL=${LOG_LEVEL}

COPY --from=builder /root/.local /root/.local
COPY src/* ./
RUN chmod +x /entrypoint.sh

EXPOSE 50051

ENTRYPOINT ["/entrypoint.sh"]


```

The Dockerfile included creates a Docker image that uses the [nvidia/cuda Docker image](https://hub.docker.com/r/nvidia/cuda/) as the base. This image will allow us to use the GPUs on the EC2 instance.

The `pip3 install` commands are particularly important when using this image. These will update and install the necessary tools to install and configure Whisper. In order to ensure the model is downloaded and installed, `python3 -c "import whisper; whisper.load_model('base')"` is run as part of the build.

As part of the `entrypoint.sh` script, we will be sure to set the correct path for the cuDNN libraries.

```bash
export LD_LIBRARY_PATH=`python3 -c 'import os; import nvidia.cublas.lib; import nvidia.cudnn.lib; print(os.path.dirname(nvidia.cublas.lib.__file__) + ":" + os.path.dirname(nvidia.cudnn.lib.__file__))'`
```

This solves the `Could not load library libcudnn_ops_infer.so.8. Error: libcudnn_ops_infer.so.8: cannot open shared object file: No such file or directory` issue that is caused by the `LD_LIBRARY_PATH` not containing the correct locations for the libraries.

Finally, we will start the server.

```bash
python3 server.py
```

## Whisper Server

The server starts with two Servicers - AudioTranscriberServicer and HealthCheckServicer. The HealthCheckServicer is used to validate to the Application Load Balancer

```python
def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    audio_transcriber_pb2_grpc.add_AudioTranscriberServiceServicer_to_server(
        AudioTranscriberServicer(), server
    )
    audio_transcriber_pb2_grpc.add_HealthCheckServiceServicer_to_server(
        HealthCheckServicer(), server
    )
    server.add_insecure_port("[::]:50051")
    server.start()
    logger.info("Server started. Listening on port 50051.")
    server.wait_for_termination()
```

The AudioTranscriberServicer will be used to process the audio stream and return the transcriptions to the gRPC client. Whisper does not natively support streaming audio, so we must break the streamed audio into chunks that Whisper can use. To do this, audio processor will process the audio frames and use VAD to detect speech segments. These frames are buffered until a non-speech frame is encountered. When this happens, the frames are joined into a chunk. If the chunk is long enough, it is processed by [faster-whisper](https://github.com/SYSTRAN/faster-whisper) using the `large-v2` model. This allows for rapid transcription of streaming audio while allowing Whisper to provide the best results.

## Notes and Warnings

Because this deployment uses GPU based instance(s), be sure to check the [prices of the instances](https://aws.amazon.com/ec2/pricing/on-demand/) that will be used.

![Pricing](/images/blog/streaming-whisper-ecs-hosted/Pricing.png)

## Testing

This demo requires a [domain hosted in Route 53](https://docs.aws.amazon.com/Route53/latest/DeveloperGuide/registrar.html) so that a certificate can be associated with the [Application Load Balancer](https://docs.aws.amazon.com/elasticloadbalancing/latest/application/load-balancer-target-groups.html#target-group-protocol-version) listener. To configure the domain within the deployment, create a `.env` file with a `DOMAIN_NAME=` associated with a [Hosted Zone Name](https://docs.aws.amazon.com/Route53/latest/DeveloperGuide/hosted-zones-working-with.html) in the account.

Once the `.env` has been configured, you can deploy the CDK from the cloned repo:

```bash
yarn deploy
```

After deploying the CDK, you can test the functionality by running the included client. This client will capture audio from your microphone and stream it to the server using gRPC. Output from the CDK will be used to configure the client in the `/client/.env` file. To run the client:

```bash
cd client
yarn build
yarn start
```

![ClientStreaming](/images/blog/streaming-whisper-ecs-hosted/CLIExample.gif)

## Cleanup

In order to delete this CDK, you should remove the Auto Scaling Group first. Once that is done, you can delete the Stack from Cloudformation.


A CDK stack that will deploy a will create an ECS hosted application for streaming audio with gRPC to Whisper and getting the transcription output

Streaming

Streaming to KVS with ffmpeg and Node.js

Using Whisper with Streaming Audio