Merge branch 'master' into m3u-packages

cmd/mjpeg-player/lex-mjpeg.js

@@ -1,12 +1,9 @@
 /*
-NAME
-  lex-mjpeg.js
-
 AUTHOR
   Trek Hopton <trek@ausocean.org>
 
 LICENSE
-  This file is Copyright (C) 2019 the Australian Ocean Lab (AusOcean)
+  This file 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
@@ -24,11 +21,14 @@ LICENSE
 
 // MJPEGLexer lexes a byte array containing MJPEG into individual JPEGs.
 class MJPEGLexer {
-    constructor(src) {
+    constructor() {
-        this.src = src;
         this.off = 0;
     }
 
+    append(data) {
+        this.src = new Uint8Array(data);
+    }
+
     // read returns the next single frame.
     read() {
         // Check if the src can contain at least the start and end flags (4B).

cmd/revid-cli/main.go

@@ -32,7 +32,9 @@ package main
 
 import (
 	"flag"
+	"io"
 	"os"
+	"path/filepath"
 	"runtime/pprof"
 	"strconv"
 	"strings"
@@ -44,10 +46,11 @@ import (
 	"bitbucket.org/ausocean/av/device/raspivid"
 	"bitbucket.org/ausocean/av/revid"
 	"bitbucket.org/ausocean/av/revid/config"
+	"bitbucket.org/ausocean/iot/pi/netlogger"
 	"bitbucket.org/ausocean/iot/pi/netsender"
 	"bitbucket.org/ausocean/iot/pi/sds"
-	"bitbucket.org/ausocean/iot/pi/smartlogger"
 	"bitbucket.org/ausocean/utils/logger"
+	"gopkg.in/natefinch/lumberjack.v2"
 )
 
 // Revid modes
@@ -72,7 +75,10 @@ const (
 var canProfile = true
 
 // The logger that will be used throughout.
-var log *logger.Logger
+var (
+	netLog *netlogger.Logger
+	log    *logger.Logger
+)
 
 const (
 	metaPreambleKey  = "copyright"
@@ -160,7 +166,20 @@ func handleFlags() config.Config {
 		cfg.LogLevel = defaultLogVerbosity
 	}
 
-	log = logger.New(cfg.LogLevel, &smartlogger.New(*logPathPtr).LogRoller, true)
+	netLog = netlogger.New()
+	log = logger.New(
+		cfg.LogLevel,
+		io.MultiWriter(
+			&lumberjack.Logger{
+				Filename:   filepath.Join(*logPathPtr, "netsender.log"),
+				MaxSize:    500, // MB
+				MaxBackups: 10,
+				MaxAge:     28, // days
+			},
+			netLog,
+		),
+		true,
+	)
 
 	cfg.Logger = log
 
@@ -296,6 +315,11 @@ func run(cfg config.Config) {
 			continue
 		}
 
+		err = netLog.Send(ns)
+		if err != nil {
+			log.Log(logger.Warning, pkg+"Logs could not be sent", "error", err.Error())
+		}
+
 		// If var sum hasn't changed we continue.
 		var vars map[string]string
 		newVs := ns.VarSum()

codec/h264/lex.go

@@ -66,7 +66,7 @@ func Lex(dst io.Writer, src io.Reader, delay time.Duration) error {
 	buf := make([]byte, len(h264Prefix), bufSize)
 	copy(buf, h264Prefix[:])
 	writeOut := false
-outer:
+
 	for {
 		var b byte
 		var err error
@@ -75,7 +75,10 @@ outer:
 			if err != io.EOF {
 				return err
 			}
-			break
+			if len(buf) != 0 {
+				return io.ErrUnexpectedEOF
+			}
+			return io.EOF
 		}
 
 		for n := 1; b == 0x0 && n < 4; n++ {
@@ -84,7 +87,7 @@ outer:
 				if err != io.EOF {
 					return err
 				}
-				break outer
+				return io.ErrUnexpectedEOF
 			}
 			buf = append(buf, b)
 
@@ -109,7 +112,7 @@ outer:
 				if err != io.EOF {
 					return err
 				}
-				break outer
+				return io.ErrUnexpectedEOF
 			}
 			buf = append(buf, b)
 
@@ -127,10 +130,4 @@ outer:
 			}
 		}
 	}
-	if len(buf) == len(h264Prefix) {
-		return nil
-	}
-	<-tick
-	_, err := dst.Write(buf)
-	return err
 }

codec/h265/lex.go

@@ -76,10 +76,14 @@ func (l *Lexer) Lex(dst io.Writer, src io.Reader, delay time.Duration) error {
 		n, err := src.Read(buf)
 		switch err {
 		case nil: // Do nothing.
-		case io.EOF:
-			return nil
 		default:
-			return fmt.Errorf("source read error: %w\n", err)
+			if err == io.EOF {
+				if l.buf.Len() == 0 {
+					return io.EOF
+				}
+				return io.ErrUnexpectedEOF
+			}
+			return err
 		}
 
 		// Get payload from RTP packet.
@@ -181,7 +185,7 @@ func (l *Lexer) handleFragmentation(d []byte) {
 	}
 }
 
-// handlePACI will handl PACI packets
+// handlePACI will handle PACI packets
 //
 // TODO: complete this
 func (l *Lexer) handlePACI(d []byte) {

codec/h265/lex_test.go

@@ -247,7 +247,9 @@ func TestLex(t *testing.T) {
 		r := &rtpReader{packets: test.packets}
 		d := &destination{}
 		err := NewLexer(test.donl).Lex(d, r, 0)
-		if err != nil {
+		switch err {
+		case nil, io.EOF: // Do nothing
+		default:
 			t.Fatalf("error lexing: %v\n", err)
 		}
 

codec/mjpeg/lex.go

@@ -60,12 +60,9 @@ func Lex(dst io.Writer, src io.Reader, delay time.Duration) error {
 		buf := make([]byte, 2, 4<<10)
 		n, err := r.Read(buf)
 		if n < 2 {
-			return nil
+			return io.ErrUnexpectedEOF
 		}
 		if err != nil {
-			if err == io.EOF {
-				return nil
-			}
 			return err
 		}
 		if !bytes.Equal(buf, []byte{0xff, 0xd8}) {
@@ -76,7 +73,7 @@ func Lex(dst io.Writer, src io.Reader, delay time.Duration) error {
 			b, err := r.ReadByte()
 			if err != nil {
 				if err == io.EOF {
-					return nil
+					return io.ErrUnexpectedEOF
 				}
 				return err
 			}

filter/filters_circleci.go

@@ -32,10 +32,10 @@ import (
 )
 
 // NewMOGFilter returns a pointer to a new NoOp struct for testing purposes only.
-func NewMOGFilter(dst io.WriteCloser, area, threshold float64, history int, debug bool) *NoOp {
+func NewMOGFilter(dst io.WriteCloser, area, threshold float64, history int, debug bool, hf int) *NoOp {
 	return &NoOp{dst: dst}
 }
 
-func NewKNNFilter(dst io.WriteCloser, area, threshold float64, history, kernelSize int, debug bool) *NoOp {
+func NewKNNFilter(dst io.WriteCloser, area, threshold float64, history, kernelSize int, debug bool, hf int) *NoOp {
 	return &NoOp{dst: dst}
 }

filter/knn.go

@@ -40,23 +40,26 @@ import (
 // KNNFilter is a filter that provides basic motion detection. KNN is short for
 // K-Nearest Neighbours method.
 type KNNFilter struct {
-	dst     io.WriteCloser
+	dst     io.WriteCloser                // Destination to which motion containing frames go.
-	area    float64
+	area    float64                       // The minimum area that a contour can be found in.
-	bs      *gocv.BackgroundSubtractorKNN
+	bs      *gocv.BackgroundSubtractorKNN // Uses the KNN algorithm to find the difference between the current and background frame.
-	knl     gocv.Mat
+	knl     gocv.Mat                      // Matrix that is used for calculations.
-	debug   bool
+	debug   bool                          // If true then debug windows with the bounding boxes and difference will be shown on the screen.
-	windows []*gocv.Window
+	windows []*gocv.Window                // Holds debug windows.
+	hold    [][]byte                      // Will hold all frames up to hf (so only every hf frame is motion detected).
+	hf      int                           // The number of frames to be held.
+	hfCount int                           // Counter for the hold array.
 }
 
 // NewKNNFilter returns a pointer to a new KNNFilter.
-func NewKNNFilter(dst io.WriteCloser, area, threshold float64, history, kernelSize int, debug bool) *KNNFilter {
+func NewKNNFilter(dst io.WriteCloser, area, threshold float64, history, kernelSize int, debug bool, hf int) *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("KNN: Bounding boxes"), gocv.NewWindow("KNN: Motion")}
 	}
-	return &KNNFilter{dst, area, &bs, k, debug, windows}
+	return &KNNFilter{dst, area, &bs, k, debug, windows, make([][]byte, hf-1), hf, 0}
 }
 
 // Implements io.Closer.
@@ -75,6 +78,13 @@ func (m *KNNFilter) Close() error {
 // 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) {
+	if m.hfCount < (m.hf - 1) {
+		m.hold[m.hfCount] = f
+		m.hfCount++
+		return len(f), nil
+	}
+	m.hfCount = 0
+
 	img, err := gocv.IMDecode(f, gocv.IMReadColor)
 	if err != nil {
 		return 0, fmt.Errorf("can't decode image: %w", err)
@@ -125,9 +135,16 @@ func (m *KNNFilter) Write(f []byte) (int, error) {
 
 	// Don't write to destination if there is no motion.
 	if len(contours) == 0 {
-		return -1, nil
+		return len(f), nil
 	}
 
-	// Write to destination.
+	// Write to destination, past 4 frames then current frame.
+	for i, h := range m.hold {
+		_, err := m.dst.Write(h)
+		m.hold[i] = nil
+		if err != nil {
+			return len(f), fmt.Errorf("could not write previous frames: %w", err)
+		}
+	}
 	return m.dst.Write(f)
 }

filter/mog.go

@@ -40,23 +40,26 @@ import (
 // MOGFilter is a filter that provides basic motion detection. MoG is short for
 // Mixture of Gaussians method.
 type MOGFilter struct {
-	dst     io.WriteCloser
+	dst     io.WriteCloser                 // Destination to which motion containing frames go.
-	area    float64
+	area    float64                        // The minimum area that a contour can be found in.
-	bs      *gocv.BackgroundSubtractorMOG2
+	bs      *gocv.BackgroundSubtractorMOG2 // Uses the MOG algorithm to find the difference between the current and background frame.
-	knl     gocv.Mat
+	knl     gocv.Mat                       // Matrix that is used for calculations.
-	debug   bool
+	debug   bool                           // If true then debug windows with the bounding boxes and difference will be shown on the screen.
-	windows []*gocv.Window
+	windows []*gocv.Window                 // Holds debug windows.
+	hold    [][]byte                       // Will hold all frames up to hf (so only every hf frame is motion detected).
+	hf      int                            // The number of frames to be held.
+	hfCount int                            // Counter for the hold array.
 }
 
 // NewMOGFilter returns a pointer to a new MOGFilter struct.
-func NewMOGFilter(dst io.WriteCloser, area, threshold float64, history int, debug bool) *MOGFilter {
+func NewMOGFilter(dst io.WriteCloser, area, threshold float64, history int, debug bool, hf int) *MOGFilter {
 	bs := gocv.NewBackgroundSubtractorMOG2WithParams(history, threshold, false)
 	k := gocv.GetStructuringElement(gocv.MorphRect, image.Pt(3, 3))
 	var windows []*gocv.Window
 	if debug {
 		windows = []*gocv.Window{gocv.NewWindow("MOG: Bounding boxes"), gocv.NewWindow("MOG: Motion")}
 	}
-	return &MOGFilter{dst, area, &bs, k, debug, windows}
+	return &MOGFilter{dst, area, &bs, k, debug, windows, make([][]byte, hf-1), hf, 0}
 }
 
 // Implements io.Closer.
@@ -75,6 +78,13 @@ func (m *MOGFilter) Close() error {
 // 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 *MOGFilter) Write(f []byte) (int, error) {
+	if m.hfCount < (m.hf - 1) {
+		m.hold[m.hfCount] = f
+		m.hfCount++
+		return len(f), nil
+	}
+
+	m.hfCount = 0
 	img, err := gocv.IMDecode(f, gocv.IMReadColor)
 	if err != nil {
 		return 0, fmt.Errorf("image can't be decoded: %w", err)
@@ -125,9 +135,16 @@ func (m *MOGFilter) Write(f []byte) (int, error) {
 
 	// Don't write to destination if there is no motion.
 	if len(contours) == 0 {
-		return 0, nil
+		return len(f), nil
 	}
 
-	// Write to destination.
+	// Write to destination, past 4 frames then current frame.
+	for i, h := range m.hold {
+		_, err := m.dst.Write(h)
+		m.hold[i] = nil
+		if err != nil {
+			return len(f), fmt.Errorf("could not write previous frames: %w", err)
+		}
+	}
 	return m.dst.Write(f)
 }

go.mod

@@ -3,7 +3,7 @@ module bitbucket.org/ausocean/av
 go 1.13
 
 require (
-	bitbucket.org/ausocean/iot v1.2.11
+	bitbucket.org/ausocean/iot v1.2.13
 	bitbucket.org/ausocean/utils v1.2.12
 	github.com/Comcast/gots v0.0.0-20190305015453-8d56e473f0f7
 	github.com/go-audio/audio v0.0.0-20181013203223-7b2a6ca21480
@@ -12,4 +12,5 @@ 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/natefinch/lumberjack.v2 v2.0.0
 )

go.sum

@@ -4,6 +4,10 @@ bitbucket.org/ausocean/iot v1.2.10 h1:TTu+ykH5gQA8wU/pN0aS55ySQ/XcGxV4s4LKx3Wye5
 bitbucket.org/ausocean/iot v1.2.10/go.mod h1:Q5FwaOKnCty3dVeVtki6DLwYa5vhNpOaeu1lwLyPCg8=
 bitbucket.org/ausocean/iot v1.2.11 h1:MwYQK1F2ESA5jPVSCB0lBUN8HBiNDHGkh/OMGJKw8Oc=
 bitbucket.org/ausocean/iot v1.2.11/go.mod h1:Q5FwaOKnCty3dVeVtki6DLwYa5vhNpOaeu1lwLyPCg8=
+bitbucket.org/ausocean/iot v1.2.12 h1:Ixf0CTmWOMJVrJ6IYMEluTrCLlu9LM1eNSBZ+ZUnDmU=
+bitbucket.org/ausocean/iot v1.2.12/go.mod h1:Q5FwaOKnCty3dVeVtki6DLwYa5vhNpOaeu1lwLyPCg8=
+bitbucket.org/ausocean/iot v1.2.13 h1:E9LcW3HYqRgJqxNhPJUCfVRvoV2IAU4B7JSDNxB/x2k=
+bitbucket.org/ausocean/iot v1.2.13/go.mod h1:Q5FwaOKnCty3dVeVtki6DLwYa5vhNpOaeu1lwLyPCg8=
 bitbucket.org/ausocean/utils v1.2.11 h1:zA0FOaPjN960ryp8PKCkV5y50uWBYrIxCVnXjwbvPqg=
 bitbucket.org/ausocean/utils v1.2.11/go.mod h1:uXzX9z3PLemyURTMWRhVI8uLhPX4uuvaaO85v2hcob8=
 bitbucket.org/ausocean/utils v1.2.12 h1:VnskjWTDM475TnQRhBQE0cNp9D6Y6OELrd4UkD2VVIQ=

revid/config/config.go

@@ -85,6 +85,7 @@ const (
 	defaultClipDuration    = 0
 	defaultAudioInputCodec = codecutil.ADPCM
 	defaultPSITime         = 2
+	defaultMotionInterval  = 5
 
 	// Ring buffer defaults.
 	defaultRBMaxElements  = 10000
@@ -275,6 +276,7 @@ type Config struct {
 	HorizontalFlip bool  // HorizontalFlip flips video horizontally for Raspivid input.
 	VerticalFlip   bool  // VerticalFlip flips video vertically for Raspivid input.
 	Filters        []int // Defines the methods of filtering to be used in between lexing and encoding.
+	MotionInterval int   // Sets the number of frames that are held before the filter is used (on the nth frame)
 	PSITime        int   // Sets the time between a packet being sent.
 
 	// Ring buffer parameters.
@@ -328,6 +330,7 @@ var TypeData = map[string]string{
 	"MOGHistory":       "uint",
 	"MOGMinArea":       "float",
 	"MOGThreshold":     "float",
+	"MotionInterval":   "int",
 	"RBCapacity":       "uint",
 	"RBMaxElements":    "uint",
 	"RBWriteTimeout":   "uint",
@@ -462,6 +465,10 @@ func (c *Config) Validate() error {
 		c.Logger.Log(logger.Info, pkg+"PSITime bad or unset, defaulting", "PSITime", defaultPSITime)
 		c.PSITime = defaultPSITime
 	}
+	if c.MotionInterval <= 0 {
+		c.Logger.Log(logger.Info, pkg+"MotionInterval bad or unset, defaulting", "MotionInterval", defaultMotionInterval)
+		c.MotionInterval = defaultMotionInterval
+	}
 
 	if c.MinFPS <= 0 {
 		c.Logger.Log(logger.Info, pkg+"MinFPS bad or unset, defaulting", "MinFPS", defaultMinFPS)

revid/revid.go

@@ -337,11 +337,11 @@ func (r *Revid) setupPipeline(mtsEnc func(dst io.WriteCloser, rate float64) (io.
 			case config.FilterNoOp:
 				r.filters[i] = filter.NewNoOp(dst)
 			case config.FilterMOG:
-				r.filters[i] = filter.NewMOGFilter(dst, r.cfg.MOGMinArea, r.cfg.MOGThreshold, int(r.cfg.MOGHistory), r.cfg.ShowWindows)
+				r.filters[i] = filter.NewMOGFilter(dst, r.cfg.MOGMinArea, r.cfg.MOGThreshold, int(r.cfg.MOGHistory), r.cfg.ShowWindows, r.cfg.MotionInterval)
 			case config.FilterVariableFPS:
-				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.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.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)
 			default:
 				panic("Undefined Filter")
 			}
@@ -671,6 +671,13 @@ func (r *Revid) Update(vars map[string]string) error {
 				}
 				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 {
@@ -847,7 +854,13 @@ 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.filters[0], read, delay)
+	err := r.lexTo(r.filters[0], read, delay)
-	r.cfg.Logger.Log(logger.Info, pkg+"finished lexing")
+	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
+	}
 	r.wg.Done()
 }