diff --git a/.circleci/config.yml b/.circleci/config.yml index 85bfbcc7..b4ad0827 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -23,9 +23,9 @@ jobs: - run: go get -d -t -v ./... - - run: go build -v ./... + - run: go build -v -tags circleci ./... - - run: go test -v ./... + - run: go test -v -tags circleci ./... - save_cache: key: v1-pkg-cache diff --git a/device/webcam/webcam.go b/device/webcam/webcam.go index 0ac08751..fbb3a91b 100644 --- a/device/webcam/webcam.go +++ b/device/webcam/webcam.go @@ -28,6 +28,7 @@ import ( "errors" "fmt" "io" + "io/ioutil" "os/exec" "strings" @@ -61,15 +62,19 @@ var ( // Webcam is an implementation of the AVDevice interface for a Webcam. Webcam // uses an ffmpeg process to pipe the video data from the webcam. type Webcam struct { - out io.ReadCloser - log config.Logger - cfg config.Config - cmd *exec.Cmd + out io.ReadCloser + log config.Logger + cfg config.Config + cmd *exec.Cmd + done chan struct{} } // New returns a new Webcam. func New(l config.Logger) *Webcam { - return &Webcam{log: l} + return &Webcam{ + log: l, + done: make(chan struct{}), + } } // Name returns the name of the device. @@ -151,6 +156,32 @@ func (w *Webcam) Start() error { return fmt.Errorf("failed to create pipe: %w", err) } + stderr, err := w.cmd.StderrPipe() + if err != nil { + return fmt.Errorf("could not pipe command error: %w", err) + } + + go func() { + for { + select { + case <-w.done: + w.cfg.Logger.Log(logger.Info, "webcam.Stop() called, finished checking stderr") + return + default: + buf, err := ioutil.ReadAll(stderr) + if err != nil { + w.cfg.Logger.Log(logger.Error, "could not read stderr", "error", err) + return + } + + if len(buf) != 0 { + w.cfg.Logger.Log(logger.Error, "error from webcam stderr", "error", string(buf)) + return + } + } + } + }() + err = w.cmd.Start() if err != nil { return fmt.Errorf("failed to start ffmpeg: %w", err) @@ -161,6 +192,7 @@ func (w *Webcam) Start() error { // Stop will kill the ffmpeg process and close the output pipe. func (w *Webcam) Stop() error { + close(w.done) if w.cmd == nil || w.cmd.Process == nil { return errors.New("ffmpeg process was never started") } diff --git a/exp/gocv-exp/main.go b/exp/gocv-exp/main.go index 9de10184..8083df5a 100644 --- a/exp/gocv-exp/main.go +++ b/exp/gocv-exp/main.go @@ -1,3 +1,5 @@ +// +build !circleci + // What it does: // // This example detects motion using a delta threshold from the first frame, diff --git a/filter/filter.go b/filter/filter.go new file mode 100644 index 00000000..ec182531 --- /dev/null +++ b/filter/filter.go @@ -0,0 +1,49 @@ +/* +NAME + filter.go + +AUTHORS + Ella Pietraroia + +LICENSE + filter.go is Copyright (C) 2019 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 provides the interface and implementations of the filters to be used +// on video input that has been lexed. +package filter + +import ( + "io" +) + +// Interface for all filters. +type Filter interface { + io.WriteCloser + //NB: Filter interface may evolve with more methods as required. +} + +// The NoOp filter will perform no operation on the data that is being recieved, +// it will pass it on to the encoder with no changes. +type NoOp struct { + dst io.Writer +} + +func NewNoOp(dst io.Writer) *NoOp { return &NoOp{dst: dst} } + +func (n *NoOp) Write(p []byte) (int, error) { return n.dst.Write(p) } + +func (n *NoOp) Close() error { return nil } diff --git a/filter/filters_circleci.go b/filter/filters_circleci.go new file mode 100644 index 00000000..e9fc155a --- /dev/null +++ b/filter/filters_circleci.go @@ -0,0 +1,41 @@ +// +build circleci + +/* +DESCRIPTION + Replaces filters that use the gocv package when Circle-CI builds revid. This + is needed because Circle-CI does not have a copy of Open CV installed. + +AUTHORS + Scott Barnard + +LICENSE + filters_circleci.go is Copyright (C) 2019 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 ( + "io" +) + +// NewMOGFilter returns a pointer to a new NoOp struct for testing purposes only. +func NewMOGFilter(dst io.WriteCloser, area, threshold float64, history, kernelSize int, debug bool) *NoOp { + return &NoOp{dst: dst} +} + +func NewKNNFilter(dst io.WriteCloser, area, threshold float64, history, kernelSize int, debug bool) *NoOp { + return &NoOp{dst: dst} +} diff --git a/filter/knn.go b/filter/knn.go new file mode 100644 index 00000000..4ce51d9e --- /dev/null +++ b/filter/knn.go @@ -0,0 +1,133 @@ +// +build !circleci + +/* +DESCRIPTION + A filter that detects motion and discards frames without motion. The + filter uses a K-Nearest Neighbours (KNN) to determine what is + background and what is foreground. + +AUTHORS + Ella Pietraroia + +LICENSE + KNN.go is Copyright (C) 2019 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" +) + +// KNNFilter is a filter that provides basic motion detection. KNN is short for +// K-Nearest Neighbours method. +type KNNFilter struct { + dst io.WriteCloser + area float64 + bs *gocv.BackgroundSubtractorKNN + knl gocv.Mat + debug bool + windows []*gocv.Window +} + +// NewKNNFilter returns a pointer to a new KNNFilter. +func NewKNNFilter(dst io.WriteCloser, area, threshold float64, history, kernelSize int, debug bool) *KNNFilter { + bs := gocv.NewBackgroundSubtractorKNNWithParams(history, threshold, false) + k := gocv.GetStructuringElement(gocv.MorphRect, image.Pt(kernelSize, kernelSize)) + var windows []*gocv.Window + if debug { + windows = []*gocv.Window{gocv.NewWindow("Debug: Bounding boxes"), gocv.NewWindow("Debug: Motion")} + } + return &KNNFilter{dst, area, &bs, k, 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 (m *KNNFilter) Close() error { + m.bs.Close() + m.knl.Close() + for _, window := range m.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 (m *KNNFilter) Write(f []byte) (int, error) { + img, err := gocv.IMDecode(f, gocv.IMReadColor) + if err != nil { + return 0, fmt.Errorf("can't decode image: %w", err) + } + defer img.Close() + + imgDelta := gocv.NewMat() + defer imgDelta.Close() + + // Seperate foreground and background. + m.bs.Apply(img, &imgDelta) + + // Threshold imgDelta. + gocv.Threshold(imgDelta, &imgDelta, 25, 255, gocv.ThresholdBinary) + + // Remove noise. + gocv.Erode(imgDelta, &imgDelta, m.knl) + gocv.Dilate(imgDelta, &imgDelta, m.knl) + + // Fill small holes. + gocv.Dilate(imgDelta, &imgDelta, m.knl) + gocv.Erode(imgDelta, &imgDelta, m.knl) + + // Find contours and reject ones with a small area. + var contours [][]image.Point + allContours := gocv.FindContours(imgDelta, gocv.RetrievalExternal, gocv.ChainApproxSimple) + for _, c := range allContours { + if gocv.ContourArea(c) > m.area { + contours = append(contours, c) + } + } + + // Draw debug information. + if m.debug { + for _, c := range contours { + rect := gocv.BoundingRect(c) + gocv.Rectangle(&img, rect, color.RGBA{0, 0, 255, 0}, 1) + } + + if len(contours) > 0 { + gocv.PutText(&img, "Motion", image.Pt(32, 32), gocv.FontHersheyPlain, 2.0, color.RGBA{255, 0, 0, 0}, 2) + } + + m.windows[0].IMShow(img) + m.windows[1].IMShow(imgDelta) + m.windows[0].WaitKey(1) + } + + // Don't write to destination if there is no motion. + if len(contours) == 0 { + return -1, nil + } + + // Write to destination. + return m.dst.Write(f) +} diff --git a/filter/mog.go b/filter/mog.go index 99fe332b..283ea034 100644 --- a/filter/mog.go +++ b/filter/mog.go @@ -1,3 +1,5 @@ +// +build !circleci + /* DESCRIPTION A filter that detects motion and discards frames without motion. The @@ -45,8 +47,8 @@ type MOGFilter struct { windows []*gocv.Window } -// NewMOGFilter returns a pointer to a new MOGFilter. -func NewMOGFilter(dst io.WriteCloser, area, threshold float64, history, kernelSize int, debug bool) *MogFilter { +// NewMOGFilter returns a pointer to a new MOGFilter struct. +func NewMOGFilter(dst io.WriteCloser, area, threshold float64, history, kernelSize int, debug bool) *MOGFilter { bs := gocv.NewBackgroundSubtractorMOG2WithParams(history, threshold, false) k := gocv.GetStructuringElement(gocv.MorphRect, image.Pt(kernelSize, kernelSize)) var windows []*gocv.Window @@ -65,7 +67,7 @@ func (m *MOGFilter) Close() error { for _, window := range m.windows { window.Close() } - return m.dst.Close() + return nil } // Implements io.Writer. diff --git a/filter/vfps.go b/filter/vfps.go new file mode 100644 index 00000000..93b6f543 --- /dev/null +++ b/filter/vfps.go @@ -0,0 +1,67 @@ +/* +DESCRIPTION + A motion filter that has a variable frame rate. When motion is detected, + the filter sends all frames and when it is not, the filter sends frames + at a reduced rate, as set by a parameter. + +AUTHORS + Scott Barnard + +LICENSE + vfps.go is Copyright (C) 2019 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 ( + "io" +) + +// VariableFPSFilter is a filter that has a variable frame rate. When motion is +// detected, the filter sends all frames and when it is not, the filter +// sends frames at a reduced framerate. +type VariableFPSFilter struct { + filter Filter + dst io.WriteCloser + frames uint + count uint +} + +// NewVariableFPSFilter returns a pointer to a new VariableFPSFilter struct. +func NewVariableFPSFilter(dst io.WriteCloser, minFPS float64, filter Filter) *VariableFPSFilter { + frames := uint(25 / minFPS) + return &VariableFPSFilter{filter, dst, frames, 0} +} + +// Implements io.Writer. +// Write applies the motion filter to the video stream. Frames are sent +// at a reduced frame rate, except when motion is detected, then all frames +// with motion are sent. +func (v *VariableFPSFilter) Write(f []byte) (int, error) { + v.count = (v.count + 1) % v.frames + + if v.count == 0 { + return v.dst.Write(f) + } + + return v.filter.Write(f) +} + +// Implements io.Closer. +// Close calls the motion filter's Close method. +func (v *VariableFPSFilter) Close() error { + return v.filter.Close() +} diff --git a/revid/config/config.go b/revid/config/config.go index 9457199b..d13157b2 100644 --- a/revid/config/config.go +++ b/revid/config/config.go @@ -108,6 +108,14 @@ const ( QualityExcellent ) +// The different media filters. +const ( + FilterNoOp = iota + FilterMOG + FilterVariableFPS + FilterKNN +) + // Config provides parameters relevant to a revid instance. A new config must // be passed to the constructor. Default values for these fields are defined // as consts above. @@ -248,6 +256,7 @@ type Config struct { HorizontalFlip bool // HorizontalFlip flips video horizontally for Raspivid input. VerticalFlip bool // VerticalFlip flips video vertically for Raspivid input. + Filter int // Defines the method of filtering to be used in between lexing and encoding. PSITime int // Sets the time between a packet being sent // RTMP ring buffer parameters. @@ -279,12 +288,13 @@ var TypeData = map[string]string{ "Input": "enum:raspivid,rtsp,v4l,file", "InputCodec": "enum:H264,MJPEG", "InputPath": "string", - "Logging": "enum:Debug,Info,Warning,Error,Fatal", + "logging": "enum:Debug,Info,Warning,Error,Fatal", "MinFrames": "uint", "MTSRBElementSize": "int", "MTSRBSize": "int", "MTSRBWriteTimeout": "int", "OutputPath": "string", + "Output": "enum:File,Http,Rtmp,Rtp", "Outputs": "string", "Quantization": "uint", "Rotation": "uint", diff --git a/revid/revid.go b/revid/revid.go index b2b399dc..fd3f83cf 100644 --- a/revid/revid.go +++ b/revid/revid.go @@ -49,6 +49,7 @@ import ( "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/ioext" @@ -75,6 +76,15 @@ const ( rtmpConnectionTimeout = 10 ) +// KNN filter properties +const ( + knnMinArea = 25.0 + knnThreshold = 300 + knnHistory = 300 + knnKernel = 9 + knnShowWindows = true +) + const pkg = "revid: " type Logger interface { @@ -110,7 +120,10 @@ type Revid struct { // lexTo, encoder and packer handle transcoding the input stream. lexTo func(dest io.Writer, src io.Reader, delay time.Duration) error - // encoders will hold the multiWriteCloser that writes to encoders from the lexer. + // filter will hold the filter interface that will write to the chosen filter from the lexer. + filter 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. @@ -324,6 +337,19 @@ func (r *Revid) setupPipeline(mtsEnc func(dst io.WriteCloser, rate float64) (io. r.encoders = multiWriter(encoders...) + switch r.cfg.Filter { + case config.FilterNoOp: + r.filter = filter.NewNoOp(r.encoders) + case config.FilterMOG: + r.filter = filter.NewMOGFilter(r.encoders, 25, 20, 500, 3, true) + case config.FilterVariableFPS: + r.filter = filter.NewVariableFPSFilter(r.encoders, 1.0, filter.NewMOGFilter(r.encoders, 25, 20, 500, 3, true)) + case config.FilterKNN: + r.filter = filter.NewKNNFilter(r.encoders, knnMinArea, knnThreshold, knnHistory, knnKernel, knnShowWindows) + default: + panic("Undefined Filter") + } + switch r.cfg.Input { case config.InputRaspivid: r.input = raspivid.New(r.cfg.Logger) @@ -432,11 +458,16 @@ func (r *Revid) Stop() { r.cfg.Logger.Log(logger.Error, pkg+"could not stop input", "error", err.Error()) } - r.cfg.Logger.Log(logger.Info, pkg+"closing pipeline") + 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()) } + + err = r.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") if r.cmd != nil && r.cmd.Process != nil { @@ -505,7 +536,7 @@ func (r *Revid) Update(vars map[string]string) error { default: r.cfg.Logger.Log(logger.Warning, pkg+"invalid InputCodec variable value", "value", value) } - case "Output": + case "Outputs": outputs := strings.Split(value, ",") r.cfg.Outputs = make([]uint8, len(outputs)) @@ -520,10 +551,25 @@ func (r *Revid) Update(vars map[string]string) error { case "Rtp": r.cfg.Outputs[i] = config.OutputRTP default: - r.cfg.Logger.Log(logger.Warning, pkg+"invalid output param", "value", value) + r.cfg.Logger.Log(logger.Warning, pkg+"invalid outputs param", "value", value) continue } } + case "Output": + r.cfg.Outputs = make([]uint8, 1) + switch 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 @@ -611,6 +657,13 @@ func (r *Revid) Update(vars map[string]string) error { default: r.cfg.Logger.Log(logger.Warning, pkg+"invalid VerticalFlip param", "value", value) } + case "Filter": + m := map[string]int{"NoOp": config.FilterNoOp, "MOG": config.FilterMOG, "VariableFPS": config.FilterVariableFPS, "KNN": config.FilterKNN} + v, ok := m[value] + if !ok { + r.cfg.Logger.Log(logger.Warning, pkg+"invalid FilterMethod param", "value", value) + } + r.cfg.Filter = v case "PSITime": v, err := strconv.Atoi(value) if err != nil || v < 0 { @@ -719,7 +772,7 @@ 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) { - r.err <- r.lexTo(r.encoders, read, delay) + r.err <- r.lexTo(r.filter, read, delay) r.cfg.Logger.Log(logger.Info, pkg+"finished lexing") r.wg.Done() }