asterisk/res/res_speech_aeap.c

/*
 * Asterisk -- An open source telephony toolkit.
 *
 * Copyright (C) 2021, Sangoma Technologies Corporation
 *
 * Kevin Harwell <kharwell@sangoma.com>
 *
 * See http://www.asterisk.org for more information about
 * the Asterisk project. Please do not directly contact
 * any of the maintainers of this project for assistance;
 * the project provides a web site, mailing lists and IRC
 * channels for your use.
 *
 * This program is free software, distributed under the terms of
 * the GNU General Public License Version 2. See the LICENSE file
 * at the top of the source tree.
 */

/*! \file
 *
 * \brief Asterisk External Application Speech Engine
 *
 */

/*** MODULEINFO
	<support_level>core</support_level>
 ***/

#include "asterisk.h"

#include "asterisk/astobj2.h"
#include "asterisk/config.h"
#include "asterisk/format.h"
#include "asterisk/format_cap.h"
#include "asterisk/json.h"
#include "asterisk/module.h"
#include "asterisk/speech.h"
#include "asterisk/sorcery.h"

#include "asterisk/res_aeap.h"
#include "asterisk/res_aeap_message.h"

#define SPEECH_AEAP_VERSION "0.1.0"
#define SPEECH_PROTOCOL "speech_to_text"

#define CONNECTION_TIMEOUT 2000

#define log_error(obj, fmt, ...) \
	ast_log(LOG_ERROR, "AEAP speech (%p): " fmt "\n", obj, ##__VA_ARGS__)

static struct ast_json *custom_fields_to_params(const struct ast_variable *variables)
{
	const struct ast_variable *i;
	struct ast_json *obj;

	if (!variables) {
		return NULL;
	}

	obj = ast_json_object_create();
	if (!obj) {
		return NULL;
	}

	for (i = variables; i; i = i->next) {
		if (i->name[0] == '@' && i->name[1]) {
			ast_json_object_set(obj, i->name + 1, ast_json_string_create(i->value));
		}
	}

	return obj;
}

/*!
 * \internal
 * \brief Create, and send a request to the external application
 *
 * Create, then sends a request to an Asterisk external application, and then blocks
 * until a response is received or a time out occurs. Since this method waits until
 * receiving a response the returned result is guaranteed to be pass/fail based upon
 * a response handler's result.
 *
 * \param aeap Pointer to an Asterisk external application object
 * \param name The name of the request to send
 * \param json The core json request data
 * \param data Optional user data to associate with request/response
 *
 * \returns 0 on success, -1 on error
 */
static int speech_aeap_send_request(struct ast_aeap *aeap, const char *name,
	struct ast_json *json, void *data)
{
	/*
	 * Wait for a response. Also since we're blocking,
	 * data is expected to be on the stack so no cleanup required.
	 */
	struct ast_aeap_tsx_params tsx_params = {
		.timeout = 1000,
		.wait = 1,
		.obj = data,
	};

	/* "steals" the json ref */
	tsx_params.msg = ast_aeap_message_create_request(
		ast_aeap_message_type_json, name, NULL, json);
	if (!tsx_params.msg) {
		return -1;
	}

	/* Send "steals" the json msg ref */
	return ast_aeap_send_msg_tsx(aeap, &tsx_params);
}

/*!
 * \internal
 * \brief Create, and send a "get" request to an external application
 *
 * Basic structure of the JSON message to send:
 *
 \verbatim
 { param: [<param>, ...] }
 \endverbatim
 *
 * \param speech The speech engine
 * \param param The name of the parameter to retrieve
 * \param data User data passed to the response handler
 *
 * \returns 0 on success, -1 on error
 */
static int speech_aeap_get(struct ast_speech *speech, const char *param, void *data)
{
	if (!param) {
		return -1;
	}

	/* send_request handles json ref */
	return speech_aeap_send_request(speech->data,
		"get", ast_json_pack("{s:[s]}", "params", param), data);
}

struct speech_param {
	const char *name;
	const char *value;
};

/*!
 * \internal
 * \brief Create, and send a "set" request to an external application
 *
 * Basic structure of the JSON message to send:
 *
 \verbatim
 { params: { <name> : <value> }  }
 \endverbatim
 *
 * \param speech The speech engine
 * \param name The name of the parameter to set
 * \param value The value of the parameter to set
 *
 * \returns 0 on success, -1 on error
 */
static int speech_aeap_set(struct ast_speech *speech, const char *name, const char *value)
{
	if (!name) {
		return -1;
	}

	/* send_request handles json ref */
	return speech_aeap_send_request(speech->data,
		"set", ast_json_pack("{s:{s:s}}", "params", name, value), NULL);
}

static int handle_response_set(struct ast_aeap *aeap, struct ast_aeap_message *message, void *data)
{
	return 0;
}

struct speech_setting {
	const char *param;
	size_t len;
	char *buf;
};

static int handle_setting(struct ast_aeap *aeap, struct ast_json_iter *iter,
	struct speech_setting *setting)
{
	const char *value;

	if (strcmp(ast_json_object_iter_key(iter), setting->param)) {
		log_error(aeap, "Unable to 'get' speech setting for '%s'", setting->param);
		return -1;
	}

	value = ast_json_string_get(ast_json_object_iter_value(iter));
	if (!value) {
		log_error(aeap, "No value for speech setting '%s'", setting->param);
		return -1;
	}

	ast_copy_string(setting->buf, value, setting->len);
	return 0;
}

static int handle_results(struct ast_aeap *aeap, struct ast_json_iter *iter,
	struct ast_speech_result **speech_results)
{
	struct ast_speech_result *result = NULL;
	struct ast_json *json_results;
	struct ast_json *json_result;
	size_t i;

	json_results = ast_json_object_iter_value(iter);
	if (!json_results || !speech_results) {
		log_error(aeap, "Unable to 'get' speech results");
		return -1;
	}

	for (i = 0; i < ast_json_array_size(json_results); ++i) {
		if (!(result = ast_calloc(1, sizeof(*result)))) {
			continue;
		}

		json_result = ast_json_array_get(json_results, i);

		result->text = ast_strdup(ast_json_object_string_get(json_result, "text"));
		result->score = ast_json_object_integer_get(json_result, "score");
		result->grammar = ast_strdup(ast_json_object_string_get(json_result, "grammar"));
		result->nbest_num = ast_json_object_integer_get(json_result, "best");
		if (*speech_results) {
			AST_LIST_NEXT(result, list) = *speech_results;
			*speech_results = result;
		} else {
			*speech_results = result;
		}
	}

	return 0;
}

/*!
 * \internal
 * \brief Handle a "get" response from an external application
 *
 * Basic structure of the expected JSON message to received:
 *
 \verbatim
 {
   response: "get"
   "params" : { <name>: <value> | [ <results> ] }
 }
 \endverbatim
 *
 * \param aeap Pointer to an Asterisk external application object
 * \param message The received message
 * \param data User data passed to the response handler
 *
 * \returns 0 on success, -1 on error
 */
static int handle_response_get(struct ast_aeap *aeap, struct ast_aeap_message *message, void *data)
{
	struct ast_json_iter *iter;

	iter = ast_json_object_iter(ast_json_object_get(ast_aeap_message_data(message), "params"));
	if (!iter) {
		log_error(aeap, "no 'get' parameters returned");
		return -1;
	}

	if (!strcmp(ast_json_object_iter_key(iter), "results")) {
		return handle_results(aeap, iter, data);
	}

	return handle_setting(aeap, iter, data);
}

static int handle_response_setup(struct ast_aeap *aeap, struct ast_aeap_message *message, void *data)
{
	struct ast_format *format = data;
	struct ast_json *json = ast_aeap_message_data(message);
	const char *codec_name;

	if (!format) {
		log_error(aeap, "no 'format' set");
		return -1;
	}

	if (!json) {
		log_error(aeap, "no 'setup' object returned");
		return -1;
	}

	json = ast_json_object_get(json, "codecs");
	if (!json || ast_json_array_size(json) == 0) {
		log_error(aeap, "no 'setup' codecs available");
		return -1;
	}

	codec_name = ast_json_object_string_get(ast_json_array_get(json, 0), "name");
	if (!codec_name || strcmp(codec_name, ast_format_get_codec_name(format))) {
		log_error(aeap, "setup  codec '%s' unsupported", ast_format_get_codec_name(format));
		return -1;
	}

	return 0;
}

static const struct ast_aeap_message_handler response_handlers[] = {
	{ "setup", handle_response_setup },
	{ "get", handle_response_get },
	{ "set", handle_response_set },
};

static int handle_request_set(struct ast_aeap *aeap, struct ast_aeap_message *message, void *data)
{
	struct ast_json_iter *iter;
	const char *error_msg = NULL;

	iter = ast_json_object_iter(ast_json_object_get(ast_aeap_message_data(message), "params"));
	if (!iter) {
		error_msg = "no parameter(s) requested";
	} else if (!strcmp(ast_json_object_iter_key(iter), "results")) {
		struct ast_speech *speech = ast_aeap_user_data_object_by_id(aeap, "speech");

		if (!speech) {
			error_msg = "no associated speech object";
		} else if (handle_results(aeap, iter, &speech->results)) {
			error_msg = "unable to handle results";
		} else {
			ast_speech_change_state(speech, AST_SPEECH_STATE_DONE);
		}
	} else {
		error_msg = "can only set 'results'";
	}

	if (error_msg) {
		log_error(aeap, "set - %s", error_msg);
		message = ast_aeap_message_create_error(ast_aeap_message_type_json,
			ast_aeap_message_name(message), ast_aeap_message_id(message), error_msg);
	} else {
		message = ast_aeap_message_create_response(ast_aeap_message_type_json,
			ast_aeap_message_name(message), ast_aeap_message_id(message), NULL);
	}

	ast_aeap_send_msg(aeap, message);

	return 0;
}

static const struct ast_aeap_message_handler request_handlers[] = {
	{ "set", handle_request_set },
};

/*!
 * \internal
 * \brief Handle an error from an external application by setting state to done
 *
 * \param aeap Pointer to an Asterisk external application object
 */
static void ast_aeap_speech_on_error(struct ast_aeap *aeap)
{
	struct ast_speech *speech = ast_aeap_user_data_object_by_id(aeap, "speech");
	if (!speech) {
		ast_log(LOG_ERROR, "aeap generated error with no associated speech object");
		return;
	}

	ast_speech_change_state(speech, AST_SPEECH_STATE_DONE);
}

static struct ast_aeap_params speech_aeap_params = {
	.response_handlers = response_handlers,
	.response_handlers_size = ARRAY_LEN(response_handlers),
	.request_handlers = request_handlers,
	.request_handlers_size = ARRAY_LEN(request_handlers),
	.on_error = ast_aeap_speech_on_error,
};

/*!
 * \internal
 * \brief Create, and connect to an external application and send initial setup
 *
 * Basic structure of the JSON message to send:
 *
 \verbatim
 {
   "request": "setup"
   "codecs": [
       {
           "name": <name>,
           "attributes": { <name>: <value>, ..., }
       },
       ...,
   ],
   "params": { <name>: <value>, ..., }
 }
 \endverbatim
 *
 * \param speech The speech engine
 * \param format The format codec to use
 *
 * \returns 0 on success, -1 on error
 */
static int speech_aeap_engine_create(struct ast_speech *speech, struct ast_format *format)
{
	struct ast_aeap *aeap;
	struct ast_variable *vars;
	struct ast_json *json;

	aeap = ast_aeap_create_and_connect_by_id(
		speech->engine->name, &speech_aeap_params, CONNECTION_TIMEOUT);
	if (!aeap) {
		return -1;
	}

	speech->data = aeap;

	/* Don't allow unloading of this module while an external application is in use */
	ast_module_ref(ast_module_info->self);

	vars = ast_aeap_custom_fields_get(speech->engine->name);

	/* While the protocol allows sending of codec attributes, for now don't */
	json = ast_json_pack("{s:s,s:[{s:s}],s:o*}", "version", SPEECH_AEAP_VERSION, "codecs",
		"name", ast_format_get_codec_name(format), "params", custom_fields_to_params(vars));

	ast_variables_destroy(vars);

	if (ast_aeap_user_data_register(aeap, "speech", speech, NULL)) {
		ast_module_unref(ast_module_info->self);
		return -1;
	}

	/* send_request handles json ref */
	if (speech_aeap_send_request(speech->data, "setup", json, format)) {
		ast_module_unref(ast_module_info->self);
		return -1;
	}

	/*
	 * Add a reference to the engine here, so if it happens to get unregistered
	 * while executing it won't disappear.
	 */
	ao2_ref(speech->engine, 1);

	return 0;
}

static int speech_aeap_engine_destroy(struct ast_speech *speech)
{
	ao2_ref(speech->engine, -1);
	ao2_cleanup(speech->data);

	ast_module_unref(ast_module_info->self);

	return 0;
}

static int speech_aeap_engine_write(struct ast_speech *speech, void *data, int len)
{
	return ast_aeap_send_binary(speech->data, data, len);
}

static int speech_aeap_engine_dtmf(struct ast_speech *speech, const char *dtmf)
{
	return speech_aeap_set(speech, "dtmf", dtmf);
}

static int speech_aeap_engine_start(struct ast_speech *speech)
{
	ast_speech_change_state(speech, AST_SPEECH_STATE_READY);

	return 0;
}

static int speech_aeap_engine_change(struct ast_speech *speech, const char *name, const char *value)
{
	return speech_aeap_set(speech, name, value);
}

static int speech_aeap_engine_get_setting(struct ast_speech *speech, const char *name,
	char *buf, size_t len)
{
	struct speech_setting setting = {
		.param = name,
		.len = len,
		.buf = buf,
	};

	return speech_aeap_get(speech, name, &setting);
}

static int speech_aeap_engine_change_results_type(struct ast_speech *speech,
	enum ast_speech_results_type results_type)
{
	return speech_aeap_set(speech, "results_type",
		ast_speech_results_type_to_string(results_type));
}

static struct ast_speech_result *speech_aeap_engine_get(struct ast_speech *speech)
{
	struct ast_speech_result *results = NULL;

	if (speech->results) {
		return speech->results;
	}

	if (speech_aeap_get(speech, "results", &results)) {
		return NULL;
	}

	return results;
}

static void speech_engine_destroy(void *obj)
{
	struct ast_speech_engine *engine = obj;

	ao2_cleanup(engine->formats);
	ast_free(engine->name);
}

static struct ast_speech_engine *speech_engine_alloc(const char *name)
{
	struct ast_speech_engine *engine;

	engine = ao2_t_alloc_options(sizeof(*engine), speech_engine_destroy,
		AO2_ALLOC_OPT_LOCK_NOLOCK, name);
	if (!engine) {
		ast_log(LOG_ERROR, "AEAP speech: unable create engine '%s'\n", name);
		return NULL;
	}

	engine->name = ast_strdup(name);
	if (!engine->name) {
		ao2_ref(engine, -1);
		return NULL;
	}

	engine->create = speech_aeap_engine_create;
	engine->destroy = speech_aeap_engine_destroy;
	engine->write = speech_aeap_engine_write;
	engine->dtmf = speech_aeap_engine_dtmf;
	engine->start = speech_aeap_engine_start;
	engine->change = speech_aeap_engine_change;
	engine->get_setting = speech_aeap_engine_get_setting;
	engine->change_results_type = speech_aeap_engine_change_results_type;
	engine->get = speech_aeap_engine_get;

	engine->formats = ast_format_cap_alloc(AST_FORMAT_CAP_FLAG_DEFAULT);

	return engine;
}

static void speech_engine_alloc_and_register(const char *name, const struct ast_format_cap *formats)
{
	struct ast_speech_engine *engine;

	engine = speech_engine_alloc(name);
	if (!engine) {
		return;
	}

	if (formats && ast_format_cap_append_from_cap(engine->formats,
			formats, AST_MEDIA_TYPE_AUDIO)) {
		ast_log(LOG_WARNING, "AEAP speech: Unable to add engine '%s' formats\n", name);
		ao2_ref(engine, -1);
		return;
	}

	if (ast_speech_register(engine)) {
		ast_log(LOG_WARNING, "AEAP speech: Unable to register engine '%s'\n", name);
		ao2_ref(engine, -1);
	}
}

#ifdef TEST_FRAMEWORK

static void speech_engine_alloc_and_register2(const char *name, const char *codec_names)
{
	struct ast_speech_engine *engine;

	engine = speech_engine_alloc(name);
	if (!engine) {
		return;
	}

	if (codec_names && ast_format_cap_update_by_allow_disallow(engine->formats, codec_names, 1)) {
		ast_log(LOG_WARNING, "AEAP speech: Unable to add engine '%s' codecs\n", name);
		ao2_ref(engine, -1);
		return;
	}

	if (ast_speech_register(engine)) {
		ast_log(LOG_WARNING, "AEAP speech: Unable to register engine '%s'\n", name);
		ao2_ref(engine, -1);
	}
}

#endif

static int unload_engine(void *obj, void *arg, int flags)
{
	if (ast_aeap_client_config_has_protocol(obj, SPEECH_PROTOCOL)) {
		ao2_cleanup(ast_speech_unregister2(ast_sorcery_object_get_id(obj)));
	}

	return 0;
}

static int load_engine(void *obj, void *arg, int flags)
{
	const char *id;
	const struct ast_format_cap *formats;
	const struct ast_speech_engine *engine;

	if (!ast_aeap_client_config_has_protocol(obj, SPEECH_PROTOCOL)) {
		return 0;
	}

	id = ast_sorcery_object_get_id(obj);
	formats = ast_aeap_client_config_codecs(obj);
	if (!formats) {
		formats = ast_format_cap_alloc(AST_FORMAT_CAP_FLAG_DEFAULT);
		if (!formats) {
			ast_log(LOG_ERROR, "AEAP speech: unable to allocate default engine format for '%s'\n", id);
			return 0;
		}
	}

	engine = ast_speech_find_engine(id);
	if (!engine) {
		speech_engine_alloc_and_register(id, formats);
		return 0;
	}

	if (ast_format_cap_identical(formats, engine->formats)) {
		/* Same name, same formats then nothing changed */
		return 0;
	}

	ao2_ref(ast_speech_unregister2(engine->name), -1);
	speech_engine_alloc_and_register(id, formats);

	return 0;
}

static int matches_engine(void *obj, void *arg, int flags)
{
	const struct ast_speech_engine *engine = arg;

	return strcmp(ast_sorcery_object_get_id(obj), engine->name) ? 0 : CMP_MATCH;
}

static int should_unregister(const struct ast_speech_engine *engine, void *data)
{
	void *obj;

	if (engine->create != speech_aeap_engine_create) {
		/* Only want to potentially unregister AEAP speech engines */
		return 0;
	}

#ifdef TEST_FRAMEWORK
	if (!strcmp("_aeap_test_speech_", engine->name)) {
		/* Don't remove the test engine */
		return 0;
	}
#endif

	obj = ao2_callback(data, 0, matches_engine, (void*)engine);

	if (obj) {
		ao2_ref(obj, -1);
		return 0;
	}

	/* If no match in given container then unregister engine */
	return 1;
}

static void speech_observer_loaded(const char *object_type)
{
	struct ao2_container *container;

	if (strcmp(object_type, AEAP_CONFIG_CLIENT)) {
		return;
	}

	container = ast_aeap_client_configs_get(SPEECH_PROTOCOL);
	if (!container) {
		return;
	}

	/*
	 * An AEAP module reload has occurred. First
	 * remove all engines that no longer exist.
	 */
	ast_speech_unregister_engines(should_unregister, container, __ao2_cleanup);

	/* Now add or update engines */
	ao2_callback(container, 0, load_engine, NULL);
	ao2_ref(container, -1);
}

/*! \brief Observer for AEAP reloads */
static const struct ast_sorcery_observer speech_observer = {
	.loaded = speech_observer_loaded,
};

static int unload_module(void)
{
	struct ao2_container *container;

#ifdef TEST_FRAMEWORK
	ao2_cleanup(ast_speech_unregister2("_aeap_test_speech_"));
#endif

	ast_sorcery_observer_remove(ast_aeap_sorcery(), AEAP_CONFIG_CLIENT, &speech_observer);

	container = ast_aeap_client_configs_get(SPEECH_PROTOCOL);
	if (container) {
		ao2_callback(container, 0, unload_engine, NULL);
		ao2_ref(container, -1);
	}

	return 0;
}

static int load_module(void)
{
	struct ao2_container *container;

	speech_aeap_params.msg_type = ast_aeap_message_type_json;

	container = ast_aeap_client_configs_get(SPEECH_PROTOCOL);
	if (container) {
		ao2_callback(container, 0, load_engine, NULL);
		ao2_ref(container, -1);
	}

	/*
	 * Add an observer since a named speech server must be created,
	 * registered, and eventually removed for all AEAP client
	 * configuration matching the "speech_to_text" protocol.
	*/
	if (ast_sorcery_observer_add(ast_aeap_sorcery(), AEAP_CONFIG_CLIENT, &speech_observer)) {
		return AST_MODULE_LOAD_DECLINE;
	}

#ifdef TEST_FRAMEWORK
	speech_engine_alloc_and_register2("_aeap_test_speech_", "ulaw");
#endif

	return AST_MODULE_LOAD_SUCCESS;
}

AST_MODULE_INFO(ASTERISK_GPL_KEY, AST_MODFLAG_LOAD_ORDER, "Asterisk External Application Speech Engine",
	.support_level = AST_MODULE_SUPPORT_CORE,
	.load = load_module,
	.unload = unload_module,
	.load_pri = AST_MODPRI_CHANNEL_DEPEND,
	.requires = "res_speech,res_aeap",
);