
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


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

Using Whisper with Streaming Audio


This demo will guide you through the process of deploying a [gRPC](https://grpc.io/) server that is capable of receiving streaming audio from a gRPC client and processing that audio with [Amazon Transcribe](https://aws.amazon.com/transcribe/) and returning the result to the client.

[GitHub Repo](https://github.com/schuettc/node-grpc-demo)

## Server Infrastructure

![gRPCServerOverview](/images/blog/node-grpc-server/gRPCServerOverview.png)

This demo will deploy a [Fargate](https://aws.amazon.com/fargate/) container that runs the gRPC server. This gRPC server will be available through an [Application Load Balancer](https://aws.amazon.com/elasticloadbalancing/application-load-balancer/) that is enabled with HTTPS through a [Certificate](https://docs.aws.amazon.com/acm/latest/userguide/gs-acm-request-public.html) generated through [AWS Certificate Manager](https://aws.amazon.com/certificate-manager/).

### Certificate Manager

```typescript
this.hostedZone = HostedZone.fromLookup(this, 'HostedZone', {
  domainName: props.domainName,
});

this.certificate = new Certificate(this, 'Certificate', {
  domainName: `grpc.${props.domainName}`,
  validation: CertificateValidation.fromDns(this.hostedZone),
});
```

Using a provided domain name, the CDK will lookup the associated [HostedZone](https://docs.aws.amazon.com/Route53/latest/DeveloperGuide/hosted-zones-working-with.html) and create a Certificate that will be associated with the Application Load Balancer.

### Application Load Balancer

```typescript
const grpcServerTargetGroup = new ApplicationTargetGroup(
  this,
  'gRPCServerTargetGroup',
  {
    vpc: props.vpc,
    port: 50051,
    protocol: ApplicationProtocol.HTTP,
    protocolVersion: ApplicationProtocolVersion.GRPC,
    targets: [this.fargateService],
    healthCheck: {
      healthyGrpcCodes: '12',
    },
  },
);

this.applicationLoadBalancer.addListener('fargateListener', {
  port: 50051,
  protocol: ApplicationProtocol.HTTPS,
  certificates: [ListenerCertificate.fromCertificateManager(props.certificate)],
  open: true,
  defaultTargetGroups: [grpcServerTargetGroup],
});

props.applicationLoadBalancerSecurityGroup.addIngressRule(
  Peer.anyIpv4(),
  Port.tcp(50051),
);
```

This will create an Application Load Balancer [Target Group](https://docs.aws.amazon.com/elasticloadbalancing/latest/application/load-balancer-target-groups.html) that uses `ApplicationProtocolVersion.GRPC` as the protocol version. Additionally, the `healthCheck` will look for gRPC codes in the response from the gRPC server. The [Listener](https://docs.aws.amazon.com/elasticloadbalancing/latest/application/load-balancer-listeners.html) associated with the Application Load Balancer uses HTTPS and the previously created Certificate. The Listener will use port `50051` and forward to port `50051` on the Fargate task. This port is also opened on the Application Load Balancer Security Group.

### Route 53

```typescript
new ARecord(this, 'grpcARecord', {
  zone: props.hostedZone,
  recordName: 'grpc',
  target: RecordTarget.fromAlias(
    new LoadBalancerTarget(this.applicationLoadBalancer),
  ),
});
```

Finally, an [A Record](https://docs.aws.amazon.com/Route53/latest/DeveloperGuide/ResourceRecordTypes.html#AFormat) will be created that points to the Application Load Balancer alias use the record name `grpc`. This will create a record `grpc.example.com` that can be used with the Application Load Balancer.

## gRPC Server

### Server initialization

```typescript
function main() {
  const server = new grpc.Server();
  server.addService(audio_proto.AudioStreamer.service, { streamAudio });
  server.bindAsync(
    '0.0.0.0:50051',
    grpc.ServerCredentials.createInsecure(),
    () => {
      server.start();
      console.log('gRPC server started on port 50051');
    },
  );
}

main();
```

The main function of the server code, is used to create the server that is listening on port `50051` with HTTP. The Application Load Balancer will provide the TLS security so HTTP can be used here.

### streamAudio Function

```typescript
function streamAudio(call: grpc.ServerWritableStream<any, any>) {
  console.log('Streaming audio data received from client.');

  const audioStream = new PassThrough();

  startTranscription(audioStream, call).catch((error) => {
    console.error('Transcription error:', error);
    call.write({ message: error });
  });

  call.on('data', (data: any) => {
    audioStream.write(data.audioData);
  });

  call.on('end', async () => {
    console.log('Streaming completed.');
    audioStream.end();
  });

  audioStream.on('end', () => {
    console.log('Transcribing ended.');
  });
}
```

Now, when a stream arrives, the `streamAudio` function will begin processing the stream. In this example, it involves three major steps:

1. Create `audioStream` PassThrough Stream
2. Start `startTranscription` function with the `audioStream`
3. Write data to the `audioStream` from the gRPC stream as it arrives

### Transcribe the stream

```typescript
const audioStream = async function* () {
  for await (const chunk of stream) {
    const buffer = Buffer.from(chunk);
    yield { AudioEvent: { AudioChunk: buffer } };
  }
};

const command = new StartStreamTranscriptionCommand({
  LanguageCode: LanguageCode.EN_US,
  MediaEncoding: MediaEncoding.PCM,
  MediaSampleRateHertz: 16000,
  AudioStream: audioStream(),
});

const response = await client.send(command);
```

This will send the stream to Transcribe for processing. Once we get a response back from Transcribe, we will write this back to the gRPC client via the stream.

```typescript
call.write({
  message: JSON.stringify(
    event.TranscriptEvent.Transcript.Results[0].Alternatives[0].Transcript,
  ),
});
```

## gRPC Client

```typescript
async function main() {
  let target: string = 'localhost:50051';
  let credentials = grpc.credentials.createInsecure();

  if (TARGET) {
    target = `${TARGET}:50051`;
    credentials = grpc.credentials.createSsl();
  }

  console.log(`Using target: ${target}`);

  if (!WAV_FILE) {
    throw new Error('WAV_FILE required');
  }

  console.log(`Using source: ${WAV_FILE}`);

  const { audioStream } = await processFfmpeg(WAV_FILE);
}
```

The main client function will determine the `TARGET` and `WAV_FILE` to be used. This client can be used both locally or with the deployed gRPC server. A `WAV_FILE` is required and can be stored in the `client\src` directory and will be copied over during the Docker build. If the `WAV_FILE` is a stereo file with each channel a separate user, they will be muxed together when sent to the gRPC server. The result will be that the transcription is the combined result of both channels. If you want to get a single channel, you can modify the `ffmpeg` to only pass one channel, or separate the `WAV_FILE` before sending.

### ffmpeg Processing

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

  ffmpeg(fileName)
    .native()
    .output(audioStream)
    .format('wav')
    .audioCodec('pcm_s16le')
    .audioBitrate(8000)
    .on('error', (error: { message: string }) => {
      console.log('Cannot process: ' + error.message);
    })
    .run();

  return { audioStream };
}
```

This demo uses `ffmpeg` to ensure the audio being streamed is sent at the correct rate and with the correct formatting.

### Streaming

```typescript
const client = new audio_proto.AudioStreamer(target, credentials);
const chunkSize = 1024;
const call = client.streamAudio();

audioStream.on('data', (chunk) => {
  let offset = 0;
  while (offset < chunk.length) {
    const end = Math.min(offset + chunkSize, chunk.length);
    const chunkToSend = chunk.slice(offset, end);
    call.write({ audioData: chunkToSend });
    offset = end;
  }
});

audioStream.on('end', () => {
  console.log('Audio stream ended.');
  call.end();
});

call.on('data', (response: any) => {
  console.log(`${Date.now()}: ${response.message}`);
});

call.on('end', () => {
  console.log('gRPC call ended.');
  process.exit(0);
});

call.on('error', (err: any) => {
  console.error('Error:', err);
});
```

Once the `audioStream` has been created through `processFfmpeg`, when data arrives, it will be written to the gRPC stream in chunks through the client. Responses that are generated on the server (the transcriptions) will be surfaced through the `call.on('data', () => {})` function.

## In Action

![InActionOverview](/images/blog/node-grpc-server/InActionOverview.png)

In order to test this demo, you can either use it locally or through the deployed Fargate container.

### Deploying

```bash
yarn launch
```

This will deploy the CDK stack to your account. You will need a HostedZone domain in order to use this. The A Record created will be written to the `client\src\.env` file so that you can use the client with this gRPC server.

### Testing

In order to test locally, you will need to start the gRPC server. In the `src\resources\gRPCServer` directory, you can start the gRPC Server Docker container:

```bash
docker-compose run --build grpc-server
```

You will need to have AWS credentials exposed in this container in order to use Transcribe.

In the `client\src` directory, you can start the gRPC Client Docker container:

```bash
docker-compose run --build -e TARGET= grpc-client
```

You can add the the `WAV_FILE` environment variable in the command line as well

```bash
docker-compose run --build -e TARGET= -e WAV_FILE=EXAMPLE.wav grpc-client
```

This will use the localhost as the `TARGET`. If you want to test with the hosted gRPC server and have already deployed it, you can use:

```bash
docker-compose run --build grpc-client
```

### Client Results

![ClientResults](/images/blog/node-grpc-server//ClientResults.png)

### Server Results

![ServerResults](/images/blog/node-grpc-server/ServerResults.png)


This demo will show you how to deploy and use a gRPC server and client to stream audio and use Amazon Transcribe to analyze the audio