From e8fd6c1d49feb1ea6b421c2589ffc7342582f83c Mon Sep 17 00:00:00 2001 From: LeonardoTrapani Date: Sat, 16 Aug 2025 20:56:56 +0200 Subject: [PATCH] general refactor for simpler code --- internal/daemon/daemon.go | 27 +-- internal/notify/notify.go | 35 --- internal/pipeline/pipeline.go | 266 ++++++++++----------- internal/recording/recording.go | 18 +- internal/transcriber/simple_transcriber.go | 2 +- internal/transcriber/transcriber.go | 12 +- 6 files changed, 147 insertions(+), 213 deletions(-) diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index e1e8148..a5d5f09 100644 --- a/internal/daemon/daemon.go +++ b/internal/daemon/daemon.go @@ -157,27 +157,28 @@ func (d *Daemon) toggle() { d.pipeline = p d.mu.Unlock() - go d.notifier.RecordingStarted() + go d.notifier.Notify("Hyprvoice", "Recording Started") go d.monitorPipelineErrors(p) case pipeline.Recording: d.stopPipeline() // aborted during recording (chunks not sent to transcriber yet) - go d.notifier.Aborted() + go d.notifier.Error("Recording Aborted") case pipeline.Transcribing: d.mu.RLock() if d.pipeline != nil { actionChan := d.pipeline.GetActionCh() + log.Printf("Daemon: Sending inject action to pipeline") d.mu.RUnlock() actionChan <- pipeline.Inject } else { d.mu.RUnlock() } - go d.notifier.RecordingEnded() + go d.notifier.Notify("Hyprvoice", "Recording Ended... Transcribing") case pipeline.Injecting: d.stopPipeline() // aborted during injection - go d.notifier.Aborted() + go d.notifier.Error("Injection Aborted") } } @@ -186,19 +187,15 @@ func (d *Daemon) monitorPipelineErrors(p pipeline.Pipeline) { for { select { case pipelineErr := <-errorCh: - d.handlePipelineError(pipelineErr) + message := pipelineErr.Message + + if pipelineErr.Err != nil { + message = fmt.Sprintf("%s: %v", message, pipelineErr.Err) + } + + d.notifier.Error(message) 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 26fa1b2..ff54b9c 100644 --- a/internal/notify/notify.go +++ b/internal/notify/notify.go @@ -6,10 +6,6 @@ import ( ) type Notifier interface { - RecordingStarted() - RecordingEnded() - Aborted() - Transcribing() Error(msg string) Notify(title, message string) } @@ -20,21 +16,10 @@ func (d Desktop) RecordingStarted() { d.Notify("Hyprvoice", "Recording Started") } -func (d Desktop) RecordingEnded() { - d.Notify("Hyprvoice", "Recording Ended") -} - func (d Desktop) Transcribing() { d.Notify("Hyprvoice", "Transcribing...") } -func (Desktop) Aborted() { - cmd := exec.Command("notify-send", "-a", "Hyprvoice", "-u", "critical", "Hyprvoice: Aborted") - if err := cmd.Run(); err != nil { - log.Printf("Failed to send abort notification: %v", err) - } -} - func (Desktop) Error(msg string) { cmd := exec.Command("notify-send", "-a", "Hyprvoice", "-u", "critical", "Hyprvoice Error", msg) if err := cmd.Run(); err != nil { @@ -51,22 +36,6 @@ func (Desktop) Notify(title, message string) { 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) } @@ -77,9 +46,5 @@ func (Log) Notify(title, message string) { type Nop struct{} -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 e771d27..1f02d2d 100644 --- a/internal/pipeline/pipeline.go +++ b/internal/pipeline/pipeline.go @@ -3,7 +3,6 @@ package pipeline import ( "context" "log" - "os" "sync" "sync/atomic" "time" @@ -50,7 +49,7 @@ type pipeline struct { cancel context.CancelFunc stopOnce sync.Once - running int32 + running atomic.Bool } func New() Pipeline { @@ -59,6 +58,88 @@ func New() Pipeline { errorCh: make(chan PipelineError, 10), } } +func (p *pipeline) Run(ctx context.Context) { + if !p.running.CompareAndSwap(false, true) { + log.Printf("Pipeline: Already running, ignoring Run() call") + return + } + + runCtx, cancel := context.WithTimeout(ctx, 5*time.Minute) + p.setCancel(cancel) + + p.wg.Add(1) + go p.run(runCtx) +} + +func (p *pipeline) run(ctx context.Context) { + defer func() { + p.running.Store(false) + p.setStatus(Idle) + p.wg.Done() + }() + + log.Printf("Pipeline: Starting recording") + p.setStatus(Recording) + + recorder := recording.NewDefaultRecorder() + frameCh, rErrCh, err := recorder.Start(ctx) + + if err != nil { + log.Printf("Pipeline: Recording error: %v", err) + p.sendError("Recording Error", "Failed to start recording", err) + return + } + + defer recorder.Stop() + + t, err := transcriber.NewDefaultTranscriber() + 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(ctx); stopErr != nil { + log.Printf("Pipeline: Error stopping transcriber: %v", stopErr) + p.sendError("Transcription Error", "Failed to stop transcriber cleanly", stopErr) + } + }() + + for { + select { + case <-frameCh: + + case action := <-p.actionCh: + switch action { + case Inject: + p.handleInjectAction(ctx, recorder, t) + return + } + + case err := <-tErrCh: + p.handleTranscriberError(err) + return + + case err := <-rErrCh: + p.handleRecordingError(err) + return + + case <-ctx.Done(): + return + } + } +} func (p *pipeline) Status() Status { p.mu.RLock() @@ -110,6 +191,43 @@ func (p *pipeline) sendError(title, message string, err error) { } } +func (p *pipeline) handleTranscriberError(err error) { + p.sendError("Transcription Error", "Transcription processing error", err) +} + +func (p *pipeline) handleRecordingError(err error) { + p.sendError("Recording Error", "Recording stream error", err) +} + +func (p *pipeline) handleInjectAction(ctx context.Context, recorder *recording.Recorder, t transcriber.Transcriber) { + status := p.Status() + + if status != Transcribing { + log.Printf("Pipeline: Inject action received, but not in transcribing state, ignoring") + return + } + + log.Printf("Pipeline: Inject action received, stopping recording and finalizing transcription") + p.setStatus(Injecting) + + recorder.Stop() + + if err := t.Stop(ctx); err != nil { + p.sendError("Transcription Error", "Failed to stop transcriber during injection", err) + } + + transcriptionText, err := t.GetFinalTranscription() + if err != nil { + p.sendError("Transcription Error", "Failed to retrieve transcription", err) + return + } + log.Printf("Pipeline: Final transcription text: %s", transcriptionText) + + log.Printf("Pipeline: Simulating injection work") + time.Sleep(10 * time.Millisecond) + log.Printf("Pipeline: Injection work done, returning to idle") +} + func (p *pipeline) Stop() { p.stopOnce.Do(func() { cancel := p.getCancel() @@ -119,147 +237,3 @@ func (p *pipeline) Stop() { }) p.wg.Wait() } - -func (p *pipeline) Run(ctx context.Context) { - if !atomic.CompareAndSwapInt32(&p.running, 0, 1) { - log.Printf("Pipeline: Already running, ignoring Run() call") - return - } - - runCtx, cancel := context.WithTimeout(ctx, 5*time.Minute) - p.setCancel(cancel) - - p.wg.Add(1) - go p.run(runCtx) -} - -func (p *pipeline) run(ctx context.Context) { - defer func() { - atomic.StoreInt32(&p.running, 0) - p.setStatus(Idle) - p.wg.Done() - }() - - log.Printf("Pipeline: Starting recording") - p.setStatus(Recording) - - recorder := recording.NewDefaultRecorder() - frameCh, errCh, err := recorder.Start(ctx) - if err != nil { - log.Printf("Pipeline: Recording error: %v", err) - p.sendError("Recording Error", "Failed to start recording", err) - return - } - - 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(ctx); stopErr != nil { - log.Printf("Pipeline: Error stopping transcriber: %v", stopErr) - p.sendError("Transcription Error", "Failed to stop transcriber cleanly", stopErr) - } - }() - - frameCount := 0 - totalBytes := 0 - - for { - select { - case frame := <-frameCh: - frameCount++ - totalBytes += len(frame.Data) - // 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) - - 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.sendError("Recording Error", "Recording stream error", err) - return - } - - case action := <-p.actionCh: - log.Printf("Pipeline: Received action: %v", action) - switch action { - case Inject: - 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 finalizing transcription") - p.setStatus(Injecting) - - // Stop the recorder first - this will close the frameCh channel - 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) - } - - // Wait for the recorder to fully stop and frameCh to be closed - // The transcriber will process any remaining frames when frameCh closes - // Drain the frameCh to ensure it's fully closed - for range frameCh { - // Continue draining until channel is closed - } - - // Stop the transcriber to ensure all buffered audio is processed - if err := t.Stop(ctx); err != nil { - log.Printf("Pipeline: Error stopping transcriber: %v", err) - p.sendError("Transcription Error", "Failed to stop transcriber during injection", err) - } - - // Now get the final transcription which includes all processed audio - 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: Final transcription text: %s", transcriptionText) - } - - log.Printf("Pipeline: Simulating injection work") - time.Sleep(10 * time.Millisecond) - log.Printf("Pipeline: Injection work done, returning to idle") - return - } - - case <-ctx.Done(): - log.Printf("Pipeline: Context cancelled, stopping") - return - } - } -} diff --git a/internal/recording/recording.go b/internal/recording/recording.go index 8d86ba4..e5d9a9c 100644 --- a/internal/recording/recording.go +++ b/internal/recording/recording.go @@ -87,21 +87,14 @@ func (r *Recorder) Start(ctx context.Context) (<-chan AudioFrame, <-chan error, return frameCh, errCh, nil } -func (r *Recorder) Stop() error { +func (r *Recorder) Stop() { if !r.recording.Load() { - return nil + return } - r.mu.Lock() - cancel := r.cancel - r.cancel = nil - r.mu.Unlock() + r.requestCancel() - if cancel != nil { - cancel() - } r.wg.Wait() - return nil } func (r *Recorder) captureLoop(ctx context.Context, frameCh chan<- AudioFrame, errCh chan<- error) { @@ -110,12 +103,7 @@ func (r *Recorder) captureLoop(ctx context.Context, frameCh chan<- AudioFrame, e close(errCh) r.recording.Store(false) - // Ensure any child process is reaped. r.mu.Lock() - if r.cmd != nil { - _ = r.cmd.Wait() - r.cmd = nil - } r.cancel = nil r.mu.Unlock() diff --git a/internal/transcriber/simple_transcriber.go b/internal/transcriber/simple_transcriber.go index 99110ae..53a7da4 100644 --- a/internal/transcriber/simple_transcriber.go +++ b/internal/transcriber/simple_transcriber.go @@ -61,7 +61,7 @@ func (t *SimpleTranscriber) Stop(ctx context.Context) error { return t.transcribeAll(ctx) } -func (t *SimpleTranscriber) GetTranscription() (string, error) { +func (t *SimpleTranscriber) GetFinalTranscription() (string, error) { t.transcriptionMu.RLock() defer t.transcriptionMu.RUnlock() return t.transcriptionText, nil diff --git a/internal/transcriber/transcriber.go b/internal/transcriber/transcriber.go index 01a1c26..7bfeede 100644 --- a/internal/transcriber/transcriber.go +++ b/internal/transcriber/transcriber.go @@ -3,6 +3,7 @@ package transcriber import ( "context" "fmt" + "os" "github.com/leonardotrapani/hyprvoice/internal/recording" ) @@ -11,7 +12,7 @@ import ( type Transcriber interface { Start(ctx context.Context, frameCh <-chan recording.AudioFrame) (<-chan error, error) Stop(ctx context.Context) error - GetTranscription() (string, error) + GetFinalTranscription() (string, error) } // Adapter interface for different transcription backends @@ -59,3 +60,12 @@ func NewTranscriber(config Config) (Transcriber, error) { return transcriber, nil } + +func NewDefaultTranscriber() (Transcriber, error) { + config := DefaultConfig() + if apiKey := os.Getenv("OPENAI_API_KEY"); apiKey != "" { + config.APIKey = apiKey + } + + return NewTranscriber(config) +}