/* 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 ) const pkg = "revid: " 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, pkg+"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 { err := r.setConfig(c) if err != nil { return err } r.cfg.Logger.SetLevel(c.LogLevel) 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, ) 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 err := config.Validate() if err != nil { return errors.New("Config struct is bad: " + err.Error()) } 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: 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: 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, pkg+"rtp connect error", "error", err.Error()) } mtsSenders = append(mtsSenders, w) case config.OutputFile: w, err := newFileSender(r.cfg.OutputPath) if err != nil { return err } mtsSenders = append(mtsSenders, w) case config.OutputRTMP: 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, pkg+"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.filters = make([]filter.Filter, l) dst := r.encoders for i := l - 1; i >= 0; i-- { switch r.cfg.Filters[i] { case config.FilterNoOp: r.filters[i] = filter.NewNoOp(dst) case config.FilterMOG: r.filters[i] = filter.NewMOG(dst, r.cfg.MOGMinArea, r.cfg.MOGThreshold, int(r.cfg.MOGHistory), r.cfg.ShowWindows, r.cfg.MotionInterval, r.cfg.MotionDownscaling) case config.FilterVariableFPS: r.filters[i] = filter.NewVariableFPS(dst, r.cfg.MinFPS, filter.NewMOG(dst, r.cfg.MOGMinArea, r.cfg.MOGThreshold, int(r.cfg.MOGHistory), r.cfg.ShowWindows, r.cfg.MotionInterval, r.cfg.MotionDownscaling)) case config.FilterKNN: r.filters[i] = filter.NewKNN(dst, r.cfg.KNNMinArea, r.cfg.KNNThreshold, int(r.cfg.KNNHistory), int(r.cfg.KNNKernel), r.cfg.ShowWindows, r.cfg.MotionInterval, r.cfg.MotionDownscaling) case config.FilterDifference: r.filters[i] = filter.NewDifference(dst, r.cfg.ShowWindows, r.cfg.DiffThreshold) case config.FilterBasic: r.filters[i] = filter.NewBasic(dst, r.cfg.ShowWindows, r.cfg.BasicThreshold, r.cfg.BasicPixels) default: panic("Undefined Filter") } dst = r.filters[i] } } switch r.cfg.Input { case config.InputRaspivid: r.input = raspivid.New(r.cfg.Logger) switch r.cfg.InputCodec { case codecutil.H264: r.lexTo = h264.Lex case codecutil.MJPEG: r.lexTo = mjpeg.Lex } case config.InputV4L: r.input = webcam.New(r.cfg.Logger) switch r.cfg.InputCodec { case codecutil.H264: r.lexTo = h264.Lex case codecutil.MJPEG: r.lexTo = mjpeg.Lex } case config.InputFile: r.input = file.New() switch r.cfg.InputCodec { case codecutil.H264: r.lexTo = h264.Lex case codecutil.MJPEG: r.lexTo = mjpeg.Lex } case config.InputRTSP: r.input = geovision.New(r.cfg.Logger) switch r.cfg.InputCodec { case codecutil.H264: r.lexTo = h264.NewExtractor().Extract case codecutil.H265: r.lexTo = h265.NewLexer(false).Lex case codecutil.MJPEG: r.lexTo = mjpeg.NewExtractor().Extract } case config.InputAudio: 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. err := r.input.Set(r.cfg) if err != nil { r.cfg.Logger.Log(logger.Warning, pkg+"errors from configuring input device", "errors", err) } return nil } // 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, pkg+"start called, but revid already running") return nil } r.mu.Lock() defer r.mu.Unlock() r.cfg.Logger.Log(logger.Info, pkg+"starting Revid") err := r.reset(r.cfg) if err != nil { r.Stop() return err } // 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.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, pkg+"stop called but revid isn't running") return } r.mu.Lock() defer r.mu.Unlock() err := r.input.Stop() if err != nil { r.cfg.Logger.Log(logger.Error, pkg+"could not stop input", "error", err.Error()) } r.cfg.Logger.Log(logger.Info, pkg+"closing pid peline") err = r.encoders.Close() if err != nil { r.cfg.Logger.Log(logger.Error, pkg+"failed to close pipeline", "error", err.Error()) } for _, filter := range r.filters { err = filter.Close() if err != nil { r.cfg.Logger.Log(logger.Error, pkg+"failed to close pipeline", "error", err.Error()) } } r.cfg.Logger.Log(logger.Info, pkg+"closed pipeline") r.cfg.Logger.Log(logger.Info, pkg+"waiting for routines to close") r.wg.Wait() r.cfg.Logger.Log(logger.Info, pkg+"revid stopped") r.running = false } // Burst starts revid, waits for time specified, and then stops revid. func (r *Revid) Burst() error { r.cfg.Logger.Log(logger.Info, pkg+"starting burst") err := r.Start() if err != nil { return fmt.Errorf("could not start revid: %w", err) } dur := time.Duration(r.cfg.BurstPeriod) * time.Second time.Sleep(dur) r.cfg.Logger.Log(logger.Info, pkg+"stopping burst") r.Stop() 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.Stop() } r.mu.Lock() defer r.mu.Unlock() //look through the vars and update revid where needed 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, pkg+"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, pkg+"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, pkg+"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, pkg+"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, pkg+"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, pkg+"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, pkg+"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, pkg+"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, pkg+"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, pkg+"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, pkg+"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, pkg+"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, pkg+"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, pkg+"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, pkg+"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, pkg+"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.FilterDifference, "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, pkg+"invalid Filters param", "value", value) } r.cfg.Filters[i] = v } case "MotionInterval": v, err := strconv.Atoi(value) if err != nil || v < 0 { r.cfg.Logger.Log(logger.Warning, pkg+"invalid MotionInterval var", "value", value) break } r.cfg.MotionInterval = v case "PSITime": v, err := strconv.Atoi(value) if err != nil || v < 0 { r.cfg.Logger.Log(logger.Warning, pkg+"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, pkg+"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, pkg+"invalid Logging param", "value", value) } case "RTMPRBMaxElements": v, err := strconv.Atoi(value) if err != nil || v < 0 { r.cfg.Logger.Log(logger.Warning, pkg+"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, pkg+"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, pkg+"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, pkg+"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, pkg+"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, pkg+"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, pkg+"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, pkg+"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, pkg+"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, pkg+"invalid CameraChan var", "value", value) break } r.cfg.CameraChan = v case "ShowWindows": switch strings.ToLower(value) { case "true": r.cfg.ShowWindows = true case "false": r.cfg.ShowWindows = false default: r.cfg.Logger.Log(logger.Warning, pkg+"invalid ShowWindows var", "value", value) break } case "MinFPS": v, err := strconv.ParseFloat(value, 64) if err != nil { r.cfg.Logger.Log(logger.Warning, pkg+"invalid MinFPS var", "value", value) break } r.cfg.MinFPS = v case "KNNMinArea": v, err := strconv.ParseFloat(value, 64) if err != nil { r.cfg.Logger.Log(logger.Warning, pkg+"invalid KNNMinArea var", "value", value) break } r.cfg.KNNMinArea = v case "KNNThreshold": v, err := strconv.ParseFloat(value, 64) if err != nil { r.cfg.Logger.Log(logger.Warning, pkg+"invalid KNNThreshold var", "value", value) break } r.cfg.KNNThreshold = v case "DiffThreshold": v, err := strconv.ParseFloat(value, 64) if err != nil { r.cfg.Logger.Log(logger.Warning, pkg+"invalid DiffThreshold var", "value", value) break } r.cfg.DiffThreshold = v case "KNNKernel": v, err := strconv.Atoi(value) if err != nil { r.cfg.Logger.Log(logger.Warning, pkg+"invalid KNNKernel var", "value", value) break } r.cfg.KNNKernel = uint(v) case "MOGMinArea": v, err := strconv.ParseFloat(value, 64) if err != nil { r.cfg.Logger.Log(logger.Warning, pkg+"invalid MOGMinArea var", "value", value) break } r.cfg.MOGMinArea = v case "MOGThreshold": v, err := strconv.ParseFloat(value, 64) if err != nil { r.cfg.Logger.Log(logger.Warning, pkg+"invalid MOGThreshold var", "value", value) break } r.cfg.MOGThreshold = v case "KNNHistory": v, err := strconv.Atoi(value) if err != nil || v <= 0 { r.cfg.Logger.Log(logger.Warning, pkg+"invalid KNNHistory var", "value", value) break } r.cfg.KNNHistory = uint(v) case "MOGHistory": v, err := strconv.Atoi(value) if err != nil || v <= 0 { r.cfg.Logger.Log(logger.Warning, pkg+"invalid MOGHistory var", "value", value) break } r.cfg.MOGHistory = uint(v) case "BasicThreshold": v, err := strconv.Atoi(value) if err != nil { r.cfg.Logger.Log(logger.Warning, pkg+"invalid BasicThreshold var", "value", value) break } r.cfg.BasicThreshold = v case "BasicPixels": v, err := strconv.Atoi(value) if err != nil { r.cfg.Logger.Log(logger.Warning, pkg+"invalid BasicPixels var", "value", value) break } r.cfg.BasicPixels = v case "FileFPS": v, err := strconv.Atoi(value) if err != nil { r.cfg.Logger.Log(logger.Warning, pkg+"invalid FileFPS var", "value", value) break } r.cfg.FileFPS = v case "mode": r.cfg.Loop = false if value == "Loop" { r.cfg.Loop = true } case "MotionDownscaling": v, err := strconv.Atoi(value) if err != nil { r.cfg.Logger.Log(logger.Warning, pkg+"invalid MotionDownscaling var", "value", value) break } r.cfg.MotionDownscaling = v } } r.cfg.Logger.Log(logger.Info, pkg+"revid config changed", "config", fmt.Sprintf("%+v", 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.Info, pkg+"lexing") err = r.lexTo(r.filters[0], in, delay) switch err { case nil, io.EOF: case io.ErrUnexpectedEOF: r.cfg.Logger.Log(logger.Info, pkg+"unexpected EOF from input") default: r.err <- err } err = in.Stop() if err != nil { r.err <- fmt.Errorf("could not stop input source: %w", err) } } r.cfg.Logger.Log(logger.Info, pkg+"finished lexing") }