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 _log; private BlockingCollection _textQueue = new(); private BlockingCollection _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 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(); } }