tactility_apps/Apps/PipecatVoice/main/Source/main.c

#include <tt_app.h>
#include <tt_lvgl.h>
#include <tt_lvgl_toolbar.h>

#include <tactility/device.h>
#include <tactility/drivers/audio_stream.h>


#include <cJSON.h>
#include <esp_log.h>
#include <esp_random.h>
#include <freertos/FreeRTOS.h>
#include <freertos/semphr.h>
#include <freertos/task.h>

#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "voice_protocol.h"
#include "websocket.h"

/* Exported by firmware 0.8.0-dev although absent from the CDN SDK header. */
struct Device* device_find_by_name(const char* name);
struct Device* device_find_first_by_type(const struct DeviceType* type);

#define TAG "PipecatVoice"
#define DEFAULT_ENDPOINT "ws://192.168.68.102:8644/api/esp32/voice/ws"
#define DEFAULT_DEVICE_ID "tactility-14c19d1a790"

typedef struct {
    AppHandle app;
    volatile bool visible;
    volatile bool streaming;
    volatile bool playing;
    volatile bool socket_failed;
    int fd;
    PvState state;
    unsigned retry_attempt;
    size_t expected_audio_bytes;
    char endpoint[128];
    char device_id[64];
    char api_key[128];
    char detail[96];
    lv_obj_t* state_label;
    lv_obj_t* detail_label;
    struct Device* stream_dev;
    AudioStreamHandle input_handle;
    AudioStreamHandle output_handle;
    TaskHandle_t worker;
    TaskHandle_t receiver;
    SemaphoreHandle_t socket_lock;
    SemaphoreHandle_t audio_lock;
} VoiceContext;

static void update_ui(VoiceContext* ctx) {
    if (!ctx->visible || !tt_lvgl_lock(pdMS_TO_TICKS(100))) return;
    lv_label_set_text(ctx->state_label, pv_state_label(ctx->state));
    lv_obj_set_style_text_color(ctx->state_label,
        ctx->state == PV_STREAMING ? lv_color_hex(0x32c86e) :
        ctx->state == PV_FAILED ? lv_color_hex(0xd94a4a) : lv_color_hex(0xe0b64a), LV_PART_MAIN);
    lv_label_set_text(ctx->detail_label, ctx->detail);
    tt_lvgl_unlock();
}

static void set_state(VoiceContext* ctx, PvState state, const char* detail) {
    ctx->state = state;
    snprintf(ctx->detail, sizeof(ctx->detail), "%s", detail);
    update_ui(ctx);
}

static bool open_input_stream(VoiceContext* ctx) {
    if (ctx->stream_dev == NULL) return false;
    if (ctx->output_handle) { audio_stream_close(ctx->output_handle); ctx->output_handle = NULL; }
    if (ctx->input_handle) return true;
    struct AudioStreamConfig cfg = {
        .sample_rate = 16000,
        .bits_per_sample = 16,
        .channels = 1,
    };
    if (audio_stream_open_input(ctx->stream_dev, &cfg, &ctx->input_handle) != ERROR_NONE) {
        ESP_LOGW(TAG, "audio_stream_open_input failed");
        return false;
    }
    audio_stream_set_mute(ctx->stream_dev, AUDIO_CODEC_DIR_INPUT, false);
    audio_stream_set_volume(ctx->stream_dev, AUDIO_CODEC_DIR_INPUT, 100.0f);
    return true;
}

static bool open_output_stream(VoiceContext* ctx) {
    if (ctx->stream_dev == NULL) return false;
    if (ctx->input_handle) { audio_stream_close(ctx->input_handle); ctx->input_handle = NULL; }
    if (ctx->output_handle) return true;
    struct AudioStreamConfig cfg = {
        .sample_rate = 16000,
        .bits_per_sample = 16,
        .channels = 1,
    };
    if (audio_stream_open_output(ctx->stream_dev, &cfg, &ctx->output_handle) != ERROR_NONE) {
        ESP_LOGW(TAG, "audio_stream_open_output failed");
        return false;
    }
    audio_stream_set_mute(ctx->stream_dev, AUDIO_CODEC_DIR_OUTPUT, false);
    audio_stream_set_volume(ctx->stream_dev, AUDIO_CODEC_DIR_OUTPUT, 80.0f);
    return true;
}

static void close_audio(VoiceContext* ctx) {
    if (ctx->output_handle) { audio_stream_close(ctx->output_handle); ctx->output_handle = NULL; }
    if (ctx->input_handle) { audio_stream_close(ctx->input_handle); ctx->input_handle = NULL; }
}

static int send_locked(VoiceContext* ctx, const uint8_t* data, size_t length, bool binary) {
    if (ctx->fd < 0 || xSemaphoreTake(ctx->socket_lock, pdMS_TO_TICKS(500)) != pdTRUE) return -1;
    int result = ws_send(ctx->fd, data, length, binary);
    xSemaphoreGive(ctx->socket_lock);
    return result;
}

static void handle_event(VoiceContext* ctx, const char* text) {
    cJSON* root = cJSON_Parse(text);
    if (root == NULL) return;
    cJSON* version = cJSON_GetObjectItem(root, "v");
    cJSON* event = cJSON_GetObjectItem(root, "event");
    if (!cJSON_IsNumber(version) || version->valueint != PV_PROTOCOL_VERSION || !cJSON_IsString(event)) {
        cJSON_Delete(root);
        return;
    }
    if (strcmp(event->valuestring, "ready") == 0) {
        ctx->streaming = true;
    } else if (strcmp(event->valuestring, "state") == 0) {
        cJSON* state = cJSON_GetObjectItem(root, "state");
        if (cJSON_IsString(state) && strcmp(state->valuestring, "listening") == 0) {
            ctx->streaming = true;
            set_state(ctx, PV_STREAMING, "Continuous microphone streaming");
        }
    } else if (strcmp(event->valuestring, "audio") == 0) {
        cJSON* format = cJSON_GetObjectItem(root, "format");
        cJSON* rate = cJSON_GetObjectItem(root, "sample_rate");
        cJSON* channels = cJSON_GetObjectItem(root, "channels");
        cJSON* width = cJSON_GetObjectItem(root, "sample_width");
        cJSON* length = cJSON_GetObjectItem(root, "byte_length");
        if (cJSON_IsString(format) && cJSON_IsNumber(rate) && cJSON_IsNumber(channels) && cJSON_IsNumber(width) &&
            cJSON_IsNumber(length) && pv_valid_downstream_audio(format->valuestring, rate->valueint, channels->valueint,
            width->valueint, (size_t)length->valueint)) {
            ctx->expected_audio_bytes = (size_t)length->valueint;
            ctx->playing = true;
            set_state(ctx, PV_STREAMING, "Playing response; microphone paused");
        } else {
            ctx->socket_failed = true;
        }
    } else if (strcmp(event->valuestring, "error") == 0) {
        /* The server-provided message is intentionally not copied to display/logs. */
        ctx->socket_failed = true;
    }
    cJSON_Delete(root);
}

static void receiver_task(void* argument) {
    VoiceContext* ctx = argument;
    uint8_t* buffer = malloc(PV_DOWNSTREAM_MAX + 1U);
    if (buffer == NULL) { ctx->socket_failed = true; ctx->receiver = NULL; vTaskDelete(NULL); }
    while (ctx->visible && ctx->fd >= 0) {
        int opcode = 0;
        bool final = false;
        int received = ws_recv(ctx->fd, &opcode, &final, buffer, PV_DOWNSTREAM_MAX);
        if (received < 0 || !final) { ctx->socket_failed = true; break; }
        if (opcode == 0x01) {
            buffer[received] = '\0';
            handle_event(ctx, (const char*)buffer);
        } else if (opcode == 0x02) {
            if (!ctx->playing || !pv_binary_matches_metadata(ctx->expected_audio_bytes, (size_t)received)) { ctx->socket_failed = true; break; }
            xSemaphoreTake(ctx->audio_lock, portMAX_DELAY);
            bool ok = open_output_stream(ctx);
            size_t written = 0;
            if (ok) ok = audio_stream_write(ctx->output_handle, buffer, (size_t)received, &written, pdMS_TO_TICKS(3000)) == ERROR_NONE;
            open_input_stream(ctx);
            xSemaphoreGive(ctx->audio_lock);
            if (!ok || written != (size_t)received) { ctx->socket_failed = true; break; }
            ctx->expected_audio_bytes = 0;
            ctx->playing = false;
        } else if (opcode == 0x09) {
            if (xSemaphoreTake(ctx->socket_lock, pdMS_TO_TICKS(500)) == pdTRUE) {
                ws_send_pong(ctx->fd, buffer, (size_t)received);
                xSemaphoreGive(ctx->socket_lock);
            }
        } else if (opcode == 0x08) {
            ctx->socket_failed = true;
            break;
        }
    }
    free(buffer);
    ctx->receiver = NULL;
    vTaskDelete(NULL);
}

static void close_session(VoiceContext* ctx) {
    int fd = ctx->fd;
    ctx->fd = -1;
    ctx->streaming = false;
    ctx->playing = false;
    ctx->expected_audio_bytes = 0;
    if (fd >= 0) {
        if (xSemaphoreTake(ctx->socket_lock, pdMS_TO_TICKS(100)) == pdTRUE) {
            ws_send_close(fd);
            xSemaphoreGive(ctx->socket_lock);
        }
        ws_close(fd);
    }
    close_audio(ctx);
}

static void worker_task(void* argument) {
    VoiceContext* ctx = argument;
    PvEndpoint endpoint;
    if (!pv_parse_endpoint(ctx->endpoint, &endpoint)) {
        set_state(ctx, PV_FAILED, "Set a private-LAN ws:// endpoint in config.json");
        ctx->worker = NULL;
        vTaskDelete(NULL);
    }
    uint8_t pcm[1024];
    while (ctx->visible) {
        set_state(ctx, ctx->retry_attempt ? PV_RECONNECTING : PV_CONNECTING,
                  ctx->retry_attempt ? "Retrying voice gateway" : "Connecting to voice gateway");
        ctx->socket_failed = false;
        ESP_LOGI(TAG, "Opening voice gateway session");
        ctx->fd = ws_connect(endpoint.host, endpoint.port, endpoint.path, ctx->device_id, ctx->api_key);
        if (ctx->fd >= 0) {
            char session[64];
            snprintf(session, sizeof(session), "pv-%08lx", (unsigned long)esp_random());
            char start[256];
            if (pv_make_start_json(start, sizeof(start), session, ctx->device_id) &&
                send_locked(ctx, (const uint8_t*)start, strlen(start), false) == 0) {
                if (!open_input_stream(ctx)) {
                    set_state(ctx, PV_FAILED, "Audio stream unavailable");
                    ctx->socket_failed = true;
                } else {
                    xTaskCreate(receiver_task, "pv_rx", 6144, ctx, 6, &ctx->receiver);
                    uint32_t stable_ticks = 0;
                    while (ctx->visible && !ctx->socket_failed) {
                        if (!ctx->streaming || ctx->playing) { vTaskDelay(pdMS_TO_TICKS(20)); continue; }
                        xSemaphoreTake(ctx->audio_lock, portMAX_DELAY);
                        bool opened = open_input_stream(ctx);
                        size_t read = 0;
                        error_t read_result = opened ? audio_stream_read(ctx->input_handle, pcm, sizeof(pcm), &read, pdMS_TO_TICKS(100)) : ERROR_RESOURCE;
                        xSemaphoreGive(ctx->audio_lock);
                        if (!opened) { ctx->socket_failed = true; break; }
                        if (read_result == ERROR_NONE && pv_valid_pcm_chunk(read) && send_locked(ctx, pcm, read, true) < 0) ctx->socket_failed = true;
                        if (++stable_ticks >= 300) ctx->retry_attempt = 0;
                    }
                }
            }
        }
        close_session(ctx);
        if (!ctx->visible) break;
        uint32_t delay = pv_retry_delay_seconds(ctx->retry_attempt++);
        ESP_LOGW(TAG, "Voice gateway session unavailable; retry in %lu seconds", (unsigned long)delay);
        set_state(ctx, PV_RECONNECTING, "Gateway unavailable; retry scheduled");
        for (uint32_t second = 0; ctx->visible && second < delay; ++second) vTaskDelay(pdMS_TO_TICKS(1000));
    }
    ctx->worker = NULL;
    vTaskDelete(NULL);
}

static void load_config(VoiceContext* ctx) {
    snprintf(ctx->endpoint, sizeof(ctx->endpoint), "%s", DEFAULT_ENDPOINT);
    snprintf(ctx->device_id, sizeof(ctx->device_id), "%s", DEFAULT_DEVICE_ID);
    ctx->api_key[0] = '\0';
    char path[256]; size_t path_size = sizeof(path);
    tt_app_get_user_data_child_path(ctx->app, "config.json", path, &path_size);
    FILE* file = fopen(path, "r");
    if (file == NULL) return;
    char json[512]; size_t bytes = fread(json, 1, sizeof(json) - 1, file); fclose(file); json[bytes] = '\0';
    cJSON* root = cJSON_Parse(json);
    cJSON* endpoint = root ? cJSON_GetObjectItem(root, "server_url") : NULL;
    cJSON* device = root ? cJSON_GetObjectItem(root, "device_id") : NULL;
    cJSON* key = root ? cJSON_GetObjectItem(root, "api_key") : NULL;
    if (cJSON_IsString(endpoint)) snprintf(ctx->endpoint, sizeof(ctx->endpoint), "%s", endpoint->valuestring);
    if (cJSON_IsString(device)) snprintf(ctx->device_id, sizeof(ctx->device_id), "%s", device->valuestring);
    if (cJSON_IsString(key)) snprintf(ctx->api_key, sizeof(ctx->api_key), "%s", key->valuestring);
    cJSON_Delete(root);
}

static void* create_data(void) { VoiceContext* ctx = calloc(1, sizeof(*ctx)); if (ctx) ctx->fd = -1; return ctx; }
static void destroy_data(void* data) { free(data); }
static void on_create(AppHandle app, void* data) { ((VoiceContext*)data)->app = app; }
static void on_destroy(AppHandle app, void* data) { (void)app; (void)data; }

static void on_show(AppHandle app, void* data, lv_obj_t* parent) {
    VoiceContext* ctx = data; ctx->visible = true; load_config(ctx);
    ctx->stream_dev = device_find_by_name("audio-stream");
    if (ctx->stream_dev == NULL) ctx->stream_dev = device_find_first_by_type(&AUDIO_STREAM_TYPE);
    ctx->socket_lock = xSemaphoreCreateMutex();
    ctx->audio_lock = xSemaphoreCreateMutex();
    lv_obj_t* toolbar = tt_lvgl_toolbar_create_for_app(parent, app);
    lv_obj_align(toolbar, LV_ALIGN_TOP_MID, 0, 0);
    ctx->state_label = lv_label_create(parent);

    lv_obj_align(ctx->state_label, LV_ALIGN_CENTER, 0, -30);
    ctx->detail_label = lv_label_create(parent);
    lv_obj_set_width(ctx->detail_label, lv_pct(88)); lv_label_set_long_mode(ctx->detail_label, LV_LABEL_LONG_WRAP);
    lv_obj_set_style_text_align(ctx->detail_label, LV_TEXT_ALIGN_CENTER, LV_PART_MAIN);
    lv_obj_align(ctx->detail_label, LV_ALIGN_CENTER, 0, 25);
    if (ctx->stream_dev == NULL || ctx->socket_lock == NULL || ctx->audio_lock == NULL) set_state(ctx, PV_FAILED, "Audio service unavailable");
    else xTaskCreate(worker_task, "pv_worker", 7168, ctx, 5, &ctx->worker);
}

static void on_hide(AppHandle app, void* data) {
    (void)app; VoiceContext* ctx = data; ctx->visible = false; close_session(ctx);
    for (unsigned i = 0; (ctx->worker || ctx->receiver) && i < 100; ++i) vTaskDelay(pdMS_TO_TICKS(10));
    close_audio(ctx);
    if (ctx->socket_lock) { vSemaphoreDelete(ctx->socket_lock); ctx->socket_lock = NULL; }
    if (ctx->audio_lock) { vSemaphoreDelete(ctx->audio_lock); ctx->audio_lock = NULL; }
}

int main(int argc, char* argv[]) {
    (void)argc; (void)argv;
    tt_app_register((AppRegistration){.createData=create_data,.destroyData=destroy_data,.onCreate=on_create,.onDestroy=on_destroy,.onShow=on_show,.onHide=on_hide});
    return 0;
}