diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index e72009c..7658eea 100644 --- a/internal/daemon/daemon.go +++ b/internal/daemon/daemon.go @@ -172,36 +172,67 @@ func (d *Daemon) handle(c net.Conn) { } func (d *Daemon) toggle() { - switch d.status { + // Capture current state and prepare action under lock + d.mu.Lock() + status := d.status + var pipelineToStop pipeline.Pipeline + var actionChan chan<- pipeline.Action + + switch status { case Idle: - // Defensive: if a prior pipeline exists, stop and wait before starting a new one + // Clean up any existing pipeline first if d.pipeline != nil { - d.pipeline.Stop() + pipelineToStop = d.pipeline d.pipeline = nil } + // Start new pipeline p := pipeline.New() statusCh, actionCh := p.Run(d.ctx) - d.pipeline = p d.actionChannel = actionCh - go d.notifier.RecordingStarted() + d.mu.Unlock() + // Stop old pipeline if needed (outside lock) + if pipelineToStop != nil { + pipelineToStop.Stop() + } + + go d.notifier.RecordingStarted() go d.startStatusReader(d.ctx, statusCh) case Recording: + pipelineToStop = d.pipeline + d.mu.Unlock() + go d.notifier.RecordingEnded() - d.pipeline.Stop() + if pipelineToStop != nil { + pipelineToStop.Stop() + } case Transcribing: - select { - case d.actionChannel <- pipeline.Inject: - default: + actionChan = d.actionChannel + d.mu.Unlock() + + // Try to inject (non-blocking) + if actionChan != nil { + select { + case actionChan <- pipeline.Inject: + default: + } } case Injecting: + pipelineToStop = d.pipeline + d.mu.Unlock() + go d.notifier.RecordingEnded() - d.pipeline.Stop() // aborted during injection + if pipelineToStop != nil { + pipelineToStop.Stop() // aborted during injection + } + + default: + d.mu.Unlock() } } diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go index 15c4713..e7f16f8 100644 --- a/internal/pipeline/pipeline.go +++ b/internal/pipeline/pipeline.go @@ -5,6 +5,8 @@ import ( "log" "sync" "time" + + "github.com/leonardotrapani/hyprvoice/internal/recording" ) type Status string @@ -61,45 +63,62 @@ func (p *pipeline) run(ctx context.Context, statusCh chan<- Status) { log.Printf("Pipeline: Starting recording") statusCh <- Recording - // Recording phase - select { - case <-time.After(2 * time.Second): - log.Printf("Pipeline: TODO start recording, and on first chunk set transcribing and start streaming with Whisper") - statusCh <- Transcribing - case <-ctx.Done(): - log.Printf("Pipeline: Stopped during recording") + recorder := recording.NewDefaultRecorder() + frameCh, errCh, err := recorder.Start(ctx) + if err != nil { + log.Printf("Pipeline: Recording error: %v", err) + statusCh <- Idle return } + defer recorder.Stop() - // TODO: when integrating the recorder, ensure its context is cancelled here on exit paths - // Wait for an action or timeout - select { - case action := <-p.actionCh: - switch action { - case Inject: - log.Printf("Pipeline: Injection started") - statusCh <- Injecting + // Track statistics for logging + frameCount := 0 + totalBytes := 0 + startTime := time.Now() + + // Main event loop + for { + select { + case frame, ok := <-frameCh: + if !ok { + // Frame channel closed, recording ended + log.Printf("Pipeline: Recording ended. Total frames: %d, Total bytes: %d, Duration: %v", + frameCount, totalBytes, time.Since(startTime)) + statusCh <- Transcribing - // Injection work - select { - case <-time.After(1 * time.Second): - log.Printf("Pipeline: Injection complete") - statusCh <- Idle - case <-ctx.Done(): - log.Printf("Pipeline: Stopped during injection") return } - default: - // Unknown action: ignore for now + // Log frame details + 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) + // TODO: pass this channel to the transcriber + + case err := <-errCh: + if err != nil { + log.Printf("Pipeline: Recording error: %v", err) + statusCh <- Idle + return + } + + case action := <-p.actionCh: + log.Printf("Pipeline: Received action: %v", action) + if action == Inject { + log.Printf("Pipeline: Inject action received, stopping recording") + + if err := recorder.Stop(); err != nil { + log.Printf("Pipeline: Error stopping recorder: %v", err) + } + statusCh <- Injecting + return + } + + case <-ctx.Done(): + log.Printf("Pipeline: Context cancelled, stopping") + return } - - case <-time.After(10 * time.Second): // use context timeout - log.Printf("Pipeline: Auto-timeout, completing") - statusCh <- Idle // Instead of Completed - - case <-ctx.Done(): - log.Printf("Pipeline: Stopped during transcription wait") - return } } diff --git a/internal/recording/recording.go b/internal/recording/recording.go index 1c311db..d1597c4 100644 --- a/internal/recording/recording.go +++ b/internal/recording/recording.go @@ -32,7 +32,7 @@ func DefaultConfig() Config { return Config{ SampleRate: 16000, Channels: 1, - Format: "s16le", + Format: "s16", BufferSize: 4096, Device: "", ChannelBufferSize: 20, @@ -175,7 +175,6 @@ func (r *Recorder) captureLoop(ctx context.Context, frameCh chan<- AudioFrame, e case frameCh <- frame: sentCount++ case <-ctx.Done(): - // Context cancelled, stop cleanly after writing any pending frames. return default: droppedCount++ @@ -266,8 +265,8 @@ func (r *Recorder) validateConfig() error { if r.config.Format == "" { return fmt.Errorf("invalid Format: empty") } - // For s16le, sample frame size is 2 bytes per sample per channel. - if r.config.Format == "s16le" { + // For s16, sample frame size is 2 bytes per sample per channel. + if r.config.Format == "s16" { frameBytes := 2 * r.config.Channels if r.config.BufferSize%frameBytes != 0 { log.Printf("Recording: BufferSize %d not aligned to frame size %d; audio frames may split",