Captioneer/src/capture/pulse.go

// Package capture reads the default PulseAudio-compatible output monitor.
package capture

import (
	"context"
	"errors"
	"fmt"
	"io"
	"os"
	"os/exec"
	"strings"
	"sync"

	"tea.chunkbyte.com/kato/captioneer/src/audio"
)

func ValidatePrograms() error {
	for _, program := range []string{"pactl", "parec"} {
		if _, err := exec.LookPath(program); err != nil {
			return fmt.Errorf("%s was not found in PATH: %w", program, err)
		}
	}
	return nil
}

func DefaultMonitorSource(ctx context.Context) (string, error) {
	output, err := exec.CommandContext(ctx, "pactl", "get-default-sink").Output()
	if err != nil {
		return "", err
	}
	return strings.TrimSpace(string(output)) + ".monitor", nil
}

func Packets(ctx context.Context, monitorSource string) (<-chan []float32, func() error, error) {
	cmd := exec.CommandContext(ctx, "parec",
		"--device="+monitorSource,
		"--format=s16le",
		fmt.Sprintf("--rate=%d", audio.SampleRate),
		fmt.Sprintf("--channels=%d", audio.Channels),
		"--raw",
		"--latency-msec=50",
	)
	cmd.Stderr = os.Stderr
	stdout, err := cmd.StdoutPipe()
	if err != nil {
		return nil, nil, err
	}
	if err := cmd.Start(); err != nil {
		return nil, nil, err
	}

	packets := make(chan []float32, 50) // Five seconds at 100 ms per packet.
	var waitOnce sync.Once
	var waitErr error
	wait := func() error {
		waitOnce.Do(func() { waitErr = cmd.Wait() })
		return waitErr
	}

	go func() {
		defer close(packets)
		packetBytes := make([]byte, audio.SamplesPerPacket()*audio.BytesPerSample)
		for {
			n, readErr := io.ReadFull(stdout, packetBytes)
			if n > 0 {
				enqueueLatest(packets, audio.PCM16LEToFloat32(packetBytes[:n]))
			}
			if readErr != nil {
				if ctx.Err() == nil && !errors.Is(readErr, io.EOF) && !errors.Is(readErr, io.ErrUnexpectedEOF) {
					fmt.Fprintf(os.Stderr, "audio capture read error: %v\n", readErr)
				}
				return
			}
		}
	}()
	return packets, wait, nil
}

func enqueueLatest(packets chan []float32, packet []float32) {
	select {
	case packets <- packet:
		return
	default:
	}

	select {
	case <-packets:
	default:
	}
	select {
	case packets <- packet:
	default:
	}
	fmt.Fprintln(os.Stderr, "warning: caption processing overloaded; dropped 100 ms of oldest audio")
}