Merged in rv (pull request #355)

cmd/rv: created revid client that only offers control via netsender

Approved-by: Scott Barnard <scott@ausocean.org>
Approved-by: Trek Hopton <trek.hopton@gmail.com>

Makefile

@@ -1,7 +1,7 @@
-all:	revid-cli audio-netsender
+all:	rv audio-netsender
 
-revid-cli:
+rv:
-	cd cmd/revid-cli; go build
+	cd cmd/rv; go build
 
 audio-netsender:
 	cd cmd/audio-netsender; go build

cmd/rv/main.go

@@ -0,0 +1,274 @@
+/*
+DESCRIPTION
+  rv is a netsender client using the revid package to perform media collection
+  and forwarding whose behaviour is controllable via the cloud interfaces
+  netreceiver and vidgrind.
+
+AUTHORS
+  Saxon A. Nelson-Milton <saxon@ausocean.org>
+  Alan Noble <alan@ausocean.org>
+  Dan Kortschak <dan@ausocean.org>
+  Jack Richardson <jack@ausocean.org>
+  Trek Hopton <trek@ausocean.org>
+  Scott Barnard <scott@ausocean.org>
+
+LICENSE
+  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
+  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
+  along with revid in gpl.txt. If not, see http://www.gnu.org/licenses.
+
+USAGE
+  There must firstly be a netsender configuration file under /etc/netsender.conf.
+  Example:
+
+  ma 00:00:00:00:00:01
+  dk 0
+  wi
+  ip V0, T0
+  op
+  mp 60
+  ap 0
+  tg
+  hw
+  sh vidgrind.appspot.com
+
+  Revid configuration is controlled by valid variables given values on netreceiver
+  or vidgrind interface. See revid/config for valid variables.
+
+  To run rv simply build and call:
+  ./rv
+*/
+
+// Package rv is a netsender client for revid.
+package main
+
+import (
+	"fmt"
+	"io"
+	"os"
+	"runtime/pprof"
+	"strconv"
+	"time"
+
+	"gopkg.in/natefinch/lumberjack.v2"
+
+	"bitbucket.org/ausocean/av/container/mts"
+	"bitbucket.org/ausocean/av/container/mts/meta"
+	"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/utils/logger"
+)
+
+// Copyright information prefixed to all metadata.
+const (
+	metaPreambleKey  = "copyright"
+	metaPreambleData = "ausocean.org/license/content2019"
+)
+
+// Logging configuration.
+const (
+	logPath      = "/var/log/netsender/netsender.log"
+	logMaxSize   = 500 // MB
+	logMaxBackup = 10
+	logMaxAge    = 28 // days
+	logVerbosity = logger.Info
+	logSuppress  = true
+)
+
+// Revid modes.
+const (
+	modeNormal = "Normal"
+	modePaused = "Paused"
+	modeBurst  = "Burst"
+	modeLoop   = "Loop"
+)
+
+// Misc constants.
+const (
+	netSendRetryTime = 5 * time.Second
+	defaultSleepTime = 60 // Seconds
+	profilePath      = "rv.prof"
+	pkg              = "rv: "
+)
+
+// This is set to true if the 'profile' build tag is provided on build.
+var canProfile = false
+
+func main() {
+	mts.Meta = meta.NewWith([][2]string{{metaPreambleKey, metaPreambleData}})
+
+	// Create lumberjack logger to handle logging to file.
+	fileLog := &lumberjack.Logger{
+		Filename:   logPath,
+		MaxSize:    logMaxSize,
+		MaxBackups: logMaxBackup,
+		MaxAge:     logMaxAge,
+	}
+
+	// Create netlogger to handle logging to cloud.
+	netLog := netlogger.New()
+
+	// Create logger that we call methods on to log, which in turn writes to the
+	// lumberjack and netloggers.
+	log := logger.New(logVerbosity, io.MultiWriter(fileLog, netLog), logSuppress)
+
+	// If rv has been built with the profile tag, then we'll start a CPU profile.
+	if canProfile {
+		profile(log)
+		defer pprof.StopCPUProfile()
+	}
+
+	var rv *revid.Revid
+
+	ns, err := netsender.New(log, nil, readPin(rv), nil, config.TypeData)
+	if err != nil {
+		log.Log(logger.Fatal, pkg+"could not initialise netsender client: "+err.Error())
+	}
+
+	rv, err = revid.New(config.Config{Logger: log}, ns)
+	if err != nil {
+		log.Log(logger.Fatal, pkg+"could not initialise revid", "error", err.Error())
+	}
+
+	run(rv, ns, log, netLog)
+}
+
+// run starts the main loop. This will run netsender on every pass of the loop
+// (sleeping inbetween), check vars, and if changed, update revid as appropriate.
+func run(rv *revid.Revid, ns *netsender.Sender, l *logger.Logger, nl *netlogger.Logger) {
+	var vs int
+	for {
+		err := ns.Run()
+		if err != nil {
+			l.Log(logger.Warning, pkg+"Run Failed. Retrying...", "error", err.Error())
+			time.Sleep(netSendRetryTime)
+			continue
+		}
+
+		err = nl.Send(ns)
+		if err != nil {
+			l.Log(logger.Warning, pkg+"Logs could not be sent", "error", err.Error())
+		}
+
+		// If var sum hasn't changed we skip rest of loop.
+		newVs := ns.VarSum()
+		if vs == newVs {
+			sleep(ns, l)
+			continue
+		}
+		vs = newVs
+
+		vars, err := ns.Vars()
+		if err != nil {
+			l.Log(logger.Error, pkg+"netSender failed to get vars", "error", err.Error())
+			time.Sleep(netSendRetryTime)
+			continue
+		}
+
+		// Configure revid based on the vars.
+		err = rv.Update(vars)
+		if err != nil {
+			l.Log(logger.Warning, pkg+"Couldn't update revid", "error", err.Error())
+			sleep(ns, l)
+			continue
+		}
+
+		switch ns.Mode() {
+		case modePaused:
+			rv.Stop()
+		case modeNormal, modeLoop:
+			err = rv.Start()
+			if err != nil {
+				l.Log(logger.Warning, pkg+"could not start revid", "error", err.Error())
+				ns.SetMode(modePaused, &vs)
+				sleep(ns, l)
+				continue
+			}
+		case modeBurst:
+			err = burst(l, rv, ns)
+			if err != nil {
+				l.Log(logger.Warning, pkg+"could not start burst", "error", err.Error())
+				ns.SetMode(modePaused, &vs)
+				sleep(ns, l)
+				continue
+			}
+			ns.SetMode(modePaused, &vs)
+		}
+
+		sleep(ns, l)
+	}
+}
+
+// profile opens a file to hold CPU profiling metrics and then starts the
+// CPU profiler.
+func profile(l *logger.Logger) {
+	f, err := os.Create(profilePath)
+	if err != nil {
+		l.Log(logger.Fatal, pkg+"could not create CPU profile", "error", err.Error())
+	}
+	if err := pprof.StartCPUProfile(f); err != nil {
+		l.Log(logger.Fatal, pkg+"could not start CPU profile", "error", err.Error())
+	}
+}
+
+// sleep uses a delay to halt the program based on the monitoring period
+// netsender parameter (mp) defined in the netsender.conf config.
+func sleep(ns *netsender.Sender, l *logger.Logger) {
+	t, err := strconv.Atoi(ns.Param("mp"))
+	if err != nil {
+		l.Log(logger.Error, pkg+"could not get sleep time, using default", "error", err)
+		t = defaultSleepTime
+	}
+	time.Sleep(time.Duration(t) * time.Second)
+}
+
+// readPin provides a callback function of consistent signature for use by
+// netsender to retrieve software defined pin values e.g. revid bitrate (X23).
+func readPin(rv *revid.Revid) func(pin *netsender.Pin) error {
+	return func(pin *netsender.Pin) error {
+		switch {
+		case pin.Name == "X23":
+			pin.Value = -1
+			if rv != nil {
+				pin.Value = rv.Bitrate()
+			}
+		case pin.Name[0] == 'X':
+			return sds.ReadSystem(pin)
+		default:
+			pin.Value = -1
+		}
+		return nil
+	}
+}
+
+// burst starts revid, waits for time specified in the Config.BurstPeriod
+// field, and then stops revid.
+//
+// TODO: move this functionality to the revid API into a Revid.Burst(time) method.
+func burst(l *logger.Logger, r *revid.Revid, s *netsender.Sender) error {
+	l.Log(logger.Info, pkg+"starting burst")
+
+	err := r.Start()
+	if err != nil {
+		return fmt.Errorf("could not start revid: %w", err)
+	}
+
+	time.Sleep(time.Duration(r.Config().BurstPeriod) * time.Second)
+	l.Log(logger.Info, pkg+"stopping burst")
+	r.Stop()
+	return nil
+}

cmd/rv/profile.go

@@ -0,0 +1,34 @@
+// +build profile
+
+/*
+DESCRIPTION
+  profile.go provides an init to change canProfile flag to true if profile tag
+  provided on build.
+
+AUTHORS
+  Dan Kortschak <dan@ausocean.org>
+
+LICENSE
+  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
+  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
+  along with revid in gpl.txt. If not, see http://www.gnu.org/licenses.
+*/
+
+package main
+
+import _ "net/http/pprof"
+
+func init() {
+	canProfile = true
+}

cmd/rv/run.sh

@@ -0,0 +1,4 @@
+#!/bin/sh
+REVIDPATH=$HOME/go/src/bitbucket.org/ausocean/av/cmd/rv
+cd $REVIDPATH
+sudo "PATH=$PATH:$REVIDPATH" ./rv &

cmd/rv/upgrade.sh

@@ -0,0 +1,56 @@
+#!/bin/bash
+# All-purpose upgrade script.
+# Upgrades source(s) to given Git tag, runs make in each directory,
+# and write tags to tags.conf upon success, exiting 0.
+# NB: Customize SrcDirs as needed to reflect dependencies.
+Usage="Usage: upgrade.sh [-d] tag"
+BaseDir=$GOPATH/src/bitbucket.org/ausocean
+VarDir=/var/netsender
+LogFile=/var/log/netsender/stream.log
+SrcDirs=($BaseDir/utils $BaseDir/iot $BaseDir/av)
+if [ "$1" == "-d" ]; then
+    set  -x
+    GitFlags=""
+    NewTag="$2"
+else
+    # capture stdout and stderr
+    exec 2> $LogFile
+    exec 1>&2
+    GitFlags="--quiet"
+    NewTag="$1"
+fi
+if [ -z "$GOPATH" ]; then
+    echo "Error: GOPATH not defined"
+    exit 1
+fi
+if [ -z "$NewTag" ]; then
+    echo "$Usage"
+    exit 1
+fi
+for dir in ${SrcDirs[@]}; do
+  pushd $dir
+  if [ ! "$?" == 0 ]; then
+    exit 1
+  fi
+  git fetch $GitFlags --depth=1 origin refs/tags/$NewTag:refs/tags/$NewTag
+  if [ ! "$?" == 0 ]; then
+    exit 1
+  fi
+  git checkout $GitFlags --force tags/$NewTag
+  if [ ! "$?" == 0 ]; then
+    exit 1
+  fi
+  if [ -e Makefile ]; then
+    make
+    if [ ! "$?" == 0 ]; then
+      exit 1
+    fi
+  fi
+  popd
+done
+if [ ! -d "$VarDir" ]; then
+  echo "Error: $VarDir does not exit."	
+  exit 1
+fi
+git tag > "$VarDir/tags.conf"
+exit $?

codec/codecutil/list.go

@@ -30,7 +30,8 @@ const numCodecs = 5
 // A global list containing all available codecs for reference in any application.
 // When adding or removing a codec from this list, the numCodecs const must be updated.
 const (
-	PCM = iota
+	Undef = iota
+	PCM
 	ADPCM
 	H264
 	H265

device/webcam/webcam.go

@@ -182,10 +182,12 @@ func (w *Webcam) Start() error {
 		}
 	}()
 
+	w.cfg.Logger.Log(logger.Info, "starting webcam")
 	err = w.cmd.Start()
 	if err != nil {
 		return fmt.Errorf("failed to start ffmpeg: %w", err)
 	}
+	w.cfg.Logger.Log(logger.Info, "webcam started")
 
 	return nil
 }

revid/revid.go

@@ -194,7 +194,7 @@ func (r *Revid) reset(c config.Config) error {
 					encOptions = append(encOptions, mts.TimeBasedPSI(time.Duration(r.cfg.PSITime)*time.Second))
 					r.cfg.CBR = true
 				default:
-					panic("unknown input codec for v4l or input file input")
+					panic(fmt.Sprintf("unknown input codec %d for v4l or input file input", r.cfg.InputCodec))
 				}
 			case config.InputRTSP:
 				switch r.cfg.InputCodec {