Compare commits

..

7 Commits

Author SHA1 Message Date
Dave Horton 15c41d8884 further fix for aws transcribe race condition 2024-03-12 12:23:58 -04:00
Dave Horton b606255206 add mod_dub (#16)
* add mod_dub

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

* remove some locks

---------

Signed-off-by: Dave Horton <daveh@beachdognet.com>
2024-03-12 09:56:49 -04:00
Hoan Luu Huu d6ae0a6a39 fix elevenlabs and whisper for multiple codec (#14)
* fix elevenlabs and whisper for multiple codec

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

* fix review comments

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

---------

Signed-off-by: Quan HL <quan.luuhoang8@gmail.com>
2024-03-12 09:52:35 -04:00
Dave Horton 06ab877f68 fix aws race condition when 2 start transcribes are sent at the same instant 2024-03-11 13:57:15 -04:00
Hoan Luu Huu 92dd3fc854 enable azure language identification mode (#15)
Signed-off-by: Quan HL <quan.luuhoang8@gmail.com>
2024-03-08 10:12:33 -05:00
Hoan Luu Huu b4f3a41913 whisper tts stream module (#8)
* whisper tts stream module

* add convert mp3 to linear

* add lmpg123 link

* wip

* whisper cache

* wip

* wip

* clean up

* add free mem and openai metric

* whisper: cache files as mp3 for simplicity

---------

Co-authored-by: Dave Horton <daveh@beachdognet.com>
2024-02-22 14:17:13 -05:00
Dave Horton bbcabdd927 update deepgram model defaults, enable interim results when using utterance_end_ms 2024-02-12 10:47:52 -05:00
36 changed files with 3337 additions and 89 deletions
-5
View File
@@ -1,5 +0,0 @@
Copyright 2023, Drachtio Communications Services, LLC
This software is provided under a dual-licensing scheme, described here: https://github.com/jambonz/freeswitch-modules/blob/main/COPYING.
Please refer to the above file and the links therein for the complete terms and conditions of each license. Your use of this software constitutes agreement to the terms of these licenses. If you have any questions regarding the licensing of this software, please consult a legal expert.
-5
View File
@@ -1,5 +0,0 @@
Copyright 2023, Drachtio Communications Services, LLC
This software is provided under a dual-licensing scheme, described here: https://github.com/jambonz/freeswitch-modules/blob/main/COPYING.
Please refer to the above file and the links therein for the complete terms and conditions of each license. Your use of this software constitutes agreement to the terms of these licenses. If you have any questions regarding the licensing of this software, please consult a legal expert.
-5
View File
@@ -1,5 +0,0 @@
Copyright 2023, Drachtio Communications Services, LLC
This software is provided under a dual-licensing scheme, described here: https://github.com/jambonz/freeswitch-modules/blob/main/COPYING.
Please refer to the above file and the links therein for the complete terms and conditions of each license. Your use of this software constitutes agreement to the terms of these licenses. If you have any questions regarding the licensing of this software, please consult a legal expert.
-5
View File
@@ -1,5 +0,0 @@
Copyright 2023, Drachtio Communications Services, LLC
This software is provided under a dual-licensing scheme, described here: https://github.com/jambonz/freeswitch-modules/blob/main/COPYING.
Please refer to the above file and the links therein for the complete terms and conditions of each license. Your use of this software constitutes agreement to the terms of these licenses. If you have any questions regarding the licensing of this software, please consult a legal expert.
+12 -1
View File
@@ -512,7 +512,18 @@ extern "C" {
// close connection and get final responses
switch_mutex_lock(cb->mutex);
GStreamer* streamer = (GStreamer *) cb->streamer;
GStreamer* streamer;
// race condition: if we just started a transcribe at same instant, the spawned thread may not have had a chance to set the streamer
int i = 0;
do {
streamer = (GStreamer *) cb->streamer;
if (streamer) break;
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);
if (streamer) {
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "aws_transcribe_session_stop: finish..%s\n", bugname);
streamer->finish();
-5
View File
@@ -1,5 +0,0 @@
Copyright 2023, Drachtio Communications Services, LLC
This software is provided under a dual-licensing scheme, described here: https://github.com/jambonz/freeswitch-modules/blob/main/COPYING.
Please refer to the above file and the links therein for the complete terms and conditions of each license. Your use of this software constitutes agreement to the terms of these licenses. If you have any questions regarding the licensing of this software, please consult a legal expert.
@@ -135,6 +135,13 @@ public:
properties.SetProperty(PropertyId::Speech_SegmentationSilenceTimeoutMs, segmentationInterval);
}
//https://learn.microsoft.com/en-us/azure/ai-services/speech-service/language-identification?tabs=once&pivots=programming-language-cpp#at-start-and-continuous-language-identification
const char* languageIdMode = switch_channel_get_variable(channel, "AZURE_LANGUAGE_ID_MODE");
if (languageIdMode) {
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(psession), SWITCH_LOG_DEBUG, "setting SpeechServiceConnection_LanguageIdMode to %s \n", languageIdMode);
properties.SetProperty(PropertyId::SpeechServiceConnection_LanguageIdMode, languageIdMode);
}
// recognition mode - readonly according to Azure docs:
// https://docs.microsoft.com/en-us/javascript/api/microsoft-cognitiveservices-speech-sdk/propertyid?view=azure-node-latest
/*
-5
View File
@@ -1,5 +0,0 @@
Copyright 2023, Drachtio Communications Services, LLC
This software is provided under a dual-licensing scheme, described here: https://github.com/jambonz/freeswitch-modules/blob/main/COPYING.
Please refer to the above file and the links therein for the complete terms and conditions of each license. Your use of this software constitutes agreement to the terms of these licenses. If you have any questions regarding the licensing of this software, please consult a legal expert.
-5
View File
@@ -1,5 +0,0 @@
Copyright 2023, Drachtio Communications Services, LLC
This software is provided under a dual-licensing scheme, described here: https://github.com/jambonz/freeswitch-modules/blob/main/COPYING.
Please refer to the above file and the links therein for the complete terms and conditions of each license. Your use of this software constitutes agreement to the terms of these licenses. If you have any questions regarding the licensing of this software, please consult a legal expert.
+20 -14
View File
@@ -37,7 +37,8 @@ namespace {
{"zh", "base"},
{"zh-CN", "base"},
{"zh-TW", "base"},
{"da", "enhanced"},
{"da", "nova-2"},
{"da-DK", "nova-2"},
{"en", "nova-2"},
{"en-US", "nova-2"},
{"en-AU", "nova-2"},
@@ -45,29 +46,32 @@ namespace {
{"en-IN", "nova-2"},
{"en-NZ", "nova-2"},
{"nl", "nova-2"},
{"nl-BE", "nova-2"},
{"fr", "nova-2"},
{"fr-CA", "nova-2"},
{"de", "nova-2"},
{"el", "nova-2"},
{"hi", "nova-2"},
{"hi-Latn", "nova-2"},
{"id", "base"},
{"it", "enhanced"},
{"id", "nova-2"},
{"it", "nova-2"},
{"ja", "enhanced"},
{"ko", "enhanced"},
{"no", "enhanced"},
{"pl", "enhanced"},
{"ko", "nova-2"},
{"ko-KR", "nova-2"},
{"no", "nova-2"},
{"pl", "nova-2"},
{"pt","nova-2"},
{"pt-BR", "nova-2"},
{"pt-PT", "enhanced"},
{"ru", "base"},
{"ru", "nova-2"},
{"es","nova-2"},
{"es-419","nova-2"},
{"es-LATAM","enhanced"},
{"sv", "enhanced"},
{"sv", "nova-2"},
{"sv-SE", "nova-2"},
{"ta", "enhanced"},
{"taq", "enhanced"},
{"tr", "base"},
{"uk", "base"}
{"tr", "nova-2"},
{"uk", "nova-2"}
};
static bool getLanguageInfo(const std::string& language, std::string& model) {
@@ -239,9 +243,6 @@ namespace {
oss << "&tag=";
oss << var;
}
if (interim) {
oss << "&interim_results=true";
}
if (var = switch_channel_get_variable(channel, "DEEPGRAM_SPEECH_ENDPOINTING")) {
oss << "&endpointing=";
oss << var;
@@ -249,6 +250,11 @@ namespace {
if (var = switch_channel_get_variable(channel, "DEEPGRAM_SPEECH_UTTERANCE_END_MS")) {
oss << "&utterance_end_ms=";
oss << var;
interim = 1;
//this requires interim_results=true (https://developers.deepgram.com/docs/understanding-end-of-speech-detection)
}
if (interim) {
oss << "&interim_results=true";
}
if (var = switch_channel_get_variable(channel, "DEEPGRAM_SPEECH_VAD_TURNOFF")) {
oss << "&vad_turnoff=";
-5
View File
@@ -1,5 +0,0 @@
Copyright 2023, Drachtio Communications Services, LLC
This software is provided under a dual-licensing scheme, described here: https://github.com/jambonz/freeswitch-modules/blob/main/COPYING.
Please refer to the above file and the links therein for the complete terms and conditions of each license. Your use of this software constitutes agreement to the terms of these licenses. If you have any questions regarding the licensing of this software, please consult a legal expert.
+10
View File
@@ -0,0 +1,10 @@
include $(top_srcdir)/build/modmake.rulesam
MODNAME=mod_dub
mod_LTLIBRARIES = mod_dub.la
mod_dub_la_SOURCES = file_loader.cpp audio_downloader.cpp mod_dub.c dub_glue.cpp
mod_dub_la_CFLAGS = $(AM_CFLAGS)
mod_dub_la_CXXFLAGS = $(AM_CXXFLAGS) -std=c++17
mod_dub_la_LIBADD = $(switch_builddir)/libfreeswitch.la
mod_dub_la_LDFLAGS = -avoid-version -module -no-undefined -shared `pkg-config --libs boost` -lstdc++ -lmpg123
+885
View File
@@ -0,0 +1,885 @@
#include "audio_downloader.h"
#include <boost/thread.hpp>
#include <boost/asio.hpp>
#include <boost/asio/ssl.hpp>
#include <boost/pool/object_pool.hpp>
#include <boost/bind/bind.hpp>
#include <boost/tokenizer.hpp>
#include <boost/foreach.hpp>
#include <boost/asio.hpp>
#include <boost/assign/list_of.hpp>
#include <boost/algorithm/string.hpp>
#include <map>
#include <mpg123.h>
#include <curl/curl.h>
#define BUFFER_GROW_SIZE (80000)
#define BUFFER_THROTTLE_LOW (40000)
#define BUFFER_THROTTLE_HIGH (160000)
static uint16_t currDownloadId = 0;
typedef struct
{
CURLM *multi;
int still_running;
} GlobalInfo_t;
static GlobalInfo_t global;
typedef enum
{
STATUS_NONE = 0,
STATUS_FAILED,
STATUS_DOWNLOAD_IN_PROGRESS,
STATUS_DOWNLOAD_PAUSED,
STATUS_DOWNLOAD_COMPLETE,
STATUS_AWAITING_RESTART,
STATUS_STOPPING,
STATUS_STOPPED
} Status_t;
static const char* status2String(Status_t status)
{
static const char* statusStrings[] = {
"STATUS_NONE",
"STATUS_FAILED",
"STATUS_DOWNLOAD_IN_PROGRESS",
"STATUS_DOWNLOAD_PAUSED",
"STATUS_DOWNLOAD_COMPLETE",
"STATUS_AWAITING_RESTART",
"STATUS_STOPPING",
"STATUS_STOPPED"
};
if (status >= 0 && status < sizeof(statusStrings) / sizeof(statusStrings[0]))
{
return statusStrings[status];
}
else
{
return "UNKNOWN_STATUS";
}
}
typedef struct
{
GlobalInfo_t *global;
CURL *easy;
switch_mutex_t* mutex;
CircularBuffer_t* buffer;
mpg123_handle *mh;
char error[CURL_ERROR_SIZE]; // curl error buffer
char *err_msg; // http server error message
char* url;
bool loop;
int rate;
boost::asio::deadline_timer *timer;
Status_t status;
downloadId_t id;
int response_code;
int gain;
} ConnInfo_t;
typedef std::map<int32_t, ConnInfo_t *> Id2ConnMap_t;
static Id2ConnMap_t id2ConnMap;
static boost::object_pool<ConnInfo_t> pool ;
static std::map<curl_socket_t, boost::asio::ip::tcp::socket *> socket_map;
static boost::asio::io_service io_service;
static boost::asio::deadline_timer timer(io_service);
static std::string fullDirPath;
static std::thread worker_thread;
/* forward declarations */
static ConnInfo_t* createDownloader(const char *url, int rate, int loop, int gain, mpg123_handle *mhm, switch_mutex_t *mutex, CircularBuffer_t *buffer);
static CURL* createEasyHandle(void);
static void destroyConnection(ConnInfo_t *conn);
static void check_multi_info(GlobalInfo_t *g) ;
static int mcode_test(const char *where, CURLMcode code);
static void event_cb(GlobalInfo_t *g, curl_socket_t s, int action, const boost::system::error_code & error, int *fdp);
static void setsock(int *fdp, curl_socket_t s, CURL *e, int act, int oldact, GlobalInfo_t *g);
static void addsock(curl_socket_t s, CURL *easy, int action, GlobalInfo_t *g);
static int sock_cb(CURL *e, curl_socket_t s, int what, void *cbp, void *sockp);
static void threadFunc();
static void timer_cb(const boost::system::error_code & error, GlobalInfo_t *g);
static int multi_timer_cb(CURLM *multi, long timeout_ms, GlobalInfo_t *g);
static std::vector<int16_t> convert_mp3_to_linear(ConnInfo_t *conn, int8_t *data, size_t len);
static void throttling_cb(const boost::system::error_code& error, ConnInfo_t* conn) ;
static void restart_cb(const boost::system::error_code& error, ConnInfo_t* conn);
static size_t write_cb(void *ptr, size_t size, size_t nmemb, ConnInfo_t *conn);
static bool parseHeader(const std::string& str, std::string& header, std::string& value) ;
static int extract_response_code(const std::string& input) ;
static size_t header_callback(char *buffer, size_t size, size_t nitems, ConnInfo_t *conn);
static curl_socket_t opensocket(void *clientp, curlsocktype purpose, struct curl_sockaddr *address);
static int close_socket(void *clientp, curl_socket_t item);
/* apis */
extern "C" {
switch_status_t init_audio_downloader() {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "init_audio_downloader loading..\n");
memset(&global, 0, sizeof(GlobalInfo_t));
global.multi = curl_multi_init();
if (!global.multi) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "init_audio_downloader curl_multi_init() failed, exiting!\n");
return SWITCH_STATUS_FALSE;
}
curl_multi_setopt(global.multi, CURLMOPT_SOCKETFUNCTION, sock_cb);
curl_multi_setopt(global.multi, CURLMOPT_SOCKETDATA, &global);
curl_multi_setopt(global.multi, CURLMOPT_TIMERFUNCTION, multi_timer_cb);
curl_multi_setopt(global.multi, CURLMOPT_TIMERDATA, &global);
curl_multi_setopt(global.multi, CURLMOPT_PIPELINING, CURLPIPE_MULTIPLEX);
if (mpg123_init() != MPG123_OK) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "init_audio_downloader: failed to initiate MPG123");
return SWITCH_STATUS_FALSE;
}
/* start worker thread */
std::thread t(threadFunc) ;
worker_thread.swap( t ) ;
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "init_audio_downloader: loaded\n");
return SWITCH_STATUS_SUCCESS;
}
switch_status_t deinit_audio_downloader() {
/* stop the ASIO IO service */
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "deinit_audio_downloader: stopping io service\n");
io_service.stop();
/* Join the worker thread */
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "deinit_audio_downloader: wait for worker thread to complete\n");
if (worker_thread.joinable()) {
worker_thread.join();
}
/* cleanup curl multi handle*/
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "deinit_audio_downloader: release curl multi\n");
curl_multi_cleanup(global.multi);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "deinit_audio_downloader: completed\n");
mpg123_exit();
return SWITCH_STATUS_SUCCESS;
}
downloadId_t start_audio_download(const char* url, int rate, int loop, int gain, switch_mutex_t* mutex, CircularBuffer_t* buffer) {
int mhError = 0;
/* allocate handle for mpeg decoding */
mpg123_handle *mh = mpg123_new("auto", &mhError);
if (!mh) {
const char *mhErr = mpg123_plain_strerror(mhError);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error allocating mpg123 handle! %s\n", switch_str_nil(mhErr));
return INVALID_DOWNLOAD_ID;
}
if (mpg123_open_feed(mh) != MPG123_OK) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error mpg123_open_feed!\n");
return INVALID_DOWNLOAD_ID;
}
if (mpg123_format_all(mh) != MPG123_OK) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error mpg123_format_all!\n");
return INVALID_DOWNLOAD_ID;
}
if (mpg123_param(mh, MPG123_FORCE_RATE, rate, 0) != MPG123_OK) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error forcing resample to 8k!\n");
return INVALID_DOWNLOAD_ID;
}
if (mpg123_param(mh, MPG123_FLAGS, MPG123_MONO_MIX, 0) != MPG123_OK) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error forcing single channel!\n");
return INVALID_DOWNLOAD_ID;
}
ConnInfo_t* conn = createDownloader(url, rate, loop, gain, mh, mutex, buffer);
if (!conn) {
return INVALID_DOWNLOAD_ID;
}
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,
"start_audio_download: starting download %d\n", conn->id);
return conn->id;
}
switch_status_t stop_audio_download(int id) {
auto it = id2ConnMap.find(id);
if (it == id2ConnMap.end()) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "stop_audio_download: id %d has already completed\n", id);
return SWITCH_STATUS_FALSE;
}
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,
"stop_audio_download: stopping download %d, status %s\n", id, status2String(it->second->status));
ConnInfo_t *conn = it->second;
auto status = conn->status;
/* past this point I shall not access either the mutex or the buffer provided */
conn->mutex = nullptr;
conn->buffer = nullptr;
/* if download is in progress set status to cancel it during next call back */
if (status == Status_t::STATUS_DOWNLOAD_PAUSED) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "stop_audio_download: resuming download %d so we can cancel it\n", id);
conn->status = Status_t::STATUS_STOPPING;
curl_easy_pause(conn->easy, CURLPAUSE_CONT);
}
if (status != Status_t::STATUS_DOWNLOAD_IN_PROGRESS) {
destroyConnection(conn);
}
conn->status = Status_t::STATUS_STOPPING;
return SWITCH_STATUS_SUCCESS;
}
}
/* internal */
ConnInfo_t* createDownloader(const char *url, int rate, int loop, int gain, mpg123_handle *mh, switch_mutex_t *mutex, CircularBuffer_t *buffer) {
ConnInfo_t *conn = pool.malloc() ;
CURL* easy = createEasyHandle();
if (!easy || !conn) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "createDownloader: failed to allocate memory\n");
return nullptr;
}
memset(conn, 0, sizeof(ConnInfo_t));
conn->easy = easy;
conn->mutex = mutex;
conn->buffer = buffer;
conn->mh = mh;
conn->loop = loop;
conn->gain = gain;
conn->rate = rate;
conn->url = strdup(url);
conn->global = &global;
conn->status = Status_t::STATUS_NONE;
conn->timer = new boost::asio::deadline_timer(io_service);
downloadId_t id = ++currDownloadId;
if (id == 0) id++;
id2ConnMap[id] = conn;
conn->id = id;
curl_easy_setopt(easy, CURLOPT_URL, url);
curl_easy_setopt(easy, CURLOPT_HTTPGET, 1L);
curl_easy_setopt(easy, CURLOPT_WRITEFUNCTION, write_cb);
curl_easy_setopt(easy, CURLOPT_WRITEDATA, conn);
curl_easy_setopt(easy, CURLOPT_ERRORBUFFER, conn->error);
curl_easy_setopt(easy, CURLOPT_PRIVATE, conn);
curl_easy_setopt(easy, CURLOPT_VERBOSE, 0L);
curl_easy_setopt(easy, CURLOPT_NOPROGRESS, 1L);
curl_easy_setopt(easy, CURLOPT_HEADERFUNCTION, header_callback);
curl_easy_setopt(easy, CURLOPT_HEADERDATA, conn);
/* call this function to get a socket */
curl_easy_setopt(easy, CURLOPT_OPENSOCKETFUNCTION, opensocket);
/* call this function to close a socket */
curl_easy_setopt(easy, CURLOPT_CLOSESOCKETFUNCTION, close_socket);
curl_easy_setopt(easy, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2_0);
/* keep the speed down so we don't have to buffer large amounts*/
curl_easy_setopt(easy, CURLOPT_MAX_RECV_SPEED_LARGE, (curl_off_t)31415);
auto rc = curl_multi_add_handle(global.multi, conn->easy);
if (mcode_test("new_conn: curl_multi_add_handle", rc) < 0) {
return nullptr;
}
conn->status = Status_t::STATUS_DOWNLOAD_IN_PROGRESS;
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "createDownloader: launched request, gain %d\n", conn->gain);
return conn;
}
void destroyConnection(ConnInfo_t *conn) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "destroyConnection\n");
/* clean up the curl handle*/
curl_multi_remove_handle(conn->global, conn->easy);
curl_easy_cleanup(conn->easy);
/* clear asio resources and free resources */
if (conn->timer) {
conn->timer->cancel();
delete conn->timer;
}
if (conn->err_msg) {
free(conn->err_msg);
}
/* free mp3 decoder */
if (conn->mh) {
mpg123_close(conn->mh);
mpg123_delete(conn->mh);
}
if (conn->url) {
free(conn->url);
}
if (conn->mutex) switch_mutex_lock(conn->mutex);
id2ConnMap.erase(conn->id);
if (conn->mutex) switch_mutex_unlock(conn->mutex);
memset(conn, 0, sizeof(ConnInfo_t));
pool.destroy(conn) ;
}
CURL* createEasyHandle(void) {
CURL* easy = curl_easy_init();
if(!easy) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "curl_easy_init() failed!\n");
return nullptr ;
}
curl_easy_setopt(easy, CURLOPT_FOLLOWLOCATION, 1L);
curl_easy_setopt(easy, CURLOPT_USERAGENT, "jambonz/0.8.5");
// set connect timeout to 3 seconds and no total timeout as files could be large
curl_easy_setopt(easy, CURLOPT_CONNECTTIMEOUT_MS, 3000L);
curl_easy_setopt(easy, CURLOPT_TIMEOUT, 0L); // no timeout
return easy ;
}
/* Check for completed transfers, and remove their easy handles */
void check_multi_info(GlobalInfo_t *g) {
CURLMsg *msg;
int msgs_left;
ConnInfo_t *conn;
CURL *easy;
CURLcode res;
while((msg = curl_multi_info_read(g->multi, &msgs_left))) {
if(msg->msg == CURLMSG_DONE) {
long response_code;
double namelookup=0, connect=0, total=0 ;
char *ct = NULL ;
easy = msg->easy_handle;
res = msg->data.result;
curl_easy_getinfo(easy, CURLINFO_PRIVATE, &conn);
curl_easy_getinfo(easy, CURLINFO_RESPONSE_CODE, &response_code);
curl_easy_getinfo(easy, CURLINFO_CONTENT_TYPE, &ct);
curl_easy_getinfo(easy, CURLINFO_NAMELOOKUP_TIME, &namelookup);
curl_easy_getinfo(easy, CURLINFO_CONNECT_TIME, &connect);
curl_easy_getinfo(easy, CURLINFO_TOTAL_TIME, &total);
downloadId_t id = conn->id;
auto mutex = conn->mutex;
auto buffer = conn->buffer;
auto rate = conn->rate;
auto loop = conn->loop;
auto gain = conn->gain;
auto oldId = conn->id;
bool restart = conn->loop && conn->status != Status_t::STATUS_STOPPING && response_code == 200;
conn->response_code = response_code;
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "curl done, response code %d, status %s\n", response_code, status2String(conn->status));
conn->status = Status_t::STATUS_DOWNLOAD_COMPLETE;
curl_multi_remove_handle(g->multi, easy);
if (restart) {
conn->status = Status_t::STATUS_AWAITING_RESTART;
conn->timer->expires_from_now(boost::posix_time::millisec(1000));
conn->timer->async_wait(boost::bind(&restart_cb, boost::placeholders::_1, conn));
//TODO: this seems to not be working from this callback; maybe start it from a timer callback?
}
else {
destroyConnection(conn);
}
}
}
}
int mcode_test(const char *where, CURLMcode code) {
if(CURLM_OK != code) {
const char *s;
switch(code) {
case CURLM_CALL_MULTI_PERFORM:
s = "CURLM_CALL_MULTI_PERFORM";
break;
case CURLM_BAD_HANDLE:
s = "CURLM_BAD_HANDLE";
break;
case CURLM_BAD_EASY_HANDLE:
s = "CURLM_BAD_EASY_HANDLE";
break;
case CURLM_OUT_OF_MEMORY:
s = "CURLM_OUT_OF_MEMORY";
break;
case CURLM_INTERNAL_ERROR:
s = "CURLM_INTERNAL_ERROR";
break;
case CURLM_UNKNOWN_OPTION:
s = "CURLM_UNKNOWN_OPTION";
break;
case CURLM_LAST:
s = "CURLM_LAST";
break;
default:
s = "CURLM_unknown";
break;
case CURLM_BAD_SOCKET:
s = "CURLM_BAD_SOCKET";
break;
}
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "mcode_test ERROR: %s returns %s:%d\n", where, s, code);
return -1;
}
return 0 ;
}
void remsock(int *f, GlobalInfo_t *g) {
if(f) {
free(f);
f = NULL;
}
}
/* Called by asio when there is an action on a socket */
void event_cb(GlobalInfo_t *g, curl_socket_t s, int action, const boost::system::error_code & error, int *fdp) {
int f = *fdp;
//switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "event_cb socket %#X has action %d\n", s, action) ;
// Socket already POOL REMOVED.
if (f == CURL_POLL_REMOVE) {
//switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "event_cb socket %#X removed\n", s);
remsock(fdp, g);
return;
}
if(socket_map.find(s) == socket_map.end()) {
//switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "event_cb: socket %#X already closed\n, s");
return;
}
/* make sure the event matches what are wanted */
if(f == action || f == CURL_POLL_INOUT) {
if(error) {
action = CURL_CSELECT_ERR;
}
CURLMcode rc = curl_multi_socket_action(g->multi, s, action, &g->still_running);
mcode_test("event_cb: curl_multi_socket_action", rc);
check_multi_info(g);
if(g->still_running <= 0) {
timer.cancel();
}
/* keep on watching.
* the socket may have been closed and/or fdp may have been changed
* in curl_multi_socket_action(), so check them both */
if(!error && socket_map.find(s) != socket_map.end() &&
(f == action || f == CURL_POLL_INOUT)) {
boost::asio::ip::tcp::socket *tcp_socket = socket_map.find(s)->second;
if(action == CURL_POLL_IN) {
tcp_socket->async_read_some(boost::asio::null_buffers(),
boost::bind(&event_cb, g, s,
action, boost::placeholders::_1, fdp));
}
if(action == CURL_POLL_OUT) {
tcp_socket->async_write_some(boost::asio::null_buffers(),
boost::bind(&event_cb, g, s,
action, boost::placeholders::_1, fdp));
}
}
}
}
/* socket functions */
void setsock(int *fdp, curl_socket_t s, CURL *e, int act, int oldact, GlobalInfo_t *g) {
std::map<curl_socket_t, boost::asio::ip::tcp::socket *>::iterator it = socket_map.find(s);
if(it == socket_map.end()) {
//switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "setsock: socket %#X not found\n, s");
return;
}
boost::asio::ip::tcp::socket * tcp_socket = it->second;
*fdp = act;
if(act == CURL_POLL_IN) {
if(oldact != CURL_POLL_IN && oldact != CURL_POLL_INOUT) {
tcp_socket->async_read_some(boost::asio::null_buffers(),
boost::bind(&event_cb, g, s,
CURL_POLL_IN, boost::placeholders::_1, fdp));
}
}
else if(act == CURL_POLL_OUT) {
if(oldact != CURL_POLL_OUT && oldact != CURL_POLL_INOUT) {
tcp_socket->async_write_some(boost::asio::null_buffers(),
boost::bind(&event_cb, g, s,
CURL_POLL_OUT, boost::placeholders::_1, fdp));
}
}
else if(act == CURL_POLL_INOUT) {
if(oldact != CURL_POLL_IN && oldact != CURL_POLL_INOUT) {
tcp_socket->async_read_some(boost::asio::null_buffers(),
boost::bind(&event_cb, g, s,
CURL_POLL_IN, boost::placeholders::_1, fdp));
}
if(oldact != CURL_POLL_OUT && oldact != CURL_POLL_INOUT) {
tcp_socket->async_write_some(boost::asio::null_buffers(),
boost::bind(&event_cb, g, s,
CURL_POLL_OUT, boost::placeholders::_1, fdp));
}
}
}
void addsock(curl_socket_t s, CURL *easy, int action, GlobalInfo_t *g) {
/* fdp is used to store current action */
int *fdp = (int *) calloc(sizeof(int), 1);
setsock(fdp, s, easy, action, 0, g);
curl_multi_assign(g->multi, s, fdp);
}
int sock_cb(CURL *e, curl_socket_t s, int what, void *cbp, void *sockp) {
GlobalInfo_t *g = &global;
int *actionp = (int *) sockp;
static const char *whatstr[] = { "none", "IN", "OUT", "INOUT", "REMOVE"};
if(what == CURL_POLL_REMOVE) {
*actionp = what;
}
else {
if(!actionp) {
addsock(s, e, what, g);
}
else {
setsock(actionp, s, e, what, *actionp, g);
}
}
return 0;
}
void threadFunc() {
/* to make sure the event loop doesn't terminate when there is no work to do */
io_service.reset() ;
boost::asio::io_service::work work(io_service);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "mod_dub threadFunc - starting\n");
for(;;) {
try {
io_service.run() ;
break ;
}
catch( std::exception& e) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "mod_dub threadFunc - Error: %s\n", e.what());
}
}
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "mod_dub threadFunc - ending\n");
}
/* Called by asio when our timeout expires */
void timer_cb(const boost::system::error_code & error, GlobalInfo_t *g)
{
//switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "timer_cb\n");
if(!error) {
CURLMcode rc = curl_multi_socket_action(g->multi, CURL_SOCKET_TIMEOUT, 0, &g->still_running);
mcode_test("timer_cb: curl_multi_socket_action", rc);
check_multi_info(g);
}
}
int multi_timer_cb(CURLM *multi, long timeout_ms, GlobalInfo_t *g) {
/* cancel running timer */
timer.cancel();
if(timeout_ms >= 0) {
// from libcurl 7.88.1-10+deb12u4 does not allow call curl_multi_socket_action or curl_multi_perform in curl_multi callback directly
timer.expires_from_now(boost::posix_time::millisec(timeout_ms ? timeout_ms : 1));
timer.async_wait(boost::bind(&timer_cb, boost::placeholders::_1, g));
}
return 0;
}
std::vector<int16_t> convert_mp3_to_linear(ConnInfo_t *conn, int8_t *data, size_t len) {
std::vector<int16_t> linear_data;
int eof = 0;
int mp3err = 0;
if(mpg123_feed(conn->mh, (const unsigned char*) data, len) == MPG123_OK) {
while(!eof) {
size_t usedlen = 0;
off_t frame_offset;
unsigned char* audio;
int decode_status = mpg123_decode_frame(conn->mh, &frame_offset, &audio, &usedlen);
switch(decode_status) {
case MPG123_NEW_FORMAT:
continue;
case MPG123_OK:
{
size_t samples = usedlen / sizeof(int16_t);
linear_data.insert(linear_data.end(), reinterpret_cast<int16_t*>(audio), reinterpret_cast<int16_t*>(audio) + samples);
}
break;
case MPG123_DONE:
case MPG123_NEED_MORE:
eof = 1;
break;
case MPG123_ERR:
default:
if(++mp3err >= 5) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Decoder Error!\n");
eof = 1;
}
}
if (eof)
break;
mp3err = 0;
}
if (conn->gain != 0) {
switch_change_sln_volume_granular(linear_data.data(), linear_data.size(), conn->gain);
}
}
return linear_data;
}
void restart_cb(const boost::system::error_code& error, ConnInfo_t* conn) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "restart_cb status is %s\n", status2String(conn->status));
if (conn->status == Status_t::STATUS_AWAITING_RESTART) {
auto url = strdup(conn->url);
auto rate = conn->rate;
auto loop = conn->loop;
auto gain = conn->gain;
auto mutex = conn->mutex;
auto buffer = conn->buffer;
auto oldId = conn->id;
destroyConnection(conn);
downloadId_t id = start_audio_download(url, rate, loop, gain, mutex, buffer);
/* re-use id since caller is tracking that id */
auto * newConnection = id2ConnMap[id];
id2ConnMap[oldId] = newConnection;
id2ConnMap.erase(id);
free(url);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "restarted looped download\n");
}
}
void throttling_cb(const boost::system::error_code& error, ConnInfo_t* conn) {
if (conn->status == Status_t::STATUS_STOPPING || !conn->mutex || !conn->buffer) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "throttling_cb: session gone, resume download so we can complete\n");
curl_easy_pause(conn->easy, CURLPAUSE_CONT);
return;
}
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "throttling_cb: status is %s\n", status2String(conn->status));
switch_mutex_lock(conn->mutex);
if (!error) {
auto size = conn->buffer->size();
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "throttling_cb: size is now %ld\n", size);
if (size < BUFFER_THROTTLE_LOW) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "throttling_cb: resuming download\n");
curl_easy_pause(conn->easy, CURLPAUSE_CONT);
switch_mutex_unlock(conn->mutex);
return;
}
// check back in 2 seconds
conn->timer->expires_from_now(boost::posix_time::millisec(2000));
conn->timer->async_wait(boost::bind(&throttling_cb, boost::placeholders::_1, conn));
} else {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "throttling_cb: error (%d): %s\n", error.value(), error.message().c_str());
// Handle any errors
}
switch_mutex_unlock(conn->mutex);
}
/* CURLOPT_WRITEFUNCTION - here is where we receive the data */
size_t write_cb(void *ptr, size_t size, size_t nmemb, ConnInfo_t *conn) {
int8_t *data = (int8_t *) ptr;
size_t bytes_received = size * nmemb;
std::vector<int16_t> pcm_data;
if (conn->status == Status_t::STATUS_STOPPING || conn->status == Status_t::STATUS_STOPPED || !conn->mutex || !conn->buffer) {
if (conn->timer) conn->timer->cancel();
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG,
"write_cb: aborting transfer, status %s, mutex %p, buffer %p\n", status2String(conn->status), conn->mutex, conn->buffer);
/* this will abort the transfer */
return 0;
}
{
switch_mutex_lock(conn->mutex);
if (conn->response_code > 0 && conn->response_code != 200) {
std::string body((char *) ptr, bytes_received);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "write_cb: received body %s\n", body.c_str());
conn->err_msg = strdup(body.c_str());
conn->status = Status_t::STATUS_FAILED;
switch_mutex_unlock(conn->mutex);
return 0;
}
/* throttle after reaching high water mark */
if (conn->buffer->size() > BUFFER_THROTTLE_HIGH) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "write_cb: throttling download, buffer size is %ld\n", conn->buffer->size());
// check back in 2 seconds
conn->timer->expires_from_now(boost::posix_time::millisec(2000));
conn->timer->async_wait(boost::bind(&throttling_cb, boost::placeholders::_1, conn));
conn->status = Status_t::STATUS_DOWNLOAD_PAUSED;
switch_mutex_unlock(conn->mutex);
return CURL_WRITEFUNC_PAUSE;
}
pcm_data = convert_mp3_to_linear(conn, data, bytes_received);
size_t bytesResampled = pcm_data.size() * sizeof(int16_t);
// Resize the buffer if necessary
if (conn->buffer->capacity() - conn->buffer->size() < (bytesResampled / sizeof(int16_t))) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "write_cb growing buffer, size now %ld\n", conn->buffer->size());
//TODO: if buffer exceeds some max size, return CURL_WRITEFUNC_ERROR to abort the transfer
conn->buffer->set_capacity(conn->buffer->size() + std::max((bytesResampled / sizeof(int16_t)), (size_t)BUFFER_GROW_SIZE));
}
/* Push the data into the buffer */
conn->buffer->insert(conn->buffer->end(), pcm_data.data(), pcm_data.data() + pcm_data.size());
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "write_cb: wrote data, buffer size is now %ld\n", conn->buffer->size());
switch_mutex_unlock(conn->mutex);
}
return bytes_received;
}
bool parseHeader(const std::string& str, std::string& header, std::string& value) {
std::vector<std::string> parts;
boost::split(parts, str, boost::is_any_of(":"), boost::token_compress_on);
if (parts.size() != 2)
return false;
header = boost::trim_copy(parts[0]);
value = boost::trim_copy(parts[1]);
return true;
}
int extract_response_code(const std::string& input) {
std::size_t space_pos = input.find(' ');
if (space_pos == std::string::npos) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Invalid HTTP response format %s\n", input.c_str());
return 0;
}
std::size_t code_start_pos = space_pos + 1;
std::size_t code_end_pos = input.find(' ', code_start_pos);
if (code_end_pos == std::string::npos) {
code_end_pos = input.length();
}
std::string code_str = input.substr(code_start_pos, code_end_pos - code_start_pos);
int response_code = std::stoi(code_str);
return response_code;
}
size_t header_callback(char *buffer, size_t size, size_t nitems, ConnInfo_t *conn) {
size_t bytes_received = size * nitems;
const std::string prefix = "HTTP/";
std::string header, value;
std::string input(buffer, bytes_received);
if (parseHeader(input, header, value)) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "recv header: %s with value %s\n", header.c_str(), value.c_str());
}
else {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "recv header: %s\n", input.c_str());
if (input.rfind(prefix, 0) == 0) {
try {
conn->response_code = extract_response_code(input);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "parsed response code: %ld\n", conn->response_code);
} catch (const std::invalid_argument& e) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "header_callback: invalid response code %s\n", input.substr(prefix.length()).c_str());
}
}
}
return bytes_received;
}
/* CURLOPT_OPENSOCKETFUNCTION */
curl_socket_t opensocket(void *clientp, curlsocktype purpose, struct curl_sockaddr *address) {
curl_socket_t sockfd = CURL_SOCKET_BAD;
//switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "opensocket: %d\n", purpose);
/* restrict to IPv4 */
if(purpose == CURLSOCKTYPE_IPCXN && address->family == AF_INET) {
/* create a tcp socket object */
boost::asio::ip::tcp::socket *tcp_socket = new boost::asio::ip::tcp::socket(io_service);
/* open it and get the native handle*/
boost::system::error_code ec;
tcp_socket->open(boost::asio::ip::tcp::v4(), ec);
if(ec) {
/* An error occurred */
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Couldn't open socket [%ld][%s]\n", ec, ec.message().c_str());
}
else {
sockfd = tcp_socket->native_handle();
/* save it for monitoring */
socket_map.insert(std::pair<curl_socket_t, boost::asio::ip::tcp::socket *>(sockfd, tcp_socket));
}
}
return sockfd;
}
/* CURLOPT_CLOSESOCKETFUNCTION */
int close_socket(void *clientp, curl_socket_t item) {
//switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "close_socket : %#X\n", item);
std::map<curl_socket_t, boost::asio::ip::tcp::socket *>::iterator it = socket_map.find(item);
if(it != socket_map.end()) {
delete it->second;
socket_map.erase(it);
}
return 0;
}
+18
View File
@@ -0,0 +1,18 @@
#ifndef __AUDIO_DOWNLOADER_H__
#define __AUDIO_DOWNLOADER_H__
#include <switch.h>
#include "common.h"
extern "C" {
switch_status_t init_audio_downloader();
switch_status_t deinit_audio_downloader();
int start_audio_download(const char* url, int rate, int loop, int gain, switch_mutex_t* mutex, CircularBuffer_t* buffer);
switch_status_t stop_audio_download(int id);
}
#endif
+12
View File
@@ -0,0 +1,12 @@
#ifndef _COMMON_H_
#define _COMMON_H_
#include <boost/circular_buffer.hpp>
typedef boost::circular_buffer<int16_t> CircularBuffer_t;
typedef int32_t downloadId_t;
#define INVALID_DOWNLOAD_ID (-1)
#endif
+222
View File
@@ -0,0 +1,222 @@
#include "mod_dub.h"
#include "audio_downloader.h"
#include "file_loader.h"
#include <string>
#include <switch.h>
#include <curl/curl.h>
#include <boost/circular_buffer.hpp>
typedef boost::circular_buffer<int16_t> CircularBuffer_t;
#define INIT_BUFFER_SIZE (80000)
extern "C" {
void init_dub_track(dub_track_t *track, char* trackName, int sampleRate) {
track->state = DUB_TRACK_STATE_READY;
track->trackName = strdup(trackName);
track->sampleRate = sampleRate;
track->circularBuffer = new CircularBuffer_t(INIT_BUFFER_SIZE);
}
switch_status_t silence_dub_track(dub_track_t *track) {
assert(track);
switch (track->generator) {
case DUB_GENERATOR_TYPE_HTTP:
stop_audio_download(track->generatorId);
break;
case DUB_GENERATOR_TYPE_FILE:
stop_file_load(track->generatorId);
break;
case DUB_GENERATOR_TYPE_TTS:
//TODO
break;
}
CircularBuffer_t* buffer = reinterpret_cast<CircularBuffer_t*>(track->circularBuffer);
buffer->clear();
track->state = DUB_TRACK_STATE_READY;
track->generator = DUB_GENERATOR_TYPE_UNKNOWN;
track->generatorId = 0;
return SWITCH_STATUS_SUCCESS;
}
switch_status_t remove_dub_track(dub_track_t *track) {
assert(track);
switch (track->generator) {
case DUB_GENERATOR_TYPE_HTTP:
stop_audio_download(track->generatorId);
break;
case DUB_GENERATOR_TYPE_FILE:
stop_file_load(track->generatorId);
break;
case DUB_GENERATOR_TYPE_TTS:
//TODO
break;
}
CircularBuffer_t* buffer = reinterpret_cast<CircularBuffer_t*>(track->circularBuffer);
if (buffer) {
delete buffer;
}
if (track->trackName) {
free(track->trackName);
}
memset(track, 0, sizeof(dub_track_t));
return SWITCH_STATUS_SUCCESS;
}
switch_status_t play_dub_track(dub_track_t *track, switch_mutex_t *mutex, char* url, int loop, int gain) {
bool isHttp = strncmp(url, "http", 4) == 0;
if (track->state != DUB_TRACK_STATE_READY) {
silence_dub_track(track);
}
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "play_dub_track: starting %s download: %s\n", (isHttp ? "HTTP" : "file"), url);
int id = isHttp ?
start_audio_download(url, track->sampleRate, loop, gain, mutex, (CircularBuffer_t*) track->circularBuffer) :
start_file_load(url, track->sampleRate, loop, gain, mutex, (CircularBuffer_t*) track->circularBuffer);
if (id == INVALID_DOWNLOAD_ID) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "play_dub_track: failed to start audio download\n");
return SWITCH_STATUS_FALSE;
}
track->state = DUB_TRACK_STATE_ACTIVE;
track->generatorId = id;
track->generator = isHttp ? DUB_GENERATOR_TYPE_HTTP : DUB_GENERATOR_TYPE_FILE;
track->gain = gain;
return SWITCH_STATUS_SUCCESS;
}
switch_status_t say_dub_track(dub_track_t *track, switch_mutex_t *mutex, char* text, int gain) {
if (track->state != DUB_TRACK_STATE_READY) {
silence_dub_track(track); // wait...shouldnt we queue says?
}
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "say_dub_track: starting TTS\n");
/**
* TODO:
* This is not implemented yet. We can play TTS using using the playOnSay function
* because jambonz can generate local audio files using TTS vendors.
* However, we should probably at least implement support for elevenlabs streaming api
* here because it is so much faster.
*
*/
/*
track->state = DUB_TRACK_STATE_ACTIVE;
track->generatorId = id;
track->generator = DUB_GENERATOR_TYPE_TTS;
track->gain = gain;
*/
return SWITCH_STATUS_SUCCESS;
}
/* module load and unload */
switch_status_t dub_init() {
switch_status_t status;
status = init_audio_downloader();
if (status == SWITCH_STATUS_SUCCESS) {
status = init_file_loader();
}
return status;
}
switch_status_t dub_cleanup() {
switch_status_t status;
status = deinit_audio_downloader();
if (status == SWITCH_STATUS_SUCCESS) {
status = deinit_file_loader();
}
return status;
}
switch_status_t dub_session_cleanup(switch_core_session_t *session, int channelIsClosing, switch_media_bug_t *bug) {
switch_channel_t *channel = switch_core_session_get_channel(session);
if (bug) {
struct cap_cb *cb = (struct cap_cb *) switch_core_media_bug_get_user_data(bug);
switch_mutex_lock(cb->mutex);
if (!switch_channel_get_private(channel, MY_BUG_NAME)) {
// race condition
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "%s Bug is not attached (race).\n", switch_channel_get_name(channel));
switch_mutex_unlock(cb->mutex);
return SWITCH_STATUS_FALSE;
}
switch_channel_set_private(channel, MY_BUG_NAME, NULL);
for (int i = 0; i < MAX_DUB_TRACKS; i++) {
dub_track_t* track = &cb->tracks[i];
if (track->state != DUB_TRACK_STATE_INACTIVE) {
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "dub_session_cleanup: cleared track %d:%s\n", i, track->trackName);
remove_dub_track(track);
}
}
if (!channelIsClosing) {
switch_core_media_bug_remove(session, &bug);
}
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "dub_session_cleanup: removed bug and cleared tracks\n");
switch_mutex_unlock(cb->mutex);
}
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "%s dub_session_cleanup: Bug is not attached.\n", switch_channel_get_name(channel));
return SWITCH_STATUS_FALSE;
}
switch_bool_t dub_speech_frame(switch_media_bug_t *bug, void* user_data) {
switch_core_session_t *session = switch_core_media_bug_get_session(bug);
struct cap_cb *cb = (struct cap_cb *) user_data;
if (switch_mutex_trylock(cb->mutex) == SWITCH_STATUS_SUCCESS) {
/* check if any tracks are actively pushing audio */
int trackCount = 0;
for (int i = 0; i < MAX_DUB_TRACKS; i++) {
if (cb->tracks[i].state == DUB_TRACK_STATE_ACTIVE) trackCount++;
}
if (trackCount == 0 && cb->gain == 0) {
switch_mutex_unlock(cb->mutex);
return SWITCH_TRUE;
}
switch_frame_t* rframe = switch_core_media_bug_get_write_replace_frame(bug);
int16_t *fp = reinterpret_cast<int16_t*>(rframe->data);
rframe->channels = 1;
rframe->datalen = rframe->samples * rframe->channels * sizeof(int16_t);
/* apply gain to audio in main channel if requested*/
if (cb->gain != 0) {
switch_change_sln_volume_granular(fp, rframe->samples, cb->gain);
}
/* now mux in the data from tracks */
for (int i = 0; i < rframe->samples; i++) {
int16_t input = fp[i];
int16_t value = input;
for (int j = 0; j < MAX_DUB_TRACKS; j++) {
dub_track_t* track = &cb->tracks[j];
if (track->state == DUB_TRACK_STATE_ACTIVE) {
CircularBuffer_t* buffer = reinterpret_cast<CircularBuffer_t*>(track->circularBuffer);
if (buffer && !buffer->empty()) {
int16_t sample = buffer->front();
buffer->pop_front();
value += sample;
}
}
}
switch_normalize_to_16bit(value);
fp[i] = (int16_t) value;
}
switch_core_media_bug_set_write_replace_frame(bug, rframe);
switch_mutex_unlock(cb->mutex);
}
return SWITCH_TRUE;
}
}
+16
View File
@@ -0,0 +1,16 @@
#ifndef __DUB_GLUE_H__
#define __DUB_GLUE_H__
switch_status_t dub_init();
switch_status_t dub_cleanup();
void init_dub_track(dub_track_t *track, char* trackName, int sampleRate);
switch_status_t silence_dub_track(dub_track_t *track);
switch_status_t remove_dub_track(dub_track_t *track);
switch_status_t play_dub_track(dub_track_t *track, switch_mutex_t *mutex, char* url, int loop, int gain);
switch_status_t say_dub_track(dub_track_t *track, switch_mutex_t *mutex, char* text, int gain);
switch_status_t dub_session_cleanup(switch_core_session_t *session, int channelIsClosing, switch_media_bug_t *bug);
switch_bool_t dub_speech_frame(switch_media_bug_t *bug, void* user_data);
#endif
+437
View File
@@ -0,0 +1,437 @@
#include "file_loader.h"
#include <boost/thread.hpp>
#include <boost/asio.hpp>
#include <boost/asio/ssl.hpp>
#include <boost/pool/object_pool.hpp>
#include <boost/bind/bind.hpp>
#include <boost/tokenizer.hpp>
#include <boost/foreach.hpp>
#include <boost/asio.hpp>
#include <boost/assign/list_of.hpp>
#include <boost/algorithm/string.hpp>
#include <map>
#include <mpg123.h>
#define INIT_BUFFER_SIZE (80000)
#define BUFFER_GROW_SIZE (80000)
#define BUFFER_THROTTLE_LOW (40000)
#define BUFFER_THROTTLE_HIGH (160000)
static uint16_t currDownloadId = 0;
typedef enum
{
STATUS_NONE = 0,
STATUS_FAILED,
STATUS_FILE_IN_PROGRESS,
STATUS_FILE_PAUSED,
STATUS_FILE_COMPLETE,
STATUS_AWAITING_RESTART,
STATUS_STOPPING,
STATUS_STOPPED
} Status_t;
typedef enum {
FILE_TYPE_MP3 = 0,
FILE_TYPE_R8
} FileType_t;
static const char* status2String(Status_t status)
{
static const char* statusStrings[] = {
"STATUS_NONE",
"STATUS_FAILED",
"STATUS_FILE_IN_PROGRESS",
"STATUS_FILE_PAUSED",
"STATUS_FILE_COMPLETE",
"STATUS_AWAITING_RESTART",
"STATUS_STOPPING",
"STATUS_STOPPED"
};
if (status >= 0 && status < sizeof(statusStrings) / sizeof(statusStrings[0]))
{
return statusStrings[status];
}
else
{
return "UNKNOWN_STATUS";
}
}
typedef struct
{
switch_mutex_t* mutex;
CircularBuffer_t* buffer;
mpg123_handle *mh;
FILE* fp;
char* path;
bool loop;
int rate;
boost::asio::deadline_timer *timer;
Status_t status;
FileType_t type;
downloadId_t id;
int gain;
} FileInfo_t;
typedef std::map<int32_t, FileInfo_t *> Id2FileMap_t;
static Id2FileMap_t id2FileMap;
static boost::object_pool<FileInfo_t> pool ;
static boost::asio::io_service io_service;
static std::thread worker_thread;
/* forward declarations */
static FileInfo_t* createFileLoader(const char *path, int rate, int loop, int gain, mpg123_handle *mhm, switch_mutex_t *mutex, CircularBuffer_t *buffer);
static void destroyFileInfo(FileInfo_t *finfo);
static void threadFunc();
static std::vector<int16_t> convert_mp3_to_linear(FileInfo_t *file, int8_t *data, size_t len);
static void read_cb(const boost::system::error_code& error, FileInfo_t* finfo) ;
/* apis */
extern "C" {
switch_status_t init_file_loader() {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "init_file_loader loading..\n");
if (mpg123_init() != MPG123_OK) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "init_file_loader: failed to initiate MPG123");
return SWITCH_STATUS_FALSE;
}
/* start worker thread */
std::thread t(threadFunc) ;
worker_thread.swap( t ) ;
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "init_file_loader: loaded\n");
return SWITCH_STATUS_SUCCESS;
}
switch_status_t deinit_file_loader() {
/* stop the ASIO IO service */
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "deinit_file_loader: stopping io service\n");
io_service.stop();
/* Join the worker thread */
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "deinit_file_loader: wait for worker thread to complete\n");
if (worker_thread.joinable()) {
worker_thread.join();
}
mpg123_exit();
return SWITCH_STATUS_SUCCESS;
}
downloadId_t start_file_load(const char* path, int rate, int loop, int gain, switch_mutex_t* mutex, CircularBuffer_t* buffer) {
int mhError = 0;
/* we only handle mp3 or r8 files atm */
const char *ext = strrchr(path, '.');
if (!ext) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "start_file_load: file %s has no extension\n", path);
return INVALID_DOWNLOAD_ID;
}
ext++;
if (0 != strcmp(ext, "mp3") && 0 != strcmp(ext, "r8")) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "start_file_load: file %s has unsupported extension %s\n", path, ext);
return INVALID_DOWNLOAD_ID;
}
/* allocate handle for mpeg decoding */
mpg123_handle *mh = mpg123_new("auto", &mhError);
if (!mh) {
const char *mhErr = mpg123_plain_strerror(mhError);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error allocating mpg123 handle! %s\n", switch_str_nil(mhErr));
return INVALID_DOWNLOAD_ID;
}
if (mpg123_open_feed(mh) != MPG123_OK) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error mpg123_open_feed!\n");
return INVALID_DOWNLOAD_ID;
}
if (mpg123_format_all(mh) != MPG123_OK) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error mpg123_format_all!\n");
return INVALID_DOWNLOAD_ID;
}
if (mpg123_param(mh, MPG123_FORCE_RATE, rate, 0) != MPG123_OK) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error forcing resample to 8k!\n");
return INVALID_DOWNLOAD_ID;
}
if (mpg123_param(mh, MPG123_FLAGS, MPG123_MONO_MIX, 0) != MPG123_OK) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error forcing single channel!\n");
return INVALID_DOWNLOAD_ID;
}
FileInfo_t* finfo = createFileLoader(path, rate, loop, gain, mh, mutex, buffer);
if (!finfo) {
return INVALID_DOWNLOAD_ID;
}
/* do the initial read in the worker thread so we don't block here */
finfo->timer->expires_from_now(boost::posix_time::millisec(1));
finfo->timer->async_wait(boost::bind(&read_cb, boost::placeholders::_1, finfo));
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,
"start_file_load: starting load %d\n", finfo->id);
return finfo->id;
}
switch_status_t stop_file_load(int id) {
auto it = id2FileMap.find(id);
if (it == id2FileMap.end()) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "stop_file_load: id %d has already completed\n", id);
return SWITCH_STATUS_FALSE;
}
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,
"stop_audio_download: stopping download %d, status %s\n", id, status2String(it->second->status));
FileInfo_t *finfo = it->second;
auto status = finfo->status;
/* past this point I shall not access either the mutex or the buffer provided */
finfo->mutex = nullptr;
finfo->buffer = nullptr;
destroyFileInfo(finfo);
finfo->status = Status_t::STATUS_STOPPED;
return SWITCH_STATUS_SUCCESS;
}
}
/* internal */
FileInfo_t* createFileLoader(const char *path, int rate, int loop, int gain, mpg123_handle *mh, switch_mutex_t *mutex, CircularBuffer_t *buffer) {
FileInfo_t *finfo = pool.malloc() ;
const char *ext = strrchr(path, '.');
memset(finfo, 0, sizeof(FileInfo_t));
finfo->mutex = mutex;
finfo->buffer = buffer;
finfo->mh = mh;
finfo->loop = loop;
finfo->gain = gain;
finfo->rate = rate;
finfo->path = strdup(path);
finfo->status = Status_t::STATUS_NONE;
finfo->timer = new boost::asio::deadline_timer(io_service);
if (0 == strcmp(ext, "mp3")) finfo->type = FileType_t::FILE_TYPE_MP3;
else if (0 == strcmp(ext, "r8")) finfo->type = FileType_t::FILE_TYPE_R8;
downloadId_t id = ++currDownloadId;
if (id == 0) id++;
id2FileMap[id] = finfo;
finfo->id = id;
finfo->status = Status_t::STATUS_AWAITING_RESTART;
finfo->fp = fopen(finfo->path, "rb");
if (finfo->fp == NULL) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "createFileLoader: failed to open file %s\n", finfo->path);
destroyFileInfo(finfo);
return nullptr;
}
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG,
"createFileLoader: launched request, loop %s, gain %d\n", (finfo->loop ? "yes": "no"), finfo->gain);
return finfo;
}
void destroyFileInfo(FileInfo_t *finfo) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "destroyFileInfo\n");
/* clear asio resources and free resources */
if (finfo->timer) {
finfo->timer->cancel();
delete finfo->timer;
}
/* free mp3 decoder */
if (finfo->mh) {
mpg123_close(finfo->mh);
mpg123_delete(finfo->mh);
}
if (finfo->path) {
free(finfo->path);
}
if (finfo->mutex) switch_mutex_lock(finfo->mutex);
id2FileMap.erase(finfo->id);
if (finfo->mutex) switch_mutex_unlock(finfo->mutex);
memset(finfo, 0, sizeof(FileInfo_t));
pool.destroy(finfo) ;
}
void threadFunc() {
/* to make sure the event loop doesn't terminate when there is no work to do */
io_service.reset() ;
boost::asio::io_service::work work(io_service);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "file_loader threadFunc - starting\n");
for(;;) {
try {
io_service.run() ;
break ;
}
catch( std::exception& e) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "file_loader threadFunc - Error: %s\n", e.what());
}
}
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "file_loader threadFunc - ending\n");
}
std::vector<int16_t> convert_mp3_to_linear(FileInfo_t *finfo, int8_t *data, size_t len) {
std::vector<int16_t> linear_data;
int eof = 0;
int mp3err = 0;
if(mpg123_feed(finfo->mh, (const unsigned char*) data, len) == MPG123_OK) {
while(!eof) {
size_t usedlen = 0;
off_t frame_offset;
unsigned char* audio;
int decode_status = mpg123_decode_frame(finfo->mh, &frame_offset, &audio, &usedlen);
switch(decode_status) {
case MPG123_NEW_FORMAT:
continue;
case MPG123_OK:
{
size_t samples = usedlen / sizeof(int16_t);
linear_data.insert(linear_data.end(), reinterpret_cast<int16_t*>(audio), reinterpret_cast<int16_t*>(audio) + samples);
}
break;
case MPG123_DONE:
case MPG123_NEED_MORE:
eof = 1;
break;
case MPG123_ERR:
default:
if(++mp3err >= 5) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Decoder Error!\n");
eof = 1;
}
}
if (eof)
break;
mp3err = 0;
}
if (finfo->gain != 0) {
switch_change_sln_volume_granular(linear_data.data(), linear_data.size(), finfo->gain);
}
}
return linear_data;
}
void read_cb(const boost::system::error_code& error, FileInfo_t* finfo) {
if (finfo->status == Status_t::STATUS_STOPPING || !finfo->mutex || !finfo->buffer) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "read_cb: %u session gone\n", finfo->id);
return;
}
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "read_cb: %u status is %s\n", finfo->id, status2String(finfo->status));
if (finfo->status == Status_t::STATUS_AWAITING_RESTART) {
finfo->status = Status_t::STATUS_FILE_IN_PROGRESS;
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "read_cb: %u starting initial read of file\n", finfo->id);
}
if (!error) {
size_t size = 0;
switch_mutex_lock(finfo->mutex);
size = finfo->buffer->size();
switch_mutex_unlock(finfo->mutex);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "read_cb: %u size is now %ld\n", finfo->id, size);
if (size < BUFFER_THROTTLE_LOW) {
std::vector<int16_t> pcm_data;
int8_t buf[INIT_BUFFER_SIZE];
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "read_cb: %u reading data\n", finfo->id);
size_t bytesRead = ::fread(buf, sizeof(int8_t), INIT_BUFFER_SIZE, finfo->fp);
if (bytesRead <= 0) {
if (::feof(finfo->fp)) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "read_cb: %u eof\n", finfo->id);
}
else if (::ferror(finfo->fp)) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "read_cb: %u error reading file\n", finfo->id);
}
else {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "read_cb: %u unknown error reading file\n", finfo->id);
}
finfo->status = Status_t::STATUS_FILE_COMPLETE;
return;
}
if (finfo->type == FileType_t::FILE_TYPE_MP3) {
pcm_data = convert_mp3_to_linear(finfo, buf, bytesRead);
} else {
pcm_data = std::vector<int16_t>(reinterpret_cast<int16_t*>(buf), reinterpret_cast<int16_t*>(buf) + bytesRead / 2);
}
switch_mutex_lock(finfo->mutex);
// Resize the buffer if necessary
if (finfo->buffer->capacity() - finfo->buffer->size() < pcm_data.size()) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "write_cb %u growing buffer, size now %ld\n", finfo->id, finfo->buffer->size());
finfo->buffer->set_capacity(finfo->buffer->size() + std::max(pcm_data.size(), (size_t)BUFFER_GROW_SIZE));
}
/* Push the data into the buffer */
finfo->buffer->insert(finfo->buffer->end(), pcm_data.data(), pcm_data.data() + pcm_data.size());
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "read_cb: %u wrote data, buffer size is now %ld\n", finfo->id, finfo->buffer->size());
switch_mutex_unlock(finfo->mutex);
if (bytesRead < INIT_BUFFER_SIZE) {
finfo->status = Status_t::STATUS_FILE_COMPLETE;
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "read_cb: %u reached end of file, status is %s\n", finfo->id, status2String(finfo->status));
}
}
if (finfo->status == Status_t::STATUS_FILE_COMPLETE && finfo->loop) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "read_cb: %u looping\n", finfo->id);
::fseek(finfo->fp, 0, SEEK_SET);
finfo->status = Status_t::STATUS_AWAITING_RESTART;
}
if (finfo->status != Status_t::STATUS_FILE_COMPLETE) {
// read more in 2 seconds
finfo->timer->expires_from_now(boost::posix_time::millisec(2000));
finfo->timer->async_wait(boost::bind(&read_cb, boost::placeholders::_1, finfo));
}
else {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "read_cb: %u file complete, status %s loop %s\n",
finfo->id, status2String(finfo->status), (finfo->loop ? "yes" : "no"));
}
} else {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "read_cb: %u error (%d): %s\n", finfo->id, error.value(), error.message().c_str());
// Handle any errors
}
}
+18
View File
@@ -0,0 +1,18 @@
#ifndef __FILE_LOADER_H__
#define __FILE_LOADER_H__
#include <switch.h>
#include "common.h"
extern "C" {
switch_status_t init_file_loader();
switch_status_t deinit_file_loader();
int start_file_load(const char* path, int rate, int loop, int gain, switch_mutex_t* mutex, CircularBuffer_t* buffer);
switch_status_t stop_file_load(int id);
}
#endif
+383
View File
@@ -0,0 +1,383 @@
/*
*
* mod_dub.c
*
*/
#include "mod_dub.h"
#include <stdlib.h>
#include <switch.h>
#include <switch_curl.h>
#include "dub_glue.h"
/* Prototypes */
SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_dub_shutdown);
SWITCH_MODULE_LOAD_FUNCTION(mod_dub_load);
SWITCH_MODULE_DEFINITION(mod_dub, mod_dub_load, mod_dub_shutdown, NULL);
static switch_bool_t capture_callback(switch_media_bug_t *bug, void *user_data, switch_abc_type_t type)
{
switch_core_session_t *session = switch_core_media_bug_get_session(bug);
switch_bool_t ret = SWITCH_TRUE;
switch (type) {
case SWITCH_ABC_TYPE_INIT:
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "Got SWITCH_ABC_TYPE_INIT.\n");
break;
case SWITCH_ABC_TYPE_CLOSE:
{
dub_session_cleanup(session, 1, bug);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "Finished SWITCH_ABC_TYPE_CLOSE.\n");
}
break;
case SWITCH_ABC_TYPE_WRITE_REPLACE:
ret = dub_speech_frame(bug, user_data);
break;
default:
break;
}
return ret;
}
static switch_status_t dub_set_gain(switch_core_session_t *session, int gain) {
switch_channel_t *channel = switch_core_session_get_channel(session);
switch_media_bug_t *bug = NULL;
struct cap_cb *cb = NULL;
if (switch_channel_pre_answer(channel) != SWITCH_STATUS_SUCCESS) {
return SWITCH_STATUS_FALSE;
}
if (!(bug = (switch_media_bug_t*) switch_channel_get_private(channel, MY_BUG_NAME))) {
cb =(struct cap_cb *) switch_core_session_alloc(session, sizeof(struct cap_cb));
memset(cb, 0, sizeof(struct cap_cb));
switch_mutex_init(&cb->mutex, SWITCH_MUTEX_NESTED, switch_core_session_get_pool(session));
}
else {
cb = (struct cap_cb *) switch_core_media_bug_get_user_data(bug);
}
cb->gain = gain;
/* check again under lock */
switch_mutex_lock(cb->mutex);
if (!(bug = (switch_media_bug_t*) switch_channel_get_private(channel, MY_BUG_NAME))) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "dub_set_gain: adding bug so we can set gain on main channel\n");
if (switch_core_media_bug_add(session, MY_BUG_NAME, NULL, capture_callback, (void *) cb, 0, SMBF_WRITE_REPLACE, &bug) != SWITCH_STATUS_SUCCESS) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "dub_set_gain: error adding bug!\n");
switch_mutex_unlock(cb->mutex);
return SWITCH_STATUS_FALSE;
}
switch_channel_set_private(channel, MY_BUG_NAME, bug);
}
else {
cb = (struct cap_cb *) switch_core_media_bug_get_user_data(bug);
cb->gain = gain;
}
switch_mutex_unlock(cb->mutex);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "dub_set_gain: setting gain to %d\n", gain);
return SWITCH_STATUS_SUCCESS;
}
static switch_status_t dub_add_track(switch_core_session_t *session, char* trackName) {
switch_channel_t *channel = switch_core_session_get_channel(session);
switch_media_bug_t *bug = NULL;
struct cap_cb *cb = NULL;
int offset = 0;
int samples_per_second;
switch_codec_implementation_t write_impl = { 0 };
if (switch_channel_pre_answer(channel) != SWITCH_STATUS_SUCCESS) {
return SWITCH_STATUS_FALSE;
}
if (!(bug = (switch_media_bug_t*) switch_channel_get_private(channel, MY_BUG_NAME))) {
/* allocate per-session memory and use track 0 */
cb =(struct cap_cb *) switch_core_session_alloc(session, sizeof(struct cap_cb));
memset(cb, 0, sizeof(struct cap_cb));
switch_mutex_init(&cb->mutex, SWITCH_MUTEX_NESTED, switch_core_session_get_pool(session));
offset = 0;
}
else {
/* retrieve the bug and search for an empty track */
cb = (struct cap_cb *) switch_core_media_bug_get_user_data(bug);
while (offset < MAX_DUB_TRACKS && cb->tracks[offset].state != DUB_TRACK_STATE_INACTIVE) {
offset++;
}
if (offset == MAX_DUB_TRACKS) {
/* all tracks are in use */
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "dub_add_track: no available tracks\n");
return SWITCH_STATUS_FALSE;
}
}
switch_core_session_get_write_impl(session, &write_impl);
samples_per_second = !strcasecmp(write_impl.iananame, "g722") ? write_impl.actual_samples_per_second : write_impl.samples_per_second;
init_dub_track(&cb->tracks[offset], trackName, samples_per_second);
if (!bug) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "dub_add_track: adding bug for track %s\n", trackName);
if (switch_core_media_bug_add(session, MY_BUG_NAME, NULL, capture_callback, (void *) cb, 0, SMBF_WRITE_REPLACE, &bug) != SWITCH_STATUS_SUCCESS) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "dub_add_track: error adding bug!\n");
return SWITCH_STATUS_FALSE;
}
switch_channel_set_private(channel, MY_BUG_NAME, bug);
}
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "dub_add_track: added track %s at offset %d\n", trackName, offset);
return SWITCH_STATUS_SUCCESS;
}
static switch_status_t dub_remove_track(switch_core_session_t *session, char* trackName) {
switch_channel_t *channel = switch_core_session_get_channel(session);
switch_media_bug_t *bug = (switch_media_bug_t*) switch_channel_get_private(channel, MY_BUG_NAME);
switch_status_t status = SWITCH_STATUS_FALSE;
dub_track_t *track = NULL;
if (bug) {
struct cap_cb *cb =(struct cap_cb *) switch_core_media_bug_get_user_data(bug);
switch_mutex_lock(cb->mutex);
for (int i = 0; i < MAX_DUB_TRACKS && track == NULL; i++) {
if (cb->tracks[i].state != DUB_TRACK_STATE_INACTIVE && strcmp(cb->tracks[i].trackName, trackName) == 0) {
track = &cb->tracks[i];
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "dub_remove_track: removing track %s at offset %d\n", trackName, i);
break;
}
}
if (track) {
int count = 0;
remove_dub_track(track);
/* check if this is the last bug */
for (int i = 0; i < MAX_DUB_TRACKS; i++) {
if (cb->tracks[i].state != DUB_TRACK_STATE_INACTIVE) count++;
}
if (count == 0) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "dub_remove_track: removing bug after removing last track\n");
dub_session_cleanup(session, 0, bug);
}
status = SWITCH_STATUS_SUCCESS;
}
else {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "dub_remove_track: track %s not found\n", trackName);
}
switch_mutex_unlock(cb->mutex);
}
else {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "dub_remove_track: bug not found\n");
}
return status;
}
static switch_status_t dub_silence_track(switch_core_session_t *session, char* trackName) {
switch_channel_t *channel = switch_core_session_get_channel(session);
switch_media_bug_t *bug = (switch_media_bug_t*) switch_channel_get_private(channel, MY_BUG_NAME);
switch_status_t status = SWITCH_STATUS_FALSE;
dub_track_t *track = NULL;
if (bug) {
struct cap_cb *cb =(struct cap_cb *) switch_core_media_bug_get_user_data(bug);
switch_mutex_lock(cb->mutex);
for (int i = 0; i < MAX_DUB_TRACKS; i++) {
if (cb->tracks[i].state != DUB_TRACK_STATE_INACTIVE && strcmp(cb->tracks[i].trackName, trackName) == 0) {
track = &cb->tracks[i];
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "dub_silence_track: silencing track %s at offset %d\n", trackName, i);
break;
}
}
if (track) {
silence_dub_track(track);
status = SWITCH_STATUS_SUCCESS;
}
switch_mutex_unlock(cb->mutex);
}
else {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "dub_silence_track: bug not found\n");
}
return status;
}
static switch_status_t dub_play_on_track(switch_core_session_t *session, char* trackName, char* url, int loop, int gain) {
switch_channel_t *channel = switch_core_session_get_channel(session);
switch_media_bug_t *bug = (switch_media_bug_t*) switch_channel_get_private(channel, MY_BUG_NAME);
switch_status_t status = SWITCH_STATUS_FALSE;
dub_track_t *track = NULL;
if (bug) {
struct cap_cb *cb =(struct cap_cb *) switch_core_media_bug_get_user_data(bug);
switch_mutex_lock(cb->mutex);
for (int i = 0; i < MAX_DUB_TRACKS; i++) {
if (cb->tracks[i].state != DUB_TRACK_STATE_INACTIVE && strcmp(cb->tracks[i].trackName, trackName) == 0) {
track = &cb->tracks[i];
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,
"dub_play_on_track: playing %s on track %s at offset %d with gain %d\n", url, trackName, i, gain);
break;
}
}
if (track) {
status = play_dub_track(track, cb->mutex, url, loop, gain);
}
switch_mutex_unlock(cb->mutex);
}
return status;
}
static switch_status_t dub_say_on_track(switch_core_session_t *session, char* trackName, char* text, int gain) {
switch_channel_t *channel = switch_core_session_get_channel(session);
switch_media_bug_t *bug = (switch_media_bug_t*) switch_channel_get_private(channel, MY_BUG_NAME);
switch_status_t status = SWITCH_STATUS_FALSE;
dub_track_t *track = NULL;
if (bug) {
struct cap_cb *cb =(struct cap_cb *) switch_core_media_bug_get_user_data(bug);
switch_mutex_lock(cb->mutex);
for (int i = 0; i < MAX_DUB_TRACKS; i++) {
if (cb->tracks[i].state != DUB_TRACK_STATE_INACTIVE && strcmp(cb->tracks[i].trackName, trackName) == 0) {
track = &cb->tracks[i];
break;
}
}
if (track) {
status = say_dub_track(track, cb->mutex, text, gain);
}
switch_mutex_unlock(cb->mutex);
}
return status;
}
#define DUB_API_SYNTAX "<uuid> [addTrack|removeTrack|silenceTrack|playOnTrack|sayOnTrack|setGain] track [url|text|gain] [gain] [loop]"
SWITCH_STANDARD_API(dub_function)
{
char *mycmd = NULL, *argv[6] = { 0 };
int argc = 0;
int error_written = 0;
switch_status_t status = SWITCH_STATUS_FALSE;
if (!zstr(cmd) && (mycmd = strdup(cmd))) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "dub_function: %s\n", mycmd);
argc = switch_separate_string(mycmd, ' ', argv, (sizeof(argv) / sizeof(argv[0])));
}
if (zstr(cmd) || argc < 3 || zstr(argv[0]) || zstr(argv[1]) || zstr(argv[2])) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error with command %s.\n", cmd);
}
else {
switch_core_session_t *session = NULL;
if ((session = switch_core_session_locate(argv[0]))) {
char* action = argv[1];
char* track = argv[2];
if (0 == strcmp(action, "setGain")) {
int gain = atoi(track);
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "setGain %d\n", gain);
status = dub_set_gain(session, gain);
}
else if (0 == strcmp(action, "addTrack")) {
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "addTrack %s\n", track);
status = dub_add_track(session, track);
}
else if (0 == strcmp(action, "removeTrack")) {
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "removeTrack %s\n", track);
status = dub_remove_track(session, track);
}
else if (0 == strcmp(action, "silenceTrack")) {
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "silenceTrack %s\n", track);
status = dub_silence_track(session, track);
}
else if (0 == strcmp(action, "playOnTrack")) {
if (argc < 4) {
stream->write_function(stream, "-USAGE: %s\n", DUB_API_SYNTAX);
error_written = 1;
}
else {
char* url = argv[3];
int loop = argc > 4 && 0 == strcmp(argv[4], "loop");
int gain = argc > 5 ? atoi(argv[5]) : 0;
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO,
"playOnTrack %s, %s gain %d\n", url, loop ? "loop" : "once", gain);
status = dub_play_on_track(session, track, url, loop, gain);
}
}
else if (0 == strcmp(action, "sayOnTrack")) {
if (argc < 4) {
stream->write_function(stream, "-USAGE: %s\n", DUB_API_SYNTAX);
error_written = 1;
}
else {
char* text = argv[3];
int gain = argc > 4 ? atoi(argv[4]) : 0;
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "sayOnTrack %s gain %d\n", text, gain);
status = dub_say_on_track(session, track, text, gain);
}
}
else {
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Error with command %s %s %s.\n", cmd, argv[0], argv[1]);
stream->write_function(stream, "-USAGE: %s\n", DUB_API_SYNTAX);
error_written = 1;
}
switch_core_session_rwunlock(session);
}
else {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error locating session for %s.\n", cmd);
stream->write_function(stream, "-ERR Invalid session\n");
error_written = 1;
}
}
if (status == SWITCH_STATUS_SUCCESS) {
stream->write_function(stream, "+OK Success\n");
} else if (!error_written){
stream->write_function(stream, "-ERR Operation Failed\n");
}
switch_safe_free(mycmd);
return SWITCH_STATUS_SUCCESS;
}
SWITCH_MODULE_LOAD_FUNCTION(mod_dub_load)
{
switch_api_interface_t *api_interface;
*module_interface = switch_loadable_module_create_module_interface(pool, modname);
dub_init();
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "mod_dub loaded\n");
SWITCH_ADD_API(api_interface, "uuid_dub", "dub mp3 track over channel audio", dub_function, DUB_API_SYNTAX);
switch_console_set_complete("add uuid_dub addTrack <trackname>");
switch_console_set_complete("add uuid_dub removeTrack <trackname>");
switch_console_set_complete("add uuid_dub silenceTrack <trackname>");
switch_console_set_complete("add uuid_dub playOnTrack <trackname> <url|file> [loop|once] [gain]");
switch_console_set_complete("add uuid_dub setGain <gain>");
switch_console_set_complete("add uuid_dub stop ");
/* indicate that the module should continue to be loaded */
return SWITCH_STATUS_SUCCESS;
}
/*
Called when the system shuts down
Macro expands to: switch_status_t mod_dub_shutdown() */
SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_dub_shutdown)
{
dub_cleanup();
return SWITCH_STATUS_SUCCESS;
}
+59
View File
@@ -0,0 +1,59 @@
#ifndef __MOD_DUB_H__
#define __MOD_DUB_H__
#include <switch.h>
#include <speex/speex_resampler.h>
#include <mpg123.h>
#include <unistd.h>
#define MY_BUG_NAME "_dub_"
#define MAX_SESSION_ID (256)
#define MAX_BUG_LEN (64)
#define MAX_URL_LEN (1024)
#define MAX_DUB_TRACKS (2)
/* per-channel data */
typedef void (*responseHandler_t)(switch_core_session_t* session, const char* json, const char* bugname, const char* details);
typedef enum {
DUB_TRACK_STATE_INACTIVE = 0,
DUB_TRACK_STATE_READY,
DUB_TRACK_STATE_ACTIVE,
DUB_TRACK_STATE_PAUSED
} dub_state_t;
typedef enum {
DUB_GENERATOR_TYPE_UNKNOWN = 0,
DUB_GENERATOR_TYPE_HTTP,
DUB_GENERATOR_TYPE_FILE,
DUB_GENERATOR_TYPE_TTS
} dub_generator_t;
typedef enum {
DUB_TRACK_EVENT_PLAY = 0,
DUB_TRACK_EVENT_STOP,
DUB_TRACK_EVENT_PAUSE,
DUB_TRACK_EVENT_RESUME
} dub_event_t;
typedef struct dub_track {
dub_state_t state;
dub_generator_t generator;
char* trackName;
int sampleRate;
int gain;
void* circularBuffer;
int generatorId;
} dub_track_t;
struct cap_cb {
switch_mutex_t *mutex;
int gain;
dub_track_t tracks[MAX_DUB_TRACKS];
};
#endif
+51 -4
View File
@@ -29,8 +29,10 @@
#include <boost/asio.hpp>
#include <boost/assign/list_of.hpp>
#include <boost/algorithm/string.hpp>
#include <boost/unordered_map.hpp>
#include "mod_elevenlabs_tts.h"
#include <speex/speex_resampler.h>
#define TXNID_LEN (255)
#define URL_LEN (1024)
@@ -820,6 +822,7 @@ extern "C" {
CURL* easy = createEasyHandle();
el->conn = (void *) conn ;
el->sample_rate = 0;
conn->elevenlabs = el;
conn->easy = easy;
conn->global = &global;
@@ -830,6 +833,23 @@ extern "C" {
el->circularBuffer = (void *) new CircularBuffer_t(8192);
if (el->session_id) {
int err;
switch_codec_implementation_t read_impl;
switch_core_session_t *psession = switch_core_session_locate(el->session_id);
switch_core_session_get_read_impl(psession, &read_impl);
uint32_t samples_per_second = !strcasecmp(read_impl.iananame, "g722") ? read_impl.actual_samples_per_second : read_impl.samples_per_second;
el->sample_rate = samples_per_second;
// elevenlabs output is PCMU 8000
if (samples_per_second != 8000 /*Hz*/) {
el->resampler = speex_resampler_init(1, 8000, samples_per_second, SWITCH_RESAMPLE_QUALITY, &err);
if (0 != err) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error initializing resampler: %s.\n", speex_resampler_strerror(err));
return SWITCH_STATUS_FALSE;
}
}
}
std::ostringstream api_key_stream;
api_key_stream << "xi-api-key: " << el->api_key;
@@ -881,7 +901,6 @@ extern "C" {
{
switch_mutex_lock(el->mutex);
ConnInfo_t *conn = (ConnInfo_t *) el->conn;
if (el->response_code > 0 && el->response_code != 200) {
switch_mutex_unlock(el->mutex);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "elevenlabs_speech_read_tts, returning failure\n") ;
@@ -901,14 +920,35 @@ extern "C" {
switch_mutex_unlock(el->mutex);
return SWITCH_STATUS_SUCCESS;
}
size_t size = std::min((*datalen/2), cBuffer->size());
size_t size = el->sample_rate ?
std::min((*datalen/(2 * el->sample_rate / 8000)), cBuffer->size()) :
std::min((*datalen/2), cBuffer->size());
pcm_data.insert(pcm_data.end(), cBuffer->begin(), cBuffer->begin() + size);
cBuffer->erase(cBuffer->begin(), cBuffer->begin() + size);
switch_mutex_unlock(el->mutex);
}
memcpy(data, pcm_data.data(), pcm_data.size() * sizeof(uint16_t));
*datalen = pcm_data.size() * sizeof(uint16_t);
size_t data_size = pcm_data.size();
if (el->resampler) {
std::vector<int16_t> in(pcm_data.begin(), pcm_data.end());
std::vector<int16_t> out((*datalen));
spx_uint32_t in_len = data_size;
spx_uint32_t out_len = out.size();
speex_resampler_process_interleaved_int(el->resampler, in.data(), &in_len, out.data(), &out_len);
if (out_len > out.size()) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Resampler output exceeded maximum buffer size!\n");
return SWITCH_STATUS_FALSE;
}
memcpy(data, out.data(), out_len * sizeof(int16_t));
*datalen = out_len * sizeof(int16_t);
} else {
memcpy(data, pcm_data.data(), pcm_data.size() * sizeof(uint16_t));
*datalen = pcm_data.size() * sizeof(uint16_t);
}
return SWITCH_STATUS_SUCCESS;
}
@@ -922,8 +962,15 @@ extern "C" {
delete cBuffer;
el->circularBuffer = nullptr ;
// destroy resampler
if (el->resampler) {
speex_resampler_destroy(el->resampler);
el->resampler = NULL;
}
if (conn) {
conn->flushed = true;
if (!download_complete) {
if (conn->file) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "closing audio cache file %s because download was interrupted\n", el->cache_filename);
+3
View File
@@ -5,6 +5,7 @@
#include <fcntl.h>
#include <sys/stat.h>
#include <unistd.h>
#include <speex/speex_resampler.h>
struct elevenlabs_data {
char *session_id;
@@ -30,6 +31,7 @@ struct elevenlabs_data {
char *cache_filename;
int rate;
uint32_t sample_rate;
void *conn;
FILE *file;
@@ -38,6 +40,7 @@ struct elevenlabs_data {
int draining;
int reads;
int cache_audio;
SpeexResamplerState *resampler;
};
typedef struct elevenlabs_data elevenlabs_t;
-5
View File
@@ -1,5 +0,0 @@
Copyright 2023, Drachtio Communications Services, LLC
This software is provided under a dual-licensing scheme, described here: https://github.com/jambonz/freeswitch-modules/blob/main/COPYING.
Please refer to the above file and the links therein for the complete terms and conditions of each license. Your use of this software constitutes agreement to the terms of these licenses. If you have any questions regarding the licensing of this software, please consult a legal expert.
-5
View File
@@ -1,5 +0,0 @@
Copyright 2023, Drachtio Communications Services, LLC
This software is provided under a dual-licensing scheme, described here: https://github.com/jambonz/freeswitch-modules/blob/main/COPYING.
Please refer to the above file and the links therein for the complete terms and conditions of each license. Your use of this software constitutes agreement to the terms of these licenses. If you have any questions regarding the licensing of this software, please consult a legal expert.
-5
View File
@@ -1,5 +0,0 @@
Copyright 2023, Drachtio Communications Services, LLC
This software is provided under a dual-licensing scheme, described here: https://github.com/jambonz/freeswitch-modules/blob/main/COPYING.
Please refer to the above file and the links therein for the complete terms and conditions of each license. Your use of this software constitutes agreement to the terms of these licenses. If you have any questions regarding the licensing of this software, please consult a legal expert.
-5
View File
@@ -1,5 +0,0 @@
Copyright 2023, Drachtio Communications Services, LLC
This software is provided under a dual-licensing scheme, described here: https://github.com/jambonz/freeswitch-modules/blob/main/COPYING.
Please refer to the above file and the links therein for the complete terms and conditions of each license. Your use of this software constitutes agreement to the terms of these licenses. If you have any questions regarding the licensing of this software, please consult a legal expert.
-5
View File
@@ -1,5 +0,0 @@
Copyright 2023, Drachtio Communications Services, LLC
This software is provided under a dual-licensing scheme, described here: https://github.com/jambonz/freeswitch-modules/blob/main/COPYING.
Please refer to the above file and the links therein for the complete terms and conditions of each license. Your use of this software constitutes agreement to the terms of these licenses. If you have any questions regarding the licensing of this software, please consult a legal expert.
-5
View File
@@ -1,5 +0,0 @@
Copyright 2023, Drachtio Communications Services, LLC
This software is provided under a dual-licensing scheme, described here: https://github.com/jambonz/freeswitch-modules/blob/main/COPYING.
Please refer to the above file and the links therein for the complete terms and conditions of each license. Your use of this software constitutes agreement to the terms of these licenses. If you have any questions regarding the licensing of this software, please consult a legal expert.
+8
View File
@@ -0,0 +1,8 @@
include $(top_srcdir)/build/modmake.rulesam
MODNAME=mod_whisper_tts
mod_LTLIBRARIES = mod_whisper_tts.la
mod_whisper_tts_la_SOURCES = mod_whisper_tts.c whisper_glue.cpp
mod_whisper_tts_la_CFLAGS = $(AM_CFLAGS)
mod_whisper_tts_la_LIBADD = $(switch_builddir)/libfreeswitch.la
mod_whisper_tts_la_LDFLAGS = -avoid-version -module -no-undefined -shared `pkg-config --libs boost` -lstdc++ -lmpg123
+3
View File
@@ -0,0 +1,3 @@
# mod_whisper_tts
A Freeswitch module that allows speak text to speech audio from whisper stream.
+164
View File
@@ -0,0 +1,164 @@
#include "mod_whisper_tts.h"
#include "whisper_glue.h"
SWITCH_MODULE_LOAD_FUNCTION(mod_whisper_tts_load);
SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_whisper_tts_shutdown);
SWITCH_MODULE_DEFINITION(mod_whisper_tts, mod_whisper_tts_load, mod_whisper_tts_shutdown, NULL);
static void clearWhisper(whisper_t* w, int freeAll) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "clearWhisper\n");
if (w->api_key) free(w->api_key);
if (w->model_id) free(w->model_id);
if (w->speed) free(w->speed);
if (w->request_id) free(w->request_id);
if (w->reported_latency) free(w->reported_latency);
if (w->ct) free(w->ct);
if (w->err_msg) free(w->err_msg);
if (w->name_lookup_time_ms) free(w->name_lookup_time_ms);
if (w->connect_time_ms) free(w->connect_time_ms);
if (w->final_response_time_ms) free(w->final_response_time_ms);
if (w->cache_filename) free(w->cache_filename);
w->api_key = NULL;
w->model_id = NULL;
w->speed = NULL;
w->request_id = NULL;
w->reported_latency = NULL;
w->ct = NULL;
w->err_msg = NULL;
w->name_lookup_time_ms = NULL;
w->connect_time_ms = NULL;
w->final_response_time_ms = NULL;
w->cache_filename = NULL;
if (freeAll) {
if (w->voice_name) free(w->voice_name);
if (w->session_id) free(w->session_id);
w->voice_name = NULL;
w->session_id = NULL;
}
}
static whisper_t * createOrRetrievePrivateData(switch_speech_handle_t *sh) {
whisper_t *w = (whisper_t *) sh->private_info;
if (!w) {
w = switch_core_alloc(sh->memory_pool, sizeof(*w));
sh->private_info = w;
memset(w, 0, sizeof(*w));
switch_mutex_init(&w->mutex, SWITCH_MUTEX_NESTED, sh->memory_pool);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "allocated whisper_t\n");
}
return w;
}
switch_status_t w_speech_open(switch_speech_handle_t *sh, const char *voice_name, int rate, int channels, switch_speech_flag_t *flags)
{
whisper_t *w = createOrRetrievePrivateData(sh);
w->voice_name = strdup(voice_name);
w->rate = rate;
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "w_speech_open voice: %s, rate %d, channels %d\n", voice_name, rate, channels);
return whisper_speech_open(w);
}
static switch_status_t w_speech_close(switch_speech_handle_t *sh, switch_speech_flag_t *flags)
{
switch_status_t rc;
whisper_t *w = createOrRetrievePrivateData(sh);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "w_speech_close\n");
switch_mutex_destroy(w->mutex);
rc = whisper_speech_close(w);
clearWhisper(w, 1);
return rc;
}
/**
* Freeswitch will call this function to feed us text to speak
*/
static switch_status_t w_speech_feed_tts(switch_speech_handle_t *sh, char *text, switch_speech_flag_t *flags)
{
whisper_t *w = createOrRetrievePrivateData(sh);
w->draining = 0;
w->reads = 0;
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "w_speech_feed_tts\n");
return whisper_speech_feed_tts(w, text, flags);
}
/**
* Freeswitch calls periodically to get some rendered audio in L16 format. We can provide up to 8k of audio at a time.
*/
static switch_status_t w_speech_read_tts(switch_speech_handle_t *sh, void *data, size_t *datalen, switch_speech_flag_t *flags)
{
whisper_t *w = createOrRetrievePrivateData(sh);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "w_speech_read_tts\n");
return whisper_speech_read_tts(w, data, datalen, flags);
}
/**
* This is called at the end, not sure exactly what we need to do here..
*/
static void w_speech_flush_tts(switch_speech_handle_t *sh)
{
whisper_t *w = createOrRetrievePrivateData(sh);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "w_speech_flush_tts\n");
whisper_speech_flush_tts(w);
clearWhisper(w, 0);
}
static void w_text_param_tts(switch_speech_handle_t *sh, char *param, const char *val)
{
whisper_t *w = createOrRetrievePrivateData(sh);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "w_text_param_tts: %s=%s\n", param, val);
if (0 == strcmp(param, "api_key")) {
if (w->api_key) free(w->api_key);
w->api_key = strdup(val);
} else if (0 == strcmp(param, "voice")) {
if (w->voice_name) free(w->voice_name);
w->voice_name = strdup(val);
} else if (0 == strcmp(param, "model_id")) {
if (w->model_id) free(w->model_id);
w->model_id = strdup(val);
} else if (0 == strcmp(param, "speed")) {
if (w->speed) free(w->speed);
w->speed = strdup(val);
} else if (0 == strcmp(param, "session-uuid")) {
if (w->session_id) free(w->session_id);
w->session_id = strdup(val);
} else if (0 == strcmp(param, "write_cache_file") && switch_true(val)) {
w->cache_audio = 1;
}
}
static void w_numeric_param_tts(switch_speech_handle_t *sh, char *param, int val)
{
}
static void w_float_param_tts(switch_speech_handle_t *sh, char *param, double val)
{
}
SWITCH_MODULE_LOAD_FUNCTION(mod_whisper_tts_load)
{
switch_speech_interface_t *speech_interface;
*module_interface = switch_loadable_module_create_module_interface(pool, modname);
speech_interface = switch_loadable_module_create_interface(*module_interface, SWITCH_SPEECH_INTERFACE);
speech_interface->interface_name = "whisper";
speech_interface->speech_open = w_speech_open;
speech_interface->speech_close = w_speech_close;
speech_interface->speech_feed_tts = w_speech_feed_tts;
speech_interface->speech_read_tts = w_speech_read_tts;
speech_interface->speech_flush_tts = w_speech_flush_tts;
speech_interface->speech_text_param_tts = w_text_param_tts;
speech_interface->speech_numeric_param_tts = w_numeric_param_tts;
speech_interface->speech_float_param_tts = w_float_param_tts;
return whisper_speech_load();
}
SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_whisper_tts_shutdown)
{
return whisper_speech_unload();
}
+33
View File
@@ -0,0 +1,33 @@
#ifndef __MOD_WHISPER_TTS_H__
#define __MOD_WHISPER_TTS_H__
#include <switch.h>
typedef struct whisper_data {
char *voice_name;
char *api_key;
char *model_id;
char *speed;
/* result data */
long response_code;
char *ct;
char *reported_latency;
char *request_id;
char *name_lookup_time_ms;
char *connect_time_ms;
char *final_response_time_ms;
char *err_msg;
char *cache_filename;
char *session_id;
int rate;
int draining;
int reads;
int cache_audio;
void *conn;
void *circularBuffer;
switch_mutex_t *mutex;
FILE *file;
} whisper_t;
#endif
+964
View File
@@ -0,0 +1,964 @@
#include "mod_whisper_tts.h"
#include <switch.h>
#include <switch_json.h>
#include <curl/curl.h>
#include <cstdlib>
#include <boost/circular_buffer.hpp>
#include <boost/thread.hpp>
#include <boost/asio.hpp>
#include <boost/asio/ssl.hpp>
#include <boost/pool/object_pool.hpp>
#include <boost/bind/bind.hpp>
#include <boost/tokenizer.hpp>
#include <boost/foreach.hpp>
#include <boost/asio.hpp>
#include <boost/assign/list_of.hpp>
#include <boost/algorithm/string.hpp>
#include "mpg123.h"
#define BUFFER_GROW_SIZE (8192)
#define MP3_DCACHE 8192 * 2
typedef boost::circular_buffer<uint16_t> CircularBuffer_t;
/* Global information, common to all connections */
typedef struct
{
CURLM *multi;
int still_running;
} GlobalInfo_t;
static GlobalInfo_t global;
/* Information associated with a specific easy handle */
typedef struct
{
CURL *easy;
whisper_t* whisper;
char* body;
struct curl_slist *hdr_list;
GlobalInfo_t *global;
mpg123_handle *mh;
char error[CURL_ERROR_SIZE];
FILE* file;
std::chrono::time_point<std::chrono::high_resolution_clock> startTime;
bool flushed;
} ConnInfo_t;
static boost::object_pool<ConnInfo_t> pool ;
static std::map<curl_socket_t, boost::asio::ip::tcp::socket *> socket_map;
static boost::asio::io_service io_service;
static boost::asio::deadline_timer timer(io_service);
static std::string fullDirPath;
static std::thread worker_thread;
std::string secondsToMillisecondsString(double seconds) {
// Convert to milliseconds
double milliseconds = seconds * 1000.0;
// Truncate to remove fractional part
long milliseconds_long = static_cast<long>(milliseconds);
// Convert to string
return std::to_string(milliseconds_long);
}
static CURL* createEasyHandle(void) {
CURL* easy = curl_easy_init();
if(!easy) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "curl_easy_init() failed!\n");
return nullptr ;
}
curl_easy_setopt(easy, CURLOPT_FOLLOWLOCATION, 1L);
curl_easy_setopt(easy, CURLOPT_USERAGENT, "jambonz/0.8.5");
// set connect timeout to 3 seconds and total timeout to 109 seconds
curl_easy_setopt(easy, CURLOPT_CONNECTTIMEOUT_MS, 3000L);
curl_easy_setopt(easy, CURLOPT_TIMEOUT, 10L);
return easy ;
}
static void cleanupConn(ConnInfo_t *conn) {
auto w = conn->whisper;
if (conn->mh) {
mpg123_close(conn->mh);
mpg123_delete(conn->mh);
}
if( conn->hdr_list ) {
curl_slist_free_all(conn->hdr_list);
conn->hdr_list = nullptr ;
}
curl_easy_cleanup(conn->easy);
if (conn->file) {
if (fclose(conn->file) != 0) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "cleanupConn: error closing audio cache file\n");
}
conn->file = nullptr ;
}
w->conn = nullptr ;
w->draining = 1;
memset(conn, 0, sizeof(ConnInfo_t));
pool.destroy(conn) ;
}
/* Check for completed transfers, and remove their easy handles */
void check_multi_info(GlobalInfo_t *g) {
CURLMsg *msg;
int msgs_left;
ConnInfo_t *conn;
CURL *easy;
CURLcode res;
while((msg = curl_multi_info_read(g->multi, &msgs_left))) {
if(msg->msg == CURLMSG_DONE) {
long response_code;
double namelookup=0, connect=0, total=0 ;
char *ct = NULL ;
easy = msg->easy_handle;
res = msg->data.result;
curl_easy_getinfo(easy, CURLINFO_PRIVATE, &conn);
curl_easy_getinfo(easy, CURLINFO_RESPONSE_CODE, &response_code);
curl_easy_getinfo(easy, CURLINFO_CONTENT_TYPE, &ct);
curl_easy_getinfo(easy, CURLINFO_NAMELOOKUP_TIME, &namelookup);
curl_easy_getinfo(easy, CURLINFO_CONNECT_TIME, &connect);
curl_easy_getinfo(easy, CURLINFO_TOTAL_TIME, &total);
auto w = conn->whisper;
w->response_code = response_code;
if (ct) w->ct = strdup(ct);
std::string name_lookup_ms = secondsToMillisecondsString(namelookup);
std::string connect_ms = secondsToMillisecondsString(connect);
std::string final_response_time_ms = secondsToMillisecondsString(total);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG,
"mod_whisper_tts: response: %ld, content-type %s,"
"dns(ms): %" CURL_FORMAT_CURL_OFF_T ".%06ld, "
"connect(ms): %" CURL_FORMAT_CURL_OFF_T ".%06ld, "
"total(ms): %" CURL_FORMAT_CURL_OFF_T ".%06ld\n",
response_code, ct,
(long)(namelookup), (long)(fmod(namelookup, 1.0) * 1000000),
(long)(connect), (long)(fmod(connect, 1.0) * 1000000),
(long)(total), (long)(fmod(total, 1.0) * 1000000));
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "name lookup time: %s\n", name_lookup_ms.c_str());
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "connect time: %s\n", connect_ms.c_str());
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "final response time: %s\n", final_response_time_ms.c_str());
w->name_lookup_time_ms = strdup(name_lookup_ms.c_str());
w->connect_time_ms = strdup(connect_ms.c_str());
w->final_response_time_ms = strdup(final_response_time_ms.c_str());
curl_multi_remove_handle(g->multi, easy);
cleanupConn(conn);
}
}
}
int mcode_test(const char *where, CURLMcode code) {
if(CURLM_OK != code) {
const char *s;
switch(code) {
case CURLM_CALL_MULTI_PERFORM:
s = "CURLM_CALL_MULTI_PERFORM";
break;
case CURLM_BAD_HANDLE:
s = "CURLM_BAD_HANDLE";
break;
case CURLM_BAD_EASY_HANDLE:
s = "CURLM_BAD_EASY_HANDLE";
break;
case CURLM_OUT_OF_MEMORY:
s = "CURLM_OUT_OF_MEMORY";
break;
case CURLM_INTERNAL_ERROR:
s = "CURLM_INTERNAL_ERROR";
break;
case CURLM_UNKNOWN_OPTION:
s = "CURLM_UNKNOWN_OPTION";
break;
case CURLM_LAST:
s = "CURLM_LAST";
break;
default:
s = "CURLM_unknown";
break;
case CURLM_BAD_SOCKET:
s = "CURLM_BAD_SOCKET";
break;
}
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "mcode_test ERROR: %s returns %s:%d\n", where, s, code);
return -1;
}
return 0 ;
}
static void remsock(int *f, GlobalInfo_t *g) {
if(f) {
free(f);
f = NULL;
}
}
/* Called by asio when there is an action on a socket */
static void event_cb(GlobalInfo_t *g, curl_socket_t s, int action, const boost::system::error_code & error, int *fdp) {
int f = *fdp;
//switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "event_cb socket %#X has action %d\n", s, action) ;
// Socket already POOL REMOVED.
if (f == CURL_POLL_REMOVE) {
//switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "event_cb socket %#X removed\n", s);
remsock(fdp, g);
return;
}
if(socket_map.find(s) == socket_map.end()) {
//switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "event_cb: socket %#X already closed\n, s");
return;
}
/* make sure the event matches what are wanted */
if(f == action || f == CURL_POLL_INOUT) {
if(error) {
action = CURL_CSELECT_ERR;
}
CURLMcode rc = curl_multi_socket_action(g->multi, s, action, &g->still_running);
mcode_test("event_cb: curl_multi_socket_action", rc);
check_multi_info(g);
if(g->still_running <= 0) {
timer.cancel();
}
/* keep on watching.
* the socket may have been closed and/or fdp may have been changed
* in curl_multi_socket_action(), so check them both */
if(!error && socket_map.find(s) != socket_map.end() &&
(f == action || f == CURL_POLL_INOUT)) {
boost::asio::ip::tcp::socket *tcp_socket = socket_map.find(s)->second;
if(action == CURL_POLL_IN) {
tcp_socket->async_read_some(boost::asio::null_buffers(),
boost::bind(&event_cb, g, s,
action, boost::placeholders::_1, fdp));
}
if(action == CURL_POLL_OUT) {
tcp_socket->async_write_some(boost::asio::null_buffers(),
boost::bind(&event_cb, g, s,
action, boost::placeholders::_1, fdp));
}
}
}
}
/* socket functions */
static void setsock(int *fdp, curl_socket_t s, CURL *e, int act, int oldact, GlobalInfo_t *g) {
std::map<curl_socket_t, boost::asio::ip::tcp::socket *>::iterator it = socket_map.find(s);
if(it == socket_map.end()) {
//switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "setsock: socket %#X not found\n, s");
return;
}
boost::asio::ip::tcp::socket * tcp_socket = it->second;
*fdp = act;
if(act == CURL_POLL_IN) {
if(oldact != CURL_POLL_IN && oldact != CURL_POLL_INOUT) {
tcp_socket->async_read_some(boost::asio::null_buffers(),
boost::bind(&event_cb, g, s,
CURL_POLL_IN, boost::placeholders::_1, fdp));
}
}
else if(act == CURL_POLL_OUT) {
if(oldact != CURL_POLL_OUT && oldact != CURL_POLL_INOUT) {
tcp_socket->async_write_some(boost::asio::null_buffers(),
boost::bind(&event_cb, g, s,
CURL_POLL_OUT, boost::placeholders::_1, fdp));
}
}
else if(act == CURL_POLL_INOUT) {
if(oldact != CURL_POLL_IN && oldact != CURL_POLL_INOUT) {
tcp_socket->async_read_some(boost::asio::null_buffers(),
boost::bind(&event_cb, g, s,
CURL_POLL_IN, boost::placeholders::_1, fdp));
}
if(oldact != CURL_POLL_OUT && oldact != CURL_POLL_INOUT) {
tcp_socket->async_write_some(boost::asio::null_buffers(),
boost::bind(&event_cb, g, s,
CURL_POLL_OUT, boost::placeholders::_1, fdp));
}
}
}
static void addsock(curl_socket_t s, CURL *easy, int action, GlobalInfo_t *g) {
/* fdp is used to store current action */
int *fdp = (int *) calloc(sizeof(int), 1);
setsock(fdp, s, easy, action, 0, g);
curl_multi_assign(g->multi, s, fdp);
}
static int sock_cb(CURL *e, curl_socket_t s, int what, void *cbp, void *sockp) {
GlobalInfo_t *g = &global;
int *actionp = (int *) sockp;
static const char *whatstr[] = { "none", "IN", "OUT", "INOUT", "REMOVE"};
if(what == CURL_POLL_REMOVE) {
*actionp = what;
}
else {
if(!actionp) {
addsock(s, e, what, g);
}
else {
setsock(actionp, s, e, what, *actionp, g);
}
}
return 0;
}
static void threadFunc() {
/* to make sure the event loop doesn't terminate when there is no work to do */
io_service.reset() ;
boost::asio::io_service::work work(io_service);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "mod_whisper_tts threadFunc - starting\n");
for(;;) {
try {
io_service.run() ;
break ;
}
catch( std::exception& e) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "mod_whisper_tts threadFunc - Error: %s\n", e.what());
}
}
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "mod_whisper_tts threadFunc - ending\n");
}
/* Called by asio when our timeout expires */
static void timer_cb(const boost::system::error_code & error, GlobalInfo_t *g)
{
//switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "timer_cb\n");
if(!error) {
CURLMcode rc = curl_multi_socket_action(g->multi, CURL_SOCKET_TIMEOUT, 0, &g->still_running);
mcode_test("timer_cb: curl_multi_socket_action", rc);
check_multi_info(g);
}
}
int multi_timer_cb(CURLM *multi, long timeout_ms, GlobalInfo_t *g) {
/* cancel running timer */
timer.cancel();
if(timeout_ms >= 0) {
// from libcurl 7.88.1-10+deb12u4 does not allow call curl_multi_socket_action or curl_multi_perform in curl_multi callback directly
timer.expires_from_now(boost::posix_time::millisec(timeout_ms ? timeout_ms : 1));
timer.async_wait(boost::bind(&timer_cb, boost::placeholders::_1, g));
}
return 0;
}
static std::vector<uint16_t> convert_mp3_to_linear(ConnInfo_t *conn, uint8_t *data, size_t len) {
std::vector<uint16_t> linear_data;
int eof = 0;
int mp3err = 0;
unsigned char decode_buf[MP3_DCACHE];
if(mpg123_feed(conn->mh, data, len) == MPG123_OK) {
while(!eof) {
size_t usedlen = 0;
off_t frame_offset;
unsigned char* audio;
int decode_status = mpg123_decode_frame(conn->mh, &frame_offset, &audio, &usedlen);
switch(decode_status) {
case MPG123_NEW_FORMAT:
continue;
case MPG123_OK:
for(size_t i = 0; i < usedlen; i += 2) {
uint16_t value = reinterpret_cast<uint16_t*>(audio)[i / 2];
linear_data.push_back(value);
}
break;
case MPG123_DONE:
case MPG123_NEED_MORE:
eof = 1;
break;
case MPG123_ERR:
default:
if(++mp3err >= 5) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Decoder Error!\n");
eof = 1;
}
}
if (eof)
break;
mp3err = 0;
}
}
return linear_data;
}
/* CURLOPT_WRITEFUNCTION */
static size_t write_cb(void *ptr, size_t size, size_t nmemb, ConnInfo_t *conn) {
bool fireEvent = false;
uint8_t *data = (uint8_t *) ptr;
size_t bytes_received = size * nmemb;
auto w = conn->whisper;
CircularBuffer_t *cBuffer = (CircularBuffer_t *) w->circularBuffer;
std::vector<uint16_t> pcm_data;
if (conn->flushed) {
/* this will abort the transfer */
return 0;
}
{
switch_mutex_lock(w->mutex);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "write_cb: received data, response %ld\n",
w->response_code);
if (w->response_code > 0 && w->response_code != 200) {
std::string body((char *) ptr, bytes_received);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "write_cb: received body %s\n", body.c_str());
w->err_msg = strdup(body.c_str());
switch_mutex_unlock(w->mutex);
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);
// 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");
//TODO: if buffer exceeds some max size, return CURL_WRITEFUNC_ERROR to abort the transfer
cBuffer->set_capacity(cBuffer->size() + std::max((bytesResampled / sizeof(uint16_t)), (size_t)BUFFER_GROW_SIZE));
}
/* Push the data into the buffer */
cBuffer->insert(cBuffer->end(), pcm_data.data(), pcm_data.data() + pcm_data.size());
if (0 == w->reads++) {
fireEvent = true;
}
switch_mutex_unlock(w->mutex);
}
if (fireEvent && w->session_id) {
auto endTime = std::chrono::high_resolution_clock::now();
auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(endTime - conn->startTime);
auto time_to_first_byte_ms = std::to_string(duration.count());
switch_core_session_t* session = switch_core_session_locate(w->session_id);
if (session) {
switch_channel_t *channel = switch_core_session_get_channel(session);
if (channel) {
switch_event_t *event;
if (switch_event_create(&event, SWITCH_EVENT_PLAYBACK_START) == SWITCH_STATUS_SUCCESS) {
switch_channel_event_set_data(channel, event);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "write_cb: firing playback-started\n");
switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "Playback-File-Type", "tts_stream");
if (w->reported_latency) {
switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_whisper_reported_latency_ms", w->reported_latency);
}
if (w->request_id) {
switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_whisper_request_id", w->request_id);
}
if (w->name_lookup_time_ms) {
switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_whisper_name_lookup_time_ms", w->name_lookup_time_ms);
}
if (w->connect_time_ms) {
switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_whisper_connect_time_ms", w->connect_time_ms);
}
if (w->final_response_time_ms) {
switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_whisper_final_response_time_ms", w->final_response_time_ms);
}
if (w->voice_name) {
switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_whisper_voice_name", w->voice_name);
}
if (w->model_id) {
switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_whisper_model_id", w->model_id);
}
if (w->cache_filename) {
switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_cache_filename", w->cache_filename);
}
switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_time_to_first_byte_ms", time_to_first_byte_ms.c_str());
switch_event_fire(&event);
}
else {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "write_cb: failed to create event\n");
}
}
else {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "write_cb: channel not found\n");
}
switch_core_session_rwunlock(session);
}
else {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "write_cb: session %s not found\n", w->session_id);
}
}
return bytes_received;
}
static bool parseHeader(const std::string& str, std::string& header, std::string& value) {
std::vector<std::string> parts;
boost::split(parts, str, boost::is_any_of(":"), boost::token_compress_on);
if (parts.size() != 2)
return false;
header = boost::trim_copy(parts[0]);
value = boost::trim_copy(parts[1]);
return true;
}
static size_t header_callback(char *buffer, size_t size, size_t nitems, ConnInfo_t *conn) {
size_t bytes_received = size * nitems;
const std::string prefix = "HTTP/2 ";
whisper_t* w = conn->whisper;
std::string header, value;
std::string input(buffer, bytes_received);
if (parseHeader(input, header, value)) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "recv header: %s with value %s\n", header.c_str(), value.c_str());
if (0 == header.compare("openai-processing-ms")) w->reported_latency = strdup(value.c_str());
else if (0 == header.compare("x-request-id")) w->request_id = strdup(value.c_str());
}
else {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "recv header: %s\n", input.c_str());
if (input.rfind(prefix, 0) == 0) {
try {
w->response_code = std::stoi(input.substr(prefix.length()));
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "parsed response code: %ld\n", w->response_code);
} catch (const std::invalid_argument& e) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "header_callback: invalid response code %s\n", input.substr(prefix.length()).c_str());
}
}
}
return bytes_received;
}
/* CURLOPT_OPENSOCKETFUNCTION */
static curl_socket_t opensocket(void *clientp, curlsocktype purpose, struct curl_sockaddr *address) {
curl_socket_t sockfd = CURL_SOCKET_BAD;
//switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "opensocket: %d\n", purpose);
/* restrict to IPv4 */
if(purpose == CURLSOCKTYPE_IPCXN && address->family == AF_INET) {
/* create a tcp socket object */
boost::asio::ip::tcp::socket *tcp_socket = new boost::asio::ip::tcp::socket(io_service);
/* open it and get the native handle*/
boost::system::error_code ec;
tcp_socket->open(boost::asio::ip::tcp::v4(), ec);
if(ec) {
/* An error occurred */
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Couldn't open socket [%ld][%s]\n", ec, ec.message().c_str());
}
else {
sockfd = tcp_socket->native_handle();
/* save it for monitoring */
socket_map.insert(std::pair<curl_socket_t, boost::asio::ip::tcp::socket *>(sockfd, tcp_socket));
}
}
return sockfd;
}
/* CURLOPT_CLOSESOCKETFUNCTION */
static int close_socket(void *clientp, curl_socket_t item) {
//switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "close_socket : %#X\n", item);
std::map<curl_socket_t, boost::asio::ip::tcp::socket *>::iterator it = socket_map.find(item);
if(it != socket_map.end()) {
delete it->second;
socket_map.erase(it);
}
return 0;
}
extern "C" {
switch_status_t whisper_speech_load() {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "whisper_speech_loading..\n");
memset(&global, 0, sizeof(GlobalInfo_t));
global.multi = curl_multi_init();
if (!global.multi) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "whisper_speech_load curl_multi_init() failed, exiting!\n");
return SWITCH_STATUS_FALSE;
}
curl_multi_setopt(global.multi, CURLMOPT_SOCKETFUNCTION, sock_cb);
curl_multi_setopt(global.multi, CURLMOPT_SOCKETDATA, &global);
curl_multi_setopt(global.multi, CURLMOPT_TIMERFUNCTION, multi_timer_cb);
curl_multi_setopt(global.multi, CURLMOPT_TIMERDATA, &global);
curl_multi_setopt(global.multi, CURLMOPT_PIPELINING, CURLPIPE_MULTIPLEX);
/* create temp folder for cache files */
const char* baseDir = std::getenv("JAMBONZ_TMP_CACHE_FOLDER");
if (!baseDir) {
baseDir = "/var/";
}
if (strcmp(baseDir, "/") == 0) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "failed to create folder %s\n", baseDir);
return SWITCH_STATUS_FALSE;
}
fullDirPath = std::string(baseDir) + "jambonz-tts-cache-files";
// Create the directory with read, write, and execute permissions for everyone
mode_t oldMask = umask(0);
int result = mkdir(fullDirPath.c_str(), S_IRWXU | S_IRWXG | S_IRWXO);
umask(oldMask);
if (result != 0) {
if (errno != EEXIST) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "failed to create folder %s\n", fullDirPath.c_str());
fullDirPath = "";
}
else switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "folder %s already exists\n", fullDirPath.c_str());
}
else {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "created folder %s\n", fullDirPath.c_str());
}
// init mgp123
if (mpg123_init() != MPG123_OK) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "failed to initiate MPG123");
return SWITCH_STATUS_FALSE;
}
/* start worker thread that handles transfers*/
std::thread t(threadFunc) ;
worker_thread.swap( t ) ;
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "whisper_speech_loaded..\n");
return SWITCH_STATUS_SUCCESS;
}
switch_status_t whisper_speech_unload() {
/* stop the ASIO IO service */
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "whisper_speech_unload: stopping io service\n");
io_service.stop();
/* Join the worker thread */
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "whisper_speech_unload: wait for worker thread to complete\n");
if (worker_thread.joinable()) {
worker_thread.join();
}
/* cleanup curl multi handle*/
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "whisper_speech_unload: release curl multi\n");
curl_multi_cleanup(global.multi);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "whisper_speech_unload: completed\n");
mpg123_exit();
return SWITCH_STATUS_SUCCESS;
}
switch_status_t whisper_speech_open(whisper_t* whisper) {
return SWITCH_STATUS_SUCCESS;
}
switch_status_t whisper_speech_feed_tts(whisper_t* w, char* text, switch_speech_flag_t *flags) {
CURLMcode rc;
const int MAX_CHARS = 20;
char tempText[MAX_CHARS + 4]; // +4 for the ellipsis and null terminator
if (strlen(text) > MAX_CHARS) {
strncpy(tempText, text, MAX_CHARS);
strcpy(tempText + MAX_CHARS, "...");
} else {
strcpy(tempText, text);
}
/* open cache file */
if (w->cache_audio && fullDirPath.length() > 0) {
switch_uuid_t uuid;
char uuid_str[SWITCH_UUID_FORMATTED_LENGTH + 1];
char outfile[512] = "";
int fd;
switch_uuid_get(&uuid);
switch_uuid_format(uuid_str, &uuid);
switch_snprintf(outfile, sizeof(outfile), "%s%s%s.mp3", fullDirPath.c_str(), SWITCH_PATH_SEPARATOR, uuid_str);
w->cache_filename = strdup(outfile);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "writing audio cache file to %s\n", w->cache_filename);
mode_t oldMask = umask(0);
fd = open(outfile, O_WRONLY | O_CREAT | O_TRUNC, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH);
umask(oldMask);
if (fd == -1 ) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error opening cache file %s: %s\n", outfile, strerror(errno));
}
else {
w->file = fdopen(fd, "wb");
if (!w->file) {
close(fd);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error opening cache file %s: %s\n", outfile, strerror(errno));
}
}
}
if (!w->api_key) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "whisper_speech_feed_tts: no api_key provided\n");
return SWITCH_STATUS_FALSE;
}
if (!w->model_id) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "whisper_speech_feed_tts: no model_id provided\n");
return SWITCH_STATUS_FALSE;
}
/* format url*/
std::string url = "https://api.openai.com/v1/audio/speech";
/* create the JSON body */
cJSON * jResult = cJSON_CreateObject();
cJSON_AddStringToObject(jResult, "model", w->model_id);
cJSON_AddStringToObject(jResult, "input", text);
cJSON_AddStringToObject(jResult, "voice", w->voice_name);
cJSON_AddStringToObject(jResult, "response_format", "mp3");
if (w->speed) {
cJSON_AddStringToObject(jResult, "speed", w->speed);
}
char *json = cJSON_PrintUnformatted(jResult);
cJSON_Delete(jResult);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "whisper_speech_feed_tts: [%s] [%s]\n", url.c_str(), tempText);
ConnInfo_t *conn = pool.malloc() ;
// COnfigure MPG123
int mhError = 0;
mpg123_handle *mh = mpg123_new("auto", &mhError);
if (!mh) {
const char *mhErr = mpg123_plain_strerror(mhError);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error allocating mpg123 handle! %s\n", switch_str_nil(mhErr));
return SWITCH_STATUS_FALSE;
}
if (mpg123_open_feed(mh) != MPG123_OK) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error mpg123_open_feed!\n");
return SWITCH_STATUS_FALSE;
}
if (mpg123_format_all(mh) != MPG123_OK) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error mpg123_format_all!\n");
return SWITCH_STATUS_FALSE;
}
if (mpg123_param(mh, MPG123_FLAGS, MPG123_MONO_MIX, 0) != MPG123_OK) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error forcing single channel!\n");
return SWITCH_STATUS_FALSE;
}
CURL* easy = createEasyHandle();
w->conn = (void *) conn ;
conn->whisper = w;
conn->easy = easy;
conn->mh = mh;
conn->global = &global;
conn->hdr_list = NULL ;
conn->file = w->file;
conn->body = json;
conn->flushed = false;
w->circularBuffer = (void *) new CircularBuffer_t(8192);
if (w->session_id) {
int err;
switch_codec_implementation_t read_impl;
switch_core_session_t *psession = switch_core_session_locate(w->session_id);
switch_core_session_get_read_impl(psession, &read_impl);
uint32_t samples_per_second = !strcasecmp(read_impl.iananame, "g722") ? read_impl.actual_samples_per_second : read_impl.samples_per_second;
if (mpg123_param(mh, MPG123_FORCE_RATE, samples_per_second /*Hz*/, 0) != MPG123_OK) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error mpg123_param!\n");
return SWITCH_STATUS_FALSE;
}
}
std::ostringstream api_key_stream;
api_key_stream << "Authorization: Bearer " << w->api_key;
curl_easy_setopt(easy, CURLOPT_URL, url.c_str());
curl_easy_setopt(easy, CURLOPT_WRITEFUNCTION, write_cb);
curl_easy_setopt(easy, CURLOPT_WRITEDATA, conn);
curl_easy_setopt(easy, CURLOPT_ERRORBUFFER, conn->error);
curl_easy_setopt(easy, CURLOPT_PRIVATE, conn);
curl_easy_setopt(easy, CURLOPT_VERBOSE, 0L);
curl_easy_setopt(easy, CURLOPT_NOPROGRESS, 1L);
curl_easy_setopt(easy, CURLOPT_HEADERFUNCTION, header_callback);
curl_easy_setopt(easy, CURLOPT_HEADERDATA, conn);
/* call this function to get a socket */
curl_easy_setopt(easy, CURLOPT_OPENSOCKETFUNCTION, opensocket);
/* call this function to close a socket */
curl_easy_setopt(easy, CURLOPT_CLOSESOCKETFUNCTION, close_socket);
conn->hdr_list = curl_slist_append(conn->hdr_list, api_key_stream.str().c_str());
conn->hdr_list = curl_slist_append(conn->hdr_list, "Content-Type: application/json");
curl_easy_setopt(easy, CURLOPT_HTTPHEADER, conn->hdr_list);
curl_easy_setopt(easy, CURLOPT_POSTFIELDS, conn->body);
//curl_easy_setopt(easy, CURLOPT_POSTFIELDSIZE, body.length());
curl_easy_setopt(easy, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2_0);
rc = curl_multi_add_handle(global.multi, conn->easy);
mcode_test("new_conn: curl_multi_add_handle", rc);
/* start a timer to measure the duration until we receive first byte of audio */
conn->startTime = std::chrono::high_resolution_clock::now();
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "whisper_speech_feed_tts: called curl_multi_add_handle\n");
return SWITCH_STATUS_SUCCESS;
}
switch_status_t whisper_speech_read_tts(whisper_t* w, void *data, size_t *datalen, switch_speech_flag_t *flags) {
CircularBuffer_t *cBuffer = (CircularBuffer_t *) w->circularBuffer;
std::vector<uint16_t> pcm_data;
{
switch_mutex_lock(w->mutex);
ConnInfo_t *conn = (ConnInfo_t *) w->conn;
if (w->response_code > 0 && w->response_code != 200) {
switch_mutex_unlock(w->mutex);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "whisper_speech_read_tts, returning failure\n") ;
return SWITCH_STATUS_FALSE;
}
if (conn && conn->flushed) {
switch_mutex_unlock(w->mutex);
return SWITCH_STATUS_BREAK;
}
if (cBuffer->empty()) {
if (w->draining) {
switch_mutex_unlock(w->mutex);
return SWITCH_STATUS_BREAK;
}
/* no audio available yet so send silence */
memset(data, 255, *datalen);
switch_mutex_unlock(w->mutex);
return SWITCH_STATUS_SUCCESS;
}
size_t size = std::min((*datalen/2), cBuffer->size());
pcm_data.insert(pcm_data.end(), cBuffer->begin(), cBuffer->begin() + size);
cBuffer->erase(cBuffer->begin(), cBuffer->begin() + size);
switch_mutex_unlock(w->mutex);
}
memcpy(data, pcm_data.data(), pcm_data.size() * sizeof(uint16_t));
*datalen = pcm_data.size() * sizeof(uint16_t);
return SWITCH_STATUS_SUCCESS;
}
switch_status_t whisper_speech_flush_tts(whisper_t* w) {
bool download_complete = w->response_code == 200;
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "whisper_speech_flush_tts, download complete? %s\n", download_complete ? "yes" : "no") ;
ConnInfo_t *conn = (ConnInfo_t *) w->conn;
CircularBuffer_t *cBuffer = (CircularBuffer_t *) w->circularBuffer;
delete cBuffer;
w->circularBuffer = nullptr ;
if (conn) {
conn->flushed = true;
if (!download_complete) {
if (conn->file) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "closing audio cache file %s because download was interrupted\n", w->cache_filename);
if (fclose(conn->file) != 0) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "error closing audio cache file\n");
}
conn->file = nullptr ;
}
if (w->cache_filename) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "removing audio cache file %s because download was interrupted\n", w->cache_filename);
if (unlink(w->cache_filename) != 0) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "cleanupConn: error removing audio cache file %s: %d:%s\n",
w->cache_filename, errno, strerror(errno));
}
free(w->cache_filename);
w->cache_filename = nullptr ;
}
}
}
if (w->session_id) {
switch_core_session_t* session = switch_core_session_locate(w->session_id);
if (session) {
switch_channel_t *channel = switch_core_session_get_channel(session);
if (channel) {
switch_event_t *event;
if (switch_event_create(&event, SWITCH_EVENT_PLAYBACK_STOP) == SWITCH_STATUS_SUCCESS) {
switch_channel_event_set_data(channel, event);
switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "Playback-File-Type", "tts_stream");
switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_whisper_response_code", std::to_string(w->response_code).c_str());
if (w->cache_filename && w->response_code == 200) {
switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_cache_filename", w->cache_filename);
}
if (w->response_code != 200 && w->err_msg) {
switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "variable_tts_error", w->err_msg);
}
switch_event_fire(&event);
}
else {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "write_cb: failed to create event\n");
}
}
else {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "write_cb: channel not found\n");
}
switch_core_session_rwunlock(session);
}
}
return SWITCH_STATUS_SUCCESS;
}
switch_status_t whisper_speech_close(whisper_t* w) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "whisper_speech_close\n") ;
return SWITCH_STATUS_SUCCESS;
}
}
+12
View File
@@ -0,0 +1,12 @@
#ifndef __WHISPER_GLUE_H__
#define __WHISPER_GLUE_H__
switch_status_t whisper_speech_load();
switch_status_t whisper_speech_open(whisper_t* whisper);
switch_status_t whisper_speech_feed_tts(whisper_t* whisper, char* text, switch_speech_flag_t *flags);
switch_status_t whisper_speech_read_tts(whisper_t* whisper, void *data, size_t *datalen, switch_speech_flag_t *flags);
switch_status_t whisper_speech_flush_tts(whisper_t* whisper);
switch_status_t whisper_speech_close(whisper_t* whisper);
switch_status_t whisper_speech_unload();
#endif