fixed different voice quality while using playht (#111)

* initial support for mark feature in bidirectional streaming

Signed-off-by: Dave Horton <daveh@beachdognet.com>

* wip

* allow max of 30 marks on any connection

* fix send multiple json in same ws text frame (#108)

* fix send multiple json in same ws text frame

Signed-off-by: Hoan HL <quan.luuhoang8@gmail.com>

* fix mark is not sent without more bidirectional audio

Signed-off-by: Hoan HL <quan.luuhoang8@gmail.com>

---------

Signed-off-by: Hoan HL <quan.luuhoang8@gmail.com>

---------

Signed-off-by: Dave Horton <daveh@beachdognet.com>
Signed-off-by: Hoan HL <quan.luuhoang8@gmail.com>
Co-authored-by: Hoan Luu Huu <110280845+xquanluu@users.noreply.github.com>

mod_audio_fork/audio_pipe.cpp

@@ -227,20 +227,21 @@ int AudioPipe::lws_callback(struct lws *wsi,
         // check for text frames to send
         {
           std::lock_guard<std::mutex> lk(ap->m_text_mutex);
-          if (ap->m_metadata.length() > 0) {
-            uint8_t buf[ap->m_metadata.length() + LWS_PRE];
-            memcpy(buf + LWS_PRE, ap->m_metadata.c_str(), ap->m_metadata.length());
-            int n = ap->m_metadata.length();
+          if (!ap->m_metadata_list.empty()) {
+            const std::string& message = ap->m_metadata_list.front();
+            uint8_t buf[message.length() + LWS_PRE];
+            memcpy(buf + LWS_PRE, message.c_str(), message.length());
+            int n = message.length();
             int m = lws_write(wsi, buf + LWS_PRE, n, LWS_WRITE_TEXT);
-            ap->m_metadata.clear();
+
             if (m < n) {
-              return -1;
+              return -1; // Failed to send the full message
             }
 
-            // there may be audio data, but only one write per writeable event
-            // get it next time
+            // Remove the message that was successfully sent
+            ap->m_metadata_list.pop_front();
+            // Request another writable event if there are more messages
             lws_callback_on_writable(wsi);
-
             return 0;
           }
         }
@@ -529,7 +530,7 @@ void AudioPipe::bufferForSending(const char* text) {
   if (m_state != LWS_CLIENT_CONNECTED) return;
   {
     std::lock_guard<std::mutex> lk(m_text_mutex);
-    m_metadata.append(text);
+    m_metadata_list.emplace_back(text);
   }
   addPendingWrite(this);
 }

mod_audio_fork/audio_pipe.hpp

@@ -130,7 +130,7 @@ namespace drachtio {
     std::string m_bugname;
     unsigned int m_port;
     std::string m_path;
-    std::string m_metadata;
+    std::list<std::string> m_metadata_list;
     std::mutex m_text_mutex;
     std::mutex m_audio_mutex;
     int m_sslFlags;

mod_audio_fork/lws_glue.cpp

@@ -27,6 +27,8 @@ typedef boost::circular_buffer<uint16_t> CircularBuffer_t;
 #define RTP_PACKETIZATION_PERIOD 20
 #define FRAME_SIZE_8000  320 /*which means each 20ms frame as 320 bytes at 8 khz (1 channel only)*/
 #define BUFFER_GROW_SIZE (16384)
+#define AUDIO_MARKER 0xFFFF
+#define MAX_MARKS (30)
 
 namespace {
   static const char *requestedBufferSecs = std::getenv("MOD_AUDIO_FORK_BUFFER_SECS");
@@ -38,6 +40,15 @@ namespace {
   static unsigned int idxCallCount = 0;
   static uint32_t playCount = 0;
 
+  static bool markCountExceeded(private_t* tech_pvt) {
+    if (nullptr != tech_pvt->pVecMarksInUse) {
+      std::deque<std::string>* pVecMarksInUse = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInUse);
+      std::deque<std::string>* pVecMarksInInventory = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInInventory);
+      return pVecMarksInUse->size()+ pVecMarksInInventory->size() >= MAX_MARKS;
+    }
+    return false;
+  }
+
   switch_status_t processIncomingBinary(private_t* tech_pvt, switch_core_session_t* session, const char* message, size_t dataLength) {
     std::vector<uint8_t> data;
 
@@ -65,12 +76,30 @@ namespace {
     // Access the prebuffer
     CircularBuffer_t* cBuffer = static_cast<CircularBuffer_t*>(tech_pvt->streamingPreBuffer);
 
+    int numMarkers = 0;
+    std::deque<std::string>* pVecMarksInInventory = nullptr;
+    std::deque<std::string>* pVecMarksInUse = nullptr;
+    if (nullptr != tech_pvt->pVecMarksInInventory) {
+      pVecMarksInInventory = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInInventory);
+      pVecMarksInUse = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInUse);
+      numMarkers = pVecMarksInInventory->size();
+
+      // move inventory to in-use
+      pVecMarksInUse->insert(pVecMarksInUse->end(), pVecMarksInInventory->begin(), pVecMarksInInventory->end());
+      pVecMarksInInventory->clear();
+    }
+  
     // Ensure the prebuffer has enough capacity
-    if (cBuffer->capacity() - cBuffer->size() < numSamples) {
-        size_t newCapacity = cBuffer->size() + std::max(numSamples, (size_t)BUFFER_GROW_SIZE);
+    if (cBuffer->capacity() - cBuffer->size() < numSamples + numMarkers) {
+        size_t newCapacity = cBuffer->size() + std::max(numSamples + numMarkers, (size_t)BUFFER_GROW_SIZE);
         cBuffer->set_capacity(newCapacity);
     }
 
+    // prepend any markers
+    while (numMarkers-- > 0) {
+      cBuffer->push_back(AUDIO_MARKER);
+    }
+
     // Append the data to the prebuffer
     cBuffer->insert(cBuffer->end(), data_uint16, data_uint16 + numSamples);
     //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Appended %zu 16-bit samples to the prebuffer.\n", numSamples);
@@ -269,6 +298,63 @@ namespace {
         // this will dump buffered incoming audio
         tech_pvt->clear_bidirectional_audio_buffer = true;
       }
+      else if (0 == type.compare("mark")) {
+        cJSON* data = cJSON_GetObjectItem(json, "data");
+        if (data) {
+          cJSON* name = cJSON_GetObjectItem(data, "name");
+          if (cJSON_IsString(name) && name->valuestring) {
+            if (markCountExceeded(tech_pvt)) {
+              switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "(%u) processIncomingMessage - mark count exceeded, discarding mark %s\n", tech_pvt->id, cJSON_GetStringValue(name));
+            }
+            else {
+              if (nullptr == tech_pvt->pVecMarksInInventory) {
+                tech_pvt->pVecMarksInInventory = static_cast<void *>(new std::deque<std::string>());
+                tech_pvt->pVecMarksInUse = static_cast<void *>(new std::deque<std::string>());
+                tech_pvt->pVecMarksCleared = static_cast<void *>(new std::deque<std::string>());
+              }
+              std::deque<std::string>* pVec = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInInventory);
+              pVec->push_back(name->valuestring);
+            }
+          }
+        }
+      }
+      else if (0 == type.compare("clearMarks")) {
+        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) processIncomingMessage - received clearMarks\n", tech_pvt->id);
+        if (nullptr != tech_pvt->pVecMarksInInventory) {
+          std::deque<std::string>* pVecMarksInInventory = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInInventory);
+          std::deque<std::string>* pVecMarksInUse = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInUse);
+          std::deque<std::string>* pVecMarksCleared = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksCleared);
+          pVecMarksCleared->insert(pVecMarksCleared->end(), pVecMarksInUse->begin(), pVecMarksInUse->end());
+          pVecMarksCleared->insert(pVecMarksCleared->end(), pVecMarksInInventory->begin(), pVecMarksInInventory->end());
+          pVecMarksInInventory->clear();
+          pVecMarksInUse->clear();
+        }
+      }
+      else if (0 == type.compare("transcription")) {
+        char* jsonString = cJSON_PrintUnformatted(jsonData);
+        tech_pvt->responseHandler(session, EVENT_TRANSCRIPTION, jsonString);
+        free(jsonString);        
+      }
+      else if (0 == type.compare("transfer")) {
+        char* jsonString = cJSON_PrintUnformatted(jsonData);
+        tech_pvt->responseHandler(session, EVENT_TRANSFER, jsonString);
+        free(jsonString);                
+      }
+      else if (0 == type.compare("disconnect")) {
+        char* jsonString = cJSON_PrintUnformatted(jsonData);
+        tech_pvt->responseHandler(session, EVENT_DISCONNECT, jsonString);
+        free(jsonString);        
+      }
+      else if (0 == type.compare("error")) {
+        char* jsonString = cJSON_PrintUnformatted(jsonData);
+        tech_pvt->responseHandler(session, EVENT_ERROR, jsonString);
+        free(jsonString);        
+      }
+      else if (0 == type.compare("json")) {
+        char* jsonString = cJSON_PrintUnformatted(json);
+        tech_pvt->responseHandler(session, EVENT_JSON, jsonString);
+        free(jsonString);
+      }
       else if (0 == type.compare("transcription")) {
         char* jsonString = cJSON_PrintUnformatted(jsonData);
         tech_pvt->responseHandler(session, EVENT_TRANSCRIPTION, jsonString);
@@ -400,6 +486,9 @@ namespace {
     }
     tech_pvt->streamingPreBufSize = 320 * tech_pvt->downscale_factor * 4; // min 80ms prebuffer
     tech_pvt->streamingPreBuffer = (void *) new CircularBuffer_t(8192);
+    tech_pvt->pVecMarksInInventory = nullptr;
+    tech_pvt->pVecMarksInUse = nullptr;
+    tech_pvt->pVecMarksCleared = nullptr;
 
     strncpy(tech_pvt->bugname, bugname, MAX_BUG_LEN);
     if (metadata) strncpy(tech_pvt->initialMetadata, metadata, MAX_METADATA_LEN);
@@ -467,6 +556,29 @@ namespace {
       delete cBuffer;
       tech_pvt->streamingPreBuffer = nullptr;
     }
+
+    if (nullptr == tech_pvt->pVecMarksInInventory) {
+      delete static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInInventory);
+      tech_pvt->pVecMarksInInventory = nullptr;
+      delete static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInUse);
+      tech_pvt->pVecMarksInUse = nullptr;
+      delete static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksCleared);
+      tech_pvt->pVecMarksCleared = nullptr;
+    }
+  }
+
+  static void send_mark_event(private_t* tech_pvt, const char* name, int cleared = false) {
+    drachtio::AudioPipe *pAudioPipe = static_cast<drachtio::AudioPipe *>(tech_pvt->pAudioPipe);
+    std::ostringstream json;
+    json << "{\"type\": \"mark\", \"data\": {\"name\":\"" << name << "\", ";
+    if (cleared) json << "\"event\": \"cleared\"}}";
+    else json << "\"event\": \"playout\"}}";
+
+    if (pAudioPipe) {
+      std::string str = json.str();
+      pAudioPipe->bufferForSending(str.c_str());
+      switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) send_mark_event: %s\n", tech_pvt->id, str.c_str());
+    }
   }
 
   void lws_logger(int level, const char *line) {
@@ -826,6 +938,28 @@ extern "C" {
         switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) dub_speech_frame - clearing buffer\n", tech_pvt->id); 
         cBuffer->clear();
         tech_pvt->clear_bidirectional_audio_buffer = false;
+
+        // send "mark" event for any queued markers
+        if (nullptr != tech_pvt->pVecMarksInInventory) {
+          std::deque<std::string>* pVecMarksInInventory = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInInventory);
+          std::deque<std::string>* pVecMarksInUse = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInUse);
+          if (pVecMarksInInventory->size() + pVecMarksInUse->size() > 0) {
+            std::deque<std::string> vec = *pVecMarksInUse;
+            vec.insert(vec.end(), pVecMarksInInventory->begin(), pVecMarksInInventory->end());
+            for (auto it = vec.begin(); it != vec.end(); ++it) {
+              switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "(%u) dub_speech_frame - Marker %s cleared\n",
+                  tech_pvt->id, it->c_str());
+              send_mark_event(tech_pvt, it->c_str(), true);
+            }
+
+            // put the "in-use" ones into the "cleared" queue so we dont notify again when they eventually come through
+            std::deque<std::string>* pVecMarksCleared = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksCleared);
+            pVecMarksCleared->insert(pVecMarksCleared->end(), pVecMarksInUse->begin(), pVecMarksInUse->end());
+
+            pVecMarksInUse->clear();
+            pVecMarksInInventory->clear();
+          }
+        }
       }
       else {
         switch_frame_t* rframe = switch_core_media_bug_get_write_replace_frame(bug);
@@ -841,11 +975,72 @@ extern "C" {
 
         //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) dub_speech_frame - samples to copy %u\n", tech_pvt->id, samplesToCopy); 
 
-        std::copy_n(cBuffer->begin(), samplesToCopy, data);
-        cBuffer->erase(cBuffer->begin(), cBuffer->begin() + samplesToCopy);
+        bool hasMarkers = false;
+        std::deque<std::string>* pVecInventory = nullptr;
+        std::deque<std::string>* pVecInUse = nullptr;
+        std::deque<std::string>* pVecCleared = nullptr;
+        if (nullptr != tech_pvt->pVecMarksInUse) {
+          pVecInventory = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInInventory);
+          pVecInUse = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksInUse);
+          pVecCleared = static_cast<std::deque<std::string>*>(tech_pvt->pVecMarksCleared);
+          hasMarkers = pVecInUse->size() + pVecCleared->size() >  0;
+        }
 
-        if (samplesToCopy > 0) {
-          vector_add(fp, data, rframe->samples);
+        if (hasMarkers) {
+          /* discard markers and send notifications */
+          auto bufferIter = cBuffer->begin();
+          auto dataIter = data;
+          for (int i = 0; i < samplesToCopy; ++i) {
+            if (*bufferIter == AUDIO_MARKER) {
+              // Marker detected, discard it and send a notice unless it was previously cleared
+              auto * pVec = pVecCleared->size() > 0 ? pVecCleared : pVecInUse;
+              if (!pVec->empty()) {
+                auto name = pVec->front();
+                pVec->pop_front();
+
+                if (pVec == pVecInUse) {
+                  send_mark_event(tech_pvt, name.c_str());
+                  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) dub_speech_frame - Marker %s detected in playout\n",
+                    tech_pvt->id, name.c_str());
+                }
+                else {
+                  switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) dub_speech_frame - Marker %s detected in playout but previously cleared\n",
+                    tech_pvt->id, name.c_str());
+                }
+              }
+            } else {
+              // Copy valid audio samplewhat 
+              *dataIter = *bufferIter;
+              ++dataIter;
+            }
+            ++bufferIter;
+          }
+
+          // Remove the processed samples (including discarded markers) from the buffer
+          cBuffer->erase(cBuffer->begin(), cBuffer->begin() + samplesToCopy);
+
+          // Adjust the number of samples copied to the output frame
+          int validSamplesCopied = std::distance(data, dataIter);
+
+          if (validSamplesCopied > 0) {
+            vector_add(fp, data, validSamplesCopied);
+          }
+        }
+        else {
+          std::copy_n(cBuffer->begin(), samplesToCopy, data);
+          cBuffer->erase(cBuffer->begin(), cBuffer->begin() + samplesToCopy);
+
+          if (samplesToCopy > 0) {
+            vector_add(fp, data, rframe->samples);
+          } else if (pVecInventory != nullptr && pVecInventory->size()) {
+            // no bidirectional audio to dub but still have some mark in inventory, send them now
+            auto name = pVecInventory->front();
+            pVecInventory->pop_front();
+
+            send_mark_event(tech_pvt, name.c_str());
+            switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) dub_speech_frame - Marker %s detected in inventory\n",
+              tech_pvt->id, name.c_str());
+          }
         }
         vector_normalize(fp, rframe->samples);
 

mod_audio_fork/mod_audio_fork.h

@@ -53,6 +53,11 @@ struct private_data {
   int graceful_shutdown:1;
   char initialMetadata[8192];
 
+  // for "mark" feature of bidirectional audio
+  void *pVecMarksInInventory;
+  void *pVecMarksInUse;
+  void *pVecMarksCleared;
+
   // bidirectional audio
   void *streamingPlayoutBuffer;
   void *streamingPreBuffer;

mod_aws_transcribe/aws_transcribe_glue.cpp

@@ -43,16 +43,16 @@ public:
     const char *sessionId,
 		const char *bugname,
 		u_int16_t channels,
-    char *lang, 
+    char *lang,
     int interim,
 		uint32_t samples_per_second,
-		const char* region, 
-		const char* awsAccessKeyId, 
+		const char* region,
+		const char* awsAccessKeyId,
 		const char* awsSecretAccessKey,
 		const char* awsSessionToken,
 		responseHandler_t responseHandler
   ) : m_sessionId(sessionId), m_bugname(bugname), m_finished(false), m_interim(interim), m_finishing(false), m_connected(false), m_connecting(false),
-	 		m_packets(0), m_responseHandler(responseHandler), m_pStream(nullptr), 
+	 		m_packets(0), m_responseHandler(responseHandler), m_pStream(nullptr),
 			m_audioBuffer(320 * (samples_per_second == 8000 ? 1 : 2), 15) {
 		Aws::Client::ClientConfiguration config;
 		if (region != nullptr && strlen(region) > 0) config.region = region;
@@ -71,7 +71,7 @@ public:
 		else {
 			m_client = Aws::MakeUnique<TranscribeStreamingServiceClient>(ALLOC_TAG, config);
 		}
-	
+
     m_handler.SetTranscriptEventCallback([this](const TranscriptEvent& ev)
     {
 			switch_core_session_t* psession = switch_core_session_locate(m_sessionId.c_str());
@@ -132,7 +132,7 @@ public:
 
 				// send any buffered audio
 				int nFrames = m_audioBuffer.getNumItems();
-				switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "GStreamer %p got stream ready, %d buffered frames\n", this, nFrames);	
+				switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "GStreamer %p got stream ready, %d buffered frames\n", this, nFrames);
 				if (nFrames) {
 					char *p;
 					do {
@@ -142,19 +142,19 @@ public:
 						}
 					} while (p);
 				}
-	
+
 				switch_core_session_rwunlock(psession);
 			}
     };
-    auto OnResponseCallback = [this](const TranscribeStreamingServiceClient* pClient, 
-			const Model::StartStreamTranscriptionRequest& request, 
-			const Model::StartStreamTranscriptionOutcome& outcome, 
+    auto OnResponseCallback = [this](const TranscribeStreamingServiceClient* pClient,
+			const Model::StartStreamTranscriptionRequest& request,
+			const Model::StartStreamTranscriptionOutcome& outcome,
 			const std::shared_ptr<const Aws::Client::AsyncCallerContext>& context)
     {
  			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "GStreamer %p stream got final response\n", this);
 			switch_core_session_t* psession = switch_core_session_locate(m_sessionId.c_str());
 			if (psession) {
-				if (!outcome.IsSuccess()) {				
+				if (!outcome.IsSuccess()) {
 					const TranscribeStreamingServiceError& err = outcome.GetError();
 					auto message = err.GetMessage();
 					auto exception = err.GetExceptionName();
@@ -186,7 +186,7 @@ public:
 
 
 	~GStreamer() {
-		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "GStreamer::~GStreamer wrote %u packets %p\n", m_packets, this);		
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "GStreamer::~GStreamer wrote %u packets %p\n", m_packets, this);
 	}
 
 	bool write(void* data, uint32_t datalen) {
@@ -228,7 +228,7 @@ public:
 		bool shutdownInitiated = false;
 		while (true) {
 			std::unique_lock<std::mutex> lk(m_mutex);
-			m_cond.wait(lk, [&, this] { 
+			m_cond.wait(lk, [&, this] {
 				return (!m_deqAudio.empty() && !m_finishing)  || m_transcript.TranscriptHasBeenSet() || m_finished  || (m_finishing && !shutdownInitiated);
 			});
 
@@ -264,7 +264,7 @@ public:
 						m_responseHandler(psession, s.str().c_str(), m_bugname.c_str());
 					}
 					TranscriptEvent empty;
-					m_transcript = empty; 
+					m_transcript = empty;
 
 					switch_core_session_rwunlock(psession);
 				}
@@ -362,7 +362,7 @@ extern "C" {
 		const char* secretAccessKey = std::getenv("AWS_SECRET_ACCESS_KEY");
 		const char* region = std::getenv("AWS_REGION");
 		if (NULL == accessKeyId && NULL == secretAccessKey) {
-			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, 
+			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE,
 				"\"AWS_ACCESS_KEY_ID\"  and/or \"AWS_SECRET_ACCESS_KEY\" env var not set; authentication will expect channel variables of same names to be set\n");
 		}
 		else {
@@ -370,7 +370,7 @@ extern "C" {
 
 		}
     Aws::SDKOptions options;
-/*		
+/*
     options.loggingOptions.logLevel = Aws::Utils::Logging::LogLevel::Trace;
 
 		Aws::Utils::Logging::InitializeAWSLogging(
@@ -381,7 +381,7 @@ extern "C" {
 
 		return SWITCH_STATUS_SUCCESS;
 	}
-	
+
 	switch_status_t aws_transcribe_cleanup() {
 		Aws::SDKOptions options;
 		/*
@@ -394,7 +394,7 @@ extern "C" {
 	}
 
 	// start transcribe on a channel
-	switch_status_t aws_transcribe_session_init(switch_core_session_t *session, responseHandler_t responseHandler, 
+	switch_status_t aws_transcribe_session_init(switch_core_session_t *session, responseHandler_t responseHandler,
           uint32_t samples_per_second, uint32_t channels, char* lang, int interim, char* bugname, void **ppUserData
 	) {
 		switch_status_t status = SWITCH_STATUS_SUCCESS;
@@ -433,7 +433,7 @@ extern "C" {
 			std::getenv("AWS_REGION")) {
 			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Using env vars for aws authentication\n");
 			strncpy(cb->awsAccessKeyId, std::getenv("AWS_ACCESS_KEY_ID"), 128);
-			strncpy(cb->awsSecretAccessKey, std::getenv("AWS_SECRET_ACCESS_KEY"), 128);		
+			strncpy(cb->awsSecretAccessKey, std::getenv("AWS_SECRET_ACCESS_KEY"), 128);
 			strncpy(cb->region, std::getenv("AWS_REGION"), MAX_REGION);
 		}
 		else {
@@ -445,7 +445,7 @@ extern "C" {
 		if (switch_mutex_init(&cb->mutex, SWITCH_MUTEX_NESTED, pool) != SWITCH_STATUS_SUCCESS) {
 			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Error initializing mutex\n");
 			status = SWITCH_STATUS_FALSE;
-			goto done; 
+			goto done;
 		}
 
 		cb->interim = interim;
@@ -455,7 +455,7 @@ extern "C" {
 		if (sampleRate != 8000) {
 			cb->resampler = speex_resampler_init(1, sampleRate, 16000, SWITCH_RESAMPLE_QUALITY, &err);
 			if (0 != err) {
-				switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "%s: Error initializing resampler: %s.\n", 
+				switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "%s: Error initializing resampler: %s.\n",
 							switch_channel_get_name(channel), speex_resampler_strerror(err));
 				status = SWITCH_STATUS_FALSE;
 				goto done;
@@ -488,7 +488,7 @@ extern "C" {
 				switch_vad_set_param(cb->vad, "silence_ms", silence_ms);
 				switch_vad_set_param(cb->vad, "voice_ms", voice_ms);
 				switch_vad_set_param(cb->vad, "debug", debug);
-				switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "%s: delaying connection until vad, voice_ms %d, mode %d\n", 
+				switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "%s: delaying connection until vad, voice_ms %d, mode %d\n",
 					switch_channel_get_name(channel), voice_ms, mode);
 			}
 		}
@@ -499,7 +499,7 @@ extern "C" {
 		switch_thread_create(&cb->thread, thd_attr, aws_transcribe_thread, cb, pool);
 
 		*ppUserData = cb;
-	
+
 	done:
 		return status;
 	}
@@ -521,7 +521,7 @@ extern "C" {
       do {
         streamer = (GStreamer *) cb->streamer;
         if (streamer) break;
-        switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, 
+        switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG,
           "aws_transcribe_session_stop: waiting for streamer to come online..%s\n", bugname);
         switch_yield(10000); // wait 10ms
       } while (i++ < 3);
@@ -557,7 +557,7 @@ extern "C" {
 		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "%s Bug is not attached.\n", switch_channel_get_name(channel));
 		return SWITCH_STATUS_FALSE;
 	}
-	
+
 	switch_bool_t aws_transcribe_frame(switch_media_bug_t *bug, void* user_data) {
 		switch_core_session_t *session = switch_core_media_bug_get_session(bug);
 		uint8_t data[SWITCH_RECOMMENDED_BUFFER_SIZE];
@@ -587,7 +587,7 @@ extern "C" {
 						}
 
 						if (cb->resampler) {
-							speex_resampler_process_interleaved_int(cb->resampler, (const spx_int16_t *) frame.data, (spx_uint32_t *) &in_len, &out[0], &out_len);						
+							speex_resampler_process_interleaved_int(cb->resampler, (const spx_int16_t *) frame.data, (spx_uint32_t *) &in_len, &out[0], &out_len);
 							streamer->write( &out[0], sizeof(spx_int16_t) * out_len);
 						}
 						else {
@@ -597,11 +597,11 @@ extern "C" {
 				}
 			}
 			else {
-				switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, 
+				switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG,
 					"aws_transcribe_frame: not sending audio because aws channel has been closed\n");
 			}
 			switch_mutex_unlock(cb->mutex);
 		}
 		return SWITCH_TRUE;
 	}
-}
+}

mod_custom_tts/custom_glue.cpp

@@ -451,12 +451,12 @@ static size_t write_cb(void *ptr, size_t size, size_t nmemb, ConnInfo_t *conn) {
       return 0;
     }
 
-    /* cache file will stay in the mp3 format for size (smaller) and simplicity */
-    if (conn->file) fwrite(data, sizeof(uint8_t), bytes_received, conn->file);
-
     pcm_data = convert_mp3_to_linear(conn, data, bytes_received);
     size_t bytesResampled = pcm_data.size() * sizeof(uint16_t);
 
+    /* cache same data to avoid streaming and cached audio quality is different*/
+    if (conn->file) fwrite(pcm_data.data(), sizeof(uint8_t), bytesResampled, conn->file);
+
     // Resize the buffer if necessary
     if (cBuffer->capacity() - cBuffer->size() < (bytesResampled / sizeof(uint16_t))) {
       //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "write_cb growing buffer\n"); 

mod_deepgram_transcribe/dg_transcribe_glue.cpp

@@ -172,7 +172,7 @@ namespace {
     else if (customModel) oss << "&model=" << customModel;
 
     if (var = switch_channel_get_variable(channel, "DEEPGRAM_SPEECH_MODEL_VERSION")) {
-     oss <<  "&version";
+     oss <<  "&version=";
      oss <<  var;
     }
     oss <<  "&language=";

mod_deepgram_tts/deepgram_glue.cpp

@@ -742,7 +742,7 @@ extern "C" {
     std::string url;
     std::ostringstream url_stream;
     // always use sample_rate=8000 for support jambonz caching system.
-    url_stream << "https://api.deepgram.com/v1/speak?model=" << d->voice_name << "&encoding=linear16&sample_rate=8000";
+    url_stream << (d->endpoint != nullptr ? d->endpoint : "https://api.deepgram.com") << "/v1/speak?model=" << d->voice_name << "&encoding=linear16&sample_rate=8000";
     url = url_stream.str();
 
     /* create the JSON body */

mod_deepgram_tts/mod_deepgram_tts.c

@@ -19,10 +19,12 @@ static void cleardeepgram(deepgram_t* d, int freeAll) {
   if (d->connect_time_ms) free(d->connect_time_ms);
   if (d->final_response_time_ms) free(d->final_response_time_ms);
   if (d->cache_filename) free(d->cache_filename);
+  if (d->endpoint) free(d->endpoint);
   
 
   d->api_key = NULL;
   d->request_id = NULL;
+  d->endpoint = NULL;
 
   d->reported_model_name = NULL;
   d->reported_model_uuid = NULL;
@@ -121,6 +123,9 @@ static void d_text_param_tts(switch_speech_handle_t *sh, char *param, const char
   if (0 == strcmp(param, "api_key")) {
     if (d->api_key) free(d->api_key);
     d->api_key = strdup(val);
+  } else if (0 == strcmp(param, "endpoint")) {
+    if (d->endpoint) free(d->endpoint);
+    d->endpoint = strdup(val);
   } else if (0 == strcmp(param, "voice")) {
     if (d->voice_name) free(d->voice_name);
     d->voice_name = strdup(val);

mod_deepgram_tts/mod_deepgram_tts.h

@@ -7,6 +7,7 @@
 typedef struct deepgram_data {
   char *voice_name;
   char *api_key;
+  char *endpoint;
 
   /* result data */
   long response_code;

mod_dub/tts_vendor_parser.cpp

@@ -378,7 +378,10 @@ switch_status_t playht_parse_text(const std::map<std::string, std::string>& para
     cJSON_AddStringToObject(jResult, "quality", quality.c_str());
   }
   if (!speed.empty()) {
-    cJSON_AddNumberToObject(jResult, "speed", atoi(speed.c_str()));
+      double val = strtod(speed.c_str(), NULL);
+      if (val != 0.0) {
+        cJSON_AddNumberToObject(jResult, "speed", val);
+      }
   }
   if (!seed.empty()) {
     cJSON_AddNumberToObject(jResult, "seed", atoi(seed.c_str()));

mod_jambonz_transcribe/audio_pipe.cpp

@@ -229,16 +229,6 @@ int AudioPipe::lws_callback(struct lws *wsi,
 }
 
 
-// static members
-static const lws_retry_bo_t retry = {
-    nullptr,   // retry_ms_table
-    0,         // retry_ms_table_count
-    0,         // conceal_count
-    UINT16_MAX,         // secs_since_valid_ping
-    UINT16_MAX,        // secs_since_valid_hangup
-    0          // jitter_percent
-};
-
 struct lws_context *AudioPipe::context = nullptr;
 std::thread AudioPipe::serviceThread;
 std::mutex AudioPipe::mutex_connects;
@@ -376,6 +366,28 @@ bool AudioPipe::lws_service_thread() {
     { NULL, NULL, 0, 0 }
   };
 
+  uint16_t secs_sinceq_valid_ping = UINT16_MAX;
+  uint16_t secs_since_valid_hangup = UINT16_MAX;
+
+  char* wsVar = std::getenv("WS_PING_INTERVAL");
+  if (wsVar != nullptr) {
+    secs_sinceq_valid_ping = std::atoi(wsVar);
+  }
+
+  wsVar = std::getenv("WS_NO_PONG_HANGUP_INTERVAL");
+  if (wsVar != nullptr) {
+    secs_since_valid_hangup = std::atoi(wsVar);
+  }
+
+  const lws_retry_bo_t retry = {
+    nullptr,                  // retry_ms_table
+    0,                        // retry_ms_table_count
+    0,                        // conceal_count
+    secs_sinceq_valid_ping,   // secs_sinceq_valid_ping
+    secs_since_valid_hangup,  // secs_since_valid_hangup
+    0                         // jitter_percent
+  };
+
   memset(&info, 0, sizeof info); 
   info.port = CONTEXT_PORT_NO_LISTEN; 
   info.options = LWS_SERVER_OPTION_DO_SSL_GLOBAL_INIT;

mod_jambonz_transcribe/jb_transcribe_glue.cpp

@@ -145,7 +145,7 @@ namespace {
     cJSON_AddStringToObject(json, "format", "raw");
     cJSON_AddStringToObject(json, "encoding", "LINEAR16");
     cJSON_AddBoolToObject(json, "interimResults", tech_pvt->interim);
-    cJSON_AddNumberToObject(json, "sampleRateHz", 8000);
+    cJSON_AddNumberToObject(json, "sampleRateHz", tech_pvt->sampling);
     if (var = switch_channel_get_variable(channel, "JAMBONZ_STT_OPTIONS")) {
       cJSON* jOptions = cJSON_Parse(var);
       if (jOptions)  {
@@ -353,7 +353,7 @@ extern "C" {
   }
 	
   switch_status_t jb_transcribe_session_init(switch_core_session_t *session, 
-    responseHandler_t responseHandler, uint32_t samples_per_second, uint32_t channels, 
+    responseHandler_t responseHandler, uint32_t samples_per_second, int desiredSampling, uint32_t channels, 
     char* lang, int interim, char* bugname, void **ppUserData)
   {    	
     int err;
@@ -365,7 +365,7 @@ extern "C" {
       return SWITCH_STATUS_FALSE;
     }
 
-    if (SWITCH_STATUS_SUCCESS != fork_data_init(tech_pvt, session, samples_per_second, 8000, channels, lang, interim, bugname, responseHandler)) {
+    if (SWITCH_STATUS_SUCCESS != fork_data_init(tech_pvt, session, samples_per_second, desiredSampling, channels, lang, interim, bugname, responseHandler)) {
       destroy_tech_pvt(tech_pvt);
       return SWITCH_STATUS_FALSE;
     }

mod_jambonz_transcribe/jb_transcribe_glue.h

@@ -6,7 +6,7 @@ int parse_ws_uri(switch_channel_t *channel, const char* szServerUri, char* host,
 switch_status_t jb_transcribe_init();
 switch_status_t jb_transcribe_cleanup();
 switch_status_t jb_transcribe_session_init(switch_core_session_t *session, responseHandler_t responseHandler, 
-		uint32_t samples_per_second, uint32_t channels, char* lang, int interim, char* bugname, void **ppUserData);
+		uint32_t samples_per_second, int desiredSampling, uint32_t channels, char* lang, int interim, char* bugname, void **ppUserData);
 switch_status_t jb_transcribe_session_stop(switch_core_session_t *session, int channelIsClosing, char* bugname);
 switch_bool_t jb_transcribe_frame(switch_core_session_t *session, switch_media_bug_t *bug);
 

mod_jambonz_transcribe/mod_jambonz_transcribe.c

@@ -73,6 +73,8 @@ static switch_status_t start_capture(switch_core_session_t *session, switch_medi
 	switch_codec_implementation_t read_impl = { 0 };
 	void *pUserData;
 	uint32_t samples_per_second;
+	uint32_t desiredSampling = 8000;
+	const char* var;
 
   if (!switch_channel_get_variable(channel, "JAMBONZ_STT_URL")) {
     switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Missing JAMBONZ_STT_URL channel var\n");
@@ -91,8 +93,11 @@ static switch_status_t start_capture(switch_core_session_t *session, switch_medi
 	}
 
 	samples_per_second = !strcasecmp(read_impl.iananame, "g722") ? read_impl.actual_samples_per_second : read_impl.samples_per_second;
-
-	if (SWITCH_STATUS_FALSE == jb_transcribe_session_init(session, responseHandler, samples_per_second, flags & SMBF_STEREO ? 2 : 1, lang, interim, bugname, &pUserData)) {
+	var = switch_channel_get_variable(channel, "JAMBONZ_STT_SAMPLING");
+	if (var != NULL) {
+		desiredSampling = atoi(var);
+	}
+	if (SWITCH_STATUS_FALSE == jb_transcribe_session_init(session, responseHandler, samples_per_second, desiredSampling, flags & SMBF_STEREO ? 2 : 1, lang, interim, bugname, &pUserData)) {
 		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error initializing jb speech session.\n");
 		return SWITCH_STATUS_FALSE;
 	}

mod_playht_tts/playht_glue.cpp

@@ -452,12 +452,12 @@ static size_t write_cb(void *ptr, size_t size, size_t nmemb, ConnInfo_t *conn) {
       return 0;
     }
 
-    /* cache file will stay in the mp3 format for size (smaller) and simplicity */
-    if (conn->file) fwrite(data, sizeof(uint8_t), bytes_received, conn->file);
-
     pcm_data = convert_mp3_to_linear(conn, data, bytes_received);
     size_t bytesResampled = pcm_data.size() * sizeof(uint16_t);
 
+    /* cache same data to avoid streaming and cached audio quality is different*/
+    if (conn->file) fwrite(pcm_data.data(), sizeof(uint8_t), bytesResampled, conn->file);
+
     // Resize the buffer if necessary
     if (cBuffer->capacity() - cBuffer->size() < (bytesResampled / sizeof(uint16_t))) {
       //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "write_cb growing buffer\n"); 
@@ -774,7 +774,10 @@ extern "C" {
       cJSON_AddStringToObject(jResult, "quality", p->quality);
     }
     if (p->speed) {
-      cJSON_AddNumberToObject(jResult, "speed", atoi(p->speed));
+      double val = strtod(p->speed, NULL);
+      if (val != 0.0) {
+        cJSON_AddNumberToObject(jResult, "speed", val);
+      }
     }
     if (p->seed) {
       cJSON_AddNumberToObject(jResult, "seed", atoi(p->seed));