/* 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 Alan Noble Dan Kortschak Jack Richardson Trek Hopton Scott Barnard Russell Stanley 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 }