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>

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.
*/

package main

import (
	"flag"
	"os"
	"runtime/pprof"
	"strconv"
	"time"

	"bitbucket.org/ausocean/av/revid"
	"bitbucket.org/ausocean/iot/pi/netsender"
	"bitbucket.org/ausocean/utils/smartlogger"
)

const (
	// progName is the program name for logging purposes.
	progName = "revid-cli"

	// Logging is set to INFO level.
	defaultLogVerbosity = smartlogger.Info
)

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

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

// The logger that will be used throughout
var logger *smartlogger.Logger

// Globals
var rv *revid.Revid

func main() {
	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 {
		// run revid for the specified duration
		startRevid(nil, cfg)
		time.Sleep(*runDurationPtr)
		stopRevid()
		return
	}

	err := run(cfg)
	if err != nil {
		logger.Log(smartlogger.Fatal, pkg+"failed to run revid", "error", err.Error())
		os.Exit(1)
	}
}

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

	var (
		cpuprofile = flag.String("cpuprofile", "", "write cpu profile to `file`")

		inputPtr              = flag.String("Input", "", "The input type: Raspivid, File")
		inputCodecPtr         = flag.String("InputCodec", "", "The codec of the input: H264, Mjpeg")
		output1Ptr            = flag.String("Output1", "", "The first output type: Http, Rtmp, File, Udp, Rtp")
		output2Ptr            = flag.String("Output2", "", "The second output type: Http, Rtmp, File, Udp, Rtp")
		rtmpMethodPtr         = flag.String("RtmpMethod", "", "The method used to send over rtmp: Ffmpeg, Librtmp")
		packetizationPtr      = flag.String("Packetization", "", "The method of data packetisation: Flv, Mpegts, None")
		quantizationModePtr   = flag.String("QuantizationMode", "", "Whether quantization if on or off (variable bitrate): On, Off")
		verbosityPtr          = flag.String("Verbosity", "", "Verbosity: Info, Warning, Error, Fatal")
		framesPerClipPtr      = flag.String("FramesPerClip", "", "Number of frames per clip sent")
		rtmpUrlPtr            = flag.String("RtmpUrl", "", "Url of rtmp endpoint")
		bitratePtr            = flag.String("Bitrate", "", "Bitrate of recorded video")
		outputFileNamePtr     = flag.String("OutputFileName", "", "The directory of the output file")
		inputFileNamePtr      = flag.String("InputFileName", "", "The directory of the input file")
		heightPtr             = flag.String("Height", "", "Height in pixels")
		widthPtr              = flag.String("Width", "", "Width in pixels")
		frameRatePtr          = flag.String("FrameRate", "", "Frame rate of captured video")
		httpAddressPtr        = flag.String("HttpAddress", "", "Destination address of http posts")
		quantizationPtr       = flag.String("Quantization", "", "Desired quantization value: 0-40")
		timeoutPtr            = flag.String("Timeout", "", "Http timeout in seconds")
		intraRefreshPeriodPtr = flag.String("IntraRefreshPeriod", "", "The IntraRefreshPeriod i.e. how many keyframes we send")
		verticalFlipPtr       = flag.String("VerticalFlip", "", "Flip video vertically: Yes, No")
		horizontalFlipPtr     = flag.String("HorizontalFlip", "", "Flip video horizontally: Yes, No")
		logPathPtr            = flag.String("LogPath", defaultLogPath, "Path for logging files (default is /var/log/netsender/)")
		rtpAddrPtr            = flag.String("RtpAddr", "", "Rtp destination address: <IP>:<port> (port is generally 6970-6999)")
	)

	flag.Parse()

	logger = smartlogger.New(defaultLogVerbosity, *logPathPtr)

	cfg.Logger = logger

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

	switch *inputPtr {
	case "Raspivid":
		cfg.Input = revid.Raspivid
	case "File":
		cfg.Input = revid.File
	case "":
	default:
		logger.Log(smartlogger.Error, pkg+"bad input argument")
	}

	switch *inputCodecPtr {
	case "H264":
		cfg.InputCodec = revid.H264
	case "":
	default:
		logger.Log(smartlogger.Error, pkg+"bad input codec argument")
	}

	switch *output1Ptr {
	case "File":
		cfg.Output1 = revid.File
	case "Http":
		cfg.Output1 = revid.Http
	case "Rtmp":
		cfg.Output1 = revid.Rtmp
	case "FfmpegRtmp":
		cfg.Output1 = revid.FfmpegRtmp
	case "Udp":
		cfg.Output1 = revid.Udp
	case "Rtp":
		cfg.Output1 = revid.Rtp
	case "":
	default:
		logger.Log(smartlogger.Error, pkg+"bad output 1 argument")
	}

	switch *output2Ptr {
	case "File":
		cfg.Output2 = revid.File
	case "Http":
		cfg.Output2 = revid.Http
	case "Rtmp":
		cfg.Output2 = revid.Rtmp
	case "FfmpegRtmp":
		cfg.Output2 = revid.FfmpegRtmp
	case "Udp":
		cfg.Output2 = revid.Udp
	case "Rtp":
		cfg.Output2 = revid.Rtp
	case "":
	default:
		logger.Log(smartlogger.Error, pkg+"bad output 2 argument")
	}

	switch *rtmpMethodPtr {
	case "Ffmpeg":
		cfg.RtmpMethod = revid.Ffmpeg
	case "LibRtmp":
		cfg.RtmpMethod = revid.LibRtmp
	case "":
	default:
		logger.Log(smartlogger.Error, pkg+"bad rtmp method argument")
	}

	switch *packetizationPtr {
	case "None":
		cfg.Packetization = revid.None
	case "Mpegts":
		cfg.Packetization = revid.Mpegts
	case "Flv":
		cfg.Packetization = revid.Flv
	case "":
	default:
		logger.Log(smartlogger.Error, pkg+"bad packetization argument")
	}

	switch *quantizationModePtr {
	case "QuantizationOn":
		cfg.QuantizationMode = revid.QuantizationOn
	case "QuantizationOff":
		cfg.QuantizationMode = revid.QuantizationOff
	case "":
	default:
		logger.Log(smartlogger.Error, pkg+"bad quantization mode argument")
	}

	switch *verbosityPtr {
	case "No":
		cfg.LogLevel = smartlogger.Fatal
	case "Yes":
		cfg.LogLevel = smartlogger.Debug
	case "":
	default:
		logger.Log(smartlogger.Error, pkg+"bad verbosity argument")
	}

	switch *horizontalFlipPtr {
	case "No":
		cfg.FlipHorizontal = false
	case "Yes":
		cfg.FlipHorizontal = true
	case "":
		cfg.FlipHorizontal = false
	default:
		logger.Log(smartlogger.Error, pkg+"bad horizontal flip option")
	}

	switch *verticalFlipPtr {
	case "No":
		cfg.FlipVertical = false
	case "Yes":
		cfg.FlipVertical = true
	case "":
		cfg.FlipVertical = false
	default:
		logger.Log(smartlogger.Error, pkg+"bad vertical flip option")
	}

	fpc, err := strconv.Atoi(*framesPerClipPtr)
	if err == nil && fpc > 0 {
		cfg.FramesPerClip = fpc
	}
	cfg.RtmpUrl = *rtmpUrlPtr
	cfg.Bitrate = *bitratePtr
	cfg.OutputFileName = *outputFileNamePtr
	cfg.InputFileName = *inputFileNamePtr
	cfg.Height = *heightPtr
	cfg.Width = *widthPtr
	cfg.FrameRate = *frameRatePtr
	cfg.HttpAddress = *httpAddressPtr
	cfg.Quantization = *quantizationPtr
	cfg.Timeout = *timeoutPtr
	cfg.IntraRefreshPeriod = *intraRefreshPeriodPtr
	cfg.RtpAddress = *rtpAddrPtr

	return cfg
}

// initialize then run the main NetSender client
func run(cfg revid.Config) error {
	// initialize NetSender and use NetSender's logger
	//config.Logger = netsender.Logger()
	logger.Log(smartlogger.Info, pkg+"running in NetSender mode")

	var ns netsender.Sender
	err := ns.Init(logger, nil, nil, nil)
	if err != nil {
		return err
	}
	vars, _ := ns.Vars()
	vs := ns.VarSum()
	paused := false
	if vars["mode"] == "Paused" {
		paused = true
	}
	if !paused {
		cfg, err = updateRevid(&ns, cfg, vars, false)
		if err != nil {
			return err
		}
	}

	for {
		if err := send(&ns); err != nil {
			logger.Log(smartlogger.Error, pkg+"polling failed", "error", err.Error())
			time.Sleep(netSendRetryTime)
			continue
		}

		if vs != ns.VarSum() {
			// vars changed
			vars, err := ns.Vars()
			if err != nil {
				logger.Log(smartlogger.Error, pkg+"netSender failed to get vars", "error", err.Error())
				time.Sleep(netSendRetryTime)
				continue
			}
			vs = ns.VarSum()
			if vars["mode"] == "Paused" {
				if !paused {
					logger.Log(smartlogger.Info, pkg+"pausing revid")
					stopRevid()
					paused = true
				}
			} else {
				cfg, err = updateRevid(&ns, cfg, vars, !paused)
				if err != nil {
					return err
				}
				if paused {
					paused = false
				}
			}
		}
		sleepTime, _ := strconv.Atoi(ns.Param("mp"))
		time.Sleep(time.Duration(sleepTime) * time.Second)
	}
}

// send implements our main NetSender client and handles NetReceiver configuration
// (as distinct from httpSender which just sends video data).
func send(ns *netsender.Sender) error {
	// populate input values, if any
	inputs := netsender.MakePins(ns.Param("ip"), "X")
	if rv != nil {
		for i, pin := range inputs {
			if pin.Name == "X23" {
				inputs[i].Value = rv.Bitrate()
			}
		}
	}

	_, reconfig, err := ns.Send(netsender.RequestPoll, inputs)
	if err != nil {
		return err
	}
	if reconfig {
		return ns.Config()
	}
	return nil
}

// wrappers for stopping and starting revid
func startRevid(ns *netsender.Sender, cfg revid.Config) (revid.Config, error) {
	var err error
	rv, err = revid.New(cfg, ns)
	if err != nil {
		return cfg, err
	}
	rv.Start()
	return cfg, nil
}

func stopRevid() {
	rv.Stop()

	// FIXME(kortschak): Is this waiting on completion of work?
	// Use a wait group and Wait method if it is.
	time.Sleep(revidStopTime)
}

func updateRevid(ns *netsender.Sender, cfg revid.Config, vars map[string]string, stop bool) (revid.Config, error) {
	if stop {
		stopRevid()
	}

	//look through the vars and update revid where needed
	for key, value := range vars {
		switch key {
		case "Output":
			switch value {
			case "File":
				cfg.Output1 = revid.File
			case "Http":
				cfg.Output1 = revid.Http
			case "Rtmp":
				cfg.Output1 = revid.Rtmp
			case "FfmpegRtmp":
				cfg.Output1 = revid.FfmpegRtmp
			default:
				logger.Log(smartlogger.Warning, pkg+"invalid Output1 param", "value", value)
				continue
			}
		case "FramesPerClip":
			fpc, err := strconv.Atoi(value)
			if fpc > 0 && err == nil {
				cfg.FramesPerClip = fpc
			} else {
				logger.Log(smartlogger.Warning, pkg+"invalid FramesPerClip param", "value", value)
			}
		case "RtmpUrl":
			cfg.RtmpUrl = value
		case "Bitrate":
			asInt, err := strconv.Atoi(value)
			if asInt > 0 && err == nil {
				cfg.Bitrate = value
			} else {
				logger.Log(smartlogger.Warning, pkg+"invalid Bitrate param", "value", value)
			}
		case "OutputFileName":
			cfg.OutputFileName = value
		case "InputFileName":
			cfg.InputFileName = value
		case "Height":
			asInt, err := strconv.Atoi(value)
			if asInt > 0 && err == nil {
				cfg.Height = value
			} else {
				logger.Log(smartlogger.Warning, pkg+"invalid Height param", "value", value)
			}
		case "Width":
			asInt, err := strconv.Atoi(value)
			if asInt > 0 && err == nil {
				cfg.Width = value
			} else {
				logger.Log(smartlogger.Warning, pkg+"invalid Width param", "value", value)
			}
		case "FrameRate":
			asInt, err := strconv.Atoi(value)
			if asInt > 0 && err == nil {
				cfg.FrameRate = value
			} else {
				logger.Log(smartlogger.Warning, pkg+"invalid FrameRate param", "value", value)
			}
		case "HttpAddress":
			cfg.HttpAddress = value
		case "Quantization":
			asInt, err := strconv.Atoi(value)
			if asInt > 0 && err == nil {
				cfg.Quantization = value
			} else {
				logger.Log(smartlogger.Warning, pkg+"invalid Quantization param", "value", value)
			}
		case "Timeout":
			asInt, err := strconv.Atoi(value)
			if asInt > 0 && err == nil {
				cfg.Timeout = value
			}
		case "IntraRefreshPeriod":
			asInt, err := strconv.Atoi(value)
			if asInt > 0 && err == nil {
				cfg.IntraRefreshPeriod = value
			}
		case "HorizontalFlip":
			switch value {
			case "Yes":
				cfg.FlipHorizontal = true
			case "No":
				cfg.FlipHorizontal = false
			default:
				logger.Log(smartlogger.Warning, pkg+"invalid HorizontalFlip param", "value", value)
			}
		case "VerticalFlip":
			switch value {
			case "Yes":
				cfg.FlipVertical = true
			case "No":
				cfg.FlipVertical = false
			default:
				logger.Log(smartlogger.Warning, pkg+"invalid VerticalFlip param", "value", value)
			}
		default:
		}
	}

	return startRevid(ns, cfg)
}