Filter interface take 2

This branch was made because we were having problems with rebasing the original filter-interface branch. Filter-interface was used to make interface for filters and adding into pipeline made a new file (filter.go) that conatins package Filter. This package has the filter interface, with one filter, NoOp. More filters can be added. The filtering stage was placed inbetween lexing and encoding by changing the LexTo function to have the filter as it's destination and making the destination of the filter to be r.encoders

filter/filter.go

@@ -0,0 +1,48 @@
+/*
+NAME
+  filter.go
+
+AUTHORS
+  Ella Pietraroia <ella@ausocean.org>
+
+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"
+)
+
+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 }

filter/mog.go

@@ -46,7 +46,7 @@ type MOGFilter struct {
 }
 
 // NewMOGFilter returns a pointer to a new MOGFilter.
-func NewMOGFilter(dst io.WriteCloser, area, threshold float64, history, kernelSize int, debug bool) *MogFilter {
+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

go.mod

@@ -12,5 +12,4 @@ require (
 	github.com/pkg/errors v0.8.1
 	github.com/yobert/alsa v0.0.0-20180630182551-d38d89fa843e
 	gocv.io/x/gocv v0.21.0
-	gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 // indirect
 )

revid/config/config.go

@@ -108,6 +108,13 @@ const (
 	QualityExcellent
 )
 
+// The different filter methods that can be used (currently these are all motion filters
+// that will only send video with motion in it)
+const (
+	FilterNoOp = iota
+	FilterMOG
+)
+
 // 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 +255,7 @@ type Config struct {
 
 	HorizontalFlip bool // HorizontalFlip flips video horizontally for Raspivid input.
 	VerticalFlip   bool // VerticalFlip flips video vertically for Raspivid input.
+	FilterMethod   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.

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"
@@ -110,7 +111,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 +328,13 @@ func (r *Revid) setupPipeline(mtsEnc func(dst io.WriteCloser, rate float64) (io.
 
 	r.encoders = multiWriter(encoders...)
 
+	switch r.cfg.FilterMethod {
+	case config.FilterNoOp:
+		r.filter = filter.NewNoOp(r.encoders)
+	default:
+		r.filter = filter.NewNoOp(r.encoders)
+	}
+
 	switch r.cfg.Input {
 	case config.InputRaspivid:
 		r.input = raspivid.New(r.cfg.Logger)
@@ -611,6 +622,13 @@ func (r *Revid) Update(vars map[string]string) error {
 			default:
 				r.cfg.Logger.Log(logger.Warning, pkg+"invalid VerticalFlip param", "value", value)
 			}
+		case "FilterMethod":
+			m := map[string]int{"NoOp": config.FilterNoOp, "MOG": config.FilterMOG}
+			v, ok := m[value]
+			if !ok {
+				r.cfg.Logger.Log(logger.Warning, pkg+"invalid FilterMethod param", "value", value)
+			}
+			r.cfg.FilterMethod = v
 		case "PSITime":
 			v, err := strconv.Atoi(value)
 			if err != nil || v < 0 {
@@ -719,7 +737,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()
 }