Merge branch 'mjpeg-player-new' of https://bitbucket.org/ausocean/av into mjpeg-player-new

.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

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")
 	}

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,

filter/filter.go

@@ -0,0 +1,49 @@
+/*
+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"
+)
+
+// 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 }

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 <scott@ausocean.org>
+
+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}
+}

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 <ella@ausocean.org>
+
+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)
+}

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.

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 <scott@ausocean.org>
+
+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()
+}

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",

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()
 }