diff --git a/Makefile b/Makefile index 47877dd1..14202f3c 100644 --- a/Makefile +++ b/Makefile @@ -1,7 +1,7 @@ -all: revid-cli audio-netsender +all: rv audio-netsender -revid-cli: - cd cmd/revid-cli; go build +rv: + cd cmd/rv; go build audio-netsender: - cd cmd/audio-netsender; go build \ No newline at end of file + cd cmd/audio-netsender; go build diff --git a/cmd/rv/main.go b/cmd/rv/main.go new file mode 100644 index 00000000..3193f92f --- /dev/null +++ b/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 + Alan Noble + Dan Kortschak + Jack Richardson + Trek Hopton + Scott Barnard + +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 +} diff --git a/cmd/rv/profile.go b/cmd/rv/profile.go new file mode 100644 index 00000000..8595b725 --- /dev/null +++ b/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 + +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 +} diff --git a/cmd/rv/run.sh b/cmd/rv/run.sh new file mode 100644 index 00000000..3521dc0c --- /dev/null +++ b/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 & diff --git a/cmd/rv/upgrade.sh b/cmd/rv/upgrade.sh new file mode 100644 index 00000000..46cfac2a --- /dev/null +++ b/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 $? diff --git a/codec/codecutil/list.go b/codec/codecutil/list.go index c270eb6b..1d4d1510 100644 --- a/codec/codecutil/list.go +++ b/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 diff --git a/device/webcam/webcam.go b/device/webcam/webcam.go index fbb3a91b..2d5e27e3 100644 --- a/device/webcam/webcam.go +++ b/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 } diff --git a/revid/revid.go b/revid/revid.go index 5eb5b7bf..65f0a295 100644 --- a/revid/revid.go +++ b/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 {