diff --git a/go.mod b/go.mod index 727fbf1..a411033 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,7 @@ go 1.24.5 require ( github.com/inconshreveable/mousetrap v1.1.0 // indirect + github.com/sashabaranov/go-openai v1.41.1 // indirect github.com/spf13/cobra v1.9.1 // indirect github.com/spf13/pflag v1.0.6 // indirect ) diff --git a/go.sum b/go.sum index ffae55e..3a7a8f1 100644 --- a/go.sum +++ b/go.sum @@ -2,6 +2,8 @@ github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6N github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= +github.com/sashabaranov/go-openai v1.41.1 h1:zf5tM+GuxpyiyD9XZg8nCqu52eYFQg9OOew0gnIuDy4= +github.com/sashabaranov/go-openai v1.41.1/go.mod h1:lj5b/K+zjTSFxVLijLSTDZuP7adOgerWeFyZLUhAKRg= github.com/spf13/cobra v1.9.1 h1:CXSaggrXdbHK9CF+8ywj8Amf7PBRmPCOJugH954Nnlo= github.com/spf13/cobra v1.9.1/go.mod h1:nDyEzZ8ogv936Cinf6g1RU9MRY64Ir93oCnqb9wxYW0= github.com/spf13/pflag v1.0.6 h1:jFzHGLGAlb3ruxLB8MhbI6A8+AQX/2eW4qeyNZXNp2o= diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index 22636bd..e1e8148 100644 --- a/internal/daemon/daemon.go +++ b/internal/daemon/daemon.go @@ -158,6 +158,7 @@ func (d *Daemon) toggle() { d.mu.Unlock() go d.notifier.RecordingStarted() + go d.monitorPipelineErrors(p) case pipeline.Recording: d.stopPipeline() // aborted during recording (chunks not sent to transcriber yet) @@ -179,3 +180,25 @@ func (d *Daemon) toggle() { go d.notifier.Aborted() } } + +func (d *Daemon) monitorPipelineErrors(p pipeline.Pipeline) { + errorCh := p.GetErrorCh() + for { + select { + case pipelineErr := <-errorCh: + d.handlePipelineError(pipelineErr) + case <-d.ctx.Done(): + return + } + } +} + +func (d *Daemon) handlePipelineError(pipelineErr pipeline.PipelineError) { + message := pipelineErr.Message + + if pipelineErr.Err != nil { + message = fmt.Sprintf("%s: %v", message, pipelineErr.Err) + } + + d.notifier.Error(message) +} diff --git a/internal/notify/notify.go b/internal/notify/notify.go index 417f6e6..26fa1b2 100644 --- a/internal/notify/notify.go +++ b/internal/notify/notify.go @@ -11,29 +11,21 @@ type Notifier interface { Aborted() Transcribing() Error(msg string) + Notify(title, message string) } type Desktop struct{} -func (Desktop) RecordingStarted() { - cmd := exec.Command("notify-send", "-a", "Hyprvoice", "Hyprvoice: Recording Started") - if err := cmd.Run(); err != nil { - log.Printf("Failed to send notification: %v", err) - } +func (d Desktop) RecordingStarted() { + d.Notify("Hyprvoice", "Recording Started") } -func (Desktop) RecordingEnded() { - cmd := exec.Command("notify-send", "-a", "Hyprvoice", "Hyprvoice: Recording Ended") - if err := cmd.Run(); err != nil { - log.Printf("Failed to send notification: %v", err) - } +func (d Desktop) RecordingEnded() { + d.Notify("Hyprvoice", "Recording Ended") } -func (Desktop) Transcribing() { - cmd := exec.Command("notify-send", "-a", "Hyprvoice", "Hyprvoice: Transcribing...") - if err := cmd.Run(); err != nil { - log.Printf("Failed to send notification: %v", err) - } +func (d Desktop) Transcribing() { + d.Notify("Hyprvoice", "Transcribing...") } func (Desktop) Aborted() { @@ -44,18 +36,50 @@ func (Desktop) Aborted() { } func (Desktop) Error(msg string) { - cmd := exec.Command("notify-send", "-a", "Hyprvoice", "-u", "critical", msg) + cmd := exec.Command("notify-send", "-a", "Hyprvoice", "-u", "critical", "Hyprvoice Error", msg) if err := cmd.Run(); err != nil { log.Printf("Failed to send error notification: %v", err) } } -// Nop is a Notifier that does absolutely nothing. -// Useful in unit tests or headless builds. +func (Desktop) Notify(title, message string) { + cmd := exec.Command("notify-send", "-a", "Hyprvoice", title, message) + if err := cmd.Run(); err != nil { + log.Printf("Failed to send notification: %v", err) + } +} + +type Log struct{} + +func (l Log) RecordingStarted() { + l.Notify("Hyprvoice", "Recording Started") +} + +func (l Log) RecordingEnded() { + l.Notify("Hyprvoice", "Recording Ended") +} + +func (l Log) Transcribing() { + l.Notify("Hyprvoice", "Transcribing...") +} + +func (l Log) Aborted() { + l.Notify("Hyprvoice", "Aborted") +} + +func (l Log) Error(msg string) { + l.Notify("Hyprvoice Error", msg) +} + +func (Log) Notify(title, message string) { + log.Printf("%s: %s", title, message) +} + type Nop struct{} -func (Nop) RecordingStarted() {} -func (Nop) RecordingEnded() {} -func (Nop) Aborted() {} -func (Nop) Transcribing() {} -func (Nop) Error(msg string) {} +func (Nop) RecordingStarted() {} +func (Nop) RecordingEnded() {} +func (Nop) Aborted() {} +func (Nop) Transcribing() {} +func (Nop) Error(msg string) {} +func (Nop) Notify(title, message string) {} diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go index ab31e4b..92d11a9 100644 --- a/internal/pipeline/pipeline.go +++ b/internal/pipeline/pipeline.go @@ -3,15 +3,24 @@ package pipeline import ( "context" "log" + "os" "sync" + "sync/atomic" "time" "github.com/leonardotrapani/hyprvoice/internal/recording" + "github.com/leonardotrapani/hyprvoice/internal/transcriber" ) type Status string type Action string +type PipelineError struct { + Title string + Message string + Err error +} + const ( Idle Status = "idle" Recording Status = "recording" @@ -28,24 +37,26 @@ type Pipeline interface { Stop() Status() Status GetActionCh() chan<- Action + GetErrorCh() <-chan PipelineError } type pipeline struct { status Status actionCh chan Action + errorCh chan PipelineError - mu sync.RWMutex - wg sync.WaitGroup - - cancel context.CancelFunc - + mu sync.RWMutex + wg sync.WaitGroup + cancel context.CancelFunc stopOnce sync.Once - running bool + + running int32 } func New() Pipeline { return &pipeline{ actionCh: make(chan Action, 1), + errorCh: make(chan PipelineError, 10), } } @@ -79,6 +90,26 @@ func (p *pipeline) GetActionCh() chan<- Action { return p.actionCh } +func (p *pipeline) GetErrorCh() <-chan PipelineError { + p.mu.RLock() + defer p.mu.RUnlock() + return p.errorCh +} + +func (p *pipeline) sendError(title, message string, err error) { + pipelineErr := PipelineError{ + Title: title, + Message: message, + Err: err, + } + + select { + case p.errorCh <- pipelineErr: + default: + log.Printf("Pipeline: Error channel full, dropping error: %s", message) + } +} + func (p *pipeline) Stop() { p.stopOnce.Do(func() { cancel := p.getCancel() @@ -90,14 +121,10 @@ func (p *pipeline) Stop() { } func (p *pipeline) Run(ctx context.Context) { - p.mu.Lock() - if p.running { - p.mu.Unlock() + if !atomic.CompareAndSwapInt32(&p.running, 0, 1) { log.Printf("Pipeline: Already running, ignoring Run() call") return } - p.running = true - p.mu.Unlock() runCtx, cancel := context.WithTimeout(ctx, 5*time.Minute) p.setCancel(cancel) @@ -108,10 +135,8 @@ func (p *pipeline) Run(ctx context.Context) { func (p *pipeline) run(ctx context.Context) { defer func() { - log.Printf("Pipeline: defered run") - p.mu.Lock() - p.running = false - p.mu.Unlock() + atomic.StoreInt32(&p.running, 0) + p.setStatus(Idle) p.wg.Done() }() @@ -122,10 +147,45 @@ func (p *pipeline) run(ctx context.Context) { frameCh, errCh, err := recorder.Start(ctx) if err != nil { log.Printf("Pipeline: Recording error: %v", err) - p.setStatus(Idle) + p.sendError("Recording Error", "Failed to start recording", err) return } - defer recorder.Stop() + + defer func() { + if stopErr := recorder.Stop(); stopErr != nil { + log.Printf("Pipeline: Error stopping recorder: %v", stopErr) + p.sendError("Recording Error", "Failed to stop recorder cleanly", stopErr) + } + }() + + config := transcriber.DefaultConfig() + if apiKey := os.Getenv("OPENAI_API_KEY"); apiKey != "" { + config.APIKey = apiKey + } + + t, err := transcriber.NewTranscriber(config) + if err != nil { + log.Printf("Pipeline: Failed to create transcriber: %v", err) + p.sendError("Transcription Error", "Failed to create transcriber", err) + return + } + + log.Printf("Pipeline: Starting transcriber") + p.setStatus(Transcribing) + + tErrCh, err := t.Start(ctx, frameCh) + if err != nil { + log.Printf("Pipeline: Transcriber error: %v", err) + p.sendError("Transcription Error", "Failed to start transcriber", err) + return + } + + defer func() { + if stopErr := t.Stop(); stopErr != nil { + log.Printf("Pipeline: Error stopping transcriber: %v", stopErr) + p.sendError("Transcription Error", "Failed to stop transcriber cleanly", stopErr) + } + }() frameCount := 0 totalBytes := 0 @@ -138,12 +198,17 @@ func (p *pipeline) run(ctx context.Context) { log.Printf("Pipeline: Received frame #%d - Size: %d bytes, Timestamp: %v, Total bytes so far: %d", frameCount, len(frame.Data), frame.Timestamp.Format("15:04:05.000"), totalBytes) - p.setStatus(Transcribing) + case err := <-tErrCh: + if err != nil { + log.Printf("Pipeline: Transcription error: %v", err) + p.sendError("Transcription Error", "Transcription processing error", err) + return + } case err := <-errCh: if err != nil { log.Printf("Pipeline: Recording error: %v", err) - p.setStatus(Idle) + p.sendError("Recording Error", "Recording stream error", err) return } @@ -151,18 +216,31 @@ func (p *pipeline) run(ctx context.Context) { log.Printf("Pipeline: Received action: %v", action) switch action { case Inject: - log.Printf("Pipeline: Inject action received, stopping recording (TODO: stop transcribing)") + if p.status != Transcribing { + log.Printf("Pipeline: Inject action received, but not in transcribing state, ignoring") + continue + } + + log.Printf("Pipeline: Inject action received, stopping recording and getting transcription") if err := recorder.Stop(); err != nil { log.Printf("Pipeline: Error stopping recorder: %v", err) + p.sendError("Recording Error", "Failed to stop recorder during injection", err) } + p.setStatus(Injecting) - // Simulate injection work then return to idle + transcriptionText, err := t.GetTranscription() + if err != nil { + log.Printf("Pipeline: Error getting transcription: %v", err) + p.sendError("Transcription Error", "Failed to retrieve transcription", err) + } else { + log.Printf("Pipeline: Transcription text: %s", transcriptionText) + } + log.Printf("Pipeline: Simulating injection work") time.Sleep(10 * time.Millisecond) log.Printf("Pipeline: Injection work done, returning to idle") - p.setStatus(Idle) return } diff --git a/internal/recording/recording.go b/internal/recording/recording.go index 95ae94e..8d86ba4 100644 --- a/internal/recording/recording.go +++ b/internal/recording/recording.go @@ -33,9 +33,9 @@ func DefaultConfig() Config { SampleRate: 16000, Channels: 1, Format: "s16", - BufferSize: 4096, + BufferSize: 8192, Device: "", - ChannelBufferSize: 20, + ChannelBufferSize: 30, } } diff --git a/internal/transcriber/openai.go b/internal/transcriber/openai.go new file mode 100644 index 0000000..a291620 --- /dev/null +++ b/internal/transcriber/openai.go @@ -0,0 +1,278 @@ +package transcriber + +import ( + "bytes" + "context" + "encoding/binary" + "fmt" + "log" + "strings" + "sync" + "time" + + "github.com/leonardotrapani/hyprvoice/internal/recording" + "github.com/sashabaranov/go-openai" +) + +type OpenAITranscriber struct { + client *openai.Client + config Config + buffer *audioBuffer + cancel context.CancelFunc + wg sync.WaitGroup + mu sync.Mutex + transcribing bool + + transcriptionMu sync.RWMutex + transcriptionText strings.Builder +} + +type audioBuffer struct { + data []byte + mu sync.Mutex + lastAdd time.Time + maxSize int +} + +func NewOpenAITranscriber(config Config) *OpenAITranscriber { + client := openai.NewClient(config.APIKey) + + buffer := &audioBuffer{ + data: make([]byte, 0, config.ChunkSize*2), + maxSize: config.ChunkSize, + } + + return &OpenAITranscriber{ + client: client, + config: config, + buffer: buffer, + } +} + +func (t *OpenAITranscriber) Start(ctx context.Context, frameCh <-chan recording.AudioFrame) (<-chan error, error) { + t.mu.Lock() + if t.transcribing { + t.mu.Unlock() + return nil, fmt.Errorf("transcriber: already transcribing") + } + t.transcribing = true + t.mu.Unlock() + + transcribeCtx, cancel := context.WithCancel(ctx) + t.cancel = cancel + + errCh := make(chan error, 1) + + t.wg.Add(1) + go t.processFrames(transcribeCtx, frameCh, errCh) + + return errCh, nil +} + +func (t *OpenAITranscriber) Stop() error { + t.mu.Lock() + if !t.transcribing { + t.mu.Unlock() + return nil + } + cancel := t.cancel + t.mu.Unlock() + + if cancel != nil { + cancel() + } + t.wg.Wait() + return nil +} + +func (t *OpenAITranscriber) GetTranscription() (string, error) { + t.transcriptionMu.RLock() + defer t.transcriptionMu.RUnlock() + + if t.transcriptionText.Len() == 0 { + return "", nil + } + + return t.transcriptionText.String(), nil +} + +func (t *OpenAITranscriber) processFrames(ctx context.Context, frameCh <-chan recording.AudioFrame, errCh chan<- error) { + defer func() { + close(errCh) + t.mu.Lock() + t.transcribing = false + t.cancel = nil + t.mu.Unlock() + t.wg.Done() + }() + + ticker := time.NewTicker(t.config.BufferTime) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + if t.buffer.hasData() { + t.transcribeBuffer(ctx, errCh) + } + return + + case frame, ok := <-frameCh: + if !ok { + if t.buffer.hasData() { + t.transcribeBuffer(ctx, errCh) + } + return + } + t.buffer.addFrame(frame) + + case <-ticker.C: + if t.buffer.shouldFlush(t.config.BufferTime) { + t.transcribeBuffer(ctx, errCh) + } + } + } +} + +func (t *OpenAITranscriber) transcribeBuffer(ctx context.Context, errCh chan<- error) { + audioData := t.buffer.flush() + if len(audioData) == 0 { + return + } + + log.Printf("transcriber: sending %d bytes to OpenAI API", len(audioData)) + + wavData, err := t.convertToWAV(audioData) + if err != nil { + log.Printf("transcriber: failed to convert audio to WAV: %v", err) + select { + case errCh <- fmt.Errorf("transcriber: convert to WAV: %w", err): + default: + } + return + } + + req := openai.AudioRequest{ + Model: t.config.Model, + Reader: bytes.NewReader(wavData), + FilePath: "audio.wav", + Language: t.config.Language, + } + + start := time.Now() + resp, err := t.client.CreateTranscription(ctx, req) + duration := time.Since(start) + + if err != nil { + log.Printf("transcriber: API call failed after %v: %v", duration, err) + select { + case errCh <- fmt.Errorf("transcriber: transcription failed: %w", err): + default: + } + return + } + + if resp.Text != "" { + log.Printf("transcriber: received result in %v: %q", duration, resp.Text) + t.transcriptionMu.Lock() + if t.transcriptionText.Len() > 0 { + t.transcriptionText.WriteString(" ") + } + t.transcriptionText.WriteString(strings.TrimSpace(resp.Text)) + t.transcriptionMu.Unlock() + } else { + log.Printf("transcriber: received empty result after %v", duration) + } +} + +func (t *OpenAITranscriber) convertToWAV(rawAudio []byte) ([]byte, error) { + var buf bytes.Buffer + + const sampleRate = 16000 + const channels = 1 + const bitsPerSample = 16 + const byteRate = sampleRate * channels * bitsPerSample / 8 + const blockAlign = channels * bitsPerSample / 8 + + dataSize := len(rawAudio) + fileSize := 36 + dataSize + + buf.WriteString("RIFF") + binary.Write(&buf, binary.LittleEndian, uint32(fileSize)) + buf.WriteString("WAVE") + + buf.WriteString("fmt ") + binary.Write(&buf, binary.LittleEndian, uint32(16)) + binary.Write(&buf, binary.LittleEndian, uint16(1)) + binary.Write(&buf, binary.LittleEndian, uint16(channels)) + binary.Write(&buf, binary.LittleEndian, uint32(sampleRate)) + binary.Write(&buf, binary.LittleEndian, uint32(byteRate)) + binary.Write(&buf, binary.LittleEndian, uint16(blockAlign)) + binary.Write(&buf, binary.LittleEndian, uint16(bitsPerSample)) + + buf.WriteString("data") + binary.Write(&buf, binary.LittleEndian, uint32(dataSize)) + buf.Write(rawAudio) + + return buf.Bytes(), nil +} + +func (b *audioBuffer) addFrame(frame recording.AudioFrame) { + b.mu.Lock() + defer b.mu.Unlock() + + b.data = append(b.data, frame.Data...) + b.lastAdd = frame.Timestamp + + if len(b.data) > b.maxSize*2 { + b.data = b.data[len(b.data)-b.maxSize:] + } +} + +func (b *audioBuffer) flush() []byte { + b.mu.Lock() + defer b.mu.Unlock() + + if len(b.data) == 0 { + return nil + } + + result := make([]byte, len(b.data)) + copy(result, b.data) + b.data = b.data[:0] + + return result +} + +func (b *audioBuffer) hasData() bool { + b.mu.Lock() + defer b.mu.Unlock() + return len(b.data) > 0 +} + +func (b *audioBuffer) shouldFlush(bufferTime time.Duration) bool { + b.mu.Lock() + defer b.mu.Unlock() + + if len(b.data) == 0 { + return false + } + + if len(b.data) >= b.maxSize { + return true + } + + return time.Since(b.lastAdd) >= bufferTime +} + +func NewTranscriber(config Config) (Transcriber, error) { + switch config.Provider { + case "openai": + if config.APIKey == "" { + return nil, fmt.Errorf("OpenAI API key required") + } + return NewOpenAITranscriber(config), nil + default: + return nil, fmt.Errorf("unsupported provider: %s", config.Provider) + } +} diff --git a/internal/transcriber/transcriber.go b/internal/transcriber/transcriber.go new file mode 100644 index 0000000..5e5248f --- /dev/null +++ b/internal/transcriber/transcriber.go @@ -0,0 +1,39 @@ +package transcriber + +import ( + "context" + "time" + + "github.com/leonardotrapani/hyprvoice/internal/recording" +) + +type TranscriptionResult struct { + Text string + Timestamp time.Time + IsFinal bool +} + +type Transcriber interface { + Start(ctx context.Context, frameCh <-chan recording.AudioFrame) (<-chan error, error) + Stop() error + GetTranscription() (string, error) +} + +type Config struct { + Provider string + APIKey string + Language string + ChunkSize int + BufferTime time.Duration + Model string +} + +func DefaultConfig() Config { + return Config{ + Provider: "openai", + Language: "en", + ChunkSize: 16384, + BufferTime: 2 * time.Second, + Model: "whisper-1", + } +}