av/revid/RevidInstance.go

349 lines
8.6 KiB
Go
Raw Normal View History

/*
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"
"fmt"
"io"
"io/ioutil"
"log"
"net"
"net/http"
"os"
"os/exec"
"strconv"
"time"
2018-01-11 09:49:33 +03:00
"bitbucket.org/ausocean/av/h264"
"bitbucket.org/ausocean/av/tsgenerator"
2018-01-10 06:57:56 +03:00
"bitbucket.org/ausocean/av/ringbuffer"
2018-01-22 08:46:56 +03:00
"bitbucket.org/ausocean/utils/smartLogger"
)
// defaults and networking consts
const (
clipDuration = 1 // s
defaultPID = 256
defaultFrameRate = 25
mp2tPacketSize = 188 // MPEG-TS packet size
mp2tMaxPackets = 2016 * clipDuration // # first multiple of 7 and 8 greater than 2000
udpPackets = 7 // # of UDP packets per ethernet frame (8 is the max)
rtpPackets = 7 // # of RTP packets per ethernet frame (7 is the max)
rtpHeaderSize = 12
rtpSSRC = 1 // any value will do
bufferSize = 100 / clipDuration
2018-01-10 04:32:16 +03:00
httpTimeOut = 5 // s
motionThreshold = "0.0025"
qscale = "3"
2018-01-10 04:32:16 +03:00
defaultRaspividCmd = "raspivid -o -"
2018-01-11 07:55:59 +03:00
framesPerSec = 25
packetsPerFrame = 7
h264BufferSize = 500000
2018-01-22 08:46:56 +03:00
bitrateTime = 60
2018-01-10 06:57:56 +03:00
)
const (
Raspivid = 0
Rtp = 1
H264Codec = 2
File = 4
HttpOut = 5
)
var cmd *exec.Cmd
var inputReader *bufio.Reader
2018-01-10 06:57:56 +03:00
type Config struct {
2018-01-11 07:55:59 +03:00
Input uint8
InputCmd string
Output uint8
2018-01-10 06:57:56 +03:00
OutputFileName string
InputFileName string
Height string
Width string
Bitrate string
2018-01-16 08:49:18 +03:00
FrameRate string
HttpAddress string
2018-01-23 06:04:37 +03:00
Quantization string
2018-01-22 08:46:56 +03:00
Logger smartLogger.LogInstance
2018-01-10 06:57:56 +03:00
}
type RevidInst interface {
2018-01-10 06:57:56 +03:00
Start()
Stop()
2018-01-23 06:15:06 +03:00
ChangeState(newconfig Config) error
GetConfigRef() *Config
}
type revidInst struct {
2018-01-10 04:32:16 +03:00
expectCC int
dumpCC int
dumpPCRBase uint64
conn net.Conn
ffmpegPath string
tempDir string
ringBuffer ringbuffer.RingBuffer
2018-01-23 06:15:06 +03:00
config Config
2018-01-10 04:32:16 +03:00
isRunning bool
Error *log.Logger
2018-01-10 06:57:56 +03:00
outputFile *os.File
2018-01-11 07:55:59 +03:00
inputFile *os.File
generator tsgenerator.TsGenerator
2018-01-16 08:49:18 +03:00
h264Parser h264.H264Parser
}
2018-01-23 06:17:02 +03:00
func (r *revidInst) GetConfigRef() *Config{
2018-01-23 06:15:06 +03:00
return &r.config
}
func NewRevidInstance(config Config) (r *revidInst, err error) {
2018-01-10 06:57:56 +03:00
r = new(revidInst)
r.ringBuffer = ringbuffer.NewRingBuffer(bufferSize, mp2tPacketSize*mp2tMaxPackets)
r.Error = log.New(os.Stderr, "ERROR: ", log.Ldate|log.Ltime|log.Lshortfile)
2018-01-09 09:35:24 +03:00
r.expectCC = -1
r.dumpCC = -1
r.dumpPCRBase = 0
2018-01-23 06:15:06 +03:00
r.ChangeState(config)
switch r.config.Output {
case File:
2018-01-23 06:15:06 +03:00
r.outputFile, err = os.Create(r.config.OutputFileName)
2018-01-10 04:32:16 +03:00
if err != nil {
return nil, err
}
}
2018-01-23 06:15:06 +03:00
switch r.config.Input {
case File:
2018-01-23 06:15:06 +03:00
r.inputFile, err = os.Open(r.config.InputFileName)
2018-01-11 07:55:59 +03:00
if err != nil {
return nil, err
}
}
r.generator = tsgenerator.NewTsGenerator(framesPerSec)
r.h264Parser = h264.H264Parser{OutputChan: r.generator.GetNalInputChan()}
r.h264Parser.InputByteChan = make(chan byte, 10000)
// TODO: Need to create constructor for parser otherwise I'm going to break
// something eventuallyl
go r.h264Parser.Parse()
go r.input()
2018-01-17 06:48:47 +03:00
go r.generator.Generate()
2018-01-23 06:15:06 +03:00
r.ErrOut("New revid instance created! config is:")
r.ErrOut(fmt.Sprintf("%v",r.config))
2018-01-10 06:57:56 +03:00
return
2018-01-10 04:32:16 +03:00
}
2018-01-23 06:15:06 +03:00
func (r *revidInst) ChangeState(newconfig Config) error {
// TODO: check that the config is G
r.config = newconfig
2018-01-10 04:32:16 +03:00
return nil
}
func (r *revidInst) ErrOut(m string){
2018-01-23 06:15:06 +03:00
r.config.Logger.Log("Debug",m)
}
func (r *revidInst)IsRunning() bool {
return r.isRunning
}
2018-01-10 06:57:56 +03:00
func (r *revidInst) Start() {
if r.isRunning {
r.ErrOut("Start() has been called but revid already running!")
return
}
2018-01-23 07:16:46 +03:00
r.ErrOut("Starting Revid!")
2018-01-23 06:15:06 +03:00
switch r.config.Input {
case Raspivid:
2018-01-22 08:46:56 +03:00
r.ErrOut("Starting raspivid!")
2018-01-16 08:49:18 +03:00
cmd = exec.Command("raspivid", "-o", "-", "-n", "-t", "0", "-b",
2018-01-23 06:15:06 +03:00
r.config.Bitrate,"-qp", r.config.Quantization, "-w", r.config.Width, "-h", r.config.Height, "-fps", r.config.FrameRate, "-ih", "-g", "100")
stdout, _ := cmd.StdoutPipe()
2018-01-10 04:32:16 +03:00
err := cmd.Start()
inputReader = bufio.NewReader(stdout)
if err != nil {
r.ErrOut(err.Error())
return
}
case File:
default:
2018-01-22 08:46:56 +03:00
r.ErrOut("Input not valid!")
}
2018-01-22 08:46:56 +03:00
r.isRunning = true
2018-01-11 09:49:33 +03:00
var h264Data []byte
2018-01-23 06:15:06 +03:00
switch r.config.Input {
case Raspivid:
go func() {
2018-01-22 08:46:56 +03:00
r.ErrOut("Reading camera data!")
for r.isRunning {
h264Data = make([]byte, 1)
_, err := io.ReadFull(inputReader, h264Data)
if err != nil {
2018-01-23 08:34:48 +03:00
if err.Error() == "EOF" && r.isRunning {
r.ErrOut("No data from camera!")
time.Sleep(5*time.Second)
} else {
r.ErrOut(err.Error())
}
} else {
r.h264Parser.InputByteChan <- h264Data[0]
2018-01-11 07:55:59 +03:00
}
}
2018-01-23 08:25:31 +03:00
r.ErrOut("Out of reading routine")
2018-01-11 09:49:33 +03:00
}()
case File:
2018-01-11 09:49:33 +03:00
stats, err := r.inputFile.Stat()
if err != nil {
2018-01-22 08:46:56 +03:00
r.ErrOut("Could not get file stats!")
return
}
2018-01-11 09:49:33 +03:00
h264Data = make([]byte, stats.Size())
_, err = r.inputFile.Read(h264Data)
if err != nil {
r.ErrOut(err.Error())
2018-01-11 09:49:33 +03:00
}
for i := range h264Data {
r.h264Parser.InputByteChan <- h264Data[i]
2018-01-11 09:49:33 +03:00
}
}
go r.output()
}
func (r *revidInst) Stop() {
if r.isRunning {
2018-01-22 08:46:56 +03:00
r.ErrOut("Stopping revid!")
r.isRunning = false
cmd.Process.Kill()
}
}
func (r *revidInst) input() {
clipSize := 0
packetCount := 0
now := time.Now()
prevTime := now
for {
2018-01-10 06:57:56 +03:00
if clip, err := r.ringBuffer.Get(); err != nil {
2018-01-22 08:46:56 +03:00
r.ErrOut(err.Error())
r.Stop()
return
} else {
for {
tsPacket := <-(r.generator.GetTsOutputChan())
byteSlice, err := tsPacket.ToByteSlice()
if err != nil {
r.ErrOut(err.Error())
2018-01-10 06:57:56 +03:00
}
upperBound := clipSize + mp2tPacketSize
copy(clip[clipSize:upperBound], byteSlice)
packetCount++
clipSize += mp2tPacketSize
// send if (1) our buffer is full or (2) 1 second has elapsed and we have % packetsPerFrame
now = time.Now()
if (packetCount == mp2tMaxPackets) ||
(now.Sub(prevTime) > clipDuration*time.Second && packetCount%packetsPerFrame == 0) {
2018-01-10 06:57:56 +03:00
if err := r.ringBuffer.DoneWriting(clipSize); err != nil {
2018-01-22 08:46:56 +03:00
r.ErrOut(err.Error())
r.ErrOut("Dropping that clip!")
}
clipSize = 0
packetCount = 0
prevTime = now
break
}
}
}
}
}
2018-01-10 06:57:56 +03:00
func (r *revidInst) output() {
now := time.Now()
prevTime := now
bytes := 0
2018-01-22 08:46:56 +03:00
delay := 0
for r.isRunning {
2018-01-22 08:46:56 +03:00
if r.ringBuffer.GetNoOfElements() < 2 {
delay++
time.Sleep(time.Duration(delay)*time.Millisecond)
} else {
if delay > 10 {
delay -= 10
}
2018-01-23 06:04:37 +03:00
}
2018-01-10 06:57:56 +03:00
if clip, err := r.ringBuffer.Read(); err == nil {
2018-01-22 08:46:56 +03:00
r.ErrOut(fmt.Sprintf("Delay is: %v\n", delay))
r.ErrOut(fmt.Sprintf("Ring buffer size: %v\n", r.ringBuffer.GetNoOfElements()))
2018-01-23 06:15:06 +03:00
switch r.config.Output {
case File:
2018-01-10 04:32:16 +03:00
r.outputFile.Write(clip)
case HttpOut:
bytes += len(clip)
2018-01-23 06:15:06 +03:00
for err := r.sendClipToHTTP(clip, r.config.HttpAddress); err != nil; {
2018-01-22 08:46:56 +03:00
r.ErrOut("Post failed trying again!")
2018-01-23 06:15:06 +03:00
err = r.sendClipToHTTP(clip, r.config.HttpAddress)
}
2018-01-10 04:32:16 +03:00
default:
r.ErrOut("No output defined!")
}
2018-01-10 06:57:56 +03:00
if err := r.ringBuffer.DoneReading(); err != nil {
r.ErrOut(err.Error())
}
now = time.Now()
deltaTime := now.Sub(prevTime)
if deltaTime > time.Duration(bitrateTime)*time.Second {
2018-01-22 08:46:56 +03:00
fmt.Printf("Bitrate: %v bytes/s\n", int64(float64(bytes) / float64(deltaTime/1e9)))
prevTime = now
bytes = 0
}
}
}
}
// sendClipToHTPP posts a video clip via HTTP, using a new TCP connection each time.
func (r *revidInst)sendClipToHTTP(clip []byte, output string) error {
timeout := time.Duration(httpTimeOut * time.Second)
client := http.Client{
Timeout: timeout,
}
2018-01-22 08:46:56 +03:00
url := output + strconv.Itoa(len(clip))
2018-01-23 07:57:15 +03:00
fmt.Printf("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", output, err)
}
defer resp.Body.Close()
body, err := ioutil.ReadAll(resp.Body)
2018-01-23 07:57:15 +03:00
if err == nil {
fmt.Printf("%s\n", body)
}
2018-01-10 06:57:56 +03:00
return nil
}