Adds a playback node to gcloud_speech_utils (#7)

* Adds a microphone playback node * Use a newer version of record_audio.cc, and adds a playback node to utils
CogRob · Oct 3, 2017 · 823fd3c · 823fd3c
1 parent 31cdffc
commit 823fd3c
Show file tree

Hide file tree

Showing 8 changed files with 146 additions and 76 deletions.
diff --git a/gcloud_speech/src/workspace/.WORKSPACE_GIT_STATUS b/gcloud_speech/src/workspace/.WORKSPACE_GIT_STATUS
@@ -1,4 +1,4 @@
-Tue Oct  3 11:31:45 PDT 2017
+Tue Oct  3 13:42:09 PDT 2017
 CogRob/workspace branch: master
-14068e6c1040ee1d401e8a98487583e36aa015fd
+f9f959353d691b92dbc9b9227b4b4235ba91c738
 Working directory clean.
diff --git a/gcloud_speech/src/workspace/cogrob/cloud/speech/record_audio.cc b/gcloud_speech/src/workspace/cogrob/cloud/speech/record_audio.cc
@@ -49,18 +49,15 @@ namespace speech {
 
 AudioRecorder::AudioRecorder(AudioQueue* output_queue) {
   queue_ = output_queue;
-  thread_.reset(new std::thread([this] {
-      LoopThread();
-  }));
+  StartRecording();
 }
 
 AudioRecorder::~AudioRecorder() {
-  stop_.store(true);
-  thread_->join();
+  StopRecording();
 }
 
-void AudioRecorder::LoopThread() {
-  PaError pa_err;
+void AudioRecorder::StartRecording() {
+  PaError pa_err = paNoError;
 
   // Intialize PortAudio
   pa_err = Pa_Initialize();
@@ -101,49 +98,53 @@ void AudioRecorder::LoopThread() {
     .hostApiSpecificStreamInfo = nullptr
   };
 
-  PaStream *pa_stream;
-  pa_err = Pa_OpenStream(&pa_stream, &input_param, nullptr, SAMPLE_RATE,
-                         SAMPLES_PER_SLICE, paNoFlag, nullptr, nullptr);
+  pa_err = Pa_OpenStream(&pa_stream_, &input_param, nullptr, SAMPLE_RATE,
+                         SAMPLES_PER_SLICE, paNoFlag,
+                         AudioRecorder::PortAudioCallback, this);
 
   if (pa_err != paNoError) {
     LOG(FATAL) << "PortAudio open stream error: " << Pa_GetErrorText(pa_err);
   }
   // Start the stream
-  Pa_StartStream(pa_stream);
+  Pa_StartStream(pa_stream_);
   if (pa_err != paNoError) {
     LOG(FATAL) << "PortAudio start stream error: " << Pa_GetErrorText(pa_err);
   }
 
-  while (!stop_.load()) {
-    std::vector<int16_t> read_buffer;
-    read_buffer.resize(SAMPLES_PER_SLICE);
-
-    // Read from the microphone
-    pa_err = Pa_ReadStream(
-        pa_stream, read_buffer.data(), SAMPLES_PER_SLICE);
-
-    std::unique_ptr<AudioSample> audio_sample(new AudioSample());
-    audio_sample->resize(SAMPLES_PER_SLICE * 2);
-
-    for (size_t i = 0; i < SAMPLES_PER_SLICE; ++i) {
-      (*audio_sample)[i * 2] = read_buffer[i] & 0xFF;
-      (*audio_sample)[i * 2 + 1] = (read_buffer[i] >> 8) & 0xFF;
-    }
-    // Put into the queue
-    queue_->push(std::move(audio_sample));
-  }
+}
 
-  // Clean up
-  Pa_StopStream(pa_stream);
+void AudioRecorder::StopRecording() {
+  PaError pa_err = paNoError;
+  Pa_StopStream(pa_stream_);
   if (pa_err != paNoError) {
     LOG(FATAL) << "PortAudio stop stream error: " << Pa_GetErrorText(pa_err);
   }
-  Pa_CloseStream(pa_stream);
+  Pa_CloseStream(pa_stream_);
   if (pa_err != paNoError) {
     LOG(FATAL) << "PortAudio close stream error: " << Pa_GetErrorText(pa_err);
   }
 }
 
+int AudioRecorder::PortAudioCallback(
+    const void* input, void* output, unsigned long frame_count,
+    const PaStreamCallbackTimeInfo* time_info,
+    PaStreamCallbackFlags status_flags, void* user_data) {
+  AudioRecorder* recorder = static_cast<AudioRecorder*>(user_data);
+
+  LOG_IF(ERROR, frame_count != SAMPLES_PER_SLICE) << "Callback frame_count is "
+      << frame_count << ", which is not " << SAMPLES_PER_SLICE;
+  LOG_IF(ERROR, status_flags) << "Callback status flag is " << status_flags;
+
+  std::unique_ptr<AudioSample> audio_sample(new AudioSample());
+  audio_sample->resize(SAMPLES_PER_SLICE * 2);
+
+  memcpy(audio_sample->data(), input, SAMPLES_PER_SLICE * 2);
+
+  recorder->queue_->push(std::move(audio_sample));
+
+  return paContinue;
+}
+
 }  // namespace speech
 }  // namespace cloud
 }  // namespace cogrob
diff --git a/gcloud_speech/src/workspace/cogrob/cloud/speech/record_audio.h b/gcloud_speech/src/workspace/cogrob/cloud/speech/record_audio.h
@@ -27,6 +27,8 @@
 #ifndef COGROB_CLOUD_SPEECH_RECORD_AUDIO_H_
 #define COGROB_CLOUD_SPEECH_RECORD_AUDIO_H_
 
+#include "portaudio.h"
+
 #include <atomic>
 #include <memory>
 #include <thread>
@@ -44,10 +46,13 @@ class AudioRecorder {
   virtual ~AudioRecorder();
  private:
   AudioQueue* queue_;
-  void LoopThread();
-  std::atomic_bool stop_{false};
-
-  std::unique_ptr<std::thread> thread_;
+  PaStream* pa_stream_ = nullptr;
+  void StartRecording();
+  void StopRecording();
+  static int PortAudioCallback(
+      const void* input, void* output, unsigned long frame_count,
+      const PaStreamCallbackTimeInfo* time_info,
+      PaStreamCallbackFlags status_flags, void* user_data);
 };
 
 }  // namespace speech

diff --git a/gcloud_speech_utils/launch/all_in_one.launch b/gcloud_speech_utils/launch/all_in_one.launch
@@ -16,4 +16,7 @@
   <node name="gcloud_speech_example_client" pkg="gcloud_speech_utils"
         type="example_client.py" respawn="true" output="screen"/>
 
+  <node name="playback_microphone_audio" pkg="gcloud_speech_utils"
+        type="playback_microphone_audio.py" />
+
 </launch>
diff --git a/gcloud_speech_utils/scripts/playback_microphone_audio.py b/gcloud_speech_utils/scripts/playback_microphone_audio.py
@@ -0,0 +1,55 @@
+#!/usr/bin/env python
+import array
+import collections
+import pyaudio
+import Queue
+import time
+
+import rospy
+import gcloud_speech_msgs.msg as gcloud_speech_msgs
+
+LinearPcm16Le16000Audio = gcloud_speech_msgs.LinearPcm16Le16000Audio
+
+FRAME_PER_SLICE=1600
+
+_audio_queue = collections.deque()
+
+
+def RosCallback(msg):
+  _audio_queue.append(msg.data)
+  rospy.loginfo("Received {} bytes.".format(len(msg.data)))
+
+
+def PaCallback(in_data, frame_count, time_info, status):
+  assert frame_count == FRAME_PER_SLICE
+
+  while True:
+    try:
+      data = _audio_queue.popleft()
+      break
+    except:
+      time.sleep(.5)
+
+  return (data, pyaudio.paContinue)
+
+
+def PlaybackMicrophoneAudio():
+    rospy.init_node('playback_microphone_audio', anonymous=True)
+    rospy.Subscriber(
+      "/cogrob/microphone_audio", LinearPcm16Le16000Audio, RosCallback,
+      queue_size=10)
+    pa = pyaudio.PyAudio()
+    stream = pa.open(
+      format=pyaudio.paInt16, channels=1, rate=16000, output=True,
+      stream_callback=PaCallback, frames_per_buffer=FRAME_PER_SLICE)
+
+    stream.start_stream()
+    rospy.spin()
+    pa.terminate()
+
+
+if __name__ == '__main__':
+  try:
+    PlaybackMicrophoneAudio()
+  except rospy.ROSInterruptException:
+    pass
diff --git a/gcloud_speech_utils/src/workspace/.WORKSPACE_GIT_STATUS b/gcloud_speech_utils/src/workspace/.WORKSPACE_GIT_STATUS
@@ -1,4 +1,4 @@
-Tue Oct  3 11:31:51 PDT 2017
+Tue Oct  3 13:41:59 PDT 2017
 CogRob/workspace branch: master
-14068e6c1040ee1d401e8a98487583e36aa015fd
+f9f959353d691b92dbc9b9227b4b4235ba91c738
 Working directory clean.
diff --git a/gcloud_speech_utils/src/workspace/cogrob/cloud/speech/record_audio.cc b/gcloud_speech_utils/src/workspace/cogrob/cloud/speech/record_audio.cc
@@ -49,18 +49,15 @@ namespace speech {
 
 AudioRecorder::AudioRecorder(AudioQueue* output_queue) {
   queue_ = output_queue;
-  thread_.reset(new std::thread([this] {
-      LoopThread();
-  }));
+  StartRecording();
 }
 
 AudioRecorder::~AudioRecorder() {
-  stop_.store(true);
-  thread_->join();
+  StopRecording();
 }
 
-void AudioRecorder::LoopThread() {
-  PaError pa_err;
+void AudioRecorder::StartRecording() {
+  PaError pa_err = paNoError;
 
   // Intialize PortAudio
   pa_err = Pa_Initialize();
@@ -101,49 +98,53 @@ void AudioRecorder::LoopThread() {
     .hostApiSpecificStreamInfo = nullptr
   };
 
-  PaStream *pa_stream;
-  pa_err = Pa_OpenStream(&pa_stream, &input_param, nullptr, SAMPLE_RATE,
-                         SAMPLES_PER_SLICE, paNoFlag, nullptr, nullptr);
+  pa_err = Pa_OpenStream(&pa_stream_, &input_param, nullptr, SAMPLE_RATE,
+                         SAMPLES_PER_SLICE, paNoFlag,
+                         AudioRecorder::PortAudioCallback, this);
 
   if (pa_err != paNoError) {
     LOG(FATAL) << "PortAudio open stream error: " << Pa_GetErrorText(pa_err);
   }
   // Start the stream
-  Pa_StartStream(pa_stream);
+  Pa_StartStream(pa_stream_);
   if (pa_err != paNoError) {
     LOG(FATAL) << "PortAudio start stream error: " << Pa_GetErrorText(pa_err);
   }
 
-  while (!stop_.load()) {
-    std::vector<int16_t> read_buffer;
-    read_buffer.resize(SAMPLES_PER_SLICE);
-
-    // Read from the microphone
-    pa_err = Pa_ReadStream(
-        pa_stream, read_buffer.data(), SAMPLES_PER_SLICE);
-
-    std::unique_ptr<AudioSample> audio_sample(new AudioSample());
-    audio_sample->resize(SAMPLES_PER_SLICE * 2);
-
-    for (size_t i = 0; i < SAMPLES_PER_SLICE; ++i) {
-      (*audio_sample)[i * 2] = read_buffer[i] & 0xFF;
-      (*audio_sample)[i * 2 + 1] = (read_buffer[i] >> 8) & 0xFF;
-    }
-    // Put into the queue
-    queue_->push(std::move(audio_sample));
-  }
+}
 
-  // Clean up
-  Pa_StopStream(pa_stream);
+void AudioRecorder::StopRecording() {
+  PaError pa_err = paNoError;
+  Pa_StopStream(pa_stream_);
   if (pa_err != paNoError) {
     LOG(FATAL) << "PortAudio stop stream error: " << Pa_GetErrorText(pa_err);
   }
-  Pa_CloseStream(pa_stream);
+  Pa_CloseStream(pa_stream_);
   if (pa_err != paNoError) {
     LOG(FATAL) << "PortAudio close stream error: " << Pa_GetErrorText(pa_err);
   }
 }
 
+int AudioRecorder::PortAudioCallback(
+    const void* input, void* output, unsigned long frame_count,
+    const PaStreamCallbackTimeInfo* time_info,
+    PaStreamCallbackFlags status_flags, void* user_data) {
+  AudioRecorder* recorder = static_cast<AudioRecorder*>(user_data);
+
+  LOG_IF(ERROR, frame_count != SAMPLES_PER_SLICE) << "Callback frame_count is "
+      << frame_count << ", which is not " << SAMPLES_PER_SLICE;
+  LOG_IF(ERROR, status_flags) << "Callback status flag is " << status_flags;
+
+  std::unique_ptr<AudioSample> audio_sample(new AudioSample());
+  audio_sample->resize(SAMPLES_PER_SLICE * 2);
+
+  memcpy(audio_sample->data(), input, SAMPLES_PER_SLICE * 2);
+
+  recorder->queue_->push(std::move(audio_sample));
+
+  return paContinue;
+}
+
 }  // namespace speech
 }  // namespace cloud
 }  // namespace cogrob
diff --git a/gcloud_speech_utils/src/workspace/cogrob/cloud/speech/record_audio.h b/gcloud_speech_utils/src/workspace/cogrob/cloud/speech/record_audio.h
@@ -27,6 +27,8 @@
 #ifndef COGROB_CLOUD_SPEECH_RECORD_AUDIO_H_
 #define COGROB_CLOUD_SPEECH_RECORD_AUDIO_H_
 
+#include "portaudio.h"
+
 #include <atomic>
 #include <memory>
 #include <thread>
@@ -44,10 +46,13 @@ class AudioRecorder {
   virtual ~AudioRecorder();
  private:
   AudioQueue* queue_;
-  void LoopThread();
-  std::atomic_bool stop_{false};
-
-  std::unique_ptr<std::thread> thread_;
+  PaStream* pa_stream_ = nullptr;
+  void StartRecording();
+  void StopRecording();
+  static int PortAudioCallback(
+      const void* input, void* output, unsigned long frame_count,
+      const PaStreamCallbackTimeInfo* time_info,
+      PaStreamCallbackFlags status_flags, void* user_data);
 };
 
 }  // namespace speech