av/cmd/revid-cli/main.go

/*
NAME
  revid-cli - command line interface for revid.

DESCRIPTION
  See Readme.md

AUTHORS
  Saxon A. Nelson-Milton <saxon@ausocean.org>
  Jack Richardson <jack@ausocean.org>
  Trek Hopton <trek@ausocean.org>

LICENSE
  revid-cli is Copyright (C) 2017-2018 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.
*/

// revid-cli is a command line interface for revid.
package main

import (
	"flag"
	"io"
	"os"
	"path/filepath"
	"runtime/pprof"
	"strconv"
	"strings"
	"time"

	"bitbucket.org/ausocean/av/codec/codecutil"
	"bitbucket.org/ausocean/av/container/mts"
	"bitbucket.org/ausocean/av/container/mts/meta"
	"bitbucket.org/ausocean/av/device/raspivid"
	"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"
	"gopkg.in/natefinch/lumberjack.v2"
)

// Revid modes
const (
	normal = "Normal"
	paused = "Paused"
	burst  = "Burst"
)

// Other misc consts
const (
	netSendRetryTime    = 5 * time.Second
	defaultRunDuration  = 24 * time.Hour
	revidStopTime       = 5 * time.Second
	defaultLogPath      = "/var/log/netsender"
	pkg                 = "revid-cli:"
	defaultLogVerbosity = logger.Info
	defaultSleepTime    = 60 // Seconds
)

// canProfile is set to false with revid-cli is built with "-tags profile".
var canProfile = true

// The logger that will be used throughout.
var (
	netLogger *netlogger.Logger
	log       *logger.Logger
)

const (
	metaPreambleKey  = "copyright"
	metaPreambleData = "ausocean.org/license/content2019"
)

func main() {
	mts.Meta = meta.NewWith([][2]string{{metaPreambleKey, metaPreambleData}})

	useNetsender := flag.Bool("NetSender", false, "Are we checking vars through netsender?")
	runDurationPtr := flag.Duration("runDuration", defaultRunDuration, "How long do you want revid to run for?")

	cfg := handleFlags()
	if !*useNetsender {
		rv, err := revid.New(cfg, nil)
		if err != nil {
			cfg.Logger.Log(logger.Fatal, pkg+"failed to initialiase revid", "error", err.Error())
		}
		if err = rv.Start(); err != nil {
			cfg.Logger.Log(logger.Fatal, pkg+"failed to start revid", "error", err.Error())
		}
		time.Sleep(*runDurationPtr)
		rv.Stop()
		return
	}

	run(cfg)
}

// handleFlags parses command line flags and returns a revid configuration
// based on them.
func handleFlags() config.Config {
	var cfg config.Config

	var (
		cpuprofile          = flag.String("cpuprofile", "", "write cpu profile to `file`")
		inputCodecPtr       = flag.String("InputCodec", "H264", "The codec of the input: H264, Mjpeg, PCM, ADPCM")
		inputPtr            = flag.String("Input", "", "The input type: Raspivid, File, v4l, Audio, RTSP")
		cameraIPPtr         = flag.String("CameraIP", "", "The IP of the RTSP server")
		verbosityPtr        = flag.String("Verbosity", "Info", "Verbosity: Debug, Info, Warning, Error, Fatal")
		rtpAddrPtr          = flag.String("RtpAddr", "", "Rtp destination address: <IP>:<port> (port is generally 6970-6999)")
		logPathPtr          = flag.String("LogPath", defaultLogPath, "The log path")
		configFilePtr       = flag.String("ConfigFile", "", "NetSender config file")
		rtmpUrlPtr          = flag.String("RtmpUrl", "", "Url of rtmp endpoint")
		outputPathPtr       = flag.String("OutputPath", "", "The directory of the output file")
		inputFilePtr        = flag.String("InputPath", "", "The directory of the input file")
		httpAddressPtr      = flag.String("HttpAddress", "", "Destination address of http posts")
		verticalFlipPtr     = flag.Bool("VerticalFlip", false, "Flip video vertically: Yes, No")
		horizontalFlipPtr   = flag.Bool("HorizontalFlip", false, "Flip video horizontally: Yes, No")
		bitratePtr          = flag.Uint("Bitrate", 0, "Bitrate of recorded video")
		heightPtr           = flag.Uint("Height", 0, "Height in pixels")
		widthPtr            = flag.Uint("Width", 0, "Width in pixels")
		frameRatePtr        = flag.Uint("FrameRate", 0, "Frame rate of captured video")
		quantizationPtr     = flag.Uint("Quantization", 0, "Desired quantization value: 0-40")
		rotationPtr         = flag.Uint("Rotation", 0, "Rotate video output. (0-359 degrees)")
		brightnessPtr       = flag.Uint("Brightness", 50, "Set brightness. (0-100) ")
		saturationPtr       = flag.Int("Saturation", 0, "Set Saturation. (100-100)")
		exposurePtr         = flag.String("Exposure", "auto", "Set exposure mode. ("+strings.Join(raspivid.ExposureModes[:], ",")+")")
		autoWhiteBalancePtr = flag.String("Awb", "auto", "Set automatic white balance mode. ("+strings.Join(raspivid.AutoWhiteBalanceModes[:], ",")+")")

		// Audio specific flags.
		sampleRatePtr = flag.Int("SampleRate", 48000, "Sample rate of recorded audio")
		channelsPtr   = flag.Int("Channels", 1, "Record in Mono or Stereo (1 or 2)")
		recPeriodPtr  = flag.Float64("recPeriod", 1, "How many seconds to record at a time")
		bitDepthPtr   = flag.Int("bitDepth", 16, "Bit Depth to record audio at.")
	)

	var outputs flagStrings
	flag.Var(&outputs, "Output", "output type: Http, Rtmp, File, Udp, Rtp (may be used more than once)")

	flag.Parse()

	switch *verbosityPtr {
	case "Debug":
		cfg.LogLevel = logger.Debug
	case "Info":
		cfg.LogLevel = logger.Info
	case "Warning":
		cfg.LogLevel = logger.Warning
	case "Error":
		cfg.LogLevel = logger.Error
	case "Fatal":
		cfg.LogLevel = logger.Fatal
	default:
		cfg.LogLevel = defaultLogVerbosity
	}

	netLogger = netlogger.New()
	log = logger.New(
		cfg.LogLevel,
		io.MultiWriter(
			&lumberjack.Logger{
				Filename:   filepath.Join(*logPathPtr, "netsender.log"),
				MaxSize:    500, // MB
				MaxBackups: 10,
				MaxAge:     28, // days
			},
			netLogger,
		),
		true,
	)

	cfg.Logger = log

	if *cpuprofile != "" {
		if canProfile {
			f, err := os.Create(*cpuprofile)
			if err != nil {
				log.Log(logger.Fatal, pkg+"could not create CPU profile", "error", err.Error())
			}
			if err := pprof.StartCPUProfile(f); err != nil {
				log.Log(logger.Fatal, pkg+"could not start CPU profile", "error", err.Error())
			}
			defer pprof.StopCPUProfile()
		} else {
			log.Log(logger.Warning, pkg+"ignoring cpuprofile flag - http/pprof built in.")
		}
	}

	switch *inputPtr {
	case "Raspivid":
		cfg.Input = config.InputRaspivid
	case "v4l":
		cfg.Input = config.InputV4L
	case "File":
		cfg.Input = config.InputFile
	case "Audio":
		cfg.Input = config.InputAudio
	case "RTSP":
		cfg.Input = config.InputRTSP
	case "":
	default:
		log.Log(logger.Error, pkg+"bad input argument")
	}

	switch *inputCodecPtr {
	case "H264":
		cfg.InputCodec = codecutil.H264
	case "PCM":
		cfg.InputCodec = codecutil.PCM
	case "ADPCM":
		cfg.InputCodec = codecutil.ADPCM
	case "MJPEG":
		cfg.InputCodec = codecutil.MJPEG
	default:
		log.Log(logger.Error, pkg+"bad input codec argument")
	}

	switch *inputPtr {
	case "Audio":
		cfg.WriteRate = 1.0 / (*recPeriodPtr)
	default:
		cfg.WriteRate = float64(*frameRatePtr)
	}

	for _, o := range outputs {
		switch o {
		case "File":
			cfg.Outputs = append(cfg.Outputs, config.OutputFile)
		case "Http":
			cfg.Outputs = append(cfg.Outputs, config.OutputHTTP)
		case "Rtmp":
			cfg.Outputs = append(cfg.Outputs, config.OutputRTMP)
		case "Rtp":
			cfg.Outputs = append(cfg.Outputs, config.OutputRTP)
		case "":
		default:
			log.Log(logger.Error, pkg+"bad output argument", "arg", o)
		}
	}

	if *configFilePtr != "" {
		netsender.ConfigFile = *configFilePtr
	}

	cfg.CameraIP = *cameraIPPtr
	cfg.Rotation = *rotationPtr
	cfg.HorizontalFlip = *horizontalFlipPtr
	cfg.VerticalFlip = *verticalFlipPtr
	cfg.RTMPURL = *rtmpUrlPtr
	cfg.Bitrate = *bitratePtr
	cfg.OutputPath = *outputPathPtr
	cfg.InputPath = *inputFilePtr
	cfg.Height = *heightPtr
	cfg.Width = *widthPtr
	cfg.FrameRate = *frameRatePtr
	cfg.HTTPAddress = *httpAddressPtr
	cfg.Quantization = *quantizationPtr
	cfg.RTPAddress = *rtpAddrPtr
	cfg.Brightness = *brightnessPtr
	cfg.Saturation = *saturationPtr
	cfg.Exposure = *exposurePtr
	cfg.AutoWhiteBalance = *autoWhiteBalancePtr
	cfg.SampleRate = *sampleRatePtr
	cfg.Channels = *channelsPtr
	cfg.RecPeriod = *recPeriodPtr
	cfg.BitDepth = *bitDepthPtr

	return cfg
}

// initialize then run the main NetSender client
func run(cfg config.Config) {
	log.Log(logger.Info, pkg+"running in NetSender mode")

	var rv *revid.Revid

	readPin := 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 // Return error only if we want NetSender to generate an error
	}

	ns, err := netsender.New(log, nil, readPin, nil, config.TypeData)
	if err != nil {
		log.Log(logger.Fatal, pkg+"could not initialise netsender client: "+err.Error())
	}

	var vs int
	for {
		err = ns.Run()
		if err != nil {
			log.Log(logger.Warning, pkg+"Run Failed. Retrying...", "error", err.Error())
			time.Sleep(netSendRetryTime)
			continue
		}

		err = netLogger.Send(ns)
		if err != nil {
			log.Log(logger.Warning, pkg+"Logs could not be sent", "error", err.Error())
		}

		// If var sum hasn't changed we continue.
		var vars map[string]string
		newVs := ns.VarSum()
		if vs == newVs {
			goto sleep
		}
		vs = newVs

		vars, err = ns.Vars()
		if err != nil {
			log.Log(logger.Error, pkg+"netSender failed to get vars", "error", err.Error())
			time.Sleep(netSendRetryTime)
			continue
		}

		if rv == nil {
			rv, err = revid.New(cfg, ns)
			if err != nil {
				log.Log(logger.Warning, pkg+"could not initialise revid", "error", err.Error())
				goto sleep
			}
		}

		err = rv.Update(vars)
		if err != nil {
			log.Log(logger.Warning, pkg+"Couldn't update revid", "error", err.Error())
			goto sleep
		}

		switch ns.Mode() {
		case paused:
			rv.Stop()
		case normal:
			err = rv.Start()
			if err != nil {
				log.Log(logger.Warning, pkg+"could not start revid", "error", err.Error())
				ns.SetMode(paused, &vs)
				goto sleep
			}
		case burst:
			log.Log(logger.Info, pkg+"Starting burst...")

			err = rv.Start()
			if err != nil {
				log.Log(logger.Warning, pkg+"could not start burst", "error", err.Error())
				ns.SetMode(paused, &vs)
				goto sleep
			}

			time.Sleep(time.Duration(rv.Config().BurstPeriod) * time.Second)
			log.Log(logger.Info, pkg+"Stopping burst...")
			rv.Stop()
			ns.SetMode(paused, &vs)
		}

	sleep:
		sleepTime, err := strconv.Atoi(ns.Param("mp"))
		if err != nil {
			log.Log(logger.Error, pkg+"could not get sleep time, using default")
			sleepTime = defaultSleepTime
		}
		time.Sleep(time.Duration(sleepTime) * time.Second)
	}
}

// flagStrings implements an appending string set flag.
type flagStrings []string

func (v *flagStrings) String() string {
	if *v != nil {
		return strings.Join(*v, ",")
	}
	return ""
}

func (v *flagStrings) Set(s string) error {
	if s == "" {
		return nil
	}
	for _, e := range *v {
		if e == s {
			return nil
		}
	}
	*v = append(*v, s)
	return nil
}

func (v *flagStrings) Get() interface{} { return *v }