Merge branch 'master' into alsa-pool

cmd/rv/main.go

@@ -11,6 +11,7 @@ AUTHORS
   Jack Richardson <jack@ausocean.org>
   Trek Hopton <trek@ausocean.org>
   Scott Barnard <scott@ausocean.org>
+  Russell Stanley <russell@ausocean.org>
 
 LICENSE
   Copyright (C) 2020 the Australian Ocean Lab (AusOcean)
@@ -103,7 +104,14 @@ const (
 	profilePath      = "rv.prof"
 	pkg              = "rv: "
 	runPreDelay      = 20 * time.Second
+)
+
+// Software define pin values.
+// See https://netreceiver.appspot.com/help, External Pin Assignments.
+const (
 	bitratePin   = "X36"
+	sharpnessPin = "X38"
+	contrastPin  = "X39"
 )
 
 // This is set to true if the 'profile' build tag is provided on build.
@@ -134,10 +142,14 @@ func main() {
 		log.Log(logger.Info, "profiling started")
 	}
 
-	var rv *revid.Revid
+	var (
+		rv *revid.Revid
+		p  *turbidityProbe
+	)
+	p, err := NewTurbidityProbe(*log, 60*time.Second)
 
 	log.Log(logger.Debug, "initialising netsender client")
-	ns, err := netsender.New(log, nil, readPin(rv), nil, createVarMap())
+	ns, err := netsender.New(log, nil, readPin(p, rv), nil, createVarMap())
 	if err != nil {
 		log.Log(logger.Fatal, pkg+"could not initialise netsender client: "+err.Error())
 	}
@@ -148,6 +160,11 @@ func main() {
 		log.Log(logger.Fatal, pkg+"could not initialise revid", "error", err.Error())
 	}
 
+	err = rv.SetProbe(p)
+	if err != nil {
+		log.Log(logger.Error, pkg+"could not set probe", "error", err.Error())
+	}
+
 	// NB: Problems were encountered with communicating with RTSP inputs. When trying to
 	// connect it would fail due to timeout; as if things had not been set up quickly
 	// enough before revid tried to do things. This delay fixes this, but there is probably
@@ -271,7 +288,7 @@ func sleep(ns *netsender.Sender, l *logger.Logger) {
 
 // readPin provides a callback function of consistent signature for use by
 // netsender to retrieve software defined pin values e.g. revid bitrate.
-func readPin(rv *revid.Revid) func(pin *netsender.Pin) error {
+func readPin(p *turbidityProbe, rv *revid.Revid) func(pin *netsender.Pin) error {
 	return func(pin *netsender.Pin) error {
 		switch {
 		case pin.Name == bitratePin:
@@ -281,6 +298,16 @@ func readPin(rv *revid.Revid) func(pin *netsender.Pin) error {
 			}
 		case pin.Name[0] == 'X':
 			return sds.ReadSystem(pin)
+		case pin.Name == sharpnessPin:
+			pin.Value = -1
+			if p != nil {
+				pin.Value = int(p.sharpness * 1000)
+			}
+		case pin.Name == contrastPin:
+			pin.Value = -1
+			if p != nil {
+				pin.Value = int(p.contrast * 100)
+			}
 		default:
 			pin.Value = -1
 		}

cmd/rv/probe.go

@@ -0,0 +1,128 @@
+//go:build !nocv
+// +build !nocv
+
+/*
+DESCRIPTION
+  Provides the methods for the turbidity probe using GoCV. Turbidity probe
+  will collect the most recent frames in a buffer and write the latest sharpness
+  and contrast scores to the probe.
+
+AUTHORS
+  Russell Stanley <russell@ausocean.org>
+
+LICENSE
+  Copyright (C) 2021-2022 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 main
+
+import (
+	"os"
+	"time"
+
+	"gocv.io/x/gocv"
+	"gonum.org/v1/gonum/stat"
+
+	"bitbucket.org/ausocean/av/turbidity"
+	"bitbucket.org/ausocean/utils/logger"
+)
+
+// Turbidity sensor constants.
+const (
+	k1, k2     = 8, 8 // Block size, must be divisible by the size template with no remainder.
+	filterSize = 3    // Sobel filter size.
+	scale      = 1.0  // Amount of scale applied to sobel filter values.
+	alpha      = 1.0  // Paramater for contrast equation.
+)
+
+// Misc constants.
+const (
+	maxImages = 10 // Max number of images read when evaluating turbidity.
+)
+
+type turbidityProbe struct {
+	sharpness, contrast float64
+	delay               time.Duration
+	ticker              time.Ticker
+	ts                  *turbidity.TurbiditySensor
+	log                 logger.Logger
+}
+
+// NewTurbidityProbe returns a new turbidity probe.
+func NewTurbidityProbe(log logger.Logger, delay time.Duration) (*turbidityProbe, error) {
+	tp := new(turbidityProbe)
+	tp.log = log
+	tp.delay = delay
+	tp.ticker = *time.NewTicker(delay)
+
+	// Create the turbidity sensor.
+	standard := gocv.IMRead("../../turbidity/images/template.jpg", gocv.IMReadGrayScale)
+	template := gocv.IMRead("../../turbidity/images/template.jpg", gocv.IMReadGrayScale)
+	ts, err := turbidity.NewTurbiditySensor(template, standard, k1, k2, filterSize, scale, alpha)
+	if err != nil {
+		log.Error("failed create turbidity sensor", "error", err.Error())
+	}
+	tp.ts = ts
+	return tp, nil
+}
+
+// Write, reads input h264 frames in the form of a byte stream and writes the the sharpness and contrast
+// scores of a video to the the turbidity probe.
+func (tp *turbidityProbe) Write(p []byte) (int, error) {
+	select {
+	case <-tp.ticker.C:
+		var imgs []gocv.Mat
+		img := gocv.NewMat()
+
+		// Write byte array to a temp file.
+		file, err := os.CreateTemp("temp", "video*.h264")
+		if err != nil {
+			tp.log.Error("failed to create temp file", "error", err.Error())
+			return 0, err
+		}
+		defer os.Remove(file.Name())
+		n, err := file.Write(p)
+		if err != nil {
+			tp.log.Error("failed to write to temporary file", "error", err.Error())
+			return n, err
+		}
+
+		// Read the file and store each frame.
+		vc, err := gocv.VideoCaptureFile(file.Name())
+		if err != nil {
+			tp.log.Error("failed to open video file", "error", err.Error())
+			return len(p), err
+		}
+		for vc.Read(&img) && len(imgs) < maxImages {
+			imgs = append(imgs, img.Clone())
+		}
+
+		// Process video data to get saturation and contrast scores.
+		res, err := tp.ts.Evaluate(imgs)
+		if err != nil {
+			tp.log.Error("evaluate failed", "errror", err.Error())
+			return len(p), err
+		}
+		tp.contrast = stat.Mean(res.Contrast, nil)
+		tp.sharpness = stat.Mean(res.Sharpness, nil)
+	default:
+	}
+	return len(p), nil
+}
+
+func (tp *turbidityProbe) Close() error {
+	return nil
+}

cmd/rv/probe_circleci.go

@@ -0,0 +1,55 @@
+//go:build nocv
+// +build nocv
+
+/*
+DESCRIPTION
+  Replaces turbidity probe implementation that uses the gocv package.
+  When Circle-CI builds revid this is needed because Circle-CI does not
+  have a copy of Open CV installed.
+
+AUTHORS
+  Russell Stanley <russell@ausocean.org>
+
+LICENSE
+  Copyright (C) 2021-2022 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 main
+
+import (
+	"time"
+
+	"bitbucket.org/ausocean/utils/logger"
+)
+
+type turbidityProbe struct {
+	sharpness, contrast float64
+}
+
+// NewTurbidityProbe returns an empty turbidity probe for CircleCI testing only.
+func NewTurbidityProbe(log logger.Logger, delay time.Duration) (*turbidityProbe, error) {
+	tp := new(turbidityProbe)
+	return tp, nil
+}
+
+// Write performs no operation for CircleCI testing only.
+func (tp *turbidityProbe) Write(p []byte) (int, error) {
+	return 0, nil
+}
+
+func (tp *turbidityProbe) Close() error {
+	return nil
+}

cmd/rv/probe_test.go

@@ -0,0 +1,62 @@
+/*
+DESCRIPTION
+  Testing function for turbidity probe.
+
+AUTHORS
+  Russell Stanley <russell@ausocean.org>
+
+LICENSE
+  Copyright (C) 2021-2022 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 main
+
+import (
+	"io"
+	"io/ioutil"
+	"testing"
+	"time"
+
+	"bitbucket.org/ausocean/utils/logger"
+	"gopkg.in/natefinch/lumberjack.v2"
+)
+
+// TestProbe reads a given video file and writes the sharpness and contrast scores to a turbidity probe.
+func TestProbe(t *testing.T) {
+	// Create lumberjack logger.
+	fileLog := &lumberjack.Logger{
+		Filename:   logPath,
+		MaxSize:    logMaxSize,
+		MaxBackups: logMaxBackup,
+		MaxAge:     logMaxAge,
+	}
+	log := logger.New(logVerbosity, io.MultiWriter(fileLog), logSuppress)
+
+	ts, err := NewTurbidityProbe(*log, time.Microsecond)
+	if err != nil {
+		t.Fatalf("failed to create turbidity probe")
+	}
+	video, err := ioutil.ReadFile("logo.h264")
+	if err != nil {
+		t.Fatalf("failed to read file: %v", err)
+	}
+
+	_, err = ts.Write(video)
+	if err != nil {
+		t.Fatalf("failed to write sharpness and contrast: %v", err)
+	}
+	t.Logf("contrast: %v, sharpness: %v\n", ts.contrast, ts.sharpness)
+}

codec/h264/h264dec/fuzz/fuzzParseLevelPrefix/fuzz.go

@@ -1,3 +1,5 @@
+// +build gofuzz
+
 /*
 DESCRIPTION
   fuzz.go provides a function with the required signature such that it may be
@@ -24,8 +26,6 @@ LICENSE
   in gpl.txt.  If not, see http://www.gnu.org/licenses.
 */
 
-// +build gofuzz
-
 package fuzzParseLevelPrefix
 
 /*

revid/audio_darwin.go

@@ -0,0 +1,33 @@
+/*
+NAME
+  audio_darwin.go
+
+AUTHORS
+  Russell Stanley <russell@ausocean.org>
+
+LICENSE
+  revid is Copyright (C) 2021 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 revid
+
+import (
+	"errors"
+)
+
+func (r *Revid) setupAudio() error {
+	return errors.New("audio not implemented on darwin(macOS)")
+}

revid/pipeline.go

@@ -283,6 +283,7 @@ func (r *Revid) setupPipeline(mtsEnc func(dst io.WriteCloser, rate float64) (io.
 		r.cfg.Logger.Log(logger.Info, "filters set up")
 	}
 
+
 	var err error
 	switch r.cfg.Input {
 	case config.InputRaspivid:
@@ -377,7 +378,13 @@ func (r *Revid) processFrom(in device.AVDevice, delay time.Duration) {
 	// 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.Debug, "lexing")
-	err = r.lexTo(r.filters[0], in, delay)
+	var w io.Writer
+	w = r.filters[0]
+	if r.probe != nil {
+		w = ioext.MultiWriteCloser(r.filters[0],r.probe)
+	}
+
+	err = r.lexTo(w, in, delay)
 	switch err {
 	case nil, io.EOF:
 		r.cfg.Logger.Log(logger.Info, "end of file")

revid/revid.go

@@ -30,6 +30,7 @@ LICENSE
 package revid
 
 import (
+	"errors"
 	"fmt"
 	"io"
 	"sync"
@@ -76,6 +77,12 @@ type Revid struct {
 	// lexTo, encoder and packer handle transcoding the input stream.
 	lexTo func(dest io.Writer, src io.Reader, delay time.Duration) error
 
+	// probe allows us to "probe" frames after being lexed before going off to
+	// later encoding stages. This is useful if we wish to perform some processing
+	// on frames to derive metrics, for example, we might like to probe frames to
+	// derive turbidity levels. This is provided through SetProbe.
+	probe io.WriteCloser
+
 	// filters will hold the filter interface that will write to the chosen filter from the lexer.
 	filters []filter.Filter
 
@@ -236,3 +243,11 @@ func (r *Revid) Update(vars map[string]string) error {
 	r.cfg.Logger.Log(logger.Debug, "config changed", "config", r.cfg)
 	return nil
 }
+
+func (r *Revid) SetProbe(p io.WriteCloser) error {
+	if r.running {
+		return errors.New("cannot set probe when revid is running")
+	}
+	r.probe = p
+	return nil
+}

turbidity/results.go

@@ -1,6 +1,3 @@
-//go:build !nocv
-// +build !nocv
-
 /*
 DESCRIPTION
   Results struct used to store results from the turbidity sensor.