thread/hyprvoice
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: fanalize straeming adapters

Browse Source
This commit is contained in:
leonardotrapani
2026-02-01 17:53:48 +01:00
parent 8df3021a9d
commit 0025bf97b6
23 changed files with 489 additions and 644 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/transcriber/adapter_deepgram.go
+68 -7
View File
@@ -32,6 +32,14 @@ type DeepgramAdapter struct {
// reconnection config
maxRetries int
retryDelays []time.Duration
// finalization signaling
finalizeDone chan struct{}
}
// deepgramCloseStream message to signal end of audio
type deepgramCloseStream struct {
Type string `json:"type"`
}
// Deepgram WebSocket response types (incoming)
@@ -77,13 +85,14 @@ type deepgramError struct {
// lang: canonical language code (will be converted to provider format)
func NewDeepgramAdapter(endpoint *provider.EndpointConfig, apiKey, model, lang string) *DeepgramAdapter {
return &DeepgramAdapter{
endpoint: endpoint,
apiKey: apiKey,
model: model,
language: lang,
resultsCh: make(chan TranscriptionResult, 100),
maxRetries: 3,
retryDelays: defaultRetryDelays,
endpoint: endpoint,
apiKey: apiKey,
model: model,
language: lang,
resultsCh: make(chan TranscriptionResult, 100),
maxRetries: 3,
retryDelays: defaultRetryDelays,
finalizeDone: make(chan struct{}, 1),
}
}
@@ -294,6 +303,11 @@ func (a *DeepgramAdapter) readLoop() {
isFinal := resp.IsFinal || resp.SpeechFinal
if isFinal {
log.Printf("deepgram: final: %q", transcript)
// signal finalization (non-blocking)
select {
case a.finalizeDone <- struct{}{}:
default:
}
}
a.resultsCh <- TranscriptionResult{Text: transcript, IsFinal: isFinal}
}
@@ -371,6 +385,53 @@ func (a *DeepgramAdapter) Results() <-chan TranscriptionResult {
return a.resultsCh
}
// Finalize sends a CloseStream message to signal end of audio and waits for final results
func (a *DeepgramAdapter) Finalize(ctx context.Context) error {
a.mu.Lock()
if !a.started {
a.mu.Unlock()
return nil
}
conn := a.conn
a.mu.Unlock()
if conn == nil {
return nil
}
// drain any previous finalize signals
select {
case <-a.finalizeDone:
default:
}
// send CloseStream message
msg := deepgramCloseStream{Type: "CloseStream"}
a.mu.Lock()
err := a.conn.WriteJSON(msg)
a.mu.Unlock()
if err != nil {
log.Printf("deepgram: finalize write error: %v", err)
return fmt.Errorf("finalize write: %w", err)
}
log.Printf("deepgram: sent CloseStream, waiting for final transcript")
// wait for final result or timeout
select {
case <-a.finalizeDone:
log.Printf("deepgram: finalize complete")
return nil
case <-ctx.Done():
log.Printf("deepgram: finalize timeout")
return ctx.Err()
case <-a.ctx.Done():
return a.ctx.Err()
}
}
// Close gracefully closes the WebSocket connection
func (a *DeepgramAdapter) Close() error {
a.mu.Lock()
internal/transcriber/adapter_elevenlabs_streaming.go
+63 -1
View File
@@ -36,6 +36,9 @@ type ElevenLabsStreamingAdapter struct {
// reconnection config
maxRetries int
retryDelays []time.Duration
// finalization signaling
commitDone chan struct{}
}
// ElevenLabs WebSocket message types (outgoing)
@@ -69,6 +72,7 @@ func NewElevenLabsStreamingAdapter(endpoint *provider.EndpointConfig, apiKey, mo
resultsCh: make(chan TranscriptionResult, 100),
maxRetries: 3,
retryDelays: defaultRetryDelays,
commitDone: make(chan struct{}, 1),
}
}
@@ -270,10 +274,15 @@ func (a *ElevenLabsStreamingAdapter) readLoop() {
case "committed_transcript", "committed_transcript_with_timestamps":
// final result
log.Printf("elevenlabs-streaming: committed: %q", msg.Text)
if msg.Text != "" {
log.Printf("elevenlabs-streaming: committed: %q", msg.Text)
a.resultsCh <- TranscriptionResult{Text: msg.Text, IsFinal: true}
}
// signal finalization is done (non-blocking)
select {
case a.commitDone <- struct{}{}:
default:
}
case "error", "auth_error", "quota_exceeded", "rate_limited",
"queue_overflow", "resource_exhausted", "session_time_limit_exceeded",
@@ -353,6 +362,59 @@ func (a *ElevenLabsStreamingAdapter) Results() <-chan TranscriptionResult {
return a.resultsCh
}
// Finalize sends a commit message to force ElevenLabs to commit any pending audio
// and waits for the committed_transcript response
func (a *ElevenLabsStreamingAdapter) Finalize(ctx context.Context) error {
a.mu.Lock()
if !a.started {
a.mu.Unlock()
return nil
}
conn := a.conn
a.mu.Unlock()
if conn == nil {
return nil
}
// drain any previous commit signals
select {
case <-a.commitDone:
default:
}
// send empty audio chunk with commit=true to force finalization
msg := elevenLabsInputAudioChunk{
MessageType: "input_audio_chunk",
AudioBase64: "",
Commit: true,
SampleRate: 16000,
}
a.mu.Lock()
err := a.conn.WriteJSON(msg)
a.mu.Unlock()
if err != nil {
log.Printf("elevenlabs-streaming: finalize write error: %v", err)
return fmt.Errorf("finalize write: %w", err)
}
log.Printf("elevenlabs-streaming: sent commit, waiting for final transcript")
// wait for committed_transcript or timeout
select {
case <-a.commitDone:
log.Printf("elevenlabs-streaming: finalize complete")
return nil
case <-ctx.Done():
log.Printf("elevenlabs-streaming: finalize timeout")
return ctx.Err()
case <-a.ctx.Done():
return a.ctx.Err()
}
}
// Close gracefully closes the WebSocket connection
func (a *ElevenLabsStreamingAdapter) Close() error {
a.mu.Lock()
internal/transcriber/adapter_openai_realtime.go
+67 -8
View File
@@ -35,6 +35,9 @@ type OpenAIRealtimeAdapter struct {
// track current item for transcription
currentItemID string
// finalization signaling
transcriptionDone chan struct{}
}
// OpenAI Realtime WebSocket message types (outgoing)
@@ -103,13 +106,14 @@ type openaiRealtimeError struct {
// lang: canonical language code (will be used for transcription config)
func NewOpenAIRealtimeAdapter(endpoint *provider.EndpointConfig, apiKey, model, lang string) *OpenAIRealtimeAdapter {
return &OpenAIRealtimeAdapter{
endpoint: endpoint,
apiKey: apiKey,
model: model,
language: lang,
resultsCh: make(chan TranscriptionResult, 100),
maxRetries: 3,
retryDelays: defaultRetryDelays,
endpoint: endpoint,
apiKey: apiKey,
model: model,
language: lang,
resultsCh: make(chan TranscriptionResult, 100),
maxRetries: 3,
retryDelays: defaultRetryDelays,
transcriptionDone: make(chan struct{}, 1),
}
}
@@ -372,10 +376,15 @@ func (a *OpenAIRealtimeAdapter) handleEvent(event openaiRealtimeServerEvent) {
case "conversation.item.input_audio_transcription.completed":
// final transcription result
log.Printf("openai-realtime: transcription completed: %q", event.Transcript)
if event.Transcript != "" {
log.Printf("openai-realtime: transcription completed: %q", event.Transcript)
a.resultsCh <- TranscriptionResult{Text: event.Transcript, IsFinal: true}
}
// signal finalization (non-blocking)
select {
case a.transcriptionDone <- struct{}{}:
default:
}
case "conversation.item.input_audio_transcription.failed":
log.Printf("openai-realtime: transcription failed for item %s", event.ItemID)
@@ -500,6 +509,56 @@ func (a *OpenAIRealtimeAdapter) Results() <-chan TranscriptionResult {
return a.resultsCh
}
// Finalize sends a commit message to force OpenAI to process any pending audio
// and waits for the transcription.completed response
func (a *OpenAIRealtimeAdapter) Finalize(ctx context.Context) error {
a.mu.Lock()
if !a.started {
a.mu.Unlock()
return nil
}
conn := a.conn
a.mu.Unlock()
if conn == nil {
return nil
}
// drain any previous transcription signals
select {
case <-a.transcriptionDone:
default:
}
// send input_audio_buffer.commit to force processing of pending audio
msg := openaiRealtimeInputAudioCommit{
Type: "input_audio_buffer.commit",
}
a.mu.Lock()
err := a.conn.WriteJSON(msg)
a.mu.Unlock()
if err != nil {
log.Printf("openai-realtime: finalize write error: %v", err)
return fmt.Errorf("finalize write: %w", err)
}
log.Printf("openai-realtime: sent commit, waiting for final transcription")
// wait for transcription.completed or timeout
select {
case <-a.transcriptionDone:
log.Printf("openai-realtime: finalize complete")
return nil
case <-ctx.Done():
log.Printf("openai-realtime: finalize timeout")
return ctx.Err()
case <-a.ctx.Done():
return a.ctx.Err()
}
}
// Close gracefully closes the WebSocket connection
func (a *OpenAIRealtimeAdapter) Close() error {
a.mu.Lock()
internal/transcriber/streaming.go
+5
View File
@@ -20,6 +20,11 @@ type StreamingAdapter interface {
// Results returns a channel that receives transcription results (partial and final)
Results() <-chan TranscriptionResult
// Finalize signals end of audio input and waits for final transcription results.
// This should be called before Close to ensure all pending audio is committed.
// The ctx controls the timeout for waiting on final results.
Finalize(ctx context.Context) error
// Close gracefully closes the streaming connection
Close() error
}
internal/transcriber/streaming_transcriber.go
+44 -7
View File
@@ -5,6 +5,7 @@ import (
"log"
"strings"
"sync"
"time"
"github.com/leonardotrapani/hyprvoice/internal/recording"
)
@@ -83,18 +84,45 @@ func (t *StreamingTranscriber) receiveResults(errCh chan<- error) {
for {
select {
case <-t.ctx.Done():
// context cancelled, drain any remaining results before exiting
t.drainRemainingResults(resultsCh)
return
case result, ok := <-resultsCh:
if !ok {
return
}
if result.Error != nil {
select {
case errCh <- result.Error:
default:
}
log.Printf("streaming transcriber: result error: %v", result.Error)
continue
t.processResult(result, errCh)
}
}
}
func (t *StreamingTranscriber) processResult(result TranscriptionResult, errCh chan<- error) {
if result.Error != nil {
select {
case errCh <- result.Error:
default:
}
log.Printf("streaming transcriber: result error: %v", result.Error)
return
}
if result.IsFinal && result.Text != "" {
t.mu.Lock()
if t.finalText.Len() > 0 {
t.finalText.WriteString(" ")
}
t.finalText.WriteString(result.Text)
t.mu.Unlock()
}
}
func (t *StreamingTranscriber) drainRemainingResults(resultsCh <-chan TranscriptionResult) {
// give a short window to collect any final results already in the channel
timeout := time.After(100 * time.Millisecond)
for {
select {
case result, ok := <-resultsCh:
if !ok {
return
}
if result.IsFinal && result.Text != "" {
t.mu.Lock()
@@ -104,11 +132,20 @@ func (t *StreamingTranscriber) receiveResults(errCh chan<- error) {
t.finalText.WriteString(result.Text)
t.mu.Unlock()
}
case <-timeout:
return
}
}
}
func (t *StreamingTranscriber) Stop(ctx context.Context) error {
// finalize adapter first to commit pending audio and wait for final results
// this must happen before canceling context so receiveResults can collect them
if err := t.adapter.Finalize(ctx); err != nil {
log.Printf("streaming transcriber: finalize error (continuing): %v", err)
}
// now cancel context to stop goroutines
if t.cancel != nil {
t.cancel()
}
internal/transcriber/transcriber_test.go
+8
View File
@@ -688,6 +688,7 @@ type MockStreamingAdapter struct {
StartFunc func(ctx context.Context, language string) error
SendChunkFunc func(audio []byte) error
ResultsFunc func() <-chan TranscriptionResult
FinalizeFunc func(ctx context.Context) error
CloseFunc func() error
resultsCh chan TranscriptionResult
@@ -720,6 +721,13 @@ func (m *MockStreamingAdapter) Results() <-chan TranscriptionResult {
return m.resultsCh
}
func (m *MockStreamingAdapter) Finalize(ctx context.Context) error {
if m.FinalizeFunc != nil {
return m.FinalizeFunc(ctx)
}
return nil
}
func (m *MockStreamingAdapter) Close() error {
if m.CloseFunc != nil {
return m.CloseFunc()