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

refactor moving status to pipeline not daemon

Browse Source
This commit is contained in:
LeonardoTrapani
2025-08-14 15:06:34 +02:00
parent 4852ad7ac1
commit 743b103857
4 changed files with 101 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/daemon/daemon.go
+24 -98
View File
@@ -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:
internal/notify/notify.go
+8
View File
@@ -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 {
internal/pipeline/pipeline.go
+34 -41
View File
@@ -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()
}
internal/recording/recording.go
+35 -38
View File
@@ -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
}
}
}
}