diff --git a/cmd/revid-cli/main.go b/cmd/revid-cli/main.go index db6e17ec..454bde10 100644 --- a/cmd/revid-cli/main.go +++ b/cmd/revid-cli/main.go @@ -128,6 +128,7 @@ func handleFlags() config.Config { httpAddressPtr = flag.String("HttpAddress", "", "Destination address of http posts") verticalFlipPtr = flag.Bool("VerticalFlip", false, "Flip video vertically: Yes, No") horizontalFlipPtr = flag.Bool("HorizontalFlip", false, "Flip video horizontally: Yes, No") + loopPtr = flag.Bool("Loop", false, "Loop input source on completion (true/false)") bitratePtr = flag.Uint("Bitrate", 0, "Bitrate of recorded video") heightPtr = flag.Uint("Height", 0, "Height in pixels") widthPtr = flag.Uint("Width", 0, "Width in pixels") @@ -138,6 +139,7 @@ func handleFlags() config.Config { saturationPtr = flag.Int("Saturation", 0, "Set Saturation. (100-100)") exposurePtr = flag.String("Exposure", "auto", "Set exposure mode. ("+strings.Join(raspivid.ExposureModes[:], ",")+")") autoWhiteBalancePtr = flag.String("Awb", "auto", "Set automatic white balance mode. ("+strings.Join(raspivid.AutoWhiteBalanceModes[:], ",")+")") + fileFPSPtr = flag.Int("FileFPS", 0, "File source frame processing FPS") // Audio specific flags. sampleRatePtr = flag.Int("SampleRate", 48000, "Sample rate of recorded audio") @@ -254,6 +256,8 @@ func handleFlags() config.Config { netsender.ConfigFile = *configFilePtr } + cfg.FileFPS = *fileFPSPtr + cfg.Loop = *loopPtr cfg.CameraIP = *cameraIPPtr cfg.Rotation = *rotationPtr cfg.HorizontalFlip = *horizontalFlipPtr diff --git a/filter/difference.go b/filter/difference.go new file mode 100644 index 00000000..d01cf0a7 --- /dev/null +++ b/filter/difference.go @@ -0,0 +1,129 @@ +// +build !circleci + +/* +DESCRIPTION + A filter that detects motion and discards frames without motion. The + algorithm calculates the absolute difference for each pixel between + two frames, then finds the mean. If the mean is above a given threshold, + then it is considered motion. + +AUTHORS + Scott Barnard + +LICENSE + difference.go is Copyright (C) 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 filter + +import ( + "fmt" + "image" + "image/color" + "io" + + "gocv.io/x/gocv" +) + +// Difference is a filter that provides basic motion detection. Difference calculates +// the absolute difference for each pixel between two frames, then finds the mean. If +// the mean is above a given threshold, then it is considered motion. +type Difference struct { + dst io.WriteCloser + thresh float64 + prev gocv.Mat + debug bool + windows []*gocv.Window +} + +// NewDifference returns a pointer to a new Difference struct. +func NewDifference(dst io.WriteCloser, debug bool, threshold float64) *Difference { + var windows []*gocv.Window + if debug { + windows = []*gocv.Window{gocv.NewWindow("Diff: Bounding boxes"), gocv.NewWindow("Diff: Motion")} + } + return &Difference{dst, threshold, gocv.NewMat(), debug, windows} +} + +// Implements io.Closer. +// Close frees resources used by gocv, because it has to be done manually, due to +// it using c-go. +func (d *Difference) Close() error { + d.prev.Close() + for _, window := range d.windows { + window.Close() + } + return nil +} + +// Implements io.Writer. +// Write applies the motion filter to the video stream. Only frames with motion +// are written to the destination encoder, frames without are discarded. +func (d *Difference) Write(f []byte) (int, error) { + if d.prev.Empty() { + var err error + d.prev, err = gocv.IMDecode(f, gocv.IMReadColor) + if err != nil { + return 0, err + } + return len(f), nil + } + + img, err := gocv.IMDecode(f, gocv.IMReadColor) + defer img.Close() + if err != nil { + return 0, err + } + + imgDelta := gocv.NewMat() + defer imgDelta.Close() + + // Seperate foreground and background. + gocv.AbsDiff(img, d.prev, &imgDelta) + gocv.CvtColor(imgDelta, &imgDelta, gocv.ColorBGRToGray) + + mean := imgDelta.Mean().Val1 + + // Update History. + d.prev = img.Clone() + + // Draw debug information. + if d.debug { + if mean >= d.thresh { + gocv.PutText( + &img, + fmt.Sprintf("motion - mean:%f", mean), + image.Pt(32, 32), + gocv.FontHersheyPlain, + 2.0, + color.RGBA{255, 0, 0, 0}, + 2, + ) + } + + d.windows[0].IMShow(img) + d.windows[1].IMShow(imgDelta) + d.windows[0].WaitKey(1) + } + + // Don't write to destination if there is no motion. + if mean < d.thresh { + return len(f), nil + } + + // Write to destination. + return d.dst.Write(f) +} diff --git a/filter/filters_circleci.go b/filter/filters_circleci.go index 374e8b0a..aaca83f6 100644 --- a/filter/filters_circleci.go +++ b/filter/filters_circleci.go @@ -36,6 +36,12 @@ func NewMOGFilter(dst io.WriteCloser, area, threshold float64, history int, debu return &NoOp{dst: dst} } +// NewKNNFilter returns a pointer to a new NoOp struct for testing purposes only. func NewKNNFilter(dst io.WriteCloser, area, threshold float64, history, kernelSize int, debug bool, hf int) *NoOp { return &NoOp{dst: dst} } + +// NewDiffference returns a pointer to a new NoOp struct for testing purposes only. +func NewDifference(dst io.WriteCloser, debug bool, threshold float64) *NoOp { + return &NoOp{dst: dst} +} diff --git a/go.mod b/go.mod index 2ed4c734..dfa6f093 100644 --- a/go.mod +++ b/go.mod @@ -4,7 +4,7 @@ go 1.13 require ( bitbucket.org/ausocean/iot v1.2.13 - bitbucket.org/ausocean/utils v1.2.12 + bitbucket.org/ausocean/utils v1.2.13 github.com/Comcast/gots v0.0.0-20190305015453-8d56e473f0f7 github.com/go-audio/audio v0.0.0-20181013203223-7b2a6ca21480 github.com/go-audio/wav v0.0.0-20181013172942-de841e69b884 diff --git a/go.sum b/go.sum index ef1fb9ff..d74fe617 100644 --- a/go.sum +++ b/go.sum @@ -12,6 +12,8 @@ bitbucket.org/ausocean/utils v1.2.11 h1:zA0FOaPjN960ryp8PKCkV5y50uWBYrIxCVnXjwbv bitbucket.org/ausocean/utils v1.2.11/go.mod h1:uXzX9z3PLemyURTMWRhVI8uLhPX4uuvaaO85v2hcob8= bitbucket.org/ausocean/utils v1.2.12 h1:VnskjWTDM475TnQRhBQE0cNp9D6Y6OELrd4UkD2VVIQ= bitbucket.org/ausocean/utils v1.2.12/go.mod h1:uXzX9z3PLemyURTMWRhVI8uLhPX4uuvaaO85v2hcob8= +bitbucket.org/ausocean/utils v1.2.13 h1:tUaIywtoMc1+zl1GCVQokX4mL5X7LNHX5O51AgAPrWA= +bitbucket.org/ausocean/utils v1.2.13/go.mod h1:uXzX9z3PLemyURTMWRhVI8uLhPX4uuvaaO85v2hcob8= github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/Comcast/gots v0.0.0-20190305015453-8d56e473f0f7 h1:LdOc9B9Bj6LEsKiXShkLA3/kpxXb6LJpH+ekU2krbzw= diff --git a/revid/config/config.go b/revid/config/config.go index 194faf33..0192b1bd 100644 --- a/revid/config/config.go +++ b/revid/config/config.go @@ -86,6 +86,7 @@ const ( defaultAudioInputCodec = codecutil.ADPCM defaultPSITime = 2 defaultMotionInterval = 5 + defaultFileFPS = 0 // Ring buffer defaults. defaultRBMaxElements = 10000 @@ -126,6 +127,7 @@ const ( FilterMOG FilterVariableFPS FilterKNN + FilterDifference ) // OS names @@ -297,6 +299,14 @@ type Config struct { MOGMinArea float64 // Used to ignore small areas of motion detection. MOGThreshold float64 // Intensity value from the KNN motion detection algorithm that is considered motion. MOGHistory uint // Length of MOG filter's history + + // If true will restart reading of input after an io.EOF. + Loop bool + + // Defines the rate at which frames from a file source are processed. + FileFPS int + // Difference filter parameters. + DiffThreshold float64 // Intensity value from the Difference motion detection algorithm that is considered motion. } // TypeData contains information about all of the variables that @@ -310,8 +320,10 @@ var TypeData = map[string]string{ "CameraIP": "string", "CBR": "bool", "ClipDuration": "uint", + "DiffThreshold": "float", "Exposure": "enum:auto,night,nightpreview,backlight,spotlight,sports,snow,beach,verylong,fixedfps,antishake,fireworks", - "Filters": "enums:NoOp,MOG,VariableFPS,KNN", + "FileFPS": "int", + "Filters": "enums:NoOp,MOG,VariableFPS,KNN,Difference", "FrameRate": "uint", "Height": "uint", "HorizontalFlip": "bool", @@ -324,20 +336,21 @@ var TypeData = map[string]string{ "KNNMinArea": "float", "KNNThreshold": "float", "logging": "enum:Debug,Info,Warning,Error,Fatal", + "Loop": "bool", "MinFPS": "float", "MinFrames": "uint", - "mode": "enum:Normal,Paused,Burst", + "mode": "enum:Normal,Paused,Burst,Loop", "MOGHistory": "uint", "MOGMinArea": "float", "MOGThreshold": "float", "MotionInterval": "int", - "RBCapacity": "uint", - "RBMaxElements": "uint", - "RBWriteTimeout": "uint", "Output": "enum:File,Http,Rtmp,Rtp", "OutputPath": "string", "Outputs": "enums:File,Http,Rtmp,Rtp", "Quantization": "uint", + "RBCapacity": "uint", + "RBMaxElements": "uint", + "RBWriteTimeout": "uint", "Rotation": "uint", "RTMPURL": "string", "RTPAddress": "string", @@ -521,6 +534,11 @@ func (c *Config) Validate() error { } } + if c.FileFPS <= 0 || (c.FileFPS > 0 && c.Input != InputFile) { + c.Logger.Log(logger.Info, pkg+"FileFPS bad or unset, defaulting", "FileFPS", defaultFileFPS) + c.FileFPS = defaultFileFPS + } + return nil } diff --git a/revid/revid.go b/revid/revid.go index 8b1b42da..5eb5b7bf 100644 --- a/revid/revid.go +++ b/revid/revid.go @@ -7,6 +7,7 @@ AUTHORS Alan Noble Dan Kortschak Trek Hopton + Scott Barnard LICENSE revid is Copyright (C) 2017-2020 the Australian Ocean Lab (AusOcean) @@ -52,6 +53,7 @@ import ( "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" @@ -115,6 +117,9 @@ type Revid struct { // 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 @@ -148,10 +153,8 @@ func (r *Revid) handleErrors() { } // Bitrate returns the result of the most recent bitrate check. -// -// TODO: get this working again. func (r *Revid) Bitrate() int { - return -1 + return r.bitrate.Bitrate() } // reset swaps the current config of a Revid with the passed @@ -266,14 +269,14 @@ func (r *Revid) setupPipeline(mtsEnc func(dst io.WriteCloser, rate float64) (io. return fmt.Errorf("could not initialise MTS ring buffer: %w", err) } w = newMtsSender( - newHttpSender(r.ns, r.cfg.Logger.Log), + 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) + 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()) } @@ -295,6 +298,7 @@ func (r *Revid) setupPipeline(mtsEnc func(dst io.WriteCloser, rate float64) (io. 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()) @@ -342,6 +346,8 @@ func (r *Revid) setupPipeline(mtsEnc func(dst io.WriteCloser, rate float64) (io. r.filters[i] = filter.NewVariableFPSFilter(dst, r.cfg.MinFPS, filter.NewMOGFilter(dst, r.cfg.MOGMinArea, r.cfg.MOGThreshold, int(r.cfg.MOGHistory), r.cfg.ShowWindows, r.cfg.MotionInterval)) case config.FilterKNN: r.filters[i] = filter.NewKNNFilter(dst, r.cfg.KNNMinArea, r.cfg.KNNThreshold, int(r.cfg.KNNHistory), int(r.cfg.KNNKernel), r.cfg.ShowWindows, r.cfg.MotionInterval) + case config.FilterDifference: + r.filters[i] = filter.NewDifference(dst, r.cfg.ShowWindows, r.cfg.DiffThreshold) default: panic("Undefined Filter") } @@ -428,13 +434,14 @@ func (r *Revid) Start() error { return err } - err = r.input.Start() - if err != nil { - return fmt.Errorf("could not start input device: %w", 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, 0) + go r.processFrom(r.input, d) r.running = true return nil @@ -662,7 +669,7 @@ func (r *Revid) Update(vars map[string]string) error { } case "Filters": filters := strings.Split(value, ",") - m := map[string]int{"NoOp": config.FilterNoOp, "MOG": config.FilterMOG, "VariableFPS": config.FilterVariableFPS, "KNN": config.FilterKNN} + m := map[string]int{"NoOp": config.FilterNoOp, "MOG": config.FilterMOG, "VariableFPS": config.FilterVariableFPS, "KNN": config.FilterKNN, "Difference": config.FilterDifference} r.cfg.Filters = make([]int, len(filters)) for i, filter := range filters { v, ok := m[filter] @@ -810,6 +817,13 @@ func (r *Revid) Update(vars map[string]string) error { 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 { @@ -845,6 +859,18 @@ func (r *Revid) Update(vars map[string]string) error { break } r.cfg.MOGHistory = uint(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 + } } } r.cfg.Logger.Log(logger.Info, pkg+"revid config changed", "config", fmt.Sprintf("%+v", r.cfg)) @@ -853,14 +879,34 @@ func (r *Revid) Update(vars map[string]string) error { // 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(read io.Reader, delay time.Duration) { - err := r.lexTo(r.filters[0], read, delay) - r.cfg.Logger.Log(logger.Debug, pkg+"finished lexing") - switch err { - case nil: // Do nothing. - case io.EOF: // TODO: handle this depending on loop mode. - default: - r.err <- err +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.wg.Done() + + r.cfg.Logger.Log(logger.Info, pkg+"finished lexing") } diff --git a/revid/senders.go b/revid/senders.go index 4871ddb5..4929bd40 100644 --- a/revid/senders.go +++ b/revid/senders.go @@ -61,19 +61,25 @@ const ( type httpSender struct { client *netsender.Sender log func(lvl int8, msg string, args ...interface{}) + report func(sent int) } // newHttpSender returns a pointer to a new httpSender. -func newHttpSender(ns *netsender.Sender, log func(lvl int8, msg string, args ...interface{})) *httpSender { +func newHttpSender(ns *netsender.Sender, log func(lvl int8, msg string, args ...interface{}), report func(sent int)) *httpSender { return &httpSender{ client: ns, log: log, + report: report, } } // Write implements io.Writer. func (s *httpSender) Write(d []byte) (int, error) { - return len(d), httpSend(d, s.client, s.log) + err := httpSend(d, s.client, s.log) + if err == nil { + s.report(len(d)) + } + return len(d), err } func (s *httpSender) Close() error { return nil } @@ -276,9 +282,10 @@ type rtmpSender struct { ring *vring.Buffer done chan struct{} wg sync.WaitGroup + report func(sent int) } -func newRtmpSender(url string, timeout uint, retries int, rb *vring.Buffer, log func(lvl int8, msg string, args ...interface{})) (*rtmpSender, error) { +func newRtmpSender(url string, timeout uint, retries int, rb *vring.Buffer, log func(lvl int8, msg string, args ...interface{}), report func(sent int)) (*rtmpSender, error) { var conn *rtmp.Conn var err error for n := 0; n < retries; n++ { @@ -299,6 +306,7 @@ func newRtmpSender(url string, timeout uint, retries int, rb *vring.Buffer, log log: log, ring: rb, done: make(chan struct{}), + report: report, } s.wg.Add(1) go s.output() @@ -364,6 +372,7 @@ func (s *rtmpSender) Write(d []byte) (int, error) { if err != nil { s.log(logger.Warning, pkg+"rtmpSender: ring buffer write error", "error", err.Error()) } + s.report(len(d)) return len(d), nil } @@ -404,9 +413,10 @@ type rtpSender struct { log func(lvl int8, msg string, args ...interface{}) encoder *rtp.Encoder data []byte + report func(sent int) } -func newRtpSender(addr string, log func(lvl int8, msg string, args ...interface{}), fps uint) (*rtpSender, error) { +func newRtpSender(addr string, log func(lvl int8, msg string, args ...interface{}), fps uint, report func(sent int)) (*rtpSender, error) { conn, err := net.Dial("udp", addr) if err != nil { return nil, err @@ -414,6 +424,7 @@ func newRtpSender(addr string, log func(lvl int8, msg string, args ...interface{ s := &rtpSender{ log: log, encoder: rtp.NewEncoder(conn, int(fps)), + report: report, } return s, nil } @@ -426,6 +437,7 @@ func (s *rtpSender) Write(d []byte) (int, error) { if err != nil { s.log(logger.Warning, pkg+"rtpSender: write error", err.Error()) } + s.report(len(d)) return len(d), nil }