From e6c95021a4c4c4671c30b357836e4fc8c8ef8718 Mon Sep 17 00:00:00 2001 From: Huaaudio Date: Sat, 21 Mar 2026 08:24:17 +0100 Subject: [PATCH] update with streaming support --- pkg/audio/sentence.go | 96 +++++++++++++++++++++++++++++++++ pkg/channels/discord/discord.go | 88 +++++++++++++++++++++++++++--- pkg/channels/discord/voice.go | 39 ++++++++++++-- 3 files changed, 213 insertions(+), 10 deletions(-) create mode 100644 pkg/audio/sentence.go diff --git a/pkg/audio/sentence.go b/pkg/audio/sentence.go new file mode 100644 index 000000000..c7a9b2f26 --- /dev/null +++ b/pkg/audio/sentence.go @@ -0,0 +1,96 @@ +package audio + +import ( + "strings" + "unicode" +) + +// SplitSentences splits text into sentence-sized chunks suitable for TTS synthesis. +// It splits on sentence-ending punctuation (.!?\n) while avoiding false splits +// on abbreviations and decimal numbers. Very short fragments are merged with +// the next sentence to prevent choppy playback. +func SplitSentences(text string) []string { + if text == "" { + return nil + } + + var sentences []string + var current strings.Builder + runes := []rune(text) + + for i := 0; i < len(runes); i++ { + r := runes[i] + current.WriteRune(r) + + if r == '\n' { + s := strings.TrimSpace(current.String()) + if s != "" { + sentences = append(sentences, s) + } + current.Reset() + continue + } + + if r == '.' || r == '!' || r == '?' { + // Avoid splitting on decimal numbers like "3.14" + if r == '.' && i > 0 && unicode.IsDigit(runes[i-1]) && + i+1 < len(runes) && unicode.IsDigit(runes[i+1]) { + continue + } + + // Consume trailing punctuation and spaces (e.g., "..." or "?!") + for i+1 < len(runes) && (runes[i+1] == '.' || runes[i+1] == '!' || runes[i+1] == '?' || runes[i+1] == ' ') { + i++ + current.WriteRune(runes[i]) + } + + s := strings.TrimSpace(current.String()) + if s != "" { + sentences = append(sentences, s) + } + current.Reset() + } + } + + // Flush remaining text + if s := strings.TrimSpace(current.String()); s != "" { + sentences = append(sentences, s) + } + + // Merge very short fragments with the next sentence + return mergeShorties(sentences, 15) +} + +// mergeShorties merges sentences shorter than minLen characters with the following sentence. +func mergeShorties(sentences []string, minLen int) []string { + if len(sentences) <= 1 { + return sentences + } + + var merged []string + var buf string + + for _, s := range sentences { + if buf != "" { + buf += " " + s + if len([]rune(buf)) >= minLen { + merged = append(merged, buf) + buf = "" + } + } else if len([]rune(s)) < minLen { + buf = s + } else { + merged = append(merged, s) + } + } + + if buf != "" { + if len(merged) > 0 { + merged[len(merged)-1] += " " + buf + } else { + merged = append(merged, buf) + } + } + + return merged +} diff --git a/pkg/channels/discord/discord.go b/pkg/channels/discord/discord.go index 08e7fe718..c32a6409c 100644 --- a/pkg/channels/discord/discord.go +++ b/pkg/channels/discord/discord.go @@ -3,6 +3,7 @@ package discord import ( "context" "fmt" + "io" "net/http" "net/url" "os" @@ -14,6 +15,7 @@ import ( "github.com/bwmarrin/discordgo" "github.com/gorilla/websocket" + "github.com/sipeed/picoclaw/pkg/audio" "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/channels" "github.com/sipeed/picoclaw/pkg/config" @@ -45,6 +47,10 @@ type DiscordChannel struct { botUserID string // stored for mention checking bus *bus.MessageBus tts tts.TTSProvider + + // TTS interruption: cancel active playback when user speaks + ttsMu sync.Mutex + cancelTTS context.CancelFunc } func NewDiscordChannel(cfg config.DiscordConfig, bus *bus.MessageBus) (*DiscordChannel, error) { @@ -151,7 +157,16 @@ func (c *DiscordChannel) Send(ctx context.Context, msg bus.OutboundMessage) erro if c.tts != nil { if ch, err := c.session.State.Channel(channelID); err == nil && ch.GuildID != "" { if vc, ok := c.session.VoiceConnections[ch.GuildID]; ok && vc != nil { - go c.playTTS(context.Background(), vc, msg.Content) + // Cancel any previous TTS playback + c.ttsMu.Lock() + if c.cancelTTS != nil { + c.cancelTTS() + } + ttsCtx, ttsCancel := context.WithCancel(context.Background()) + c.cancelTTS = ttsCancel + c.ttsMu.Unlock() + + go c.playTTS(ttsCtx, vc, msg.Content) } } } @@ -651,14 +666,73 @@ func (c *DiscordChannel) listenVoiceControl(ctx context.Context) { } func (c *DiscordChannel) playTTS(ctx context.Context, vc *discordgo.VoiceConnection, text string) { - stream, err := c.tts.Synthesize(ctx, text) - if err != nil { - logger.ErrorCF("discord", "TTS synthesize failed", map[string]any{"error": err.Error()}) + sentences := audio.SplitSentences(text) + if len(sentences) == 0 { return } - defer stream.Close() - if err := streamOggOpusToDiscord(vc, stream); err != nil { - logger.ErrorCF("discord", "TTS playback failed", map[string]any{"error": err.Error()}) + logger.InfoCF("discord", "Starting streamed TTS", map[string]any{"sentences": len(sentences)}) + + // Pipeline: prefetch next sentence's audio while playing current + type ttResult struct { + stream io.ReadCloser + err error + } + + var prefetch chan ttResult + + for i, sentence := range sentences { + // Check for cancellation (interruption) + select { + case <-ctx.Done(): + logger.InfoCF("discord", "TTS interrupted", map[string]any{"at_sentence": i}) + return + default: + } + + // Start prefetching the NEXT sentence while we process the current one + var nextPrefetch chan ttResult + if i+1 < len(sentences) { + nextPrefetch = make(chan ttResult, 1) + nextSentence := sentences[i+1] + go func() { + s, e := c.tts.Synthesize(ctx, nextSentence) + nextPrefetch <- ttResult{s, e} + }() + } + + // Get the current sentence's audio + var stream io.ReadCloser + var err error + + if prefetch != nil { + // Use prefetched result from previous iteration + result := <-prefetch + stream, err = result.stream, result.err + } else { + // First sentence: synthesize directly + stream, err = c.tts.Synthesize(ctx, sentence) + } + + if err != nil { + logger.ErrorCF("discord", "TTS synthesize failed", map[string]any{"error": err.Error(), "sentence": i}) + prefetch = nextPrefetch + continue + } + + if err := streamOggOpusToDiscord(ctx, vc, stream); err != nil { + logger.ErrorCF("discord", "TTS playback failed", map[string]any{"error": err.Error(), "sentence": i}) + } + stream.Close() + + prefetch = nextPrefetch + } + + // Drain any leftover prefetch + if prefetch != nil { + result := <-prefetch + if result.stream != nil { + result.stream.Close() + } } } diff --git a/pkg/channels/discord/voice.go b/pkg/channels/discord/voice.go index 72f767b85..4c37e952f 100644 --- a/pkg/channels/discord/voice.go +++ b/pkg/channels/discord/voice.go @@ -2,6 +2,7 @@ package discord import ( "bytes" + "context" "fmt" "io" "time" @@ -46,7 +47,7 @@ func VoiceReceiveActive(vc *discordgo.VoiceConnection) bool { return vc != nil && vc.OpusRecv != nil } -func streamOggOpusToDiscord(vc *discordgo.VoiceConnection, r io.Reader) error { +func streamOggOpusToDiscord(ctx context.Context, vc *discordgo.VoiceConnection, r io.Reader) error { // Wait for the speaking transition to register vc.Speaking(true) defer vc.Speaking(false) @@ -55,6 +56,13 @@ func streamOggOpusToDiscord(vc *discordgo.VoiceConnection, r io.Reader) error { header := make([]byte, 27) for { + // Check for interruption + select { + case <-ctx.Done(): + return ctx.Err() + default: + } + if _, err := io.ReadFull(r, header); err != nil { if err == io.EOF || err == io.ErrUnexpectedEOF { return nil @@ -84,8 +92,11 @@ func streamOggOpusToDiscord(vc *discordgo.VoiceConnection, r io.Reader) error { if len(packet) > 0 { // Ignore Ogg Opus headers if !bytes.HasPrefix(packet, []byte("OpusHead")) && !bytes.HasPrefix(packet, []byte("OpusTags")) { - // Pacing is handled natively by vc.OpusSend blocking (it has an internal ticker) - vc.OpusSend <- packet + select { + case <-ctx.Done(): + return ctx.Err() + case vc.OpusSend <- packet: + } } // Start new packet packet = nil @@ -118,6 +129,8 @@ func (c *DiscordChannel) receiveVoice(vc *discordgo.VoiceConnection, guildID str }) var sequence uint64 = 0 + var interruptCount int + var lastInterruptAt time.Time for { select { @@ -139,6 +152,26 @@ func (c *DiscordChannel) receiveVoice(vc *discordgo.VoiceConnection, guildID str continue } + // Interruption detection: if user sends voice while TTS is playing, + // cancel TTS after a short debounce (3 packets in 200ms) + now := time.Now() + if now.Sub(lastInterruptAt) > 500*time.Millisecond { + interruptCount = 0 + } + interruptCount++ + lastInterruptAt = now + + if interruptCount >= 3 { + c.ttsMu.Lock() + if c.cancelTTS != nil { + c.cancelTTS() + c.cancelTTS = nil + logger.InfoCF("discord", "TTS interrupted by user voice", nil) + } + c.ttsMu.Unlock() + interruptCount = 0 + } + sequence++ chunk := bus.AudioChunk{