From 513eee1260011ed27e3082de19fee9b81371515a Mon Sep 17 00:00:00 2001 From: Ella Pietraroia Date: Fri, 31 Jan 2020 10:26:25 +1030 Subject: [PATCH] finding errors --- cmd/revid-cli/main.go | 2 +- cmd/rv/main.go | 274 ++++++++++++++++++++++++++++++++++++++++++ codec/mjpeg/lex.go | 1 - filter/basic.go | 1 - 4 files changed, 275 insertions(+), 3 deletions(-) create mode 100644 cmd/rv/main.go diff --git a/cmd/revid-cli/main.go b/cmd/revid-cli/main.go index bced0a4f..1b973d3a 100644 --- a/cmd/revid-cli/main.go +++ b/cmd/revid-cli/main.go @@ -331,7 +331,7 @@ func run(cfg config.Config) { case normal: err = rv.Start() if err != nil { - log.Log(logger.Warning, pkg+"could not start revid", "error", err.Error()) + log.Log(logger.Error, pkg+"could not start revid", "error", err.Error()) ns.SetMode(paused, &vs) goto sleep } diff --git a/cmd/rv/main.go b/cmd/rv/main.go new file mode 100644 index 00000000..fb8dd4b1 --- /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.Error, 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/codec/mjpeg/lex.go b/codec/mjpeg/lex.go index f1166022..46ab4ce8 100644 --- a/codec/mjpeg/lex.go +++ b/codec/mjpeg/lex.go @@ -86,7 +86,6 @@ func Lex(dst io.Writer, src io.Reader, delay time.Duration) error { } last = b } - fmt.Print("lex: just before write") <-tick _, err = dst.Write(buf) if err != nil { diff --git a/filter/basic.go b/filter/basic.go index 47e64e1e..8f984f1b 100644 --- a/filter/basic.go +++ b/filter/basic.go @@ -139,7 +139,6 @@ func (b *BasicFilter) Process(j int, wg *sync.WaitGroup) { // Write applies the motion filter to the video stream. Only frames with motion // are written to the destination encoder, frames without are discarded. func (b *BasicFilter) Write(f []byte) (int, error) { - fmt.Print("in basic write\n") t0 := time.Now() //decode MJPEG r := bytes.NewReader(f)