mirror of https://bitbucket.org/ausocean/av.git
revid: added missing files for audio input
This commit is contained in:
Trek H
parent
faa6246a51
commit
2bb3a6b82f
|
|
@ -0,0 +1,325 @@
|
|||
package revid
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"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"
|
||||
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.
|
||||
mode string // operating mode, either "Normal" or "Paused"
|
||||
source string // name of audio source, or empty for the default source
|
||||
|
||||
dev *alsa.Device // audio input device
|
||||
ab alsa.Buffer // ALSA's buffer
|
||||
rb *ring.Buffer // our buffer
|
||||
chunkSize int
|
||||
vs int // our "var sum" to track var changes
|
||||
|
||||
*AudioConfig
|
||||
}
|
||||
|
||||
// AudioConfig provides parameters used by AudioInput.
|
||||
type AudioConfig struct {
|
||||
SampleRate int
|
||||
Channels int
|
||||
BitDepth int
|
||||
RecPeriod int
|
||||
Codec uint8
|
||||
}
|
||||
|
||||
// NewAudioInput starts recording audio and returns an AudioInput struct which the audio can be read from.
|
||||
func NewAudioInput(cfg *AudioConfig) *AudioInput {
|
||||
|
||||
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")
|
||||
}
|
||||
|
||||
a := &AudioInput{}
|
||||
a.AudioConfig = cfg
|
||||
|
||||
// Open the requested audio device.
|
||||
err := a.open()
|
||||
if err != nil {
|
||||
log.Log(logger.Fatal, "alsa.open failed", "error", err.Error())
|
||||
}
|
||||
|
||||
// Capture audio in periods of a.RecPeriod seconds, and buffer rbDuration seconds in total.
|
||||
a.ab = a.dev.NewBufferDuration(time.Second * time.Duration(a.RecPeriod))
|
||||
a.chunkSize = (((len(a.ab.Data) / a.dev.BufferFormat().Channels) * a.Channels) / a.dev.BufferFormat().Rate) * a.SampleRate
|
||||
rbLen := rbDuration / a.RecPeriod
|
||||
a.rb = ring.NewBuffer(rbLen, a.chunkSize, rbTimeout)
|
||||
|
||||
go a.input()
|
||||
|
||||
return a
|
||||
}
|
||||
|
||||
func (a *AudioInput) ChunkSize() int {
|
||||
return a.chunkSize
|
||||
}
|
||||
|
||||
// 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 (a *AudioInput) open() error {
|
||||
if a.dev != nil {
|
||||
log.Log(logger.Debug, "Closing", "source", a.source)
|
||||
a.dev.Close()
|
||||
a.dev = nil
|
||||
}
|
||||
log.Log(logger.Debug, "Opening", "source", a.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 == a.source || a.source == "" {
|
||||
a.dev = dev
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if a.dev == nil {
|
||||
return errors.New("No audio source found")
|
||||
}
|
||||
log.Log(logger.Debug, "Found audio source", "source", a.dev.Title)
|
||||
|
||||
// ToDo: time out if Open takes too long.
|
||||
err = a.dev.Open()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
log.Log(logger.Debug, "Opened audio source")
|
||||
|
||||
// 2 channels is what most devices need to record in. If mono is requested,
|
||||
// the recording will be converted in formatBuffer().
|
||||
_, err = a.dev.NegotiateChannels(2)
|
||||
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] < a.SampleRate {
|
||||
continue
|
||||
}
|
||||
if rates[i]%a.SampleRate == 0 {
|
||||
_, err = a.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", a.SampleRate)
|
||||
_, err = a.dev.NegotiateRate(defaultSampleRate)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
log.Log(logger.Debug, "Sample rate set", "rate", defaultSampleRate)
|
||||
}
|
||||
|
||||
var fmt alsa.FormatType
|
||||
switch a.BitDepth {
|
||||
case 16:
|
||||
fmt = alsa.S16_LE
|
||||
case 32:
|
||||
fmt = alsa.S32_LE
|
||||
default:
|
||||
return errors.New("Unsupported sample bits")
|
||||
}
|
||||
_, err = a.dev.NegotiateFormat(fmt)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Either 8192 or 16384 bytes is a reasonable ALSA buffer size.
|
||||
_, err = a.dev.NegotiateBufferSize(8192, 16384)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err = a.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 (a *AudioInput) input() {
|
||||
for {
|
||||
a.mu.Lock()
|
||||
mode := a.mode
|
||||
a.mu.Unlock()
|
||||
if mode == "Paused" {
|
||||
time.Sleep(time.Duration(a.RecPeriod) * time.Second)
|
||||
continue
|
||||
}
|
||||
log.Log(logger.Debug, "Recording audio for period", "seconds", a.RecPeriod)
|
||||
a.mu.Lock()
|
||||
err := a.dev.Read(a.ab.Data)
|
||||
a.mu.Unlock()
|
||||
if err != nil {
|
||||
log.Log(logger.Debug, "Device.Read failed", "error", err.Error())
|
||||
a.mu.Lock()
|
||||
err = a.open() // re-open
|
||||
if err != nil {
|
||||
log.Log(logger.Fatal, "alsa.open failed", "error", err.Error())
|
||||
}
|
||||
a.mu.Unlock()
|
||||
continue
|
||||
}
|
||||
|
||||
toWrite := a.formatBuffer()
|
||||
|
||||
log.Log(logger.Debug, "Audio format conversion has been performed where needed")
|
||||
|
||||
fmt.Printf("Writing %v bytes to ringbuffer\n", len(toWrite.Data))
|
||||
var n int
|
||||
n, err = a.rb.Write(toWrite.Data)
|
||||
fmt.Printf("Wrote %v bytes to ringbuffer\n", n)
|
||||
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 (a AudioInput) Read(p []byte) (n int, err error) {
|
||||
fmt.Println("Performing AudioInput read...")
|
||||
chunk, err := a.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
|
||||
}
|
||||
fmt.Printf("Reading %v bytes from ringbuffer\n", chunk.Len())
|
||||
n, err = io.ReadFull(a.rb, p[:chunk.Len()])
|
||||
fmt.Printf("Read %v bytes from ringbuffer\n", n)
|
||||
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 (a *AudioInput) formatBuffer() alsa.Buffer {
|
||||
var err error
|
||||
a.mu.Lock()
|
||||
wantChannels := a.Channels
|
||||
wantRate := a.SampleRate
|
||||
a.mu.Unlock()
|
||||
|
||||
// If nothing needs to be changed, return the original.
|
||||
if a.ab.Format.Channels == wantChannels && a.ab.Format.Rate == wantRate {
|
||||
return a.ab
|
||||
}
|
||||
|
||||
formatted := alsa.Buffer{Format: a.ab.Format}
|
||||
bufCopied := false
|
||||
if a.ab.Format.Channels != wantChannels {
|
||||
|
||||
// Convert channels.
|
||||
if a.ab.Format.Channels == 2 && wantChannels == 1 {
|
||||
if formatted.Data, err = pcm.StereoToMono(a.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 a.ab.Format.Rate != wantRate {
|
||||
|
||||
// Convert rate.
|
||||
if bufCopied {
|
||||
formatted.Data, err = pcm.Resample(formatted, wantRate)
|
||||
} else {
|
||||
formatted.Data, err = pcm.Resample(a.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
|
||||
}
|
||||
|
|
@ -0,0 +1,37 @@
|
|||
package revid
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"bitbucket.org/ausocean/iot/pi/netsender"
|
||||
)
|
||||
|
||||
func TestAudioInput(t *testing.T) {
|
||||
|
||||
var logger testLogger
|
||||
ns, err := netsender.New(&logger, nil, nil, nil)
|
||||
if err != nil {
|
||||
t.Errorf("netsender.New failed with error %v", err)
|
||||
}
|
||||
|
||||
var c Config
|
||||
c.Logger = &logger
|
||||
c.Input = Audio
|
||||
c.Outputs = make([]uint8, 1)
|
||||
|
||||
rv, err := New(c, ns)
|
||||
if err != nil {
|
||||
t.Errorf("revid.New failed with error %v", err)
|
||||
}
|
||||
|
||||
err = rv.Start()
|
||||
if err != nil {
|
||||
t.Errorf("revid.Start failed with error %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// testLogger implements a netsender.Logger.
|
||||
type testLogger struct{}
|
||||
|
||||
// SetLevel normally sets the logging level, but it is a no-op in our case.
|
||||
func (tl *testLogger) SetLevel(level int8) {}
|
||||
Loading…
Reference in New Issue