OpsLog/internal/audio/engine_linux.go

//go:build linux

package audio

// engine_linux.go — the four calls the rest of the package makes into the sound
// card, implemented on PulseAudio. The Windows half of this pair is engine.go
// (WASAPI); nothing above these functions knows which one it is talking to.
//
// Capture is fixed at 16 kHz mono 16-bit, the format the DVK, the recorder and
// the CW tap all share (see wav.go). We ask the server for it and let the
// server resample from whatever the device really runs at — the same division
// of labour as WASAPI's AUTOCONVERTPCM, and for the same reason: a rig codec
// that only does 48 kHz must still feed a 16 kHz pipeline, and the sound
// server's converter filters before it decimates, where a naive one folds the
// receiver hiss above 8 kHz straight back on top of the voice.

import (
	"fmt"
	"io"
	"time"

	"github.com/jfreymuth/pulse"
	"github.com/jfreymuth/pulse/proto"
)

// chunkFrames is how much audio a playback reader hands over at once (20 ms).
// It bounds how much silence a padded underrun can queue ahead of real audio,
// which is what keeps live monitoring from drifting seconds behind the rig.
const chunkFrames = sampleRate / 50

// channelMap describes n channels to the server. Only mono and stereo occur
// here — capture is always mono, and playback follows the WAV being played.
func channelMap(n int) proto.ChannelMap {
	if n >= 2 {
		return proto.ChannelMap{proto.ChannelLeft, proto.ChannelRight}
	}
	return proto.ChannelMap{proto.ChannelMono}
}

// chunkWriter turns the record stream's byte deliveries into onChunk calls.
// The server reuses its buffer between deliveries, so every chunk is copied
// before it leaves: the recorder keeps the slices it is given.
type chunkWriter struct{ onChunk func([]byte) }

func (w chunkWriter) Write(p []byte) (int, error) {
	if len(p) > 0 && w.onChunk != nil {
		cp := make([]byte, len(p))
		copy(cp, p)
		w.onChunk(cp)
	}
	return len(p), nil
}

// recordPCM captures from a device into 16 kHz mono 16-bit PCM bytes until the
// stop channel is closed.
func recordPCM(deviceID string, stop <-chan struct{}) ([]byte, error) {
	out := make([]byte, 0, bytesPerSec*4)
	err := captureStream(deviceID, stop, func(chunk []byte) { out = append(out, chunk...) })
	return out, err
}

// captureStream opens a device and calls onChunk with freshly-captured 16 kHz
// mono 16-bit PCM as it arrives, until stop closes. onChunk receives a private
// copy it may retain.
func captureStream(deviceID string, stop <-chan struct{}, onChunk func([]byte)) error {
	c, err := pulseClient()
	if err != nil {
		return err
	}
	defer c.Close()

	// Rate and channels first, then latency — the latency option sizes its
	// buffer from both, so setting it earlier would size it from the defaults.
	//
	// 50 ms of fragment: the CW decoder is downstream of this and works on the
	// chunks as they arrive, so a server-chosen fragment of a quarter of a
	// second would make it decide about a dit long after the dit was over.
	opts := []pulse.RecordOption{
		pulse.RecordSampleRate(sampleRate),
		pulse.RecordChannels(channelMap(channels)),
		pulse.RecordLatency(0.05),
		pulse.RecordMediaName("OpsLog capture"),
	}
	// An empty id means "whatever the desktop calls the default", which is also
	// what an operator who has never opened the audio settings expects.
	if deviceID != "" {
		src, err := c.SourceByID(deviceID)
		if err != nil {
			return fmt.Errorf("no audio input %q: %w", deviceID, err)
		}
		opts = append(opts, pulse.RecordSource(src))
	}
	st, err := c.NewRecord(pulse.NewWriter(chunkWriter{onChunk}, proto.FormatInt16LE), opts...)
	if err != nil {
		return fmt.Errorf("open capture: %w", err)
	}
	defer st.Close()

	st.Start()
	<-stop
	st.Stop()
	return st.Error()
}

// playPCM plays a fixed buffer to a device and returns when it has been heard
// (or when stop closes, which cuts it short).
func playPCM(deviceID string, pcm []byte, rate, ch, bits int, stop <-chan struct{}) error {
	if len(pcm) == 0 {
		return nil
	}
	format, err := pulseFormat(bits)
	if err != nil {
		return err
	}
	frameBytes := ch * bits / 8
	if frameBytes <= 0 || rate <= 0 {
		return fmt.Errorf("bad audio format")
	}

	c, err := pulseClient()
	if err != nil {
		return err
	}
	defer c.Close()

	// The reader hands out the buffer a slice at a time and ends the stream
	// with EndOfData — the library's own sentinel. io.EOF would work as an end
	// too, but it is recorded as the stream's error, and a message finishing
	// normally must not look like a fault in the log.
	pos := 0
	read := func(buf []byte) (int, error) {
		select {
		case <-stop:
			return 0, pulse.EndOfData
		default:
		}
		if pos >= len(pcm) {
			return 0, pulse.EndOfData
		}
		n := copy(buf, pcm[pos:])
		n -= n % frameBytes // never hand the server a partial frame
		if n == 0 {
			return 0, pulse.EndOfData
		}
		pos += n
		return n, nil
	}

	opts := []pulse.PlaybackOption{
		pulse.PlaybackSampleRate(rate),
		pulse.PlaybackChannels(channelMap(ch)),
		pulse.PlaybackLatency(0.1),
		pulse.PlaybackMediaName("OpsLog playback"),
	}
	if deviceID != "" {
		sink, err := c.SinkByID(deviceID)
		if err != nil {
			return fmt.Errorf("no audio output %q: %w", deviceID, err)
		}
		opts = append(opts, pulse.PlaybackSink(sink))
	}
	st, err := c.NewPlayback(pulse.NewReader(readerFunc(read), format), opts...)
	if err != nil {
		return fmt.Errorf("open playback: %w", err)
	}
	defer st.Close()

	st.Start()

	// Drain blocks until the server has played everything queued. Waiting on it
	// in a goroutine keeps stop responsive: a voice message must cut off the
	// instant the operator unkeys, not at the end of the buffer.
	drained := make(chan struct{})
	go func() { st.Drain(); close(drained) }()
	select {
	case <-drained:
	case <-stop:
		st.Stop()
	}
	return st.Error()
}

// renderStream continuously renders PCM pulled from src to a device until stop
// closes — the streaming counterpart to playPCM's fixed buffer. On underrun it
// writes silence rather than glitching, keeping the server's clock steady so
// live monitor audio flows smoothly even when the source stalls briefly.
func renderStream(deviceID string, rate, ch, bits int, stop <-chan struct{}, src *pcmRing) error {
	format, err := pulseFormat(bits)
	if err != nil {
		return err
	}
	frameBytes := ch * bits / 8
	if frameBytes <= 0 || rate <= 0 || src == nil {
		return fmt.Errorf("bad audio format")
	}

	c, err := pulseClient()
	if err != nil {
		return err
	}
	defer c.Close()

	// Never return 0 bytes without an error: the library's playback loop would
	// spin on it. A stalled source therefore yields silence, which is also the
	// behaviour that keeps the clock running.
	chunk := chunkFrames * frameBytes
	read := func(buf []byte) (int, error) {
		select {
		case <-stop:
			return 0, pulse.EndOfData
		default:
		}
		n := len(buf)
		if n > chunk {
			n = chunk
		}
		n -= n % frameBytes
		if n == 0 {
			n = frameBytes
		}
		got := copy(buf[:n], src.pull(n))
		for i := got; i < n; i++ {
			buf[i] = 0
		}
		return n, nil
	}

	opts := []pulse.PlaybackOption{
		pulse.PlaybackSampleRate(rate),
		pulse.PlaybackChannels(channelMap(ch)),
		pulse.PlaybackLatency(0.1),
		pulse.PlaybackMediaName("OpsLog monitor"),
	}
	if deviceID != "" {
		sink, err := c.SinkByID(deviceID)
		if err != nil {
			return fmt.Errorf("no audio output %q: %w", deviceID, err)
		}
		opts = append(opts, pulse.PlaybackSink(sink))
	}
	st, err := c.NewPlayback(pulse.NewReader(readerFunc(read), format), opts...)
	if err != nil {
		return fmt.Errorf("open monitor: %w", err)
	}
	defer st.Close()

	st.Start()
	<-stop
	st.Stop()
	// Give the server a moment to notice the stream stopped before the client
	// socket goes away, so the last fragment is heard instead of clipped.
	time.Sleep(20 * time.Millisecond)
	return st.Error()
}

// pulseFormat maps a WAV bit depth onto the server's sample formats. 8 and 16
// bit cover everything OpsLog produces or reads; anything else is refused by
// name rather than played as noise.
func pulseFormat(bits int) (byte, error) {
	switch bits {
	case 8:
		return proto.FormatUint8, nil
	case 16:
		return proto.FormatInt16LE, nil
	default:
		return 0, fmt.Errorf("unsupported sample size %d-bit (8 or 16 expected)", bits)
	}
}

// readerFunc adapts a read closure to io.Reader.
type readerFunc func([]byte) (int, error)

func (f readerFunc) Read(p []byte) (int, error) { return f(p) }

var _ io.Reader = readerFunc(nil)