mirror of https://bitbucket.org/ausocean/av.git
530 lines
14 KiB
Go
530 lines
14 KiB
Go
/*
|
|
NAME
|
|
revid - a testbed for re-muxing and re-directing video streams as MPEG-TS over various protocols.
|
|
|
|
DESCRIPTION
|
|
See Readme.md
|
|
|
|
AUTHORS
|
|
Alan Noble <anoble@gmail.com>
|
|
Saxon A. Nelson-Milton <saxon.milton@gmail.com>
|
|
|
|
LICENSE
|
|
revid is Copyright (C) 2017 Alan Noble.
|
|
|
|
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 [GNU licenses](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 (
|
|
"bufio"
|
|
"bytes"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"io/ioutil"
|
|
"net/http"
|
|
"os"
|
|
"os/exec"
|
|
"strconv"
|
|
"time"
|
|
|
|
//"bitbucket.org/ausocean/av/parser"
|
|
//"bitbucket.org/ausocean/av/tsgenerator"
|
|
"../generator"
|
|
"../parser"
|
|
|
|
//"bitbucket.org/ausocean/av/ringbuffer"
|
|
//"bitbucket.org/ausocean/utils/smartLogger"
|
|
"../ringbuffer"
|
|
"../rtmp"
|
|
)
|
|
|
|
// Misc constants
|
|
const (
|
|
clipDuration = 1 // s
|
|
mp2tPacketSize = 188 // MPEG-TS packet size
|
|
mp2tMaxPackets = 2016 * clipDuration // # first multiple of 7 and 8 greater than 2000
|
|
ringBufferSize = 1000 / clipDuration
|
|
ringBufferElementSize = 10000000
|
|
maxClipSize = 100000
|
|
httpTimeOut = 5 // s
|
|
packetsPerFrame = 7
|
|
h264BufferSize = 1000000
|
|
bitrateTime = 60
|
|
mjpegParserInChanLen = 100000
|
|
ffmpegPath = "/home/saxon/bin/ffmpeg"
|
|
)
|
|
|
|
// Log Types
|
|
const (
|
|
Error = "Error"
|
|
Warning = "Warning"
|
|
Info = "Info"
|
|
Debug = "Debug"
|
|
)
|
|
|
|
// RevidInst provides methods to control a revidInst session; providing methods
|
|
// to start, stop and change the state of an instance using the Config struct.
|
|
type RevidInst interface {
|
|
Start()
|
|
Stop()
|
|
ChangeState(newconfig Config) error
|
|
GetConfigRef() *Config
|
|
Log(logType, m string)
|
|
IsRunning() bool
|
|
}
|
|
|
|
// The revidInst struct provides fields to describe the state of a RevidInst.
|
|
type revidInst struct {
|
|
ffmpegPath string
|
|
tempDir string
|
|
ringBuffer ringbuffer.RingBuffer
|
|
config Config
|
|
isRunning bool
|
|
outputFile *os.File
|
|
inputFile *os.File
|
|
generator generator.Generator
|
|
parser parser.Parser
|
|
cmd *exec.Cmd
|
|
ffmpegCmd *exec.Cmd
|
|
inputReader *bufio.Reader
|
|
ffmpegStdin io.WriteCloser
|
|
outputChan chan []byte
|
|
setupInput func() error
|
|
setupOutput func() error
|
|
getFrame func() []byte
|
|
flushData func()
|
|
sendClip func(clip []byte) error
|
|
rtmpSession rtmp.RTMPSession
|
|
}
|
|
|
|
// NewRevidInstance returns a pointer to a new revidInst with the desired
|
|
// configuration, and/or an error if construction of the new instant was not
|
|
// successful.
|
|
func NewRevidInstance(config Config) (r *revidInst, err error) {
|
|
r = new(revidInst)
|
|
r.ringBuffer = ringbuffer.NewRingBuffer(ringBufferSize, ringBufferElementSize)
|
|
err = r.ChangeState(config)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
r.outputChan = make(chan []byte, 10000)
|
|
r.parser.Start()
|
|
go r.packClips()
|
|
r.Log(Info, "New revid instance created! config is:")
|
|
r.Log(Info, fmt.Sprintf("%v", r.config))
|
|
return
|
|
}
|
|
|
|
// GetConfigRef returns a pointer to the revidInst's Config struct object
|
|
func (r *revidInst) GetConfigRef() *Config {
|
|
return &r.config
|
|
}
|
|
|
|
// ChangeState swaps the current config of a revidInst with the passed
|
|
// configuration; checking validity and returning errors if not valid.
|
|
func (r *revidInst) ChangeState(config Config) error {
|
|
r.config.Logger = config.Logger
|
|
err := config.Validate(r)
|
|
if err != nil {
|
|
return errors.New("Config struct is bad!: " + err.Error())
|
|
}
|
|
r.config = config
|
|
switch r.config.Output {
|
|
case File:
|
|
r.sendClip = r.sendClipToFile
|
|
r.setupOutput = r.setupOutputForFile
|
|
case Rtmp:
|
|
switch r.config.RtmpMethod {
|
|
case Ffmpeg:
|
|
r.setupOutput = r.setupOutputForFfmpegRtmp
|
|
r.sendClip = r.sendClipToFfmpegRtmp
|
|
case LibRtmp:
|
|
r.setupOutput = r.setupOutputForLibRtmp
|
|
r.sendClip = r.sendClipToLibRtmp
|
|
}
|
|
case Http:
|
|
r.sendClip = r.sendClipToHTTP
|
|
}
|
|
switch r.config.Input {
|
|
case Raspivid:
|
|
r.setupInput = r.setupInputForRaspivid
|
|
case File:
|
|
r.setupInput = r.setupInputForFile
|
|
}
|
|
switch r.config.InputCodec {
|
|
case H264:
|
|
r.Log(Info, "Using H264 parser!")
|
|
r.parser = parser.NewH264Parser()
|
|
fmt.Println("here")
|
|
case Mjpeg:
|
|
r.Log(Info, "Using MJPEG parser!")
|
|
r.parser = parser.NewMJPEGParser(mjpegParserInChanLen)
|
|
}
|
|
if r.config.Packetization == None {
|
|
r.parser.SetOutputChan(r.outputChan)
|
|
r.getFrame = r.getFrameNoPacketization
|
|
} else {
|
|
switch r.config.Packetization {
|
|
case Mpegts:
|
|
frameRateAsInt, _ := strconv.Atoi(r.config.FrameRate)
|
|
r.generator = generator.NewTsGenerator(uint(frameRateAsInt))
|
|
case Flv:
|
|
frameRateAsInt, _ := strconv.Atoi(r.config.FrameRate)
|
|
r.generator = generator.NewFlvGenerator(false, true, uint(frameRateAsInt))
|
|
}
|
|
r.getFrame = r.getFramePacketization
|
|
r.parser.SetOutputChan(r.generator.GetInputChan())
|
|
r.generator.Start()
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Log takes a logtype and message and tries to send this information to the
|
|
// logger provided in the revidInst config - if there is one, otherwise the message
|
|
// is sent to stdout
|
|
func (r *revidInst) Log(logType, m string) {
|
|
if r.config.Logger != nil {
|
|
r.config.Logger.Log(logType, m)
|
|
} else {
|
|
fmt.Println(logType + ": " + m)
|
|
}
|
|
}
|
|
|
|
// IsRunning returns true if the revidInst is currently running and false otherwise
|
|
func (r *revidInst) IsRunning() bool {
|
|
return r.isRunning
|
|
}
|
|
|
|
// Start invokes a revidInst to start processing video from a defined input
|
|
// and packetising to a defined output.
|
|
func (r *revidInst) Start() {
|
|
if r.isRunning {
|
|
r.Log(Warning, "revidInst.Start() called but revid already running!")
|
|
return
|
|
}
|
|
r.Log(Info, "Starting Revid!")
|
|
r.setupOutput()
|
|
r.setupInput()
|
|
go r.outputClips()
|
|
r.isRunning = true
|
|
}
|
|
|
|
// Stop halts any processing of video data from a camera or file
|
|
func (r *revidInst) Stop() {
|
|
if r.isRunning {
|
|
r.Log(Info, "Stopping revid!")
|
|
r.isRunning = false
|
|
if r.cmd != nil {
|
|
r.cmd.Process.Kill()
|
|
}
|
|
} else {
|
|
r.Log(Warning, "revidInst.Stop() called but revid not running!")
|
|
}
|
|
}
|
|
|
|
// Start invokes a revidInst to start processing video from a defined input
|
|
// and packetising to a defined output.
|
|
func (r *revidInst) getFrameNoPacketization() []byte {
|
|
return <-r.outputChan
|
|
}
|
|
|
|
// Start invokes a revidInst to start processing video from a defined input
|
|
// and packetising to a defined output.
|
|
func (r *revidInst) getFramePacketization() []byte {
|
|
return <-(r.generator.GetOutputChan())
|
|
}
|
|
|
|
// Start invokes a revidInst to start processing video from a defined input
|
|
// and packetising to a defined output.
|
|
func (r *revidInst) flushDataNoPacketisation() {
|
|
for len(r.outputChan) > 0 {
|
|
<-(r.outputChan)
|
|
}
|
|
}
|
|
|
|
// Start invokes a revidInst to start processing video from a defined input
|
|
func (r *revidInst) flushDataMpegtsPacketisation() {
|
|
for len(r.generator.GetOutputChan()) > 0 {
|
|
<-(r.generator.GetOutputChan())
|
|
}
|
|
}
|
|
|
|
// packClips takes data segments; whether that be tsPackets or jpeg frames and
|
|
// packs them into clips 1s long.
|
|
func (r *revidInst) packClips() {
|
|
clipSize := 0
|
|
packetCount := 0
|
|
for {
|
|
if clip, err := r.ringBuffer.Get(); err != nil {
|
|
r.Log(Error, err.Error())
|
|
r.Log(Warning, "Clearing output chan!")
|
|
r.flushData()
|
|
} else {
|
|
for {
|
|
frame := r.getFrame()
|
|
lenOfFrame := len(frame)
|
|
upperBound := clipSize + lenOfFrame
|
|
copy(clip[clipSize:upperBound], frame)
|
|
packetCount++
|
|
clipSize += lenOfFrame
|
|
if packetCount >= r.config.FramesPerClip {
|
|
if err := r.ringBuffer.DoneWriting(clipSize); err != nil {
|
|
r.Log(Error, err.Error())
|
|
r.Log(Warning, "Dropping clip!")
|
|
}
|
|
clipSize = 0
|
|
packetCount = 0
|
|
break
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// outputClips takes the clips produced in the packClips method and outputs them
|
|
// to the desired output defined in the revidInst config
|
|
func (r *revidInst) outputClips() {
|
|
now := time.Now()
|
|
prevTime := now
|
|
bytes := 0
|
|
delay := 0
|
|
for r.isRunning {
|
|
// Here we slow things down as much as we can to decrease cpu usage
|
|
switch {
|
|
case r.ringBuffer.GetNoOfElements() < 2:
|
|
delay++
|
|
time.Sleep(time.Duration(delay) * time.Millisecond)
|
|
case delay > 10:
|
|
delay -= 10
|
|
}
|
|
// If the ringbuffer has something we can read and send off
|
|
if clip, err := r.ringBuffer.Read(); err == nil {
|
|
r.Log(Debug, fmt.Sprintf("Delay is: %v\n", delay))
|
|
r.Log(Debug, fmt.Sprintf("Ring buffer size: %v\n", r.ringBuffer.GetNoOfElements()))
|
|
// Output clip to the output specified in the configuration struct
|
|
bytes += len(clip)
|
|
for err := r.sendClip(clip); err != nil; {
|
|
r.Log(Error, err.Error())
|
|
r.Log(Warning, "Send failed trying again!")
|
|
err = r.sendClip(clip)
|
|
}
|
|
if err := r.ringBuffer.DoneReading(); err != nil {
|
|
r.Log(Error, err.Error())
|
|
}
|
|
// Log some information regarding bitrate and ring buffer size if it's time
|
|
now = time.Now()
|
|
deltaTime := now.Sub(prevTime)
|
|
if deltaTime > time.Duration(bitrateTime)*time.Second {
|
|
r.Log(Info, fmt.Sprintf("Bitrate: %v bits/s\n", int64(float64(bytes*8)/float64(deltaTime/1e9))))
|
|
r.Log(Info, fmt.Sprintf("Ring buffer size: %v\n", r.ringBuffer.GetNoOfElements()))
|
|
prevTime = now
|
|
bytes = 0
|
|
}
|
|
} else {
|
|
r.Log(Debug, err.Error())
|
|
time.Sleep(1*time.Second)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Start invokes a revidInst to start processing video from a defined input
|
|
// and packetising to a defined output.
|
|
func (r *revidInst) sendClipToFile(clip []byte) error {
|
|
_,err := r.outputFile.Write(clip)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// sendClipToHTTP takes a clip and an output url and posts through http.
|
|
func (r *revidInst) sendClipToHTTP(clip []byte) error {
|
|
timeout := time.Duration(httpTimeOut * time.Second)
|
|
client := http.Client{
|
|
Timeout: timeout,
|
|
}
|
|
url := r.config.HttpAddress+strconv.Itoa(len(clip))
|
|
r.Log(Debug, fmt.Sprintf("Posting %s (%d bytes)\n", url, len(clip)))
|
|
resp, err := client.Post(url, "video/mp2t", bytes.NewReader(clip)) // lighter than NewBuffer
|
|
if err != nil {
|
|
return fmt.Errorf("Error posting to %s: %s", url, err)
|
|
}
|
|
defer resp.Body.Close()
|
|
body, err := ioutil.ReadAll(resp.Body)
|
|
if err == nil {
|
|
r.Log(Debug, fmt.Sprintf("%s\n", body))
|
|
} else {
|
|
r.Log(Error, err.Error())
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Start invokes a revidInst to start processing video from a defined input
|
|
// and packetising to a defined output.
|
|
func (r *revidInst) sendClipToFfmpegRtmp(clip []byte) error {
|
|
fmt.Println("Outputting!")
|
|
_, err := r.ffmpegStdin.Write(clip)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *revidInst) sendClipToLibRtmp(clip []byte) error {
|
|
return r.rtmpSession.WriteFrame(clip,len(clip))
|
|
}
|
|
|
|
// Start invokes a revidInst to start processing video from a defined input
|
|
// and packetising to a defined output.
|
|
func (r *revidInst) setupOutputForFfmpegRtmp() error {
|
|
r.ffmpegCmd = exec.Command(ffmpegPath,
|
|
"-f", "h264",
|
|
"-r", r.config.FrameRate,
|
|
"-i", "-",
|
|
"-f", "lavfi",
|
|
"-i", "aevalsrc=0",
|
|
"-fflags", "nobuffer",
|
|
"-vcodec", "copy",
|
|
"-acodec", "aac",
|
|
"-map", "0:0",
|
|
"-map", "1:0",
|
|
"-strict", "experimental",
|
|
"-f", "flv",
|
|
r.config.RtmpUrl,
|
|
)
|
|
var err error
|
|
r.ffmpegStdin, err = r.ffmpegCmd.StdinPipe()
|
|
if err != nil {
|
|
r.Log(Error, err.Error())
|
|
r.Stop()
|
|
return err
|
|
}
|
|
err = r.ffmpegCmd.Start()
|
|
if err != nil {
|
|
r.Log(Error, err.Error())
|
|
r.Stop()
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *revidInst) setupOutputForLibRtmp() error {
|
|
r.rtmpSession = NewRTMPSession(r.config.RtmpUrl, rtmpConnectionTimout)
|
|
return rtmp.StartSession()
|
|
}
|
|
|
|
func (r *revidInst) setupOutputForFile() (err error) {
|
|
r.outputFile, err = os.Create(r.config.OutputFileName)
|
|
return
|
|
}
|
|
|
|
// Start invokes a revidInst to start processing video from a defined input
|
|
// and packetising to a defined output.
|
|
func (r *revidInst) setupInputForRaspivid() error {
|
|
r.Log(Info, "Starting raspivid!")
|
|
switch r.config.InputCodec {
|
|
case H264:
|
|
r.cmd = exec.Command("raspivid",
|
|
"-cd", "H264",
|
|
"-o", "-",
|
|
"-n",
|
|
"-t", r.config.Timeout,
|
|
"-b", r.config.Bitrate,
|
|
"-qp", r.config.Quantization,
|
|
"-w", r.config.Width,
|
|
"-h", r.config.Height,
|
|
"-fps", r.config.FrameRate,
|
|
"-ih",
|
|
"-g", r.config.IntraRefreshPeriod,
|
|
)
|
|
case Mjpeg:
|
|
r.cmd = exec.Command("raspivid",
|
|
"-cd", "MJPEG",
|
|
"-o", "-",
|
|
"-n",
|
|
"-t", r.config.Timeout,
|
|
"-fps", r.config.FrameRate,
|
|
)
|
|
}
|
|
stdout, _ := r.cmd.StdoutPipe()
|
|
err := r.cmd.Start()
|
|
r.inputReader = bufio.NewReader(stdout)
|
|
if err != nil {
|
|
r.Log(Error, err.Error())
|
|
return err
|
|
}
|
|
go r.readCamera()
|
|
return nil
|
|
}
|
|
|
|
// Start invokes a revidInst to start processing video from a defined input
|
|
func (r *revidInst) setupInputForFile() error {
|
|
go r.readFile()
|
|
return nil
|
|
}
|
|
|
|
// readCamera reads data from the defined camera while the revidInst is running.
|
|
func (r *revidInst) readCamera() {
|
|
r.Log(Info, "Reading camera data!")
|
|
for r.isRunning {
|
|
data := make([]byte, 1)
|
|
_, err := io.ReadFull(r.inputReader, data)
|
|
switch {
|
|
case err != nil && err.Error() == "EOF" && r.isRunning:
|
|
r.Log(Error, "No data from camera!")
|
|
time.Sleep(5 * time.Second)
|
|
case err != nil && r.isRunning:
|
|
r.Log(Error, err.Error())
|
|
default:
|
|
r.parser.GetInputChan() <- data[0]
|
|
}
|
|
}
|
|
r.Log(Info, "Out of reading routine!")
|
|
}
|
|
|
|
// readFile reads data from the defined file while the revidInst is running.
|
|
func (r *revidInst) readFile() error {
|
|
if len(r.parser.GetInputChan()) == 0 {
|
|
var err error
|
|
r.inputFile, err = os.Open(r.config.InputFileName)
|
|
if err != nil {
|
|
r.Log(Error, err.Error())
|
|
r.Stop()
|
|
return err
|
|
}
|
|
stats, err := r.inputFile.Stat()
|
|
if err != nil {
|
|
r.Log(Error, "Could not get input file stats!")
|
|
r.Stop()
|
|
return err
|
|
}
|
|
data := make([]byte, stats.Size())
|
|
_, err = r.inputFile.Read(data)
|
|
if err != nil {
|
|
r.Log(Error, err.Error())
|
|
r.Stop()
|
|
return err
|
|
}
|
|
for i := range data {
|
|
r.parser.GetInputChan() <- data[i]
|
|
}
|
|
r.inputFile.Close()
|
|
}
|
|
return nil
|
|
}
|