revid: started modifying audio-netsender to be a general audio input

This commit is contained in:
Trek H 2019-04-18 16:59:48 +09:30
parent 58b9458ff4
commit 8a1f35c0a5
1 changed files with 28 additions and 94 deletions

View File

@ -1,46 +1,7 @@
/* package revid
NAME
audio-netsender - NetSender client for sending audio to NetReceiver
AUTHORS
Alan Noble <alan@ausocean.org>
Trek Hopton <trek@ausocean.org>
ACKNOWLEDGEMENTS
A special thanks to Joel Jensen for his Go ALSA package.
LICENSE
audio-netsender is Copyright (C) 2018 the Australian Ocean Lab (AusOcean).
It is free software: you can redistribute it and/or modify them under
the terms of the GNU General Public License as published by the
Free Software Foundation, either version 3 of the License, or (at your
option) any later version.
It is distributed in the hope that it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
for more details.
You should have received a copy of the GNU General Public License
along with https://bitbucket.org/ausocean/iot/src/master/gpl.txt.
If not, see http://www.gnu.org/licenses.
*/
// audio-netsender is a NetSender client for sending audio to
// NetReceiver. Audio is captured by means of an ALSA recording
// device, specified by the NetReceiver "source" variable. It sent via
// HTTP to NetReceiver in raw audio form, i.e., as PCM data, where it
// is stored as BinaryData objects. Other NetReceiver variables are
// "rate", "period", "channels" and "bits", for specifiying the frame
// rate (Hz), audio period (seconds), number of channels and sample
// bit size respectively. For a description of NetReceiver see
// http://netreceiver.appspot.com/help.
package main
import ( import (
"errors" "errors"
"flag"
"io" "io"
"strconv" "strconv"
"sync" "sync"
@ -51,38 +12,33 @@ import (
"bitbucket.org/ausocean/av/codec/pcm" "bitbucket.org/ausocean/av/codec/pcm"
"bitbucket.org/ausocean/iot/pi/netsender" "bitbucket.org/ausocean/iot/pi/netsender"
"bitbucket.org/ausocean/iot/pi/sds" "bitbucket.org/ausocean/iot/pi/sds"
"bitbucket.org/ausocean/iot/pi/smartlogger"
"bitbucket.org/ausocean/utils/logger" "bitbucket.org/ausocean/utils/logger"
"bitbucket.org/ausocean/utils/ring" "bitbucket.org/ausocean/utils/ring"
) )
const ( const (
progName = "audio-netsender" defaultRate = 48000
logPath = "/var/log/netsender" defaultPeriod = 5 // seconds
retryPeriod = 5 * time.Second defaultChannels = 2
defaultFrameRate = 48000 defaultBits = 16
defaultPeriod = 5 // seconds rbDuration = 300 // seconds
defaultChannels = 2 rbTimeout = 100 * time.Millisecond
defaultBits = 16 rbNextTimeout = 100 * time.Millisecond
rbDuration = 300 // seconds
rbTimeout = 100 * time.Millisecond
rbNextTimeout = 100 * time.Millisecond
) )
// audioClient holds everything we need to know about the client. // 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 // 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. // results in PCM data chunks of 882000 bytes! A longer period exceeds datastore's 1MB blob limit.
type audioClient struct { type audioInput struct {
mu sync.Mutex // mu protects the audioClient. mu sync.Mutex // mu protects the audioInput.
parameters parameters
// internals // internals
dev *alsa.Device // audio input device dev *alsa.Device // audio input device
ab alsa.Buffer // ALSA's buffer ab alsa.Buffer // ALSA's buffer
rb *ring.Buffer // our buffer rb *ring.Buffer // our buffer //TODO: change this to output stream, doesn't have to be ring buffer
ns *netsender.Sender // our NetSender vs int // our "var sum" to track var changes
vs int // our "var sum" to track var changes
} }
type parameters struct { type parameters struct {
@ -94,32 +50,8 @@ type parameters struct {
bits int // sample bit size, 16 by default bits int // sample bit size, 16 by default
} }
var log *logger.Logger func NewAudioInput() {
var ac audioInput
func main() {
var logLevel int
flag.IntVar(&logLevel, "LogLevel", int(logger.Debug), "Specifies log level")
flag.Parse()
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 audioClient
var err error
ac.ns, err = netsender.New(log, nil, sds.ReadSystem, nil)
if err != nil {
log.Log(logger.Fatal, "netsender.Init failed", "error", err.Error())
}
// Get audio params and store the current var sum. // Get audio params and store the current var sum.
vars, err := ac.ns.Vars() vars, err := ac.ns.Vars()
@ -144,11 +76,13 @@ func main() {
go ac.input() go ac.input()
ac.output() ac.output()
return stream
} }
// params extracts audio params from corresponding NetReceiver vars and returns true if anything has changed. // params extracts audio params from corresponding NetReceiver vars and returns true if anything has changed.
// See audioClient for a description of the params and their limits. // See audioInput for a description of the params and their limits.
func (ac *audioClient) params(vars map[string]string) bool { func (ac *audioInput) params(vars map[string]string) bool {
// We are the only writers to this field // We are the only writers to this field
// so we don't need to lock here. // so we don't need to lock here.
p := ac.parameters p := ac.parameters
@ -166,7 +100,7 @@ func (ac *audioClient) params(vars map[string]string) bool {
} }
val, err := strconv.Atoi(vars["rate"]) val, err := strconv.Atoi(vars["rate"])
if err != nil { if err != nil {
val = defaultFrameRate val = defaultRate
} }
if p.rate != val { if p.rate != val {
p.rate = val p.rate = val
@ -209,7 +143,7 @@ func (ac *audioClient) params(vars map[string]string) bool {
// open or re-open the recording device with the given name and prepare it to record. // 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. // If name is empty, the first recording device is used.
func (ac *audioClient) open() error { func (ac *audioInput) open() error {
if ac.dev != nil { if ac.dev != nil {
log.Log(logger.Debug, "Closing", "source", ac.source) log.Log(logger.Debug, "Closing", "source", ac.source)
ac.dev.Close() ac.dev.Close()
@ -279,11 +213,11 @@ func (ac *audioClient) open() error {
// If no easily divisible rate is found, then use the default rate. // If no easily divisible rate is found, then use the default rate.
if !foundRate { if !foundRate {
log.Log(logger.Warning, "No available device sample-rates are divisible by the requested rate. Default rate will be used. Resampling may fail.", "rateRequested", ac.rate) log.Log(logger.Warning, "No available device sample-rates are divisible by the requested rate. Default rate will be used. Resampling may fail.", "rateRequested", ac.rate)
_, err = ac.dev.NegotiateRate(defaultFrameRate) _, err = ac.dev.NegotiateRate(defaultRate)
if err != nil { if err != nil {
return err return err
} }
log.Log(logger.Debug, "Sample rate set", "rate", defaultFrameRate) log.Log(logger.Debug, "Sample rate set", "rate", defaultRate)
} }
var fmt alsa.FormatType var fmt alsa.FormatType
@ -318,7 +252,7 @@ func (ac *audioClient) open() error {
// Spends a lot of time sleeping in Paused mode. // Spends a lot of time sleeping in Paused mode.
// ToDo: Currently, reading audio and writing to the ringbuffer are synchronous. // 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. // Need a way to asynchronously read from the ALSA buffer, i.e., _while_ it is recording to avoid any gaps.
func (ac *audioClient) input() { func (ac *audioInput) input() {
for { for {
ac.mu.Lock() ac.mu.Lock()
mode := ac.mode mode := ac.mode
@ -369,7 +303,7 @@ func (ac *audioClient) input() {
// since cycling more frequently is pointless. // since cycling more frequently is pointless.
// Finally while audio data is sent every audio period, other data is reported only every monitor period. // Finally while audio data is sent every audio period, other data is reported only every monitor period.
// This function also handles NetReceiver configuration requests and updating of NetReceiver vars. // This function also handles NetReceiver configuration requests and updating of NetReceiver vars.
func (ac *audioClient) output() { func (ac *audioInput) output() {
// Calculate the size of the output data based on wanted channels and rate. // Calculate the size of the output data based on wanted channels and rate.
outLen := (((len(ac.ab.Data) / ac.ab.Format.Channels) * ac.channels) / ac.ab.Format.Rate) * ac.rate outLen := (((len(ac.ab.Data) / ac.ab.Format.Channels) * ac.channels) / ac.ab.Format.Rate) * ac.rate
buf := make([]byte, outLen) buf := make([]byte, outLen)
@ -510,7 +444,7 @@ func read(rb *ring.Buffer, buf []byte) (int, error) {
// formatBuffer returns an ALSA buffer that has the recording data from the ac's original ALSA buffer but stored // 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. // in the desired format specified by the ac's parameters.
func (ac *audioClient) formatBuffer() alsa.Buffer { func (ac *audioInput) formatBuffer() alsa.Buffer {
var err error var err error
ac.mu.Lock() ac.mu.Lock()
wantChannels := ac.channels wantChannels := ac.channels