revid: added rate flags to config and cli, started writing test

This commit is contained in:
Trek H 2019-04-23 14:51:17 +09:30
parent 46ca3e2611
commit a60c65a6cf
3 changed files with 24 additions and 333 deletions

View File

@ -60,7 +60,9 @@ const (
defaultLogPath = "/var/log/netsender"
pkg = "revid-cli:"
defaultLogVerbosity = logger.Info
defaultSleepTime = 60 // Seconds
defaultSleepTime = 60 // Seconds
sampleSize = 2 // Bytes
blockSize = 16000 // Bytes
)
// canProfile is set to false with revid-cli is built with "-tags profile".
@ -105,7 +107,7 @@ func handleFlags() revid.Config {
var (
cpuprofile = flag.String("cpuprofile", "", "write cpu profile to `file`")
inputPtr = flag.String("Input", "", "The input type: Raspivid, File, Webcam")
inputPtr = flag.String("Input", "", "The input type: Raspivid, File, v4l, Audio")
inputCodecPtr = flag.String("InputCodec", "", "The codec of the input: H264, Mjpeg")
rtmpMethodPtr = flag.String("RtmpMethod", "", "The method used to send over rtmp: Ffmpeg, Librtmp")
quantizePtr = flag.Bool("Quantize", false, "Quantize input (non-variable bitrate)")
@ -125,6 +127,7 @@ func handleFlags() revid.Config {
heightPtr = flag.Uint("Height", 0, "Height in pixels")
widthPtr = flag.Uint("Width", 0, "Width in pixels")
frameRatePtr = flag.Uint("FrameRate", 0, "Frame rate of captured video")
sampleRatePtr = flag.Uint("SampleRate", 0, "Sample rate of recorded audio")
quantizationPtr = flag.Uint("Quantization", 0, "Desired quantization value: 0-40")
intraRefreshPeriodPtr = flag.Uint("IntraRefreshPeriod", 0, "The IntraRefreshPeriod i.e. how many keyframes we send")
rotationPtr = flag.Uint("Rotation", 0, "Rotate video output. (0-359 degrees)")
@ -180,11 +183,20 @@ func handleFlags() revid.Config {
cfg.Input = revid.V4L
case "File":
cfg.Input = revid.File
case "Audio":
cfg.Input = revid.Audio
case "":
default:
log.Log(logger.Error, pkg+"bad input argument")
}
switch *inputPtr {
case "Audio":
cfg.Rate = float64(*sampleRatePtr*sampleSize) / float64(blockSize)
default:
cfg.Rate = *frameRatePtr
}
switch *inputCodecPtr {
case "H264":
cfg.InputCodec = revid.H264

View File

@ -1,329 +0,0 @@
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
}

View File

@ -62,7 +62,8 @@ type Config struct {
Height uint
Width uint
FrameRate uint
Rate uint
SampleRate uint
Rate float64
HttpAddress string
Quantization uint
IntraRefreshPeriod uint
@ -113,6 +114,7 @@ const (
Raspivid
V4L
H264Codec
Audio
File
Http
H264
@ -139,6 +141,7 @@ const (
defaultOutput = Http
defaultPacketization = Flv
defaultFrameRate = 25
defaultSampleRate = 48000
defaultRate = 25
defaultWidth = 1280
defaultHeight = 720
@ -174,7 +177,7 @@ func (c *Config) Validate(r *Revid) error {
}
switch c.Input {
case Raspivid, V4L, File:
case Raspivid, V4L, File, Audio:
case NothingDefined:
c.Logger.Log(logger.Info, pkg+"no input type defined, defaulting", "input",
defaultInput)
@ -276,6 +279,11 @@ func (c *Config) Validate(r *Revid) error {
c.FrameRate = defaultFrameRate
}
if c.SampleRate == 0 {
c.Logger.Log(logger.Info, pkg+"no sample rate defined, defaulting", "sampleRate", defaultSampleRate)
c.SampleRate = defaultSampleRate
}
if c.Bitrate == 0 {
c.Logger.Log(logger.Info, pkg+"no bitrate defined, defaulting", "bitrate", defaultBitrate)
c.Bitrate = defaultBitrate