mirror of https://bitbucket.org/ausocean/av.git
942 lines
28 KiB
Go
942 lines
28 KiB
Go
/*
|
|
NAME
|
|
revid.go
|
|
|
|
AUTHORS
|
|
Saxon A. Nelson-Milton <saxon@ausocean.org>
|
|
Alan Noble <alan@ausocean.org>
|
|
Dan Kortschak <dan@ausocean.org>
|
|
Trek Hopton <trek@ausocean.org>
|
|
Scott Barnard <scott@ausocean.org>
|
|
|
|
LICENSE
|
|
revid is Copyright (C) 2017-2020 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
|
|
in gpl.txt. If not, see http://www.gnu.org/licenses.
|
|
*/
|
|
|
|
// Package revid provides an API for reading, transcoding, and writing audio/video streams and files.
|
|
package revid
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"bitbucket.org/ausocean/av/codec/codecutil"
|
|
"bitbucket.org/ausocean/av/codec/h264"
|
|
"bitbucket.org/ausocean/av/codec/h265"
|
|
"bitbucket.org/ausocean/av/codec/mjpeg"
|
|
"bitbucket.org/ausocean/av/container/flv"
|
|
"bitbucket.org/ausocean/av/container/mts"
|
|
"bitbucket.org/ausocean/av/device"
|
|
"bitbucket.org/ausocean/av/device/file"
|
|
"bitbucket.org/ausocean/av/device/geovision"
|
|
"bitbucket.org/ausocean/av/device/raspivid"
|
|
"bitbucket.org/ausocean/av/device/webcam"
|
|
"bitbucket.org/ausocean/av/filter"
|
|
"bitbucket.org/ausocean/av/revid/config"
|
|
"bitbucket.org/ausocean/iot/pi/netsender"
|
|
"bitbucket.org/ausocean/utils/bitrate"
|
|
"bitbucket.org/ausocean/utils/ioext"
|
|
"bitbucket.org/ausocean/utils/logger"
|
|
"bitbucket.org/ausocean/utils/vring"
|
|
)
|
|
|
|
// RTMP connection properties.
|
|
const (
|
|
rtmpConnectionMaxTries = 5
|
|
rtmpConnectionTimeout = 10
|
|
)
|
|
|
|
type Logger interface {
|
|
SetLevel(int8)
|
|
Log(level int8, message string, params ...interface{})
|
|
}
|
|
|
|
// Revid provides methods to control a revid session; providing methods
|
|
// to start, stop and change the state of an instance using the Config struct.
|
|
type Revid struct {
|
|
// config holds the Revid configuration.
|
|
// For historical reasons it also handles logging.
|
|
// FIXME(kortschak): The relationship of concerns
|
|
// in config/ns is weird.
|
|
cfg config.Config
|
|
|
|
// ns holds the netsender.Sender responsible for HTTP.
|
|
ns *netsender.Sender
|
|
|
|
// input will capture audio or video from which we can read data.
|
|
input device.AVDevice
|
|
|
|
// closeInput holds the cleanup function return from setupInput and is called
|
|
// in Revid.Stop().
|
|
closeInput func() error
|
|
|
|
// lexTo, encoder and packer handle transcoding the input stream.
|
|
lexTo func(dest io.Writer, src io.Reader, delay time.Duration) error
|
|
|
|
// filters will hold the filter interface that will write to the chosen filter from the lexer.
|
|
filters []filter.Filter
|
|
|
|
// encoders will hold the multiWriteCloser that writes to encoders from the filter.
|
|
encoders io.WriteCloser
|
|
|
|
// running is used to keep track of revid's running state between methods.
|
|
running bool
|
|
|
|
// mu is used to protect isRunning during concurrent use.
|
|
mu sync.Mutex
|
|
|
|
// wg will be used to wait for any processing routines to finish.
|
|
wg sync.WaitGroup
|
|
|
|
// err will channel errors from revid routines to the handle errors routine.
|
|
err chan error
|
|
|
|
// bitrate is used for bitrate calculations.
|
|
bitrate bitrate.Calculator
|
|
}
|
|
|
|
// New returns a pointer to a new Revid with the desired configuration, and/or
|
|
// an error if construction of the new instance was not successful.
|
|
func New(c config.Config, ns *netsender.Sender) (*Revid, error) {
|
|
r := Revid{ns: ns, err: make(chan error)}
|
|
err := r.setConfig(c)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("could not set config, failed with error: %w", err)
|
|
}
|
|
r.cfg.Logger.SetLevel(c.LogLevel)
|
|
go r.handleErrors()
|
|
return &r, nil
|
|
}
|
|
|
|
// Config returns a copy of revids current config.
|
|
//
|
|
// Config is not safe for concurrent use.
|
|
func (r *Revid) Config() config.Config {
|
|
return r.cfg
|
|
}
|
|
|
|
// TODO(Saxon): put more thought into error severity and how to handle these.
|
|
func (r *Revid) handleErrors() {
|
|
for {
|
|
err := <-r.err
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Error, "async error", "error", err.Error())
|
|
}
|
|
}
|
|
}
|
|
|
|
// Bitrate returns the result of the most recent bitrate check.
|
|
func (r *Revid) Bitrate() int {
|
|
return r.bitrate.Bitrate()
|
|
}
|
|
|
|
// reset swaps the current config of a Revid with the passed
|
|
// configuration; checking validity and returning errors if not valid. It then
|
|
// sets up the data pipeline accordingly to this configuration.
|
|
func (r *Revid) reset(c config.Config) error {
|
|
r.cfg.Logger.Log(logger.Debug, "setting config")
|
|
err := r.setConfig(c)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
r.cfg.Logger.Log(logger.Info, "config set")
|
|
|
|
r.cfg.Logger.Log(logger.Debug, "setting up revid pipeline")
|
|
err = r.setupPipeline(
|
|
func(dst io.WriteCloser, fps float64) (io.WriteCloser, error) {
|
|
var st int
|
|
var encOptions []func(*mts.Encoder) error
|
|
|
|
switch r.cfg.Input {
|
|
case config.InputRaspivid:
|
|
switch r.cfg.InputCodec {
|
|
case codecutil.H264:
|
|
st = mts.EncodeH264
|
|
case codecutil.MJPEG:
|
|
st = mts.EncodeMJPEG
|
|
encOptions = append(encOptions, mts.TimeBasedPSI(time.Duration(r.cfg.PSITime)*time.Second))
|
|
r.cfg.CBR = true
|
|
default:
|
|
panic("unknown input codec for raspivid input")
|
|
}
|
|
case config.InputFile, config.InputV4L:
|
|
switch r.cfg.InputCodec {
|
|
case codecutil.H264:
|
|
st = mts.EncodeH264
|
|
case codecutil.MJPEG:
|
|
st = mts.EncodeMJPEG
|
|
encOptions = append(encOptions, mts.TimeBasedPSI(time.Duration(r.cfg.PSITime)*time.Second))
|
|
r.cfg.CBR = true
|
|
default:
|
|
panic(fmt.Sprintf("unknown input codec %d for v4l or input file input", r.cfg.InputCodec))
|
|
}
|
|
case config.InputRTSP:
|
|
switch r.cfg.InputCodec {
|
|
case codecutil.H265:
|
|
st = mts.EncodeH265
|
|
case codecutil.H264:
|
|
st = mts.EncodeH264
|
|
case codecutil.MJPEG:
|
|
st = mts.EncodeMJPEG
|
|
encOptions = append(encOptions, mts.TimeBasedPSI(time.Duration(r.cfg.PSITime)*time.Second))
|
|
r.cfg.CBR = true
|
|
default:
|
|
panic("unknown input codec for RTSP input")
|
|
}
|
|
case config.InputAudio:
|
|
st = mts.EncodeAudio
|
|
default:
|
|
panic("unknown input type")
|
|
}
|
|
|
|
return mts.NewEncoder(dst, float64(fps), st, encOptions...)
|
|
},
|
|
func(dst io.WriteCloser, fps int) (io.WriteCloser, error) {
|
|
return flv.NewEncoder(dst, true, true, fps)
|
|
},
|
|
ioext.MultiWriteCloser,
|
|
)
|
|
r.cfg.Logger.Log(logger.Info, "finished setting pipeline")
|
|
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// setConfig takes a config, checks it's validity and then replaces the current
|
|
// revid config.
|
|
func (r *Revid) setConfig(config config.Config) error {
|
|
r.cfg.Logger = config.Logger
|
|
r.cfg.Logger.Log(logger.Debug, "validating config")
|
|
err := config.Validate()
|
|
if err != nil {
|
|
return errors.New("Config struct is bad: " + err.Error())
|
|
}
|
|
r.cfg.Logger.Log(logger.Info, "config validated")
|
|
r.cfg = config
|
|
return nil
|
|
}
|
|
|
|
// setupPipeline constructs the revid dataPipeline. Inputs, encoders and
|
|
// senders are created and linked based on the current revid config.
|
|
//
|
|
// mtsEnc and flvEnc will be called to obtain an mts encoder and flv encoder
|
|
// respectively. multiWriter will be used to create an ioext.multiWriteCloser
|
|
// so that encoders can write to multiple senders.
|
|
func (r *Revid) setupPipeline(mtsEnc func(dst io.WriteCloser, rate float64) (io.WriteCloser, error), flvEnc func(dst io.WriteCloser, rate int) (io.WriteCloser, error), multiWriter func(...io.WriteCloser) io.WriteCloser) error {
|
|
// encoders will hold the encoders that are required for revid's current
|
|
// configuration.
|
|
var encoders []io.WriteCloser
|
|
|
|
// mtsSenders will hold the senders the require MPEGTS encoding, and flvSenders
|
|
// will hold senders that require FLV encoding.
|
|
var mtsSenders, flvSenders []io.WriteCloser
|
|
|
|
// We will go through our outputs and create the corresponding senders to add
|
|
// to mtsSenders if the output requires MPEGTS encoding, or flvSenders if the
|
|
// output requires FLV encoding.
|
|
var w io.WriteCloser
|
|
for _, out := range r.cfg.Outputs {
|
|
switch out {
|
|
case config.OutputHTTP:
|
|
r.cfg.Logger.Log(logger.Debug, "using HTTP output")
|
|
rb, err := vring.NewBuffer(r.cfg.RBMaxElements, r.cfg.RBCapacity, time.Duration(r.cfg.RBWriteTimeout)*time.Second)
|
|
if err != nil {
|
|
return fmt.Errorf("could not initialise MTS ring buffer: %w", err)
|
|
}
|
|
w = newMtsSender(
|
|
newHttpSender(r.ns, r.cfg.Logger.Log, r.bitrate.Report),
|
|
r.cfg.Logger.Log,
|
|
rb,
|
|
r.cfg.ClipDuration,
|
|
)
|
|
mtsSenders = append(mtsSenders, w)
|
|
case config.OutputRTP:
|
|
r.cfg.Logger.Log(logger.Debug, "using RTP output")
|
|
w, err := newRtpSender(r.cfg.RTPAddress, r.cfg.Logger.Log, r.cfg.FrameRate, r.bitrate.Report)
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Warning, "rtp connect error", "error", err.Error())
|
|
}
|
|
mtsSenders = append(mtsSenders, w)
|
|
case config.OutputFile:
|
|
r.cfg.Logger.Log(logger.Debug, "using File output")
|
|
w, err := newFileSender(r.cfg.OutputPath)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
mtsSenders = append(mtsSenders, w)
|
|
case config.OutputRTMP:
|
|
r.cfg.Logger.Log(logger.Debug, "using RTMP output")
|
|
rb, err := vring.NewBuffer(r.cfg.RBMaxElements, r.cfg.RBCapacity, time.Duration(r.cfg.RBWriteTimeout)*time.Second)
|
|
if err != nil {
|
|
return fmt.Errorf("could not initialise RTMP ring buffer: %w", err)
|
|
}
|
|
w, err := newRtmpSender(
|
|
r.cfg.RTMPURL,
|
|
rtmpConnectionTimeout,
|
|
rtmpConnectionMaxTries,
|
|
rb,
|
|
r.cfg.Logger.Log,
|
|
r.bitrate.Report,
|
|
)
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Warning, "rtmp connect error", "error", err.Error())
|
|
}
|
|
flvSenders = append(flvSenders, w)
|
|
}
|
|
}
|
|
|
|
// If we have some senders that require MPEGTS encoding then add an MPEGTS
|
|
// encoder to revid's encoder slice, and give this encoder the mtsSenders
|
|
// as a destination.
|
|
if len(mtsSenders) != 0 {
|
|
mw := multiWriter(mtsSenders...)
|
|
e, _ := mtsEnc(mw, r.cfg.WriteRate)
|
|
encoders = append(encoders, e)
|
|
}
|
|
|
|
// If we have some senders that require FLV encoding then add an FLV
|
|
// encoder to revid's encoder slice, and give this encoder the flvSenders
|
|
// as a destination.
|
|
if len(flvSenders) != 0 {
|
|
mw := multiWriter(flvSenders...)
|
|
e, err := flvEnc(mw, int(r.cfg.FrameRate))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
encoders = append(encoders, e)
|
|
}
|
|
|
|
r.encoders = multiWriter(encoders...)
|
|
|
|
l := len(r.cfg.Filters)
|
|
r.filters = []filter.Filter{filter.NewNoOp(r.encoders)}
|
|
if l != 0 {
|
|
r.cfg.Logger.Log(logger.Debug, "setting up filters", "filters", r.cfg.Filters)
|
|
r.filters = make([]filter.Filter, l)
|
|
dst := r.encoders
|
|
|
|
for i := l - 1; i >= 0; i-- {
|
|
switch r.cfg.Filters[i] {
|
|
case config.FilterNoOp:
|
|
r.cfg.Logger.Log(logger.Debug, "using NoOp filter")
|
|
r.filters[i] = filter.NewNoOp(dst)
|
|
case config.FilterMOG:
|
|
r.cfg.Logger.Log(logger.Debug, "using MOG filter")
|
|
r.filters[i] = filter.NewMOG(dst, r.cfg)
|
|
case config.FilterVariableFPS:
|
|
r.cfg.Logger.Log(logger.Debug, "using Variable FPS MOG filter")
|
|
r.filters[i] = filter.NewVariableFPS(dst, r.cfg.MinFPS, filter.NewMOG(dst, r.cfg))
|
|
case config.FilterKNN:
|
|
r.cfg.Logger.Log(logger.Debug, "using KNN filter")
|
|
r.filters[i] = filter.NewKNN(dst, r.cfg)
|
|
case config.FilterDiff:
|
|
r.cfg.Logger.Log(logger.Debug, "using gocv difference filter")
|
|
r.filters[i] = filter.NewDiff(dst, r.cfg)
|
|
case config.FilterBasic:
|
|
r.cfg.Logger.Log(logger.Debug, "using go difference filter")
|
|
r.filters[i] = filter.NewBasic(dst, r.cfg)
|
|
default:
|
|
panic("Undefined Filter")
|
|
}
|
|
dst = r.filters[i]
|
|
}
|
|
r.cfg.Logger.Log(logger.Info, "filters set up")
|
|
}
|
|
|
|
switch r.cfg.Input {
|
|
case config.InputRaspivid:
|
|
r.cfg.Logger.Log(logger.Debug, "using raspivid input")
|
|
r.input = raspivid.New(r.cfg.Logger)
|
|
r.setLexer(r.cfg.InputCodec, false)
|
|
|
|
case config.InputV4L:
|
|
r.cfg.Logger.Log(logger.Debug, "using V4L input")
|
|
r.input = webcam.New(r.cfg.Logger)
|
|
r.setLexer(r.cfg.InputCodec, false)
|
|
|
|
case config.InputFile:
|
|
r.cfg.Logger.Log(logger.Debug, "using file input")
|
|
r.input = file.New()
|
|
r.setLexer(r.cfg.InputCodec, false)
|
|
|
|
case config.InputRTSP:
|
|
r.cfg.Logger.Log(logger.Debug, "using RTSP input")
|
|
r.input = geovision.New(r.cfg.Logger)
|
|
r.setLexer(r.cfg.InputCodec, true)
|
|
|
|
case config.InputAudio:
|
|
r.cfg.Logger.Log(logger.Debug, "using audio input")
|
|
err := r.setupAudio()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
// Configure the input device. We know that defaults are set, so no need to
|
|
// return error, but we should log.
|
|
r.cfg.Logger.Log(logger.Debug, "configuring input device")
|
|
err := r.input.Set(r.cfg)
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Warning, "errors from configuring input device", "errors", err)
|
|
}
|
|
r.cfg.Logger.Log(logger.Info, "input device configured")
|
|
|
|
return nil
|
|
}
|
|
|
|
// setLexer sets the revid input lexer based on input codec and whether input
|
|
// is RTSP or not, in which case an RTP/<codec> extractor is used.
|
|
func (r *Revid) setLexer(c uint8, isRTSP bool) {
|
|
switch c {
|
|
case codecutil.H264:
|
|
r.cfg.Logger.Log(logger.Debug, "using H.264 codec")
|
|
r.lexTo = h264.Lex
|
|
if isRTSP {
|
|
r.lexTo = h264.NewExtractor().Extract
|
|
}
|
|
case codecutil.H265:
|
|
r.cfg.Logger.Log(logger.Debug, "using H.265 codec")
|
|
r.lexTo = h265.NewLexer(false).Lex
|
|
if !isRTSP {
|
|
panic("byte stream H.265 lexing not implemented")
|
|
}
|
|
case codecutil.MJPEG:
|
|
r.cfg.Logger.Log(logger.Debug, "using MJPEG codec")
|
|
r.lexTo = mjpeg.Lex
|
|
if isRTSP {
|
|
r.lexTo = mjpeg.NewExtractor().Extract
|
|
}
|
|
default:
|
|
panic("unrecognised codec")
|
|
}
|
|
}
|
|
|
|
// Start invokes a Revid to start processing video from a defined input
|
|
// and packetising (if theres packetization) to a defined output.
|
|
//
|
|
// Start is safe for concurrent use.
|
|
func (r *Revid) Start() error {
|
|
if r.IsRunning() {
|
|
r.cfg.Logger.Log(logger.Warning, "start called, but revid already running")
|
|
return nil
|
|
}
|
|
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
|
|
r.cfg.Logger.Log(logger.Debug, "resetting revid")
|
|
err := r.reset(r.cfg)
|
|
if err != nil {
|
|
r.Stop()
|
|
return err
|
|
}
|
|
r.cfg.Logger.Log(logger.Info, "revid reset")
|
|
|
|
// Calculate delay between frames based on FileFPS.
|
|
d := time.Duration(0)
|
|
if r.cfg.FileFPS != 0 {
|
|
d = time.Duration(1000/r.cfg.FileFPS) * time.Millisecond
|
|
}
|
|
|
|
r.cfg.Logger.Log(logger.Debug, "starting input processing routine")
|
|
r.wg.Add(1)
|
|
go r.processFrom(r.input, d)
|
|
|
|
r.running = true
|
|
return nil
|
|
}
|
|
|
|
// Stop closes down the pipeline. This closes encoders and sender output routines,
|
|
// connections, and/or files.
|
|
//
|
|
// Stop is safe for concurrent use.
|
|
func (r *Revid) Stop() {
|
|
if !r.IsRunning() {
|
|
r.cfg.Logger.Log(logger.Warning, "stop called but revid isn't running")
|
|
return
|
|
}
|
|
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
|
|
r.cfg.Logger.Log(logger.Debug, "stopping input")
|
|
err := r.input.Stop()
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Error, "could not stop input", "error", err.Error())
|
|
} else {
|
|
r.cfg.Logger.Log(logger.Info, "input stopped")
|
|
}
|
|
|
|
r.cfg.Logger.Log(logger.Debug, "closing pipeline")
|
|
err = r.encoders.Close()
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Error, "failed to close pipeline", "error", err.Error())
|
|
} else {
|
|
r.cfg.Logger.Log(logger.Info, "pipeline closed")
|
|
}
|
|
|
|
for _, filter := range r.filters {
|
|
err = filter.Close()
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Error, "failed to close filters", "error", err.Error())
|
|
} else {
|
|
r.cfg.Logger.Log(logger.Info, "filters closed")
|
|
}
|
|
}
|
|
|
|
r.cfg.Logger.Log(logger.Debug, "waiting for routines to finish")
|
|
r.wg.Wait()
|
|
r.cfg.Logger.Log(logger.Info, "routines finished")
|
|
|
|
r.running = false
|
|
}
|
|
|
|
// Burst starts revid, waits for time specified, and then stops revid.
|
|
func (r *Revid) Burst() error {
|
|
r.cfg.Logger.Log(logger.Debug, "starting revid")
|
|
err := r.Start()
|
|
if err != nil {
|
|
return fmt.Errorf("could not start revid: %w", err)
|
|
}
|
|
r.cfg.Logger.Log(logger.Info, "revid started")
|
|
|
|
dur := time.Duration(r.cfg.BurstPeriod) * time.Second
|
|
time.Sleep(dur)
|
|
|
|
r.cfg.Logger.Log(logger.Debug, "stopping revid")
|
|
r.Stop()
|
|
r.cfg.Logger.Log(logger.Info, "revid stopped")
|
|
|
|
return nil
|
|
}
|
|
|
|
func (r *Revid) IsRunning() bool {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
return r.running
|
|
}
|
|
|
|
// Update takes a map of variables and their values and edits the current config
|
|
// if the variables are recognised as valid parameters.
|
|
//
|
|
// Update is safe for concurrent use.
|
|
func (r *Revid) Update(vars map[string]string) error {
|
|
if r.IsRunning() {
|
|
r.cfg.Logger.Log(logger.Debug, "revid running; stopping for re-config")
|
|
r.Stop()
|
|
r.cfg.Logger.Log(logger.Info, "revid was running; stopped for re-config")
|
|
}
|
|
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
//look through the vars and update revid where needed
|
|
r.cfg.Logger.Log(logger.Debug, "checking vars from server", "vars", vars)
|
|
for key, value := range vars {
|
|
switch key {
|
|
case "Input":
|
|
v, ok := map[string]uint8{"raspivid": config.InputRaspivid, "rtsp": config.InputRTSP, "v4l": config.InputV4L, "file": config.InputFile}[strings.ToLower(value)]
|
|
if !ok {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid input var", "value", value)
|
|
break
|
|
}
|
|
r.cfg.Input = v
|
|
case "Saturation":
|
|
s, err := strconv.Atoi(value)
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid saturation param", "value", value)
|
|
break
|
|
}
|
|
r.cfg.Saturation = int(s)
|
|
case "Brightness":
|
|
b, err := strconv.Atoi(value)
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid brightness param", "value", value)
|
|
break
|
|
}
|
|
r.cfg.Brightness = uint(b)
|
|
case "Exposure":
|
|
r.cfg.Exposure = value
|
|
case "AutoWhiteBalance":
|
|
r.cfg.AutoWhiteBalance = value
|
|
case "InputCodec":
|
|
switch value {
|
|
case "H264":
|
|
r.cfg.InputCodec = codecutil.H264
|
|
case "MJPEG":
|
|
r.cfg.InputCodec = codecutil.MJPEG
|
|
default:
|
|
r.cfg.Logger.Log(logger.Warning, "invalid InputCodec variable value", "value", value)
|
|
}
|
|
case "Outputs":
|
|
outputs := strings.Split(value, ",")
|
|
r.cfg.Outputs = make([]uint8, len(outputs))
|
|
|
|
for i, output := range outputs {
|
|
switch output {
|
|
case "File":
|
|
r.cfg.Outputs[i] = config.OutputFile
|
|
case "Http":
|
|
r.cfg.Outputs[i] = config.OutputHTTP
|
|
case "Rtmp":
|
|
r.cfg.Outputs[i] = config.OutputRTMP
|
|
case "Rtp":
|
|
r.cfg.Outputs[i] = config.OutputRTP
|
|
default:
|
|
r.cfg.Logger.Log(logger.Warning, "invalid outputs param", "value", value)
|
|
continue
|
|
}
|
|
}
|
|
case "Output":
|
|
r.cfg.Outputs = make([]uint8, 1)
|
|
switch strings.ToLower(value) {
|
|
case "file":
|
|
r.cfg.Outputs[0] = config.OutputFile
|
|
case "http":
|
|
r.cfg.Outputs[0] = config.OutputHTTP
|
|
case "rtmp":
|
|
r.cfg.Outputs[0] = config.OutputRTMP
|
|
case "rtp":
|
|
r.cfg.Outputs[0] = config.OutputRTP
|
|
default:
|
|
r.cfg.Logger.Log(logger.Warning, "invalid output param", "value", value)
|
|
continue
|
|
}
|
|
|
|
case "RtmpUrl":
|
|
r.cfg.RTMPURL = value
|
|
case "RtpAddress":
|
|
r.cfg.RTPAddress = value
|
|
case "Bitrate":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid framerate param", "value", value)
|
|
break
|
|
}
|
|
r.cfg.Bitrate = uint(v)
|
|
case "OutputPath":
|
|
r.cfg.OutputPath = value
|
|
case "InputPath":
|
|
r.cfg.InputPath = value
|
|
case "Height":
|
|
h, err := strconv.Atoi(value)
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid height param", "value", value)
|
|
break
|
|
}
|
|
r.cfg.Height = uint(h)
|
|
case "Width":
|
|
w, err := strconv.Atoi(value)
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid width param", "value", value)
|
|
break
|
|
}
|
|
r.cfg.Width = uint(w)
|
|
case "FrameRate":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid framerate param", "value", value)
|
|
break
|
|
}
|
|
r.cfg.FrameRate = uint(v)
|
|
case "Rotation":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil || v > 359 {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid rotation param", "value", value)
|
|
break
|
|
}
|
|
r.cfg.Rotation = uint(v)
|
|
case "HttpAddress":
|
|
r.cfg.HTTPAddress = value
|
|
case "Quantization":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid quantization param", "value", v)
|
|
break
|
|
}
|
|
r.cfg.Quantization = uint(v)
|
|
case "MinFrames":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid MinFrames param", "value", value)
|
|
break
|
|
}
|
|
r.cfg.MinFrames = uint(v)
|
|
|
|
case "ClipDuration":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid ClipDuration param", "value", value)
|
|
break
|
|
}
|
|
r.cfg.ClipDuration = time.Duration(v) * time.Second
|
|
|
|
case "HorizontalFlip":
|
|
switch strings.ToLower(value) {
|
|
case "true":
|
|
r.cfg.HorizontalFlip = true
|
|
case "false":
|
|
r.cfg.HorizontalFlip = false
|
|
default:
|
|
r.cfg.Logger.Log(logger.Warning, "invalid HorizontalFlip param", "value", value)
|
|
}
|
|
case "VerticalFlip":
|
|
switch strings.ToLower(value) {
|
|
case "true":
|
|
r.cfg.VerticalFlip = true
|
|
case "false":
|
|
r.cfg.VerticalFlip = false
|
|
default:
|
|
r.cfg.Logger.Log(logger.Warning, "invalid VerticalFlip param", "value", value)
|
|
}
|
|
case "Filters":
|
|
filters := strings.Split(value, ",")
|
|
m := map[string]int{"NoOp": config.FilterNoOp, "MOG": config.FilterMOG, "VariableFPS": config.FilterVariableFPS, "KNN": config.FilterKNN, "Difference": config.FilterDiff, "Basic": config.FilterBasic}
|
|
r.cfg.Filters = make([]int, len(filters))
|
|
for i, filter := range filters {
|
|
v, ok := m[filter]
|
|
if !ok {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid Filters param", "value", value)
|
|
}
|
|
r.cfg.Filters[i] = v
|
|
}
|
|
case "PSITime":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil || v < 0 {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid PSITime var", "value", value)
|
|
break
|
|
}
|
|
r.cfg.PSITime = v
|
|
case "BurstPeriod":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid BurstPeriod param", "value", value)
|
|
break
|
|
}
|
|
r.cfg.BurstPeriod = uint(v)
|
|
case "Logging":
|
|
switch value {
|
|
case "Debug":
|
|
r.cfg.LogLevel = logger.Debug
|
|
case "Info":
|
|
r.cfg.LogLevel = logger.Info
|
|
case "Warning":
|
|
r.cfg.LogLevel = logger.Warning
|
|
case "Error":
|
|
r.cfg.LogLevel = logger.Error
|
|
case "Fatal":
|
|
r.cfg.LogLevel = logger.Fatal
|
|
default:
|
|
r.cfg.Logger.Log(logger.Warning, "invalid Logging param", "value", value)
|
|
}
|
|
case "RTMPRBMaxElements":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil || v < 0 {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid RTMPRBMaxElements var", "value", value)
|
|
break
|
|
}
|
|
r.cfg.RBMaxElements = v
|
|
case "RTMPRBCapacity":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil || v < 0 {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid RTMPRBCapacity var", "value", value)
|
|
break
|
|
}
|
|
r.cfg.RBCapacity = v
|
|
case "RTMPRBWriteTimeout":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil || v <= 0 {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid RTMPRBWriteTimeout var", "value", value)
|
|
break
|
|
}
|
|
r.cfg.RBWriteTimeout = v
|
|
case "MTSRBMaxElements":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil || v < 0 {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid MTSRBMaxElements var", "value", value)
|
|
break
|
|
}
|
|
r.cfg.RBMaxElements = v
|
|
case "MTSRBCapacity":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil || v < 0 {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid MTSRBCapacity var", "value", value)
|
|
break
|
|
}
|
|
r.cfg.RBCapacity = v
|
|
case "MTSRBWriteTimeout":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil || v <= 0 {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid MTSRBWriteTimeout var", "value", value)
|
|
break
|
|
}
|
|
r.cfg.RBWriteTimeout = v
|
|
case "CBR":
|
|
v, ok := map[string]bool{"true": true, "false": false}[strings.ToLower(value)]
|
|
if !ok {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid CBR var", "value", value)
|
|
break
|
|
}
|
|
r.cfg.CBR = v
|
|
case "CameraIP":
|
|
r.cfg.CameraIP = value
|
|
case "VBRQuality":
|
|
v, ok := map[string]config.Quality{"standard": config.QualityStandard, "fair": config.QualityFair, "good": config.QualityGood, "great": config.QualityGreat, "excellent": config.QualityExcellent}[strings.ToLower(value)]
|
|
if !ok {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid VBRQuality var", "value", value)
|
|
break
|
|
}
|
|
r.cfg.VBRQuality = v
|
|
case "VBRBitrate":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil || v <= 0 {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid VBRBitrate var", "value", value)
|
|
break
|
|
}
|
|
r.cfg.VBRBitrate = v
|
|
case "CameraChan":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil || (v != 1 && v != 2) {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid CameraChan var", "value", value)
|
|
break
|
|
}
|
|
r.cfg.CameraChan = v
|
|
case "MinFPS":
|
|
v, err := strconv.ParseFloat(value, 64)
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid MinFPS var", "value", value)
|
|
break
|
|
}
|
|
r.cfg.MinFPS = v
|
|
case "MotionMinArea":
|
|
v, err := strconv.ParseFloat(value, 64)
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid MotionMinArea var", "value", value)
|
|
break
|
|
}
|
|
r.cfg.MotionMinArea = v
|
|
case "MotionThreshold":
|
|
v, err := strconv.ParseFloat(value, 64)
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid MotionThreshold var", "value", value)
|
|
break
|
|
}
|
|
r.cfg.MotionThreshold = v
|
|
case "MotionKernel":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid MotionKernel var", "value", value)
|
|
break
|
|
}
|
|
r.cfg.MotionKernel = uint(v)
|
|
case "MotionHistory":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil || v <= 0 {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid MotionHistory var", "value", value)
|
|
break
|
|
}
|
|
r.cfg.MotionHistory = uint(v)
|
|
case "MotionPadding":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil || v <= 0 {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid MotionPadding var", "value", value)
|
|
break
|
|
}
|
|
r.cfg.MotionPadding = uint(v)
|
|
case "MotionPixels":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid MotionPixels var", "value", value)
|
|
break
|
|
}
|
|
r.cfg.MotionPixels = v
|
|
case "MotionDownscaling":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid MotionDownscaling var", "value", value)
|
|
break
|
|
}
|
|
r.cfg.MotionDownscaling = v
|
|
case "MotionInterval":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil || v < 0 {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid MotionInterval var", "value", value)
|
|
break
|
|
}
|
|
r.cfg.MotionInterval = v
|
|
case "FileFPS":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil {
|
|
r.cfg.Logger.Log(logger.Warning, "invalid FileFPS var", "value", value)
|
|
break
|
|
}
|
|
r.cfg.FileFPS = v
|
|
case "mode":
|
|
r.cfg.Loop = false
|
|
if value == "Loop" {
|
|
r.cfg.Loop = true
|
|
}
|
|
}
|
|
}
|
|
r.cfg.Logger.Log(logger.Info, "finished reconfig")
|
|
r.cfg.Logger.Log(logger.Debug, "config changed", "config", r.cfg)
|
|
return nil
|
|
}
|
|
|
|
// processFrom is run as a routine to read from a input data source, lex and
|
|
// then send individual access units to revid's encoders.
|
|
func (r *Revid) processFrom(in device.AVDevice, delay time.Duration) {
|
|
defer r.wg.Done()
|
|
|
|
for l := true; l; l = r.cfg.Loop {
|
|
err := in.Start()
|
|
if err != nil {
|
|
r.err <- fmt.Errorf("could not start input device: %w", err)
|
|
return
|
|
}
|
|
|
|
// Lex data from input device, in, until finished or an error is encountered.
|
|
// For a continuous source e.g. a camera or microphone, we should remain
|
|
// in this call indefinitely unless in.Stop() is called and an io.EOF is forced.
|
|
r.cfg.Logger.Log(logger.Debug, "lexing")
|
|
err = r.lexTo(r.filters[0], in, delay)
|
|
switch err {
|
|
case nil, io.EOF:
|
|
r.cfg.Logger.Log(logger.Info, "end of file")
|
|
case io.ErrUnexpectedEOF:
|
|
r.cfg.Logger.Log(logger.Info, "unexpected EOF from input")
|
|
default:
|
|
r.err <- err
|
|
}
|
|
r.cfg.Logger.Log(logger.Info, "finished reading input")
|
|
|
|
r.cfg.Logger.Log(logger.Debug, "stopping input")
|
|
err = in.Stop()
|
|
if err != nil {
|
|
r.err <- fmt.Errorf("could not stop input source: %w", err)
|
|
} else {
|
|
r.cfg.Logger.Log(logger.Info, "input stopped")
|
|
}
|
|
}
|
|
}
|