How to write voice data stream to RecognizeStream object ?

[sychopath]

My application is getting data stream as an input which I need to send to google STT service. I need to do asynchronous recognition of user input.

I am referring to following link to build my application.
https://github.com/GoogleCloudPlatform/cpp-samples/blob/main/speech/api/streaming_transcribe.cc

Instead of extracting data from file, I am getting data stream as my input source. I am trying to prepare StreamingRecognizeRequest request object as below :

// Writing configuration 
speech::v1::StreamingRecognizeRequest request;
auto& streaming_config = *request.mutable_streaming_config();
google::cloud::speech::v1::RecognitionConfig config;
config.set_language_code("en");
config.set_sample_rate_hertz(8000);
config.set_encoding(RecognitionConfig::LINEAR16);
// Assign config here
*streaming_config.mutable_config() = config;

// Begin a stream.
auto stream =client.AsyncStreamingRecognize(google::cloud::ExperimentalTag{});

I have started another thread where I pass RecognizeStream object by reference :

// mpf_frame_t frame & RecognizeStream& stream
if(frame->codec_frame.size > 0)
{
	request.set_audio_content(frame->codec_frame.buffer, frame->codec_frame.size);
	if (!stream.Write(request, grpc::WriteOptions()).get()) return;
		
}

There error I see on my google dashboard is 400/499. Looks like I am not able to form the request correctly here.

[coryan]

Is there a message accompanying the error? Usually that includes some details about what might be wrong.

You may also be able to use the GOOGLE_CLOUD_CPP_ENABLE_CLOG=yes and GOOGLE_CLOUD_CPP_ENABLE_TRACING=rpc,rpc-streams to get more detailed logs of the requests.

Also, check the documentation for the service and its requests:

https://github.com/googleapis/googleapis/blob/30517fdc372db88146dd4e72b6309cc0b85e2e01/google/cloud/speech/v1/cloud_speech.proto#L114

And https://cloud.google.com/speech-to-text/quotas#content

[sychopath]

Hi @coryan , I did a bit re-organisation of my code. I have created a shared pointer for my stream.

std::shared_ptr<RecognizeStream> pushStream;

I am adding my configuration and adding to my stream as below :

auto recognizer = speech::SpeechClient(speech::MakeSpeechConnection());
resource->pushStream = recognizer.AsyncStreamingRecognize(google::cloud::ExperimentalTag{});  

speech::v1::StreamingRecognizeRequest request;
  auto& streaming_config = *request.mutable_streaming_config();
google::cloud::speech::v1::RecognitionConfig config;
config.set_language_code("en");
 config.set_sample_rate_hertz(8000);
// Assign config here
*streaming_config.mutable_config() = std::move(config);

resource->pushStream->Start();
apt_log(RECOG_LOG_MARK, APT_PRIO_INFO, " helper_ms_start_async_listen() stream started ");
	
if(!resource->pushStream->Write(request, grpc::WriteOptions{}).get())
{
	apt_log(RECOG_LOG_MARK, APT_PRIO_INFO, " helper_ms_start_async_listen() configuration written ");
	throw resource->pushStream->Finish().get();
}

During runtime, I encounter a strange error :

INFO] <140676400469760> Enabled logging for gRPC calls (../google/cloud/speech/internal/speech_stub_factory.cc:53)
2022-09-22T13:53:12.713548636Z [DEBUG] <140676400469760> AsyncStreamingRecognize(1) (../google/cloud/speech/internal/speech_logging_decorator.cc:77)
2022-09-22T13:53:12.713605793Z [DEBUG] <140676400469760> Start(1) << (../google/cloud/internal/async_read_write_stream_logging.h:46)
2022-09-22 13:53:12:714076 [INFO]   [JIO version 30 ----- gs_recog_engine.cpp] helper_ms_start_async_listen() stream started 
2022-09-22T13:53:12.714656093Z [DEBUG] <140676400469760> Write(1) << google.cloud.speech.v1.StreamingRecognizeRequest { streaming_config { config { sample_rate_hertz: 8000 language_code: "en" } } } (../google/cloud/internal/async_read_write_stream_logging.h:73)
E0922 13:53:12.714707139   23327 call_op_set.h:985]          assertion failed: false

In call_op_set.h the comment says

  // A failure here indicates an API misuse; for example, doing a Write
  // while another Write is already pending on the same RPC or invoking
  // WritesDone multiple times

I do not understand what did I do wrong here.

If I comment out my code of writing stream config, my code executes. I am able to write data on my request object and push it to my stream :

speech::v1::StreamingRecognizeRequest request;
request.set_audio_content(frame->codec_frame.buffer, frame->codec_frame.size);
recog_channel->resource->pushStream->Write(request, grpc::WriteOptions());

[coryan]

Start() returns a future, the operation is not completed until the future is satisfied. You cannot call Read() or Write() until the Start() operation completes.

You are missing something like:

if (!resource->pushStream->Start().get()) {
  apt_log(RECOG_LOG_MARK, APT_PRIO_INFO, " helper_ms_start_async_listen() start failed ");
  throw resource->pushStream->Finish().get();
}
apt_log(RECOG_LOG_MARK, APT_PRIO_INFO, " helper_ms_start_async_listen() stream started ");

[sychopath]

Start() returns a future, the operation is not completed until the future is satisfied. You cannot call Read() or Write() until the Start() operation completes.

Thanks @coryan for pointing out this mistake. I am able to write config to my stream object.
However, I end up with a different error message while writing my voice data to stream now. The code is pretty similar, but I get a different error message this time -

speech::v1::StreamingRecognizeRequest request;
request.set_audio_content(frame->codec_frame.buffer, frame->codec_frame.size);
if(!recog_channel->resource->pushStream->Write(request, grpc::WriteOptions{}).get())
{
	throw recog_channel->resource->pushStream->Finish().get();
}

Below is the log I get

[DEBUG] <140682339587840> Write(1) << google.cloud.speech.v1.StreamingRecognizeRequest { streaming_config { config { encoding: LINEAR16 sample_rate_hertz: 8000 language_code: "en" } } } (../google/cloud/internal/async_read_write_stream_logging.h:73)
2022-09-22T18:31:10.732710999Z [DEBUG] <140682089969408> Write(1) >> true (../google/cloud/internal/async_read_write_stream_logging.h:78)
2022-09-22 18:31:10:732898 [INFO]   [version 30 ----- gs_recog_engine.cpp] vt_ms_recog_stream_write() calling helper_ms_start_async_listen  recog_channel->timers_started 0
2022-09-22T18:31:10.732958043Z [DEBUG] <140682339587840> Write(1) << google.cloud.speech.v1.StreamingRecognizeRequest { audio_content: "`\000\010\000\000\000\034\377\330\377\204\000h\000\030\000\220\377\250\377\340\377x\0000\000\\\377\034\3774\376\324\375t\376\234\376\254\376t\376\324\375\024\376\364\375\334\376X\000\340\377\340\377\370\377\264\000l\002\234\003\234\003\234\003\\\004\234\004\334\004\334\003\014\002\224\000<\377\334\376l\377P\000p\000\304\000d\001\254\001d\0014\001\224\000\034\377\214\376\274\376\254\376T\376t\376\350\377\244\0008\000|\377\024\376\264\375\364\375...<truncated>..." } (../google/cloud/internal/async_read_write_stream_logging.h:73)
2022-09-22T18:31:10.733026992Z [DEBUG] <140682339587840> Write(1) >> false (../google/cloud/internal/async_read_write_stream_logging.h:78)
2022-09-22 18:31:10:733058 [INFO]   [version 30 ----- gs_recog_engine.cpp] writing data to google stream threw an error ----- 
2022-09-22T18:31:10.733079360Z [DEBUG] <140682339587840> Finish(1) << (../google/cloud/internal/async_read_write_stream_logging.h:96)
2022-09-22T18:31:10.733143873Z [DEBUG] <140682339587840> Finish(1) >> OK (../google/cloud/internal/async_read_write_stream_logging.h:99)
terminate called after throwing an instance of 'google::cloud::v2_2_1::Status'

The code where I am getting a false as a future bool value is :

  future<bool> Write(Request const& request,
                     grpc::WriteOptions options) override {
    auto prefix = std::string(__func__) + "(" + request_id_ + ")";
    GCP_LOG(DEBUG) << prefix << " << "
                   << DebugString(request, tracing_options_);
    return child_->Write(request, std::move(options))
        .then([prefix](future<bool> f) {
          auto r = f.get();
          GCP_LOG(DEBUG) << prefix << " >> " << (r ? "true" : "false");
          return r;
        });
  }

As per my understanding, my stream is getting closed for some reason. I am quite new to grpc and might be missing some basic points. Thanks for helping me out here !!

[coryan]

Looks like the connection is gracefully terminated by the service. I am not sure why that would happen. Make sure you read the the documentation about how LINEAR16 data is supposed to be represented (2-bytes in little-endian order).

A while ago I was writing (but never finished nor polished) a demo using a microphone as the audio source, maybe that is helpful:

https://github.com/coryan/cpp-samples/blob/feat-infinite-transcription-stream/speech/streaming_transcribe/streaming_transcribe.cc

Beyond that, I think you have exhausted my limited knowledge of this API. I am completely out of ideas 😞

You may may need to reach out to other forums, maybe stack overflow, the GCP community slack:

https://join.slack.com/t/googlecloud-community/shared_invite/zt-1czezyjlf-e33_CD2tWECRY5pecjVpTA

The GCP forums:

https://www.googlecloudcommunity.com/gc/forums/filteredbylabelpage/board-id/cloud-ai-ml/label-name/Speech-to-Text

Or customer support:

https://cloud.google.com/support

[sychopath]

I have figured out the problem was at my end. It was my connection object going out of scope resulting in closing of connection.

You may may need to reach out to other forums, maybe stack overflow, the GCP community slack:

Thanks for sharing these links.

Beyond that, I think you have exhausted my limited knowledge of this API. I am completely out of ideas

Sorry , you had to go after weird issues because of mistakes from my end. Your help is really appreciated. Thanks a lot for your responses.

https://github.com/coryan/cpp-samples/blob/feat-infinite-transcription-stream/speech/streaming_transcribe/streaming_transcribe.cc

I am already aware about this repository. I am using them to structure my application. Thanks for sharing !!