diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index 7658eea..0313436 100644 --- a/internal/daemon/daemon.go +++ b/internal/daemon/daemon.go @@ -16,25 +16,14 @@ import ( "github.com/leonardotrapani/hyprvoice/internal/pipeline" ) -type Status = pipeline.Status - -const ( - Idle = pipeline.Idle - Recording = pipeline.Recording - Transcribing = pipeline.Transcribing - Injecting = pipeline.Injecting -) - type Daemon struct { mu sync.RWMutex - status Status notifier notify.Notifier ctx context.Context cancel context.CancelFunc - pipeline pipeline.Pipeline - actionChannel chan<- pipeline.Action + pipeline pipeline.Pipeline } func New(n notify.Notifier) *Daemon { @@ -46,48 +35,16 @@ func New(n notify.Notifier) *Daemon { notifier: n, ctx: ctx, cancel: cancel, - status: Idle, } return d } -func (d *Daemon) Status() Status { - d.mu.RLock() - defer d.mu.RUnlock() - return d.status -} - -func (d *Daemon) startStatusReader(ctx context.Context, statusCh <-chan Status) { - defer func() { - // Always clean up when this goroutine exits - d.mu.Lock() - d.status = Idle - d.pipeline = nil - d.actionChannel = nil - d.mu.Unlock() - }() - - for { - select { - case status, ok := <-statusCh: - if !ok { - return // Channel closed - } - - d.mu.Lock() - oldStatus := d.status - d.status = status - d.mu.Unlock() - - if oldStatus != status { - log.Printf("Status changed: %s -> %s", oldStatus, status) - } - - case <-ctx.Done(): - return // Context cancelled - } +func (d *Daemon) status() pipeline.Status { + if d.pipeline == nil { + return pipeline.Idle } + return d.pipeline.Status() } func (d *Daemon) Run() error { @@ -158,7 +115,7 @@ func (d *Daemon) handle(c net.Conn) { d.toggle() fmt.Fprint(c, "OK toggled\n") case 's': - status := d.Status() + status := d.status() fmt.Fprintf(c, "STATUS status=%s\n", status) case 'v': fmt.Fprintf(c, "STATUS proto=%s\n", bus.ProtoVer) @@ -172,64 +129,33 @@ func (d *Daemon) handle(c net.Conn) { } func (d *Daemon) toggle() { - // 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: - // Clean up any existing pipeline first + switch d.status() { + case pipeline.Idle: + p := pipeline.New() + p.Run(d.ctx) + d.pipeline = p + go d.notifier.RecordingStarted() + + case pipeline.Recording: + go d.notifier.Aborted() if d.pipeline != nil { - pipelineToStop = d.pipeline + d.pipeline.Stop() // aborted during recording (chunks not sent to transcriber yet) d.pipeline = nil } - // Start new pipeline - p := pipeline.New() - statusCh, actionCh := p.Run(d.ctx) - d.pipeline = p - d.actionChannel = actionCh - - 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() - + case pipeline.Transcribing: go d.notifier.RecordingEnded() - if pipelineToStop != nil { - pipelineToStop.Stop() - } + actionChan = d.pipeline.Actions() + actionChan <- pipeline.Inject - case Transcribing: - actionChan = d.actionChannel - d.mu.Unlock() + case pipeline.Injecting: + go d.notifier.Aborted() - // 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() - if pipelineToStop != nil { - pipelineToStop.Stop() // aborted during injection + if d.pipeline != nil { + d.pipeline.Stop() // aborted during injection + d.pipeline = nil } default: diff --git a/internal/notify/notify.go b/internal/notify/notify.go index 2f98f93..80fd9a6 100644 --- a/internal/notify/notify.go +++ b/internal/notify/notify.go @@ -8,6 +8,7 @@ import ( type Notifier interface { RecordingStarted() RecordingEnded() + Aborted() Transcribing() Error(msg string) } @@ -35,6 +36,13 @@ func (Desktop) 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", msg) if err := cmd.Run(); err != nil { diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go index e7f16f8..363cc4b 100644 --- a/internal/pipeline/pipeline.go +++ b/internal/pipeline/pipeline.go @@ -10,6 +10,7 @@ import ( ) type Status string +type Action string const ( Idle Status = "idle" @@ -18,21 +19,19 @@ const ( Injecting Status = "injecting" ) -// Action represents commands sent to the pipeline to drive state -// transitions or trigger work. Keeping this separate from Status -// preserves directionality: callers send Actions, pipeline emits Statuses. -type Action int - const ( - Inject Action = iota + Inject Action = "inject" ) type Pipeline interface { - Run(ctx context.Context) (<-chan Status, chan<- Action) + Run(ctx context.Context) Stop() + Status() Status + Actions() chan<- Action } type pipeline struct { + status Status actionCh chan Action wg sync.WaitGroup cancel context.CancelFunc @@ -44,74 +43,75 @@ func New() Pipeline { } } -func (p *pipeline) Run(ctx context.Context) (<-chan Status, chan<- Action) { - statusCh := make(chan Status, 1) +func (p *pipeline) Status() Status { + return p.status +} + +func (p *pipeline) Actions() chan<- Action { + return p.actionCh +} + +func (p *pipeline) Stop() { + if p.cancel != nil { + p.cancel() + } + p.wg.Wait() +} + +func (p *pipeline) Run(ctx context.Context) { runCtx, cancel := context.WithTimeout(ctx, 5*time.Minute) p.cancel = cancel p.wg.Add(1) - go p.run(runCtx, statusCh) - return statusCh, p.actionCh + go p.run(runCtx) } -func (p *pipeline) run(ctx context.Context, statusCh chan<- Status) { +func (p *pipeline) run(ctx context.Context) { defer func() { - // Ensure status channel is closed and wg is decremented - close(statusCh) p.wg.Done() }() log.Printf("Pipeline: Starting recording") - statusCh <- Recording + p.status = Recording recorder := recording.NewDefaultRecorder() frameCh, errCh, err := recorder.Start(ctx) if err != nil { log.Printf("Pipeline: Recording error: %v", err) - statusCh <- Idle + p.status = Idle return } defer recorder.Stop() - // 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 - - return - } - // Log frame details + 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) - // TODO: pass this channel to the transcriber + + p.status = Transcribing case err := <-errCh: if err != nil { log.Printf("Pipeline: Recording error: %v", err) - statusCh <- Idle + p.status = Idle return } case action := <-p.actionCh: log.Printf("Pipeline: Received action: %v", action) - if action == Inject { - log.Printf("Pipeline: Inject action received, stopping recording") + switch action { + case Inject: + log.Printf("Pipeline: Inject action received, stopping recording (TODO: stop transcribing)") if err := recorder.Stop(); err != nil { log.Printf("Pipeline: Error stopping recorder: %v", err) } - statusCh <- Injecting + p.status = Injecting return } @@ -121,10 +121,3 @@ func (p *pipeline) run(ctx context.Context, statusCh chan<- Status) { } } } - -func (p *pipeline) Stop() { - if p.cancel != nil { - p.cancel() - } - p.wg.Wait() -} diff --git a/internal/recording/recording.go b/internal/recording/recording.go index d1597c4..95ae94e 100644 --- a/internal/recording/recording.go +++ b/internal/recording/recording.go @@ -71,7 +71,6 @@ func (r *Recorder) Start(ctx context.Context) (<-chan AudioFrame, <-chan error, return nil, nil, fmt.Errorf("PipeWire not available: %w", err) } - // Create a cancellable context specific to this recording session. recordingCtx, cancel := context.WithCancel(ctx) frameCh := make(chan AudioFrame, r.config.ChannelBufferSize) @@ -150,7 +149,6 @@ func (r *Recorder) captureLoop(ctx context.Context, frameCh chan<- AudioFrame, e r.cmd = cmd r.mu.Unlock() - // Log stderr lines to aid diagnostics. go func() { scanner := bufio.NewScanner(stderr) for scanner.Scan() { @@ -158,47 +156,46 @@ func (r *Recorder) captureLoop(ctx context.Context, frameCh chan<- AudioFrame, e } }() - buffer := make([]byte, r.config.BufferSize) - var sentCount int - var droppedCount int - lastDropLog := time.Now() - for { - n, readErr := stdout.Read(buffer) - if n > 0 { - frameData := make([]byte, n) - copy(frameData, buffer[:n]) - - frame := AudioFrame{Data: frameData, Timestamp: time.Now()} - - select { - case frameCh <- frame: - sentCount++ - case <-ctx.Done(): - return - default: - droppedCount++ - if time.Since(lastDropLog) > time.Second { - log.Printf("Recording: dropped %d frames due to backpressure", droppedCount) - lastDropLog = time.Now() - droppedCount = 0 - } - } - } - - if readErr != nil { - if errors.Is(readErr, io.EOF) { - return - } - r.emitErr(errCh, fmt.Errorf("read audio: %w", readErr)) - r.requestCancel() - return - } - select { case <-ctx.Done(): return default: + buffer := make([]byte, r.config.BufferSize) + var sentCount int + var droppedCount int + lastDropLog := time.Now() + n, readErr := stdout.Read(buffer) + if n > 0 { + frameData := make([]byte, n) + copy(frameData, buffer[:n]) + + frame := AudioFrame{Data: frameData, Timestamp: time.Now()} + + select { + case frameCh <- frame: + sentCount++ + case <-ctx.Done(): + return + default: + droppedCount++ + if time.Since(lastDropLog) > time.Second { + log.Printf("Recording: dropped %d frames due to backpressure", droppedCount) + lastDropLog = time.Now() + droppedCount = 0 + } + } + } + + if readErr != nil { + if errors.Is(readErr, io.EOF) { + return + } + r.emitErr(errCh, fmt.Errorf("read audio: %w", readErr)) + r.requestCancel() + return + } + } } }