av/cmd/rv/main.go

/*
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>
  Russell Stanley <russell@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"
	"os/exec"
	"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/utils/logging"
)

// Current software version.
const version = "v1.4.5"

// 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 = logging.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: "
	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.
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 := logging.New(logVerbosity, io.MultiWriter(fileLog, netLog), logSuppress)

	log.Info("starting rv", "version", version)

	// If rv has been built with the profile tag, then we'll start a CPU profile.
	if canProfile {
		profile(log)
		defer pprof.StopCPUProfile()
		log.Info("profiling started")
	}

	var (
		rv *revid.Revid
		p  *turbidityProbe
	)

	p, err := NewTurbidityProbe(log, 60*time.Second)
	if err != nil {
		log.Fatal("could not create new turbidity probe", "error", err.Error())
	}

	log.Debug("initialising netsender client")
	ns, err := netsender.New(
		log,
		nil,
		readPin(p, rv, log),
		nil,
		netsender.WithVarTypes(createVarMap()),
		netsender.WithUpgrader(upgrade),
	)
	if err != nil {
		log.Fatal(pkg + "could not initialise netsender client: " + err.Error())
	}

	log.Debug("initialising revid")
	rv, err = revid.New(config.Config{Logger: log}, ns)
	if err != nil {
		log.Fatal(pkg+"could not initialise revid", "error", err.Error())
	}

	err = rv.SetProbe(p)
	if err != nil {
		log.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
	// a better way to solve this problem.
	time.Sleep(runPreDelay)

	log.Debug("beginning main loop")
	run(rv, ns, log, netLog, p)
}

// 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 logging.Logger, nl *netlogger.Logger, p *turbidityProbe) {
	var vs int
	for {
		l.Debug("running netsender")
		err := ns.Run()
		if err != nil {
			l.Warning(pkg+"Run Failed. Retrying...", "error", err.Error())
			time.Sleep(netSendRetryTime)
			continue
		}

		l.Debug("sending logs")
		err = nl.Send(ns)
		if err != nil {
			l.Warning(pkg+"Logs could not be sent", "error", err.Error())
		}

		l.Debug("checking varsum")
		newVs := ns.VarSum()
		if vs == newVs {
			sleep(ns, l)
			continue
		}
		vs = newVs
		l.Info("varsum changed", "vs", vs)

		l.Debug("getting new vars")
		vars, err := ns.Vars()
		if err != nil {
			l.Error(pkg+"netSender failed to get vars", "error", err.Error())
			time.Sleep(netSendRetryTime)
			continue
		}
		l.Debug("got new vars", "vars", vars)

		// Configure revid based on the vars.
		l.Debug("updating revid's configuration")
		err = rv.Update(vars)
		if err != nil {
			l.Warning(pkg+"couldn't update revid", "error", err.Error())
			sleep(ns, l)
			continue
		}
		l.Info("revid successfully reconfigured")

		// Update transform matrix based on new revid variables.
		err = p.Update(rv.Config().TransformMatrix)
		if err != nil {
			l.Error("could not update turbidity probe", "error", err.Error())
		}

		l.Debug("checking mode")
		switch ns.Mode() {
		case modePaused:
			l.Debug("mode is Paused, stopping revid")
			rv.Stop()
		case modeNormal, modeLoop:
			l.Debug("mode is Normal or Loop, starting revid")
			err = rv.Start()
			if err != nil {
				l.Error(pkg+"could not start revid", "error", err.Error())
				ns.SetMode(modePaused, &vs)
				sleep(ns, l)
				continue
			}
		case modeBurst:
			l.Debug("mode is Burst, bursting revid")
			err = rv.Burst()
			if err != nil {
				l.Warning(pkg+"could not start burst", "error", err.Error())
				ns.SetMode(modePaused, &vs)
				sleep(ns, l)
				continue
			}
			ns.SetMode(modePaused, &vs)
		}
		l.Info("revid updated with new mode")

		sleep(ns, l)
	}
}

func createVarMap() map[string]string {
	m := make(map[string]string)
	for _, v := range config.Variables {
		m[v.Name] = v.Type
	}
	return m
}

// profile opens a file to hold CPU profiling metrics and then starts the
// CPU profiler.
func profile(l logging.Logger) {
	f, err := os.Create(profilePath)
	if err != nil {
		l.Fatal(pkg+"could not create CPU profile", "error", err.Error())
	}
	if err := pprof.StartCPUProfile(f); err != nil {
		l.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 logging.Logger) {
	l.Debug("sleeping")
	t, err := strconv.Atoi(ns.Param("mp"))
	if err != nil {
		l.Error(pkg+"could not get sleep time, using default", "error", err)
		t = defaultSleepTime
	}
	time.Sleep(time.Duration(t) * time.Second)
	l.Debug("finished sleeping")
}

// readPin provides a callback function of consistent signature for use by
// netsender to retrieve software defined pin values e.g. revid bitrate.
func readPin(p *turbidityProbe, rv *revid.Revid, l logging.Logger) func(pin *netsender.Pin) error {
	return func(pin *netsender.Pin) error {
		switch {
		case pin.Name == bitratePin:
			pin.Value = -1
			if rv != nil {
				pin.Value = rv.Bitrate()
			}
		case pin.Name == sharpnessPin:
			pin.Value = -1
			if p != nil {
				l.Debug("setting sharpness value", "sharpness", p.sharpness*1000)
				pin.Value = int(p.sharpness * 1000)
			}
		case pin.Name == contrastPin:
			pin.Value = -1
			if p != nil {
				l.Debug("setting contrast pin", "contrast", p.contrast)
				pin.Value = int(p.contrast * 100)
			}
		}
		return nil
	}
}

// upgrade is a callback to be provided to the netsender client initialiser through the
// netsender.WithUpgrader() option function. This function is called if the
// netsender client gets a remote upgrade request. The tag is used in this case
// to fetch, checkout and then build.
func upgrade(tag, user, gopath string) error {
	srcDir := gopath + "/src/bitbucket.org/ausocean/av"
	instrs := []struct {
		Cmd, Dir string
		Args     []string
	}{
		{Cmd: "git", Dir: srcDir, Args: []string{"fetch", "--depth=1", "origin", "refs/tags/" + tag + ":refs/tags/" + tag}},
		{Cmd: "git", Dir: srcDir, Args: []string{"checkout", "--force", "tags/" + tag}},
		{Cmd: "make", Dir: srcDir + "/init", Args: []string{"rebuild"}},
	}
	for _, instr := range instrs {
		cmd := exec.Command(instr.Cmd, instr.Args...)
		cmd.Dir = instr.Dir
		out, err := cmd.CombinedOutput()
		if err != nil {
			return fmt.Errorf("could not perform upgrade instruction: %w (%+v), out: %s", err, instr, string(out))
		}
	}
	return nil
}