Robovoice/Robovoice.App/VoicePipeline.cs

using System.Collections.Concurrent;
using Robovoice.Core;
using Robovoice.Tts.LibPiper;

namespace Robovoice.App;

internal sealed record TextItem(uint Session, string Text);

internal sealed record AudioItem(uint Session, float[]? Samples);

internal sealed class VoicePipeline : IDisposable
{
    private readonly LibPiperTtsEngine _tts;
    private readonly AudioOutput _audioOutput;
    private readonly Action<string> _log;

    private BlockingCollection<TextItem> _textQueue = new();
    private BlockingCollection<AudioItem> _audioQueue = new();
    private ManualResetEventSlim _gate = new(false);
    private volatile uint _currentSession;
    private volatile bool _running;

    private Thread? _synthThread;
    private Thread? _playerThread;
    private bool _disposed;

    public string OutputDeviceName { get; set; } = string.Empty;

    public VoicePipeline(
        LibPiperTtsEngine tts,
        AudioOutput audioOutput,
        Action<string> log)
    {
        _tts = tts;
        _audioOutput = audioOutput;
        _audioOutput.Log = log;
        _log = log;
    }

    public void Start()
    {
        _running = true;
        _synthThread = new Thread(SynthLoop) { IsBackground = true, Name = "VoicePipeline-Synth" };
        _playerThread = new Thread(PlayerLoop) { IsBackground = true, Name = "VoicePipeline-Player" };
        _synthThread.Start();
        _playerThread.Start();
    }

    public void EnqueueSegment(string text)
    {
        if (!_running) return;
        uint session = _currentSession;
        _textQueue.Add(new TextItem(session, text));
    }

    public void EnqueueFinal(string text)
    {
        if (!_running) return;
        uint session = _currentSession;
        if (!string.IsNullOrEmpty(text))
            _textQueue.Add(new TextItem(session, text));
        _textQueue.Add(new TextItem(session, ""));
        _gate.Reset();
    }

    public void OnPttPressed()
    {
        _currentSession++;
        _gate.Reset();
        _audioOutput.Stop();
    }

    public void OnPttReleased()
    {
        _gate.Set();
    }

    private void SynthLoop()
    {
        foreach (var item in _textQueue.GetConsumingEnumerable())
        {
            if (!_running) break;

            if (item.Session != _currentSession)
                continue;

            if (item.Text.Length == 0)
            {
                // :F sentinel — signal end of utterance to player
                _audioQueue.Add(new AudioItem(item.Session, null));
                continue;
            }

            try
            {
                foreach (var chunk in _tts.SynthesizeSync(item.Text))
                {
                    if (item.Session != _currentSession)
                    {
                        // Session changed mid-synthesis — drain piper cleanly
                        _tts.SynthesizeDrain();
                        break;
                    }

                    _audioQueue.Add(new AudioItem(item.Session, chunk.Samples));
                }
            }
            catch (Exception ex)
            {
                _log($"TTS error: {ex.Message}");
            }
        }
    }

    private void PlayerLoop()
    {
        while (_running)
        {
            _gate.Wait();
            if (!_running) break;

            // Drain stale items, then wait for real audio for current session
            AudioItem firstItem = default!;
            while (_running)
            {
                try { firstItem = _audioQueue.Take(); }
                catch (InvalidOperationException) { return; }

                if (firstItem.Session != _currentSession)
                    continue;

                break;
            }

            if (!_running) break;

            if (firstItem.Samples == null)
            {
                // :F with no audio — nothing to play
                _gate.Reset();
                continue;
            }

            _audioOutput.Start(_tts.SampleRate, OutputDeviceName);
            _audioOutput.WriteSamples(firstItem.Samples);
            _log($"TTS: playback started (session {firstItem.Session})");

            foreach (var item in _audioQueue.GetConsumingEnumerable())
            {
                if (!_running) break;

                if (item.Session != _currentSession)
                    break;

                if (item.Samples == null)
                    break;

                _audioOutput.WriteSamples(item.Samples);
            }

            _audioOutput.Flush();
            _log("TTS: playback finished");
            _gate.Reset();
        }
    }

    public void Dispose()
    {
        if (_disposed) return;
        _disposed = true;
        _running = false;

        _gate.Set();
        _textQueue.CompleteAdding();
        _audioQueue.CompleteAdding();

        _synthThread?.Join(5000);
        _playerThread?.Join(5000);

        _textQueue.Dispose();
        _audioQueue.Dispose();
        _gate.Dispose();

        _tts.DisposeAsync().AsTask().Wait();
        _audioOutput.Dispose();
    }
}