av/revid/audio.go

package revid

import (
	"errors"
	"io"
	"sync"
	"time"

	"github.com/yobert/alsa"

	"bitbucket.org/ausocean/av/codec/pcm"
	"bitbucket.org/ausocean/iot/pi/smartlogger"
	"bitbucket.org/ausocean/utils/logger"
	"bitbucket.org/ausocean/utils/ring"
)

const (
	logPath         = "/var/log/netsender"
	defaultSampRate = 48000
	defaultPeriod   = 5 // seconds
	defaultChannels = 2
	defaultBits     = 16
	rbDuration      = 300 // seconds
	rbTimeout       = 100 * time.Millisecond
	rbNextTimeout   = 100 * time.Millisecond
)

var log *logger.Logger

// audioInput holds everything we need to know about the audio input stream.
// NB: At 44100 Hz frame rate, 2 channels and 16-bit samples, a period of 5 seconds
// results in PCM data chunks of 882000 bytes! A longer period exceeds datastore's 1MB blob limit.
type audioInput struct {
	mu sync.Mutex // mu protects the audioInput.

	parameters

	// internals
	dev *alsa.Device // audio input device
	ab  alsa.Buffer  // ALSA's buffer
	rb  *ring.Buffer // our buffer
	vs  int          // our "var sum" to track var changes
}

type parameters struct {
	mode     string // operating mode, either "Normal" or "Paused"
	source   string // name of audio source, or empty for the default source
	rate     int    // frame rate in Hz, 44100Hz by default
	period   int    // audio period in seconds, 5s by default
	channels int    // number of audio channels, 1 for mono, 2 for stereo
	bits     int    // sample bit size, 16 by default
}

// NewAudioInput starts recording audio and returns an AudioInput which the audio can be read from.
func NewAudioInput() io.Reader {
	logLevel := int(logger.Debug)

	validLogLevel := true
	if logLevel < int(logger.Debug) || logLevel > int(logger.Fatal) {
		logLevel = int(logger.Info)
		validLogLevel = false
	}

	logSender := smartlogger.New(logPath)
	log = logger.New(int8(logLevel), &logSender.LogRoller)
	log.Log(logger.Info, "log-netsender: Logger Initialized")
	if !validLogLevel {
		log.Log(logger.Error, "Invalid log level was defaulted to Info")
	}

	var ac audioInput

	ac.setParams()

	// Open the requested audio device.
	err := ac.open()
	if err != nil {
		log.Log(logger.Fatal, "alsa.open failed", "error", err.Error())
	}

	// Capture audio in periods of ac.period seconds, and buffer rbDuration seconds in total.
	ac.ab = ac.dev.NewBufferDuration(time.Second * time.Duration(ac.period))
	recSize := (((len(ac.ab.Data) / ac.dev.BufferFormat().Channels) * ac.channels) / ac.dev.BufferFormat().Rate) * ac.rate
	rbLen := rbDuration / ac.period
	ac.rb = ring.NewBuffer(rbLen, recSize, rbTimeout)

	go ac.input()

	return ac
}

func (ac *audioInput) setParams() {
	p := ac.parameters

	p.rate = defaultSampRate
	p.period = defaultPeriod
	p.channels = defaultChannels
	p.bits = defaultBits

	ac.mu.Lock()
	ac.parameters = p
	ac.mu.Unlock()
}

// open or re-open the recording device with the given name and prepare it to record.
// If name is empty, the first recording device is used.
func (ac *audioInput) open() error {
	if ac.dev != nil {
		log.Log(logger.Debug, "Closing", "source", ac.source)
		ac.dev.Close()
		ac.dev = nil
	}
	log.Log(logger.Debug, "Opening", "source", ac.source)

	cards, err := alsa.OpenCards()
	if err != nil {
		return err
	}
	defer alsa.CloseCards(cards)

	for _, card := range cards {
		devices, err := card.Devices()
		if err != nil {
			return err
		}
		for _, dev := range devices {
			if dev.Type != alsa.PCM || !dev.Record {
				continue
			}
			if dev.Title == ac.source || ac.source == "" {
				ac.dev = dev
				break
			}
		}
	}

	if ac.dev == nil {
		return errors.New("No audio source found")
	}
	log.Log(logger.Debug, "Found audio source", "source", ac.dev.Title)

	// ToDo: time out if Open takes too long.
	err = ac.dev.Open()
	if err != nil {
		return err
	}
	log.Log(logger.Debug, "Opened audio source")

	_, err = ac.dev.NegotiateChannels(defaultChannels)
	if err != nil {
		return err
	}

	// Try to negotiate a rate to record in that is divisible by the wanted rate
	// so that it can be easily downsampled to the wanted rate.
	// Note: if a card thinks it can record at a rate but can't actually, this can cause a failure. Eg.
	// the audioinjector is supposed to record at 8000Hz and 16000Hz but it can't due to a firmware issue,
	// to fix this 8000 and 16000 must be removed from this slice.
	rates := [8]int{8000, 16000, 32000, 44100, 48000, 88200, 96000, 192000}
	foundRate := false
	for i := 0; i < len(rates) && !foundRate; i++ {
		if rates[i] < ac.rate {
			continue
		}
		if rates[i]%ac.rate == 0 {
			_, err = ac.dev.NegotiateRate(rates[i])
			if err == nil {
				foundRate = true
				log.Log(logger.Debug, "Sample rate set", "rate", rates[i])
			}
		}
	}

	// If no easily divisible rate is found, then use the default rate.
	if !foundRate {
		log.Log(logger.Warning, "Unable to sample at requested rate, default used.", "rateRequested", ac.rate)
		_, err = ac.dev.NegotiateRate(defaultSampRate)
		if err != nil {
			return err
		}
		log.Log(logger.Debug, "Sample rate set", "rate", defaultSampRate)
	}

	var fmt alsa.FormatType
	switch ac.bits {
	case 16:
		fmt = alsa.S16_LE
	case 32:
		fmt = alsa.S32_LE
	default:
		return errors.New("Unsupported sample bits")
	}
	_, err = ac.dev.NegotiateFormat(fmt)
	if err != nil {
		return err
	}

	// Either 8192 or 16384 bytes is a reasonable ALSA buffer size.
	_, err = ac.dev.NegotiateBufferSize(8192, 16384)
	if err != nil {
		return err
	}

	if err = ac.dev.Prepare(); err != nil {
		return err
	}
	log.Log(logger.Debug, "Successfully negotiated ALSA params")
	return nil
}

// input continously records audio and writes it to the ringbuffer.
// Re-opens the device and tries again if ASLA returns an error.
// Spends a lot of time sleeping in Paused mode.
// ToDo: Currently, reading audio and writing to the ringbuffer are synchronous.
// Need a way to asynchronously read from the ALSA buffer, i.e.,  _while_ it is recording to avoid any gaps.
func (ac *audioInput) input() {
	for {
		ac.mu.Lock()
		mode := ac.mode
		ac.mu.Unlock()
		if mode == "Paused" {
			time.Sleep(time.Duration(ac.period) * time.Second)
			continue
		}
		log.Log(logger.Debug, "Recording audio for period", "seconds", ac.period)
		ac.mu.Lock()
		err := ac.dev.Read(ac.ab.Data)
		ac.mu.Unlock()
		if err != nil {
			log.Log(logger.Debug, "Device.Read failed", "error", err.Error())
			ac.mu.Lock()
			err = ac.open() // re-open
			if err != nil {
				log.Log(logger.Fatal, "alsa.open failed", "error", err.Error())
			}
			ac.mu.Unlock()
			continue
		}

		toWrite := ac.formatBuffer()

		log.Log(logger.Debug, "Audio format conversion has been performed where needed")

		var n int
		n, err = ac.rb.Write(toWrite.Data)
		switch err {
		case nil:
			log.Log(logger.Debug, "Wrote audio to ringbuffer", "length", n)
		case ring.ErrDropped:
			log.Log(logger.Warning, "Dropped audio")
		default:
			log.Log(logger.Error, "Unexpected ringbuffer error", "error", err.Error())
			return
		}
	}
}

// read reads a full PCM chunk from the ringbuffer, returning the number of bytes read upon success.
// Any errors returned are unexpected and should be considered fatal.
func (ac audioInput) Read(p []byte) (n int, err error) {
	chunk, err := ac.rb.Next(rbNextTimeout)
	switch err {
	case nil:
		// Do nothing.
	case ring.ErrTimeout:
		return 0, nil
	case io.EOF:
		log.Log(logger.Error, "Unexpected EOF from ring.Next")
		return 0, io.ErrUnexpectedEOF
	default:
		log.Log(logger.Error, "Unexpected error from ring.Next", "error", err.Error())
		return 0, err
	}

	n, err = io.ReadFull(ac.rb, p[:chunk.Len()])
	if err != nil {
		log.Log(logger.Error, "Unexpected error from ring.Read", "error", err.Error())
		return n, err
	}

	log.Log(logger.Debug, "Read audio from ringbuffer", "length", n)
	return n, nil
}

// formatBuffer returns an ALSA buffer that has the recording data from the ac's original ALSA buffer but stored
// in the desired format specified by the ac's parameters.
func (ac *audioInput) formatBuffer() alsa.Buffer {
	var err error
	ac.mu.Lock()
	wantChannels := ac.channels
	wantRate := ac.rate
	ac.mu.Unlock()

	// If nothing needs to be changed, return the original.
	if ac.ab.Format.Channels == wantChannels && ac.ab.Format.Rate == wantRate {
		return ac.ab
	}

	formatted := alsa.Buffer{Format: ac.ab.Format}
	bufCopied := false
	if ac.ab.Format.Channels != wantChannels {

		// Convert channels.
		if ac.ab.Format.Channels == 2 && wantChannels == 1 {
			if formatted.Data, err = pcm.StereoToMono(ac.ab); err != nil {
				log.Log(logger.Warning, "Channel conversion failed, audio has remained stereo", "error", err.Error())
			} else {
				formatted.Format.Channels = 1
			}
			bufCopied = true
		}
	}

	if ac.ab.Format.Rate != wantRate {

		// Convert rate.
		if bufCopied {
			formatted.Data, err = pcm.Resample(formatted, wantRate)
		} else {
			formatted.Data, err = pcm.Resample(ac.ab, wantRate)
		}
		if err != nil {
			log.Log(logger.Warning, "Rate conversion failed, audio has remained original rate", "error", err.Error())
		} else {
			formatted.Format.Rate = wantRate
		}
	}
	return formatted
}