alice-backend/station/capture.go

package main

import (
	"context"
	"encoding/binary"
	"errors"
	"io"
	"os"
	"os/exec"
	"strings"
	"time"
)

const pcmBlock = hop * 2
const rawBlock = hop * 3 * 16
const spotterBlock = pcmBlock + 1

// downsample is retained for manual raw-mic tests on the Mini 2.
func downsample(raw []byte) []int16 {
	out := make([]int16, len(raw)/(8*2*3))
	for i := range out {
		var sum int32
		for j := 0; j < 3; j++ {
			off := (i*3 + j) * 16
			sum += int32(int16(binary.LittleEndian.Uint16(raw[off:])))
		}
		out[i] = int16(sum / 3)
	}
	return out
}

func pcm(samples []int16) []byte {
	data := make([]byte, len(samples)*2)
	for i, sample := range samples {
		binary.LittleEndian.PutUint16(data[i*2:], uint16(sample))
	}
	return data
}

func capture(device, vqeHelper, spotterModel string, localCommand func(byte)) (*liveReader, func(), error) {
	channels, period, buffer, block := "8", "512", "2048", rawBlock
	if vqeHelper != "" {
		device, channels, buffer, block = "plug:default", "6", "4096", pcmBlock
	}
	recorder := exec.Command("arecord", "-q", "-D", device, "-t", "raw", "-f", "S16_LE", "-r", "48000",
		"-c", channels, "--period-size="+period, "--buffer-size="+buffer)
	recorder.Stderr = os.Stderr
	terminateWithClient(recorder)
	pipe, err := recorder.StdoutPipe()
	if err != nil {
		return nil, nil, err
	}
	if err = recorder.Start(); err != nil {
		pipe.Close()
		return nil, nil, err
	}
	var processor *exec.Cmd
	stop := func() {
		if processor != nil {
			processor.Process.Kill()
			processor.Wait()
		}
		recorder.Process.Kill()
		recorder.Wait()
	}
	if vqeHelper != "" {
		helper := exec.Command(vqeHelper)
		if spotterModel != "" {
			helper.Args = append(helper.Args, spotterModel, mutePath)
			block = spotterBlock
		}
		helper.Stdin = pipe.(*os.File)
		helper.Stderr = os.Stderr
		terminateWithClient(helper)
		output, err := helper.StdoutPipe()
		if err != nil {
			stop()
			return nil, nil, err
		}
		if err = helper.Start(); err != nil {
			output.Close()
			stop()
			return nil, nil, err
		}
		processor, pipe = helper, output
	}
	return newLiveReaderWithCommands(pipe.(*os.File), block, localCommand), stop, nil
}

// Keep draining ALSA during playback and network startup. When the queue is
// full, old frames can be dropped, but a recording must never splice a gap.
type liveReader struct {
	chunks chan audioFrame
	err    error
	last   uint64
}

type audioFrame struct {
	seq         uint64
	pcm         []int16
	wake, muted bool
}

func newLiveReaderWithCommands(pipe *os.File, block int, localCommand func(byte)) *liveReader {
	live := &liveReader{chunks: make(chan audioFrame, 300)}
	go func() {
		defer close(live.chunks)
		buf := make([]byte, block)
		for seq := uint64(1); ; seq++ {
			_ = pipe.SetReadDeadline(time.Now().Add(2 * time.Second))
			if _, live.err = io.ReadFull(pipe, buf); live.err != nil {
				return
			}
			var samples []int16
			if block == rawBlock {
				samples = downsample(buf)
			} else {
				samples = make([]int16, hop)
				for i := range samples {
					samples[i] = int16(binary.LittleEndian.Uint16(buf[i*2:]))
				}
			}
			frame := audioFrame{seq: seq, pcm: samples}
			if block == spotterBlock {
				flags := buf[pcmBlock]
				if flags > 31 {
					live.err = errors.New("invalid VQE spotter flags")
					return
				}
				frame.wake = flags&1 != 0
				frame.muted = flags&2 != 0
				if frame.muted && localCommand != nil {
					localCommand(1)
				}
				if command := flags >> 2; command != 0 && localCommand != nil && !frame.muted && unmuted() && !frame.wake {
					localCommand(command)
				}
			}
			select {
			case live.chunks <- frame:
			default:
				select {
				case <-live.chunks:
				default:
				}
				live.chunks <- frame
			}
		}
	}()
	return live
}

func (l *liveReader) next() (audioFrame, error) {
	return l.nextContext(context.Background())
}

func (l *liveReader) nextContext(ctx context.Context) (audioFrame, error) {
	var frame audioFrame
	var ok bool
	select {
	case <-ctx.Done():
		return frame, ctx.Err()
	case frame, ok = <-l.chunks:
	}
	if !ok {
		return audioFrame{}, l.err
	}
	last := l.last
	l.last = frame.seq
	if last != 0 && frame.seq != last+1 {
		return audioFrame{}, errors.New("microphone buffer overrun")
	}
	return frame, nil
}

func (l *liveReader) discardPending() {
	l.last = 0
	for {
		select {
		case _, ok := <-l.chunks:
			if !ok {
				return
			}
		default:
			return
		}
	}
}

var mutePath string
var errMuted = errors.New("microphone muted or mute state unavailable")
var errNoSpeech = errors.New("no speech detected")

func unmuted() bool {
	data, err := os.ReadFile(mutePath)
	return err == nil && strings.TrimSpace(string(data)) == "1"
}