mirror
/
av
mirror of https://bitbucket.org/ausocean/av.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Updating branch with master 

Merge branch 'master' into improve-ts-encoder-performance
This commit is contained in:
saxon 2019-01-03 12:43:58 +10:30
parent 1436d1f256 6d61ff7c25
commit d037b21753
6 changed files with 226 additions and 325 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

206
cmd/revid-cli/main.go
View File

@ -33,11 +33,13 @@ import (
"os"
"runtime/pprof"
"strconv"
"strings"
"time"
"bitbucket.org/ausocean/av/revid"
"bitbucket.org/ausocean/iot/pi/netsender"
"bitbucket.org/ausocean/utils/smartlogger"
"bitbucket.org/ausocean/iot/pi/smartlogger"
"bitbucket.org/ausocean/utils/logger"
)
const (
@ -45,7 +47,7 @@ const (
progName = "revid-cli"
// Logging is set to INFO level.
defaultLogVerbosity = smartlogger.Debug
defaultLogVerbosity = logger.Debug
)
// Other misc consts
@ -61,7 +63,7 @@ const (
var canProfile = true
// The logger that will be used throughout
var logger *smartlogger.Logger
var log *logger.Logger
func main() {
useNetsender := flag.Bool("NetSender", false, "Are we checking vars through netsender?")
@ -73,7 +75,7 @@ func main() {
// run revid for the specified duration
rv, _, err := startRevid(nil, cfg)
if err != nil {
cfg.Logger.Log(smartlogger.Fatal, pkg+"failed to start revid", err.Error())
cfg.Logger.Log(logger.Fatal, pkg+"failed to start revid", err.Error())
}
time.Sleep(*runDurationPtr)
stopRevid(rv)
@ -82,7 +84,7 @@ func main() {
err := run(nil, cfg)
if err != nil {
logger.Log(smartlogger.Fatal, pkg+"failed to run revid", "error", err.Error())
log.Log(logger.Fatal, pkg+"failed to run revid", "error", err.Error())
os.Exit(1)
}
}
@ -101,44 +103,42 @@ func handleFlags() revid.Config {
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")
quantizePtr = flag.Bool("Quantize", false, "Quantize input (non-variable bitrate)")
verbosityPtr = flag.String("Verbosity", "", "Verbosity: Info, Warning, Error, Fatal")
framesPerClipPtr = flag.String("FramesPerClip", "", "Number of frames per clip sent")
framesPerClipPtr = flag.Uint("FramesPerClip", 0, "Number of frames per clip sent")
rtmpUrlPtr = flag.String("RtmpUrl", "", "Url of rtmp endpoint")
bitratePtr = flag.String("Bitrate", "", "Bitrate of recorded video")
bitratePtr = flag.Uint("Bitrate", 0, "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")
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")
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/)")
quantizationPtr = flag.Uint("Quantization", 0, "Desired quantization value: 0-40")
intraRefreshPeriodPtr = flag.Uint("IntraRefreshPeriod", 0, "The IntraRefreshPeriod i.e. how many keyframes we send")
verticalFlipPtr = flag.Bool("VerticalFlip", false, "Flip video vertically: Yes, No")
horizontalFlipPtr = flag.Bool("HorizontalFlip", false, "Flip video horizontally: Yes, No")
rtpAddrPtr = flag.String("RtpAddr", "", "Rtp destination address: <IP>:<port> (port is generally 6970-6999)")
)
flag.Parse()
logger = smartlogger.New(defaultLogVerbosity, *logPathPtr)
log = logger.New(defaultLogVerbosity, &smartlogger.New("/var/log/netsender").LogRoller)
cfg.Logger = logger
cfg.Logger = log
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())
log.Log(logger.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())
log.Log(logger.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.")
log.Log(logger.Warning, pkg+"ignoring cpuprofile flag - http/pprof built in.")
}
}
@ -149,7 +149,7 @@ func handleFlags() revid.Config {
cfg.Input = revid.File
case "":
default:
logger.Log(smartlogger.Error, pkg+"bad input argument")
log.Log(logger.Error, pkg+"bad input argument")
}
switch *inputCodecPtr {
@ -157,7 +157,7 @@ func handleFlags() revid.Config {
cfg.InputCodec = revid.H264
case "":
default:
logger.Log(smartlogger.Error, pkg+"bad input codec argument")
log.Log(logger.Error, pkg+"bad input codec argument")
}
switch *output1Ptr {
@ -175,7 +175,7 @@ func handleFlags() revid.Config {
cfg.Output1 = revid.Rtp
case "":
default:
logger.Log(smartlogger.Error, pkg+"bad output 1 argument")
log.Log(logger.Error, pkg+"bad output 1 argument")
}
switch *output2Ptr {
@ -193,7 +193,7 @@ func handleFlags() revid.Config {
cfg.Output2 = revid.Rtp
case "":
default:
logger.Log(smartlogger.Error, pkg+"bad output 2 argument")
log.Log(logger.Error, pkg+"bad output 2 argument")
}
switch *rtmpMethodPtr {
@ -203,7 +203,7 @@ func handleFlags() revid.Config {
cfg.RtmpMethod = revid.LibRtmp
case "":
default:
logger.Log(smartlogger.Error, pkg+"bad rtmp method argument")
log.Log(logger.Error, pkg+"bad rtmp method argument")
}
switch *packetizationPtr {
@ -215,56 +215,24 @@ func handleFlags() revid.Config {
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")
log.Log(logger.Error, pkg+"bad packetization argument")
}
switch *verbosityPtr {
case "No":
cfg.LogLevel = smartlogger.Fatal
cfg.LogLevel = logger.Fatal
case "Debug":
cfg.LogLevel = smartlogger.Debug
//logger.SetLevel(smartlogger.Debug)
cfg.LogLevel = logger.Debug
//logger.SetLevel(logger.Debug)
case "":
default:
logger.Log(smartlogger.Error, pkg+"bad verbosity argument")
log.Log(logger.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.Quantize = *quantizePtr
cfg.FlipHorizontal = *horizontalFlipPtr
cfg.FlipVertical = *verticalFlipPtr
cfg.FramesPerClip = *framesPerClipPtr
cfg.RtmpUrl = *rtmpUrlPtr
cfg.Bitrate = *bitratePtr
cfg.OutputFileName = *outputFileNamePtr
@ -274,7 +242,6 @@ func handleFlags() revid.Config {
cfg.FrameRate = *frameRatePtr
cfg.HttpAddress = *httpAddressPtr
cfg.Quantization = *quantizationPtr
cfg.Timeout = *timeoutPtr
cfg.IntraRefreshPeriod = *intraRefreshPeriodPtr
cfg.RtpAddress = *rtpAddrPtr
@ -285,10 +252,10 @@ func handleFlags() revid.Config {
func run(rv *revid.Revid, cfg revid.Config) error {
// initialize NetSender and use NetSender's logger
//config.Logger = netsender.Logger()
logger.Log(smartlogger.Info, pkg+"running in NetSender mode")
log.Log(logger.Info, pkg+"running in NetSender mode")
var ns netsender.Sender
err := ns.Init(logger, nil, nil, nil)
err := ns.Init(log, nil, nil, nil)
if err != nil {
return err
}
@ -307,7 +274,7 @@ func run(rv *revid.Revid, cfg revid.Config) error {
for {
if err := send(&ns, rv); err != nil {
logger.Log(smartlogger.Error, pkg+"polling failed", "error", err.Error())
log.Log(logger.Error, pkg+"polling failed", "error", err.Error())
time.Sleep(netSendRetryTime)
continue
}
@ -316,14 +283,14 @@ func run(rv *revid.Revid, cfg revid.Config) error {
// vars changed
vars, err := ns.Vars()
if err != nil {
logger.Log(smartlogger.Error, pkg+"netSender failed to get vars", "error", err.Error())
log.Log(logger.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")
log.Log(logger.Info, pkg+"pausing revid")
stopRevid(rv)
paused = true
}
@ -402,86 +369,83 @@ func updateRevid(ns *netsender.Sender, rv *revid.Revid, cfg revid.Config, vars m
case "FfmpegRtmp":
cfg.Output1 = revid.FfmpegRtmp
default:
logger.Log(smartlogger.Warning, pkg+"invalid Output1 param", "value", value)
log.Log(logger.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)
f, err := strconv.ParseUint(value, 10, 0)
if err != nil {
log.Log(logger.Warning, pkg+"invalid framesperclip param", "value", value)
break
}
cfg.FramesPerClip = uint(f)
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)
r, err := strconv.ParseUint(value, 10, 0)
if err != nil {
log.Log(logger.Warning, pkg+"invalid framerate param", "value", value)
break
}
cfg.Bitrate = uint(r)
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)
h, err := strconv.ParseUint(value, 10, 0)
if err != nil {
log.Log(logger.Warning, pkg+"invalid height param", "value", value)
break
}
cfg.Height = uint(h)
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)
w, err := strconv.ParseUint(value, 10, 0)
if err != nil {
log.Log(logger.Warning, pkg+"invalid width param", "value", value)
break
}
cfg.Width = uint(w)
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)
r, err := strconv.ParseUint(value, 10, 0)
if err != nil {
log.Log(logger.Warning, pkg+"invalid framerate param", "value", value)
break
}
cfg.FrameRate = uint(r)
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
q, err := strconv.ParseUint(value, 10, 0)
if err != nil {
log.Log(logger.Warning, pkg+"invalid quantization param", "value", value)
break
}
cfg.Quantization = uint(q)
case "IntraRefreshPeriod":
asInt, err := strconv.Atoi(value)
if asInt > 0 && err == nil {
cfg.IntraRefreshPeriod = value
p, err := strconv.ParseUint(value, 10, 0)
if err != nil {
log.Log(logger.Warning, pkg+"invalid intrarefreshperiod param", "value", value)
break
}
cfg.IntraRefreshPeriod = uint(p)
case "HorizontalFlip":
switch value {
case "Yes":
switch strings.ToLower(value) {
case "true":
cfg.FlipHorizontal = true
case "No":
case "false":
cfg.FlipHorizontal = false
default:
logger.Log(smartlogger.Warning, pkg+"invalid HorizontalFlip param", "value", value)
log.Log(logger.Warning, pkg+"invalid HorizontalFlip param", "value", value)
}
case "VerticalFlip":
switch value {
case "Yes":
switch strings.ToLower(value) {
case "true":
cfg.FlipVertical = true
case "No":
case "false":
cfg.FlipVertical = false
default:
logger.Log(smartlogger.Warning, pkg+"invalid VerticalFlip param", "value", value)
log.Log(logger.Warning, pkg+"invalid VerticalFlip param", "value", value)
}
default:
}

7
revid/cmd/h264-file-to-flv-rtmp/main.go
View File

@ -33,7 +33,8 @@ import (
"time"
"bitbucket.org/ausocean/av/revid"
"bitbucket.org/ausocean/utils/smartlogger"
"bitbucket.org/ausocean/iot/pi/smartlogger"
"bitbucket.org/ausocean/utils/logger"
)
const (
@ -61,11 +62,11 @@ func main() {
RtmpMethod: revid.LibRtmp,
RtmpUrl: *rtmpUrlPtr,
Packetization: revid.Flv,
Logger: smartlogger.New(smartlogger.Info, logPath),
Logger: logger.New(logger.Info, &smartlogger.New(logPath).LogRoller),
}
revidInst, err := revid.New(config, nil)
if err != nil {
config.Logger.Log(smartlogger.Error, "Should not have got an error!: ", err.Error())
config.Logger.Log(logger.Error, "Should not have got an error!: ", err.Error())
return
}
revidInst.Start()

7
revid/cmd/h264-file-to-mpgets-file/main.go
View File

@ -31,7 +31,8 @@ import (
"time"
"bitbucket.org/ausocean/av/revid"
"bitbucket.org/ausocean/utils/smartlogger"
"bitbucket.org/ausocean/iot/pi/smartlogger"
"bitbucket.org/ausocean/utils/logger"
)
const (
@ -52,11 +53,11 @@ func main() {
Output1: revid.File,
OutputFileName: outputFile,
Packetization: revid.Mpegts,
Logger: smartlogger.New(smartlogger.Info, logPath),
Logger: logger.New(logger.Info, &smartlogger.New(logPath).LogRoller),
}
revidInst, err := revid.New(config, nil)
if err != nil {
config.Logger.Log(smartlogger.Error, "Should not have got an error!:", err.Error())
config.Logger.Log(logger.Error, "Should not have got an error!:", err.Error())
return
}
revidInst.Start()

188
revid/config.go
View File

@ -29,40 +29,43 @@ package revid
import (
"errors"
"strconv"
"bitbucket.org/ausocean/utils/smartlogger"
"bitbucket.org/ausocean/utils/logger"
)
// Config provides parameters relevant to a revid instance. A new config must
// be passed to the constructor.
type Config struct {
Input uint8
InputCodec uint8
Output1 uint8
Output2 uint8
RtmpMethod uint8
Packetization uint8
QuantizationMode uint8
LogLevel int8
LogLevel int8
Input uint8
InputCodec uint8
Output1 uint8
Output2 uint8
RtmpMethod uint8
Packetization uint8
// Quantize specifies whether the input to
// revid will have constant or variable
// bitrate.
Quantize bool
// FlipHorizonatla and FlipVertical specify
// whether video frames should be flipped.
FlipHorizontal bool
FlipVertical bool
FramesPerClip int
FramesPerClip uint
RtmpUrl string
Bitrate string
Bitrate uint
OutputFileName string
InputFileName string
Height string
Width string
FrameRate string
Height uint
Width uint
FrameRate uint
HttpAddress string
Quantization string
Timeout string
IntraRefreshPeriod string
Quantization uint
IntraRefreshPeriod uint
RtpAddress string
Logger Logger
SendRetry bool
@ -98,20 +101,18 @@ const (
defaultInput = Raspivid
defaultOutput = Http
defaultPacketization = Flv
defaultFrameRate = "25"
defaultWidth = "1280"
defaultHeight = "720"
defaultIntraRefreshPeriod = "100"
defaultTimeout = "0"
defaultQuantization = "40"
defaultBitrate = "400000"
defaultFrameRate = 25
defaultWidth = 1280
defaultHeight = 720
defaultIntraRefreshPeriod = 100
defaultTimeout = 0
defaultQuantization = 40
defaultBitrate = 400000
defaultQuantizationMode = QuantizationOff
defaultFramesPerClip = 1
defaultVerticalFlip = No
defaultHorizontalFlip = No
httpFramesPerClip = 560
defaultInputCodec = H264
defaultVerbosity = No
defaultVerbosity = No // FIXME(kortschak): This makes no sense whatsoever. No is currently 15.
defaultRtpAddr = "localhost:6970"
)
@ -123,28 +124,17 @@ func (c *Config) Validate(r *Revid) error {
case No:
case NothingDefined:
c.LogLevel = defaultVerbosity
c.Logger.Log(smartlogger.Warning, pkg+"no LogLevel mode defined, defaulting",
c.Logger.Log(logger.Warning, pkg+"no LogLevel mode defined, defaulting",
"LogLevel", defaultVerbosity)
default:
return errors.New("bad LogLevel defined in config")
}
switch c.QuantizationMode {
case QuantizationOn:
case QuantizationOff:
case NothingDefined:
c.Logger.Log(smartlogger.Warning, pkg+"no quantization mode defined, defaulting",
"quantizationMode", QuantizationOff)
c.QuantizationMode = QuantizationOff
default:
return errors.New("bad QuantizationMode defined in config")
}
switch c.Input {
case Raspivid:
case File:
case NothingDefined:
c.Logger.Log(smartlogger.Warning, pkg+"no input type defined, defaulting", "input",
c.Logger.Log(logger.Warning, pkg+"no input type defined, defaulting", "input",
defaultInput)
c.Input = defaultInput
default:
@ -153,36 +143,31 @@ func (c *Config) Validate(r *Revid) error {
switch c.InputCodec {
case H264:
if c.Bitrate != "" && c.Quantization != "" {
bitrate, err := strconv.Atoi(c.Bitrate)
if err != nil {
return errors.New("bitrate not an integer")
}
quantization, err := strconv.Atoi(c.Quantization)
if err != nil {
return errors.New("quantization not an integer")
}
if (bitrate > 0 && quantization > 0) || (bitrate == 0 && quantization == 0) {
return errors.New("bad bitrate and quantization combination for H264 input")
}
// FIXME(kortschak): This is not really what we want.
// Configuration really needs to be rethought here.
if c.Quantize && c.Quantization == 0 {
c.Quantization = defaultQuantization
} else {
c.Bitrate = defaultBitrate
}
if (c.Bitrate > 0 && c.Quantization > 0) || (c.Bitrate == 0 && c.Quantization == 0) {
return errors.New("bad bitrate and quantization combination for H264 input")
}
case Mjpeg:
if c.Quantization != "" {
quantization, err := strconv.Atoi(c.Quantization)
if err != nil {
return errors.New("quantization not an integer")
}
if quantization > 0 || c.Bitrate == "" {
return errors.New("bad bitrate or quantization for mjpeg input")
}
if c.Quantization > 0 || c.Bitrate == 0 {
return errors.New("bad bitrate or quantization for mjpeg input")
}
case NothingDefined:
c.Logger.Log(smartlogger.Warning, pkg+"no input codec defined, defaulting",
c.Logger.Log(logger.Warning, pkg+"no input codec defined, defaulting",
"inputCodec", defaultInputCodec)
c.InputCodec = defaultInputCodec
c.Logger.Log(smartlogger.Warning, pkg+"defaulting quantization", "quantization",
c.Logger.Log(logger.Warning, pkg+"defaulting quantization", "quantization",
defaultQuantization)
c.Quantization = defaultQuantization
default:
return errors.New("bad input codec defined in config")
}
@ -192,20 +177,20 @@ func (c *Config) Validate(r *Revid) error {
case Udp:
case Rtmp, FfmpegRtmp:
if c.RtmpUrl == "" {
c.Logger.Log(smartlogger.Info, pkg+"no RTMP URL: falling back to HTTP")
c.Logger.Log(logger.Info, pkg+"no RTMP URL: falling back to HTTP")
c.Output1 = Http
break
}
c.Logger.Log(smartlogger.Info, pkg+"defaulting frames per clip for rtmp out",
c.Logger.Log(logger.Info, pkg+"defaulting frames per clip for rtmp out",
"framesPerClip", defaultFramesPerClip)
c.FramesPerClip = defaultFramesPerClip
case NothingDefined:
c.Logger.Log(smartlogger.Warning, pkg+"no output defined, defaulting", "output",
c.Logger.Log(logger.Warning, pkg+"no output defined, defaulting", "output",
defaultOutput)
c.Output1 = defaultOutput
fallthrough
case Http, Rtp:
c.Logger.Log(smartlogger.Info, pkg+"defaulting frames per clip for http out",
c.Logger.Log(logger.Info, pkg+"defaulting frames per clip for http out",
"framesPerClip", httpFramesPerClip)
c.FramesPerClip = httpFramesPerClip
default:
@ -218,7 +203,7 @@ func (c *Config) Validate(r *Revid) error {
case Udp:
case Rtmp, FfmpegRtmp:
if c.RtmpUrl == "" {
c.Logger.Log(smartlogger.Info, pkg+"no RTMP URL: falling back to HTTP")
c.Logger.Log(logger.Info, pkg+"no RTMP URL: falling back to HTTP")
c.Output2 = Http
break
}
@ -229,78 +214,41 @@ func (c *Config) Validate(r *Revid) error {
}
if c.FramesPerClip < 1 {
c.Logger.Log(smartlogger.Warning, pkg+"no FramesPerClip defined, defaulting",
c.Logger.Log(logger.Warning, pkg+"no FramesPerClip defined, defaulting",
"framesPerClip", defaultFramesPerClip)
c.FramesPerClip = defaultFramesPerClip
}
if c.Width == "" {
c.Logger.Log(smartlogger.Warning, pkg+"no width defined, defaulting", "width",
defaultWidth)
if c.Width == 0 {
c.Logger.Log(logger.Warning, pkg+"no width defined, defaulting", "width", defaultWidth)
c.Width = defaultWidth
} else {
if integer, err := strconv.Atoi(c.Width); integer < 0 || err != nil {
return errors.New("width not unsigned integer")
}
}
if c.Height == "" {
c.Logger.Log(smartlogger.Warning, pkg+"no height defined, defaulting", "height",
defaultHeight)
if c.Height == 0 {
c.Logger.Log(logger.Warning, pkg+"no height defined, defaulting", "height", defaultHeight)
c.Height = defaultHeight
} else {
if integer, err := strconv.Atoi(c.Height); integer < 0 || err != nil {
return errors.New("height not unsigned integer")
}
}
if c.FrameRate == "" {
c.Logger.Log(smartlogger.Warning, pkg+"no frame rate defined, defaulting", "fps",
defaultFrameRate)
if c.FrameRate == 0 {
c.Logger.Log(logger.Warning, pkg+"no frame rate defined, defaulting", "fps", defaultFrameRate)
c.FrameRate = defaultFrameRate
} else {
if integer, err := strconv.Atoi(c.FrameRate); integer < 0 || err != nil {
return errors.New("frame rate not unsigned integer")
}
}
if c.Bitrate == "" {
c.Logger.Log(smartlogger.Warning, pkg+"no bitrate defined, defaulting", "bitrate",
defaultBitrate)
if c.Bitrate == 0 {
c.Logger.Log(logger.Warning, pkg+"no bitrate defined, defaulting", "bitrate", defaultBitrate)
c.Bitrate = defaultBitrate
} else {
if integer, err := strconv.Atoi(c.Bitrate); integer < 0 || err != nil {
return errors.New("bitrate not unsigned integer")
}
}
if c.Timeout == "" {
c.Logger.Log(smartlogger.Warning, pkg+"no timeout defined, defaulting", "timeout", defaultTimeout)
c.Timeout = defaultTimeout
} else {
if integer, err := strconv.Atoi(c.Timeout); integer < 0 || err != nil {
return errors.New("timeout not unsigned integer")
}
}
if c.IntraRefreshPeriod == "" {
c.Logger.Log(smartlogger.Warning, pkg+"no intra refresh defined, defaulting", "intraRefresh",
defaultIntraRefreshPeriod)
if c.IntraRefreshPeriod == 0 {
c.Logger.Log(logger.Warning, pkg+"no intra refresh defined, defaulting", "intraRefresh", defaultIntraRefreshPeriod)
c.IntraRefreshPeriod = defaultIntraRefreshPeriod
} else {
if integer, err := strconv.Atoi(c.IntraRefreshPeriod); integer < 0 || err != nil {
return errors.New("intra refresh not unsigned integer")
}
}
if c.Quantization == "" {
c.Logger.Log(smartlogger.Warning, pkg+"no quantization defined, defaulting", "quantization",
defaultQuantization)
if c.Quantization == 0 {
c.Logger.Log(logger.Warning, pkg+"no quantization defined, defaulting", "quantization", defaultQuantization)
c.Quantization = defaultQuantization
} else {
if integer, err := strconv.Atoi(c.Quantization); integer < 0 || integer > 51 || err != nil {
return errors.New("quantisation not unsigned integer or is over threshold")
}
} else if c.Quantization > 51 {
return errors.New("quantisation is over threshold")
}
if c.RtpAddress == "" {

121
revid/revid.go
View File

@ -44,8 +44,8 @@ import (
"bitbucket.org/ausocean/av/stream/lex"
"bitbucket.org/ausocean/av/stream/mts"
"bitbucket.org/ausocean/iot/pi/netsender"
"bitbucket.org/ausocean/utils/logger"
"bitbucket.org/ausocean/utils/ring"
"bitbucket.org/ausocean/utils/smartlogger"
)
// Misc constants
@ -133,7 +133,7 @@ var prevTime = now
type packer struct {
owner *Revid
packetCount int
packetCount uint
}
// Write implements the io.Writer interface.
@ -143,21 +143,21 @@ type packer struct {
// write may include a dropped frame.
func (p *packer) Write(frame []byte) (int, error) {
if len(frame) > ringBufferElementSize {
p.owner.config.Logger.Log(smartlogger.Warning, pkg+"frame was too big", "frame size", len(frame))
p.owner.config.Logger.Log(logger.Warning, pkg+"frame was too big", "frame size", len(frame))
return len(frame), nil
}
n, err := p.owner.buffer.Write(frame)
if err != nil {
if err == ring.ErrDropped {
p.owner.config.Logger.Log(smartlogger.Warning, pkg+"dropped frame", "frame size", len(frame))
p.owner.config.Logger.Log(logger.Warning, pkg+"dropped frame", "frame size", len(frame))
return len(frame), nil
}
p.owner.config.Logger.Log(smartlogger.Error, pkg+"unexpected ring buffer write error", "error", err.Error())
p.owner.config.Logger.Log(logger.Error, pkg+"unexpected ring buffer write error", "error", err.Error())
return n, err
}
p.packetCount++
now = time.Now()
if now.Sub(prevTime) > clipDuration && p.packetCount%7 == 0 {
if (p.owner.config.Output1 != Rtmp && now.Sub(prevTime) > clipDuration && p.packetCount%7 == 0) || p.owner.config.Output1 == Rtmp {
p.owner.buffer.Flush()
p.packetCount = 0
prevTime = now
@ -227,7 +227,7 @@ func (r *Revid) reset(config Config) error {
}
r.destination[outNo] = s
case FfmpegRtmp:
s, err := newFfmpegSender(config.RtmpUrl, r.config.FrameRate)
s, err := newFfmpegSender(config.RtmpUrl, fmt.Sprint(r.config.FrameRate))
if err != nil {
return err
}
@ -247,10 +247,7 @@ func (r *Revid) reset(config Config) error {
}
r.destination[outNo] = s
case Rtp:
// TODO: framerate in config should probably be an int, make conversions early
// when setting config fields in revid-cli
fps, _ := strconv.Atoi(r.config.FrameRate)
s, err := newRtpSender(r.config.RtpAddress, r.config.Logger.Log, fps)
s, err := newRtpSender(r.config.RtpAddress, r.config.Logger.Log, r.config.FrameRate)
if err != nil {
return err
}
@ -266,10 +263,10 @@ func (r *Revid) reset(config Config) error {
}
switch r.config.InputCodec {
case H264:
r.config.Logger.Log(smartlogger.Info, pkg+"using H264 lexer")
r.config.Logger.Log(logger.Info, pkg+"using H264 lexer")
r.lexTo = lex.H264
case Mjpeg:
r.config.Logger.Log(smartlogger.Info, pkg+"using MJPEG lexer")
r.config.Logger.Log(logger.Info, pkg+"using MJPEG lexer")
r.lexTo = lex.MJPEG
}
@ -291,13 +288,11 @@ func (r *Revid) reset(config Config) error {
}
r.encoder = stream.NopEncoder(&r.packer)
case Mpegts:
r.config.Logger.Log(smartlogger.Info, pkg+"using MPEGTS packetisation")
frameRate, _ := strconv.ParseFloat(r.config.FrameRate, 64)
r.encoder = mts.NewEncoder(&r.packer, frameRate)
r.config.Logger.Log(logger.Info, pkg+"using MPEGTS packetisation")
r.encoder = mts.NewEncoder(&r.packer, float64(r.config.FrameRate))
case Flv:
r.config.Logger.Log(smartlogger.Info, pkg+"using FLV packetisation")
frameRate, _ := strconv.Atoi(r.config.FrameRate)
r.encoder, err = flv.NewEncoder(&r.packer, true, true, frameRate)
r.config.Logger.Log(logger.Info, pkg+"using FLV packetisation")
r.encoder, err = flv.NewEncoder(&r.packer, true, true, int(r.config.FrameRate))
if err != nil {
return err
}
@ -315,29 +310,29 @@ func (r *Revid) IsRunning() bool {
// and packetising (if theres packetization) to a defined output.
func (r *Revid) Start() {
if r.isRunning {
r.config.Logger.Log(smartlogger.Warning, pkg+"revid.Start() called but revid already running")
r.config.Logger.Log(logger.Warning, pkg+"revid.Start() called but revid already running")
return
}
r.config.Logger.Log(smartlogger.Info, pkg+"starting Revid")
r.config.Logger.Log(smartlogger.Debug, pkg+"setting up output")
r.config.Logger.Log(logger.Info, pkg+"starting Revid")
r.config.Logger.Log(logger.Debug, pkg+"setting up output")
r.isRunning = true
r.config.Logger.Log(smartlogger.Info, pkg+"starting output routine")
r.config.Logger.Log(logger.Info, pkg+"starting output routine")
go r.outputClips()
r.config.Logger.Log(smartlogger.Info, pkg+"setting up input and receiving content")
r.config.Logger.Log(logger.Info, pkg+"setting up input and receiving content")
go r.setupInput()
}
// Stop halts any processing of video data from a camera or file
func (r *Revid) Stop() {
if !r.isRunning {
r.config.Logger.Log(smartlogger.Warning, pkg+"revid.Stop() called but revid not running")
r.config.Logger.Log(logger.Warning, pkg+"revid.Stop() called but revid not running")
return
}
r.config.Logger.Log(smartlogger.Info, pkg+"stopping revid")
r.config.Logger.Log(logger.Info, pkg+"stopping revid")
r.isRunning = false
r.config.Logger.Log(smartlogger.Info, pkg+"killing input proccess")
r.config.Logger.Log(logger.Info, pkg+"killing input proccess")
// If a cmd process is running, we kill!
if r.cmd != nil && r.cmd.Process != nil {
r.cmd.Process.Kill()
@ -357,53 +352,53 @@ loop:
case nil:
// Do nothing.
case ring.ErrTimeout:
r.config.Logger.Log(smartlogger.Warning, pkg+"ring buffer read timeout")
r.config.Logger.Log(logger.Warning, pkg+"ring buffer read timeout")
continue
default:
r.config.Logger.Log(smartlogger.Error, pkg+"unexpected error", "error", err.Error())
r.config.Logger.Log(logger.Error, pkg+"unexpected error", "error", err.Error())
fallthrough
case io.EOF:
break loop
}
count += chunk.Len()
r.config.Logger.Log(smartlogger.Debug, pkg+"about to send")
r.config.Logger.Log(logger.Debug, pkg+"about to send")
for i, dest := range r.destination {
err = dest.load(chunk)
if err != nil {
r.config.Logger.Log(smartlogger.Error, pkg+"failed to load clip to output"+strconv.Itoa(i))
r.config.Logger.Log(logger.Error, pkg+"failed to load clip to output"+strconv.Itoa(i))
}
}
for i, dest := range r.destination {
err = dest.send()
if err == nil {
r.config.Logger.Log(smartlogger.Debug, pkg+"sent clip to output "+strconv.Itoa(i))
r.config.Logger.Log(logger.Debug, pkg+"sent clip to output "+strconv.Itoa(i))
} else if r.config.SendRetry == false {
r.config.Logger.Log(smartlogger.Warning, pkg+"send to output "+strconv.Itoa(i)+"failed", "error", err.Error())
r.config.Logger.Log(logger.Warning, pkg+"send to output "+strconv.Itoa(i)+"failed", "error", err.Error())
} else {
r.config.Logger.Log(smartlogger.Error, pkg+"send to output "+strconv.Itoa(i)+
r.config.Logger.Log(logger.Error, pkg+"send to output "+strconv.Itoa(i)+
"failed, trying again", "error", err.Error())
err = dest.send()
if err != nil && chunk.Len() > 11 {
r.config.Logger.Log(smartlogger.Error, pkg+"second send attempted failed, restarting connection", "error", err.Error())
r.config.Logger.Log(logger.Error, pkg+"second send attempted failed, restarting connection", "error", err.Error())
for err != nil {
time.Sleep(sendFailedDelay)
if rs, ok := dest.(restarter); ok {
r.config.Logger.Log(smartlogger.Debug, pkg+"restarting session", "session", rs)
r.config.Logger.Log(logger.Debug, pkg+"restarting session", "session", rs)
err = rs.restart()
if err != nil {
r.config.Logger.Log(smartlogger.Error, pkg+"failed to restart rtmp session", "error", err.Error())
r.config.Logger.Log(logger.Error, pkg+"failed to restart rtmp session", "error", err.Error())
r.isRunning = false
return
}
r.config.Logger.Log(smartlogger.Info, pkg+"restarted rtmp session")
r.config.Logger.Log(logger.Info, pkg+"restarted rtmp session")
}
err = dest.send()
if err != nil {
r.config.Logger.Log(smartlogger.Error, pkg+"send failed again, with error", "error", err.Error())
r.config.Logger.Log(logger.Error, pkg+"send failed again, with error", "error", err.Error())
}
}
}
@ -414,7 +409,7 @@ loop:
for _, dest := range r.destination {
dest.release()
}
r.config.Logger.Log(smartlogger.Debug, pkg+"done reading that clip from ring buffer")
r.config.Logger.Log(logger.Debug, pkg+"done reading that clip from ring buffer")
// Log some information regarding bitrate and ring buffer size if it's time
now := time.Now()
@ -422,17 +417,17 @@ loop:
if deltaTime > bitrateTime {
// FIXME(kortschak): For subsecond deltaTime, this will give infinite bitrate.
r.bitrate = int(float64(count*8) / float64(deltaTime/time.Second))
r.config.Logger.Log(smartlogger.Debug, pkg+"bitrate (bits/s)", "bitrate", r.bitrate)
r.config.Logger.Log(smartlogger.Debug, pkg+"ring buffer size", "value", r.buffer.Len())
r.config.Logger.Log(logger.Debug, pkg+"bitrate (bits/s)", "bitrate", r.bitrate)
r.config.Logger.Log(logger.Debug, pkg+"ring buffer size", "value", r.buffer.Len())
lastTime = now
count = 0
}
}
r.config.Logger.Log(smartlogger.Info, pkg+"not outputting clips anymore")
r.config.Logger.Log(logger.Info, pkg+"not outputting clips anymore")
for i, dest := range r.destination {
err := dest.close()
if err != nil {
r.config.Logger.Log(smartlogger.Error, pkg+"failed to close output"+strconv.Itoa(i)+" destination", "error", err.Error())
r.config.Logger.Log(logger.Error, pkg+"failed to close output"+strconv.Itoa(i)+" destination", "error", err.Error())
}
}
}
@ -440,17 +435,17 @@ loop:
// startRaspivid sets up things for input from raspivid i.e. starts
// a raspivid process and pipes it's data output.
func (r *Revid) startRaspivid() error {
r.config.Logger.Log(smartlogger.Info, pkg+"starting raspivid")
r.config.Logger.Log(logger.Info, pkg+"starting raspivid")
const disabled = "0"
args := []string{
"--output", "-",
"--nopreview",
"--timeout", disabled,
"--width", r.config.Width,
"--height", r.config.Height,
"--bitrate", r.config.Bitrate,
"--framerate", r.config.FrameRate,
"--width", fmt.Sprint(r.config.Width),
"--height", fmt.Sprint(r.config.Height),
"--bitrate", fmt.Sprint(r.config.Bitrate),
"--framerate", fmt.Sprint(r.config.FrameRate),
}
if r.config.FlipHorizontal {
args = append(args, "--hflip")
@ -465,48 +460,40 @@ func (r *Revid) startRaspivid() error {
args = append(args,
"--codec", "H264",
"--inline",
"--intra", r.config.IntraRefreshPeriod,
"--intra", fmt.Sprint(r.config.IntraRefreshPeriod),
)
if r.config.QuantizationMode == QuantizationOn {
args = append(args, "-qp", r.config.Quantization)
if r.config.Quantize {
args = append(args, "-qp", fmt.Sprint(r.config.Quantization))
}
case Mjpeg:
args = append(args, "--codec", "MJPEG")
}
r.config.Logger.Log(smartlogger.Info, pkg+"raspivid args", "raspividArgs", strings.Join(args, " "))
r.config.Logger.Log(logger.Info, pkg+"raspivid args", "raspividArgs", strings.Join(args, " "))
r.cmd = exec.Command("raspivid", args...)
d, err := strconv.Atoi(r.config.FrameRate)
if err != nil {
return err
}
delay := time.Second / time.Duration(d)
stdout, err := r.cmd.StdoutPipe()
if err != nil {
return err
}
err = r.cmd.Start()
if err != nil {
return err
r.config.Logger.Log(logger.Fatal, pkg+"cannot start raspivid", "error", err.Error())
}
r.config.Logger.Log(smartlogger.Info, pkg+"reading camera data")
r.config.Logger.Log(logger.Info, pkg+"reading camera data")
delay := time.Second / time.Duration(r.config.FrameRate)
err = r.lexTo(r.encoder, stdout, delay)
r.config.Logger.Log(smartlogger.Info, pkg+"finished reading camera data")
r.config.Logger.Log(logger.Info, pkg+"finished reading camera data")
return err
}
// setupInputForFile sets things up for getting input from a file
func (r *Revid) setupInputForFile() error {
fps, err := strconv.Atoi(r.config.FrameRate)
if err != nil {
return err
}
delay := time.Second / time.Duration(fps)
delay := time.Second / time.Duration(r.config.FrameRate)
f, err := os.Open(r.config.InputFileName)
if err != nil {
r.config.Logger.Log(smartlogger.Error, err.Error())
r.config.Logger.Log(logger.Error, err.Error())
r.Stop()
return err
}

22
revid/senders.go
View File

@ -39,8 +39,8 @@ import (
"bitbucket.org/ausocean/av/stream/mts"
"bitbucket.org/ausocean/av/stream/rtp"
"bitbucket.org/ausocean/iot/pi/netsender"
"bitbucket.org/ausocean/utils/logger"
"bitbucket.org/ausocean/utils/ring"
"bitbucket.org/ausocean/utils/smartlogger"
)
// loadSender is a destination to send a *ring.Chunk to.
@ -166,18 +166,18 @@ func (s *httpSender) extractMeta(r string) error {
// Extract time from reply
t, err := dec.Int("ts")
if err != nil {
s.log(smartlogger.Warning, pkg+"No timestamp in reply")
s.log(logger.Warning, pkg+"No timestamp in reply")
} else {
s.log(smartlogger.Debug, fmt.Sprintf("%v got timestamp: %v", pkg, t))
s.log(logger.Debug, fmt.Sprintf("%v got timestamp: %v", pkg, t))
mts.SetTimeStamp(uint64(t))
}
// Extract location from reply
g, err := dec.String("ll")
if err != nil {
s.log(smartlogger.Warning, pkg+"No location in reply")
s.log(logger.Warning, pkg+"No location in reply")
} else {
s.log(smartlogger.Debug, fmt.Sprintf("%v got location: %v", pkg, g))
s.log(logger.Debug, fmt.Sprintf("%v got location: %v", pkg, g))
mts.SetLocation(g)
}
@ -269,10 +269,10 @@ func newRtmpSender(url string, timeout uint, retries int, log func(lvl int8, msg
if err == nil {
break
}
log(smartlogger.Error, err.Error())
log(logger.Error, err.Error())
sess.Close()
if n < retries-1 {
log(smartlogger.Info, pkg+"retry rtmp connection")
log(logger.Info, pkg+"retry rtmp connection")
}
}
if err != nil {
@ -315,10 +315,10 @@ func (s *rtmpSender) restart() error {
if err == nil {
break
}
s.log(smartlogger.Error, err.Error())
s.log(logger.Error, err.Error())
s.sess.Close()
if n < s.retries-1 {
s.log(smartlogger.Info, pkg+"retry rtmp connection")
s.log(logger.Info, pkg+"retry rtmp connection")
}
}
return err
@ -371,14 +371,14 @@ type rtpSender struct {
encoder *rtp.Encoder
}
func newRtpSender(addr string, log func(lvl int8, msg string, args ...interface{}), fps int) (*rtpSender, error) {
func newRtpSender(addr string, log func(lvl int8, msg string, args ...interface{}), fps uint) (*rtpSender, error) {
conn, err := net.Dial("udp", addr)
if err != nil {
return nil, err
}
s := &rtpSender{
log: log,
encoder: rtp.NewEncoder(conn, fps),
encoder: rtp.NewEncoder(conn, int(fps)),
}
return s, nil
}