av/revid/revid.go

518 lines
15 KiB
Go
Raw Normal View History

/*
NAME
2018-06-09 08:28:33 +03:00
revid.go
DESCRIPTION
See Readme.md
AUTHORS
2018-06-07 14:50:57 +03:00
Saxon A. Nelson-Milton <saxon@ausocean.org>
Alan Noble <alan@ausocean.org>
LICENSE
2018-06-08 03:02:13 +03:00
revid 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
2018-06-07 14:50:57 +03:00
along with revid in gpl.txt. If not, see http://www.gnu.org/licenses.
*/
// revid is a testbed for re-muxing and re-directing video streams as MPEG-TS over various protocols.
package revid
import (
2018-01-31 04:00:03 +03:00
"errors"
"fmt"
"io"
"os"
"os/exec"
"strconv"
2018-09-19 15:32:50 +03:00
"strings"
2018-04-16 07:54:21 +03:00
"time"
2018-02-14 10:02:57 +03:00
2018-08-20 01:19:02 +03:00
"bitbucket.org/ausocean/av/stream"
"bitbucket.org/ausocean/av/stream/flv"
"bitbucket.org/ausocean/av/stream/lex"
2018-08-20 01:19:02 +03:00
"bitbucket.org/ausocean/av/stream/mts"
"bitbucket.org/ausocean/iot/pi/netsender"
2018-06-27 01:20:05 +03:00
"bitbucket.org/ausocean/utils/ring"
"bitbucket.org/ausocean/utils/smartlogger"
)
// Misc constants
const (
2018-06-08 06:18:11 +03:00
clipDuration = 1 * time.Second
mp2tPacketSize = 188 // MPEG-TS packet size
mp2tMaxPackets = int(clipDuration * 2016 / time.Second) // # first multiple of 7 and 8 greater than 2000
ringBufferSize = 10000
2018-05-06 11:38:45 +03:00
ringBufferElementSize = 150000
writeTimeout = 10 * time.Millisecond
readTimeout = 10 * time.Millisecond
2018-06-08 06:18:11 +03:00
httpTimeout = 5 * time.Second
bitrateTime = 1 * time.Minute
2018-05-06 11:38:45 +03:00
mjpegParserInChanLen = 100000
ffmpegPath = "/usr/local/bin/ffmpeg"
rtmpConnectionTimout = 10
outputChanSize = 1000
cameraRetryPeriod = 5 * time.Second
sendFailedDelay = 5 * time.Millisecond
2018-04-24 08:38:30 +03:00
maxSendFailedErrorCount = 500
2018-05-06 11:38:45 +03:00
clipSizeThreshold = 11
rtmpConnectionMaxTries = 5
raspividNoOfTries = 3
2018-06-08 06:18:11 +03:00
sendingWaitTime = 5 * time.Millisecond
pkg = "revid:"
2018-01-10 06:57:56 +03:00
)
2018-01-24 07:12:22 +03:00
// Log Types
2018-01-24 04:40:03 +03:00
const (
2018-01-24 07:12:22 +03:00
Error = "Error"
2018-01-24 04:40:03 +03:00
Warning = "Warning"
2018-01-24 07:12:22 +03:00
Info = "Info"
Debug = "Debug"
Detail = "Detail"
2018-01-24 04:40:03 +03:00
)
type Logger interface {
SetLevel(int8)
Log(level int8, message string, params ...interface{})
}
2018-04-16 08:12:16 +03:00
// Revid provides methods to control a revid session; providing methods
// to start, stop and change the state of an instance using the Config struct.
2018-06-09 05:01:21 +03:00
type Revid struct {
// config holds the Revid configuration.
// For historical reasons it also handles logging.
// FIXME(kortschak): The relationship of concerns
// in config/ns is weird.
config Config
// ns holds the netsender.Sender responsible for HTTP.
ns *netsender.Sender
// setupInput holds the current approach to setting up
// the input stream.
setupInput func() error
// cmd is the exec'd process that may be used to produce
// the input stream.
// FIXME(kortschak): This should not exist. Replace this
// with a context.Context cancellation.
cmd *exec.Cmd
// lexTo, encoder and packer handle transcoding the input stream.
lexTo func(dst stream.Encoder, src io.Reader, delay time.Duration) error
encoder stream.Encoder
packer packer
// buffer handles passing frames from the transcoder
// to the target destination.
buffer *ring.Buffer
// destination is the target endpoint.
destination []loadSender
// bitrate hold the last send bitrate calculation result.
bitrate int
// isRunning is a loaded and cocked foot-gun.
isRunning bool
2018-01-23 06:15:06 +03:00
}
var now = time.Now()
var prevTime = now
// packer takes data segments and packs them into clips
// of the number frames specified in the owners config.
type packer struct {
owner *Revid
packetCount int
}
// Write implements the io.Writer interface.
//
// Unless the ring buffer returns an error, all writes
// are deemed to be successful, although a successful
// 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))
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))
return len(frame), nil
}
p.owner.config.Logger.Log(smartlogger.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 {
p.owner.buffer.Flush()
p.packetCount = 0
prevTime = now
}
return len(frame), nil
}
// New returns a pointer to a new Revid with the desired configuration, and/or
// an error if construction of the new instance was not successful.
func New(c Config, ns *netsender.Sender) (*Revid, error) {
r := Revid{ns: ns}
r.buffer = ring.NewBuffer(ringBufferSize, ringBufferElementSize, writeTimeout)
r.packer.owner = &r
2018-06-09 05:01:21 +03:00
err := r.reset(c)
if err != nil {
2018-06-09 05:01:21 +03:00
return nil, err
}
2018-06-09 05:01:21 +03:00
return &r, nil
}
2018-06-27 01:20:05 +03:00
// Bitrate returns the result of the most recent bitrate check.
func (r *Revid) Bitrate() int {
return r.bitrate
2018-04-24 08:38:30 +03:00
}
2018-06-17 14:38:27 +03:00
// Config returns the Revid's config.
func (r *Revid) Config() *Config {
// FIXME(kortschak): This is a massive footgun and should not exist.
// Since the config's fields are accessed in running goroutines, any
// mutation is a data race. With bad luck a data race is possible by
// reading the returned value since it is possible for the running
// Ravid to mutate the config it holds.
return &r.config
}
2018-06-09 05:01:21 +03:00
// reset swaps the current config of a Revid with the passed
// configuration; checking validity and returning errors if not valid.
2018-06-09 05:01:21 +03:00
func (r *Revid) reset(config Config) error {
r.config.Logger = config.Logger
2018-02-10 10:08:14 +03:00
err := config.Validate(r)
if err != nil {
return errors.New("Config struct is bad: " + err.Error())
}
r.config = config
for _, dest := range r.destination {
if dest != nil {
err = dest.close()
if err != nil {
return err
}
2018-06-09 07:38:48 +03:00
}
}
n := 1
if r.config.Output2 != 0 {
n = 2
}
r.destination = make([]loadSender, n)
for outNo, outType := range []uint8{r.config.Output1, r.config.Output2} {
switch outType {
case File:
s, err := newFileSender(config.OutputFileName)
if err != nil {
return err
}
r.destination[outNo] = s
case FfmpegRtmp:
s, err := newFfmpegSender(config.RtmpUrl, r.config.FrameRate)
if err != nil {
return err
}
r.destination[outNo] = s
case Rtmp:
s, err := newRtmpSender(config.RtmpUrl, rtmpConnectionTimout, rtmpConnectionMaxTries, r.config.Logger.Log)
if err != nil {
return err
}
r.destination[outNo] = s
case Http:
r.destination[outNo] = newHttpSender(r.ns, r.config.Logger.Log)
case Udp:
s, err := newUdpSender(r.config.RtpAddress, r.config.Logger.Log)
if err != nil {
return err
}
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)
if err != nil {
return err
}
r.destination[outNo] = s
}
}
switch r.config.Input {
case Raspivid:
2018-06-24 15:01:09 +03:00
r.setupInput = r.startRaspivid
case File:
r.setupInput = r.setupInputForFile
2018-01-10 04:32:16 +03:00
}
switch r.config.InputCodec {
case H264:
r.config.Logger.Log(smartlogger.Info, pkg+"using H264 lexer")
r.lexTo = lex.H264
case Mjpeg:
r.config.Logger.Log(smartlogger.Info, pkg+"using MJPEG lexer")
r.lexTo = lex.MJPEG
}
switch r.config.Packetization {
case None:
// no packetisation - Revid output chan grabs raw data straight from parser
r.lexTo = func(dst stream.Encoder, src io.Reader, _ time.Duration) error {
for {
var b [4 << 10]byte
n, rerr := src.Read(b[:])
werr := dst.Encode(b[:n])
if rerr != nil {
return rerr
}
if werr != nil {
return werr
}
}
}
r.encoder = stream.NopEncoder(&r.packer)
case Mpegts:
r.config.Logger.Log(smartlogger.Info, pkg+"using MPEGTS packetisation")
2018-08-19 14:09:57 +03:00
frameRate, _ := strconv.ParseFloat(r.config.FrameRate, 64)
r.encoder = mts.NewEncoder(&r.packer, frameRate)
case Flv:
r.config.Logger.Log(smartlogger.Info, pkg+"using FLV packetisation")
2018-07-07 08:57:59 +03:00
frameRate, _ := strconv.Atoi(r.config.FrameRate)
r.encoder, err = flv.NewEncoder(&r.packer, true, true, frameRate)
if err != nil {
return err
}
}
return nil
2018-01-10 04:32:16 +03:00
}
2018-06-17 14:41:33 +03:00
// IsRunning returns whether the receiver is running.
2018-06-09 05:01:21 +03:00
func (r *Revid) IsRunning() bool {
return r.isRunning
}
2018-06-09 05:01:21 +03:00
// Start invokes a Revid to start processing video from a defined input
// and packetising (if theres packetization) to a defined output.
2018-06-09 05:01:21 +03:00
func (r *Revid) Start() {
if r.isRunning {
r.config.Logger.Log(smartlogger.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")
2018-05-06 11:03:44 +03:00
r.isRunning = true
r.config.Logger.Log(smartlogger.Info, pkg+"starting output routine")
2018-05-06 11:03:44 +03:00
go r.outputClips()
r.config.Logger.Log(smartlogger.Info, pkg+"setting up input and receiving content")
2018-05-06 11:03:44 +03:00
go r.setupInput()
}
// Stop halts any processing of video data from a camera or file
2018-06-09 05:01:21 +03:00
func (r *Revid) Stop() {
2018-03-17 16:29:42 +03:00
if !r.isRunning {
r.config.Logger.Log(smartlogger.Warning, pkg+"revid.Stop() called but revid not running")
2018-03-17 16:29:42 +03:00
return
}
2018-05-07 05:53:50 +03:00
r.config.Logger.Log(smartlogger.Info, pkg+"stopping revid")
2018-04-16 07:54:21 +03:00
r.isRunning = false
2018-05-06 10:18:17 +03:00
r.config.Logger.Log(smartlogger.Info, pkg+"killing input proccess")
2018-04-16 07:54:21 +03:00
// If a cmd process is running, we kill!
if r.cmd != nil && r.cmd.Process != nil {
r.cmd.Process.Kill()
}
}
// outputClips takes the clips produced in the packClips method and outputs them
2018-04-16 08:12:16 +03:00
// to the desired output defined in the revid config
2018-06-09 05:01:21 +03:00
func (r *Revid) outputClips() {
lastTime := time.Now()
var count int
loop:
for r.isRunning {
// If the ring buffer has something we can read and send off
chunk, err := r.buffer.Next(readTimeout)
switch err {
case nil:
// Do nothing.
case ring.ErrTimeout:
r.config.Logger.Log(smartlogger.Warning, pkg+"ring buffer read timeout")
continue
default:
r.config.Logger.Log(smartlogger.Error, pkg+"unexpected error", "error", err.Error())
fallthrough
case io.EOF:
break loop
}
2018-05-03 07:58:14 +03:00
count += chunk.Len()
r.config.Logger.Log(smartlogger.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))
}
}
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))
2018-12-14 13:20:26 +03:00
} else if r.config.SendRetry == false {
r.config.Logger.Log(smartlogger.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)+
"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())
for err != nil {
time.Sleep(sendFailedDelay)
if rs, ok := dest.(restarter); ok {
r.config.Logger.Log(smartlogger.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.isRunning = false
return
}
r.config.Logger.Log(smartlogger.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())
}
2018-06-09 07:38:48 +03:00
}
}
}
}
// Release the chunk back to the ring buffer for use
for _, dest := range r.destination {
dest.release()
}
r.config.Logger.Log(smartlogger.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()
deltaTime := now.Sub(lastTime)
2018-06-08 06:18:11 +03:00
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())
lastTime = now
count = 0
}
}
r.config.Logger.Log(smartlogger.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())
}
}
}
2018-06-24 15:01:09 +03:00
// startRaspivid sets up things for input from raspivid i.e. starts
// a raspivid process and pipes it's data output.
2018-06-24 15:01:09 +03:00
func (r *Revid) startRaspivid() error {
r.config.Logger.Log(smartlogger.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,
}
if r.config.FlipHorizontal {
args = append(args, "--hflip")
}
if r.config.FlipVertical {
args = append(args, "--vflip")
}
switch r.config.InputCodec {
default:
return fmt.Errorf("revid: invalid input codec: %v", r.config.InputCodec)
case H264:
args = append(args,
"--codec", "H264",
"--inline",
"--intra", r.config.IntraRefreshPeriod,
)
2018-05-06 11:38:45 +03:00
if r.config.QuantizationMode == QuantizationOn {
2018-09-19 15:32:50 +03:00
args = append(args, "-qp", r.config.Quantization)
2018-05-05 07:15:07 +03:00
}
case Mjpeg:
args = append(args, "--codec", "MJPEG")
}
r.config.Logger.Log(smartlogger.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)
2018-06-24 15:01:09 +03:00
stdout, err := r.cmd.StdoutPipe()
if err != nil {
return err
}
err = r.cmd.Start()
if err != nil {
return err
}
r.config.Logger.Log(smartlogger.Info, pkg+"reading camera data")
err = r.lexTo(r.encoder, stdout, delay)
r.config.Logger.Log(smartlogger.Info, pkg+"finished reading camera data")
return err
}
2018-06-24 15:01:09 +03:00
// 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)
2018-06-24 15:01:09 +03:00
f, err := os.Open(r.config.InputFileName)
if err != nil {
r.config.Logger.Log(smartlogger.Error, err.Error())
r.Stop()
return err
}
2018-06-24 15:01:09 +03:00
defer f.Close()
// TODO(kortschak): Maybe we want a context.Context-aware parser that we can stop.
return r.lexTo(r.encoder, f, delay)
}