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

Merge branch 'RevidAPI' of https://bitbucket.org/ausocean/av into RevidAPI

This commit is contained in:
Saxon Milton 2018-01-11 22:37:07 +10:30
parent a87f59db34 86dc14fa7a
commit 1d856ad82d
18 changed files with 536 additions and 1805 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

104
camstreamer/camstreamer.go
View File

@ -63,3 +63,107 @@ func (cs *CamStreamer)Connect()(session *RtpSession, err error){
// let's create a session that will store useful stuff from the connections
rtpSession := NewSession(rtpConn, rtcpConn)
}
/*******************************************************
Testing stuff related to connection i.e. rtsp, rtp, rtcp
********************************************************/
const (
rtpPort = 17300
rtcpPort = 17319
rtspUrl = "rtsp://192.168.0.50:8554/CH002.sdp"
rtpUrl = "rtsp://192.168.0.50:8554/CH002.sdp/track1"
)
/* Let's see if we can connect to an rtsp device then read an rtp stream,
and then convert the rtp packets to mpegts packets and output. */
func TestRTSP(t *testing.T) {
sess := rtsp.NewSession()
res, err := sess.Options(rtspUrl)
if err != nil {
t.Errorf("Shouldn't have got error: %v\n", err)
}
res, err = sess.Describe(rtspUrl)
if err != nil {
log.Fatalln(err)
t.Errorf("Shouldn't have got error: %v\n", err)
}
p, err := rtsp.ParseSdp(&io.LimitedReader{R: res.Body, N: res.ContentLength})
if err != nil {
t.Errorf("Shouldn't have got error: %v\n", err)
}
log.Printf("%+v", p)
res, err = sess.Setup(rtpUrl, fmt.Sprintf("RTP/AVP;unicast;client_port=%d-%d", rtpPort, rtcpPort))
if err != nil {
t.Errorf("Shouldn't have got error: %v\n", err)
}
log.Println(res)
res, err = sess.Play(rtspUrl, res.Header.Get("Session"))
if err != nil {
t.Errorf("Shouldn't have got error: %v\n", err)
}
log.Println(res)
}
func TestRTP(t *testing.T) {
sess := rtsp.NewSession()
res, err := sess.Options(rtspUrl)
if err != nil {
t.Errorf("Shouldn't have got error: %v\n", err)
}
res, err = sess.Describe(rtspUrl)
if err != nil {
log.Fatalln(err)
t.Errorf("Shouldn't have got error: %v\n", err)
}
p, err := rtsp.ParseSdp(&io.LimitedReader{R: res.Body, N: res.ContentLength})
if err != nil {
t.Errorf("Shouldn't have got error: %v\n", err)
}
log.Printf("%+v", p)
res, err = sess.Setup(rtpUrl, fmt.Sprintf("RTP/AVP;unicast;client_port=%d-%d", rtpPort, rtcpPort))
if err != nil {
t.Errorf("Shouldn't have got error: %v\n", err)
}
log.Println(res)
res, err = sess.Play(rtspUrl, res.Header.Get("Session"))
if err != nil {
t.Errorf("Shouldn't have got error: %v\n", err)
}
log.Println(res)
// create udp connection for rtp stuff
rtpLaddr, err := net.ResolveUDPAddr("udp", "192.168.0.109:17300")
if err != nil {
t.Errorf("Local rtp addr not set! %v\n", err)
}
rtpAddr, err := net.ResolveUDPAddr("udp", "192.168.0.50:17300")
if err != nil {
t.Errorf("Resolving rtp address didn't work! %v\n", err)
}
rtpConn, err := net.DialUDP("udp", rtpLaddr, rtpAddr)
if err != nil {
t.Errorf("Conncection not established! %v\n", err)
}
// Create udp connection for rtcp stuff
rtcpLaddr, err := net.ResolveUDPAddr("udp", "192.168.0.109:17319")
if err != nil {
t.Errorf("Local RTCP address not resolved! %v\n", err)
}
rtcpAddr, err := net.ResolveUDPAddr("udp", "192.168.0.50:17301")
if err != nil {
t.Errorf("Remote RTCP address not resolved! %v\n", err)
}
rtcpConn, err := net.DialUDP("udp", rtcpLaddr, rtcpAddr)
if err != nil {
t.Errorf("Connection not established! %v\n", err)
}
// let's create a session that will store useful stuff from the connections
rtpSession := NewSession(rtpConn, rtcpConn)
time.Sleep(2 * time.Second)
select {
default:
t.Errorf("Should have got rtpPacket!")
case rtpPacket := <-rtpSession.RtpChan:
fmt.Printf("RTP packet: %v\n", rtpPacket)
}
}

62
h264/H264Writer.go
View File

@ -26,11 +26,16 @@ LICENSE
along with revid in gpl.txt. If not, see [GNU licenses](http://www.gnu.org/licenses).
*/
package packets
package h264
import (
_"fmt"
"os"
"bitbucket.org/ausocean/av/mpegts"
"bitbucket.org/ausocean/av/rtp"
"bitbucket.org/ausocean/av/tools"
"bitbucket.org/ausocean/av/itut"
)
type RtpToH264Converter interface {
@ -38,23 +43,22 @@ type RtpToH264Converter interface {
}
type rtpToH264Converter struct {
TsChan <-chan *MpegTsPacket
tsChan chan<- *MpegTsPacket
InputChan chan<- RtpPacket
inputChan <-chan RtpPacket
currentTsPacket *MpegTsPacket
TsChan <-chan *mpegts.MpegTsPacket
tsChan chan<- *mpegts.MpegTsPacket
InputChan chan<- rtp.RtpPacket
inputChan <-chan rtp.RtpPacket
currentTsPacket *mpegts.MpegTsPacket
payloadByteChan chan byte
currentCC byte
}
//func parseH264File()
func NewRtpToH264Converter() (c *rtpToH264Converter) {
c = new(rtpToH264Converter)
tsChan := make(chan *MpegTsPacket,100)
tsChan := make(chan *mpegts.MpegTsPacket,100)
c.TsChan = tsChan
c.tsChan = tsChan
inputChan := make(chan RtpPacket,100)
inputChan := make(chan rtp.RtpPacket,100)
c.InputChan = inputChan
c.inputChan = inputChan
c.currentCC = 0
@ -63,7 +67,7 @@ func NewRtpToH264Converter() (c *rtpToH264Converter) {
func (c* rtpToH264Converter) Convert() {
file,_ := os.Create("video")
var rtpBuffer [](*RtpPacket)
var rtpBuffer [](*rtp.RtpPacket)
for {
select {
default:
@ -83,7 +87,7 @@ func (c* rtpToH264Converter) Convert() {
}
if len(rtpBuffer) > 200 {
// Discard everything before a type 7
for GetOctectType(rtpBuffer[0]) != 7 {
for tools.GetOctectType(rtpBuffer[0]) != 7 {
rtpBuffer = rtpBuffer[1:]
}
// get sps
@ -99,26 +103,22 @@ func (c* rtpToH264Converter) Convert() {
copy(sei[:],rtpBuffer[0].Payload[:])
rtpBuffer = rtpBuffer[1:]
// while we haven't reached the next sps in the buffer
for GetOctectType(rtpBuffer[0]) != 7 {
switch(GetOctectType(rtpBuffer[0])){
for tools.GetOctectType(rtpBuffer[0]) != 7 {
switch(tools.GetOctectType(rtpBuffer[0])){
case 28:
if GetStartBit(rtpBuffer[0]) == 1{
if tools.GetStartBit(rtpBuffer[0]) == 1{
var buffer []byte
buffer = append(buffer, []byte{0x00,0x00,0x01}...)
buffer = append(buffer, []byte{0x09,0x10}...)
buffer = append(buffer, []byte{0x00,0x00,0x01}...)
buffer = append(buffer, sps...)
buffer = append(buffer, []byte{0x00,0x00,0x01}...)
buffer = append(buffer, pps...)
buffer = append(buffer, []byte{0x00,0x00,0x01}...)
buffer = append(buffer, sei...)
buffer = append(buffer, []byte{0x00,0x00,0x01}...)
buffer = append(buffer, append(itut.StartCode1(),itut.AUD()...)...)
buffer = append(buffer, append(itut.StartCode1(),sps...)...)
buffer = append(buffer, append(itut.StartCode1(),pps...)...)
buffer = append(buffer, append(itut.StartCode1(),sei...)...)
buffer = append(buffer, itut.StartCode1()...)
buffer = append(buffer, rtpBuffer[0].Payload[0] & 0xE0 | rtpBuffer[0].Payload[1] & 0x1F )
buffer = append(buffer, rtpBuffer[0].Payload[2:]...)
rtpBuffer = rtpBuffer[1:]
for {
buffer = append(buffer, rtpBuffer[0].Payload[2:]...)
if getEndBit(rtpBuffer[0]) == 1 {
if tools.GetEndBit(rtpBuffer[0]) == 1 {
rtpBuffer = rtpBuffer[1:]
file.Write(buffer)
break
@ -128,15 +128,11 @@ func (c* rtpToH264Converter) Convert() {
}
case 1:
var buffer []byte
buffer = append(buffer, []byte{0x00,0x00,0x01}...)
buffer = append(buffer, []byte{0x09,0x10}...)
buffer = append(buffer, []byte{0x00,0x00,0x01}...)
buffer = append(buffer, sps...)
buffer = append(buffer, []byte{0x00,0x00,0x01}...)
buffer = append(buffer, pps...)
buffer = append(buffer, []byte{0x00,0x00,0x01}...)
buffer = append(buffer, sei...)
buffer = append(buffer, []byte{0x00,0x00,0x01}...)
buffer = append(buffer, append(itut.StartCode1(), itut.AUD()...)...)
buffer = append(buffer, append(itut.StartCode1(), sps...)...)
buffer = append(buffer, append(itut.StartCode1(),pps...)...)
buffer = append(buffer, append(itut.StartCode1(),sei...)...)
buffer = append(buffer, itut.StartCode1()...)
buffer = append(buffer, rtpBuffer[0].Payload[0] & 0xE0 | rtpBuffer[0].Payload[1] & 0x1F )
buffer = append(buffer, rtpBuffer[0].Payload[2:]...)
rtpBuffer = rtpBuffer[1:]

77
h264/h264Parser.go
View File

@ -30,47 +30,56 @@ package h264
import (
"../itut"
"log"
"sync"
_"fmt"
_"time"
)
type H264Parser {
const (
acceptedLength = 1000
)
var (
Info *log.Logger
mutex *sync.Mutex
)
type H264Parser struct {
inputBuffer []byte
isParsing bool
}
func (p* H264Parser)SendInputData(someData []byte){
inputBuffer = append(input, someData)
OutputChan chan<- []byte
InputByteChan chan byte
}
func (p* H264Parser)Stop(){
isParsing = false
p.isParsing = false
}
func (p* H264Parser)Parse(outputChan chan<- []byte) {
isParsing = true
buffer = b.InputBuffer
for isParsing {
for i := 0 ;; i++{
var start bool
i, start = func() (int,bool) {
switch{
case reflect.DeepEqual(buffer[i:i+3],itut.StartCode1()):
return i+3, true
case reflect.DeepEqual(buffer[i:i+4],itut.StartCode2()):
return i+4, true
}
return i, false
}()
if nalType := buffer[i] & 0x1F; start && ( nalType == 1 || nalType == 5) {
for ; i < len(buffer) && !(i+3 < len(buffer) && ( reflect.DeepEqual(buffer[i:i+3],itut.StartCode1()) ||
reflect.DeepEqual(buffer[i:i+4],startCode2))); i++ {}
outputChan<-append(append(itut.StartCode1(),itut.AUD()),buffer[:i]...)
buffer = buffer[i:]
i=0
}
if i >= len(buffer) {
inputBuffer = []byte{}
break
}
}
}
func (p* H264Parser)Parse() {
p.isParsing = true
outputBuffer := []byte{}
searchingForEnd := false
p.InputByteChan = make(chan byte, 10000)
for p.isParsing {
aByte := <-p.InputByteChan
outputBuffer = append(outputBuffer, aByte)
for i:=1; aByte == 0x00 && i != 4; i++ {
aByte = <-p.InputByteChan
outputBuffer = append(outputBuffer, aByte)
if ( aByte == 0x01 && i == 2 ) || ( aByte == 0x01 && i == 3 ) {
if searchingForEnd {
output := append(append(itut.StartCode1(),itut.AUD()...),outputBuffer[:len(outputBuffer)-(i+1)]...)
p.OutputChan<-output
outputBuffer = outputBuffer[len(outputBuffer)-1-i:]
searchingForEnd = false
}
aByte = <-p.InputByteChan
outputBuffer = append(outputBuffer, aByte)
if nalType := aByte & 0x1F; nalType == 1 || nalType == 5 {
searchingForEnd = true
}
}
}
}
}

2
itut/standards.go
View File

@ -30,4 +30,4 @@ package itut
func StartCode1() []byte { return []byte{0x00, 0x00, 0x01} }
func StartCode2() []byte { return []byte{0x00, 0x00, 0x00, 0x01} }
func AUD() []byte { return []byte{0x09, 0x10} }
func AUD() []byte { return []byte{0x09, 0xF0} }

2
mpegts/MpegTs.go
View File

@ -29,7 +29,7 @@ LICENSE
package mpegts
import (
"../tools"
"bitbucket.org/ausocean/av/tools"
"errors"
_"fmt"
)

2
pes/Pes.go
View File

@ -27,7 +27,7 @@ LICENSE
package pes
import (
"../tools"
"bitbucket.org/ausocean/av/tools"
)
/*
The below data struct encapsulates the fields of an PES packet. Below is

296
revid/RevidInstance.go Normal file
View File

@ -0,0 +1,296 @@
/*
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"
"crypto/md5"
"encoding/hex"
"fmt"
"io/ioutil"
"log"
"net"
"net/http"
"os"
"os/exec"
"strconv"
"time"
"io"
"../h264"
"../tsgenerator"
"bitbucket.org/ausocean/av/ringbuffer"
)
// 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 = 1000 / clipDuration
httpTimeOut = 5 // s
motionThreshold = "0.0025"
qscale = "3"
defaultRaspividCmd = "raspivid -o -"
framesPerSec = 25
packetsPerFrame = 7
h264BufferSize = 500000
)
const (
raspivid = 0
rtp = 1
h264Codec = 2
file = 4
httpOut = 5
)
type Config struct {
Input uint8
InputCmd string
Output uint8
OutputFileName string
InputFileName string
}
type RevidInst interface {
Start()
Stop()
ChangeState(newConfig Config) error
}
type revidInst struct {
expectCC int
dumpCC int
dumpPCRBase uint64
conn net.Conn
ffmpegPath string
tempDir string
ringBuffer ringbuffer.RingBuffer
config Config
isRunning bool
Error *log.Logger
outputFile *os.File
inputFile *os.File
}
func NewRevidInstance(config Config) (r *revidInst, err error) {
r = new(revidInst)
r.ringBuffer = ringbuffer.NewRingBuffer(bufferSize, mp2tPacketSize*mp2tMaxPackets)
r.Error = log.New(os.Stderr, "ERROR: ", log.Ldate|log.Ltime|log.Lshortfile)
r.expectCC = -1
r.dumpCC = -1
r.dumpPCRBase = 0
r.ChangeState(config)
switch r.config.Output {
case file:
r.outputFile, err = os.Create(r.config.OutputFileName)
if err != nil {
return nil, err
}
}
switch r.config.Input {
case file:
r.inputFile, err = os.Open(r.config.InputFileName)
if err != nil {
return nil, err
}
}
return
}
func (r *revidInst) ChangeState(newConfig Config) error {
// TODO: check that the config is G
r.config = newConfig
return nil
}
func (r *revidInst) Start() {
r.isRunning = true
go r.input()
go r.output()
}
func (r *revidInst) Stop() {
r.isRunning = false
}
func (r *revidInst) input() {
generator := tsgenerator.NewTsGenerator(framesPerSec)
go generator.Generate()
h264Parser := h264.H264Parser{OutputChan: generator.NalInputChan}
// TODO: Need to create constructor for parser otherwise I'm going to break
// something eventuallyl
go h264Parser.Parse()
var inputReader *bufio.Reader
switch r.config.Input {
case raspivid:
cmd := exec.Command("raspivid", "-o", "-", "-n", "-t", "0")
stdout, _ := cmd.StdoutPipe()
err := cmd.Start()
inputReader = bufio.NewReader(stdout)
if err != nil {
r.Error.Println(err.Error())
return
}
case file:
default:
r.Error.Println("Input not valid!")
}
clipSize := 0
packetCount := 0
now := time.Now()
prevTime := now
startPackets := [][]byte{
{71, 64, 17, 16, 0, 66, 240, 65, 0, 1, 193, 0, 0, 255, 1, 255, 0, 1, 252, 128, 48, 72, 46, 1, 6, 70, 70, 109, 112, 101, 103, 37, 115, 116, 114, 101, 97, 109, 101, 100, 32, 98, 121, 32, 116, 104, 101, 32, 71, 101, 111, 86, 105, 115, 105, 111, 110, 32, 82, 116, 115, 112, 32, 83, 101, 114, 118, 101, 114, 99, 176, 214, 195, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255},
{71, 64, 0, 16, 0, 0, 176, 13, 0, 1, 193, 0, 0, 0, 1, 240, 0, 42, 177, 4, 178, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255},
/*PMT*/ {71, 80, 0, 16,
/*Start of payload*/
0, 2, 176, 18, 0, 1, 193, 0, 0, 0xE1, 0x00, 0xF0, 0, 0x1B, 0xE1, 0, 0xF0, 0, 0x15, 0xBD, 0x4D, 0x56, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255},
}
donePSI := false
ii := 0
for r.isRunning {
fmt.Println("reading")
var h264Data []byte
switch(r.config.Input){
case raspivid:
go func(){
for {
h264Data = make([]byte, 1)
io.ReadFull(inputReader, h264Data)
h264Parser.InputByteChan<-h264Data[0]
}
}()
case file:
stats, err := r.inputFile.Stat()
if err != nil {
panic("Could not get file stats!")
}
h264Data = make([]byte, stats.Size())
_, err = r.inputFile.Read(h264Data)
if err != nil {
r.Error.Println(err.Error())
}
for i := range h264Data {
h264Parser.InputByteChan<-h264Data[i]
}
}
if clip, err := r.ringBuffer.Get(); err != nil {
r.Error.Println(err.Error())
return
} else {
for {
upperBound := clipSize + mp2tPacketSize
if ii < 3 && !donePSI {
packetByteSlice := startPackets[ii]
copy(clip[clipSize:upperBound], packetByteSlice)
ii++
} else {
donePSI = true
if err != nil {
fmt.Println(err)
}
tsPacket := <-generator.TsChan
byteSlice, err := tsPacket.ToByteSlice()
if err != nil {
r.Error.Println(err.Error())
}
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) {
if err := r.ringBuffer.DoneWriting(clipSize); err != nil {
r.Error.Println(err.Error())
return
}
clipSize = 0
packetCount = 0
prevTime = now
break
}
}
}
}
}
func (r *revidInst) output() {
for r.isRunning {
if clip, err := r.ringBuffer.Read(); err == nil {
switch r.config.Output {
case file:
r.outputFile.Write(clip)
default:
r.Error.Println("No output?")
}
if err := r.ringBuffer.DoneReading(); err != nil {
r.Error.Println(err.Error())
}
}
}
}
// sendClipToHTPP posts a video clip via HTTP, using a new TCP connection each time.
func sendClipToHTTP(clip []byte, output string, _ net.Conn) error {
timeout := time.Duration(httpTimeOut * time.Second)
client := http.Client{
Timeout: timeout,
}
hash := md5.Sum(clip)
url := output + strconv.Itoa(len(clip)) + "." + hex.EncodeToString(hash[:]) // NB: append size.digest to output
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)
if err == nil {
fmt.Printf("%s\n", body)
}
return nil
}

46
revid/config.go
View File

@ -1,46 +0,0 @@
/*
NAME
RtpToTsConverter.go - provides utilities for the conversion of Rtp packets
to equivalent MpegTs packets.
DESCRIPTION
See Readme.md
AUTHOR
Saxon Nelson-Milton <saxon.milton@gmail.com>
LICENSE
RtpToTsConverter.go is Copyright (C) 2017 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 [GNU licenses](http://www.gnu.org/licenses).
*/
package revid
const (
// input types
raspivid = 0
rtp = 1
// input format
h264 = 2
rtp = 3
// output types
file = 4
http = 5
)
type Config struct {
input uint8
output uint8
}

BIN
revid/mpegtsCompare
View File

Binary file not shown.

597
revid/mpegtsCompare.go
View File

@ -1,597 +0,0 @@
/*
NAME
revid - a testbed for re-muxing and re-directing video streams as MPEG-TS over various protocols.
DESCRIPTION
See Readme.md
AUTHOR
Alan Noble <anoble@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 main
import (
"bufio"
"bytes"
"crypto/md5"
"encoding/binary"
"encoding/hex"
"flag"
"fmt"
"io/ioutil"
"log"
"math/rand"
"net"
"net/http"
"runtime"
"strconv"
"strings"
"time"
"os"
"../tscreator"
"../tools"
"bitbucket.org/ausocean/av/ringbuffer"
"github.com/Comcast/gots/packet"
"github.com/Comcast/gots/packet/adaptationfield"
"github.com/Comcast/gots/psi"
)
// defaults and networking consts
const (
clipDuration = 1 // s
defaultPID = 256
defaultFrameRate = 25
defaultHTTPOutput = "http://localhost:8080?"
defaultUDPOutput = "udp://0.0.0.0:16384"
defaultRTPOutput = "rtp://0.0.0.0:16384"
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 = 1000 / clipDuration
bitrateOutputDelay = 60 // s
httpTimeOut = 5 // s
motionThreshold = "0.0025"
qscale = "3"
rtpPort = 17300
rtcpPort = 17319
rtspUrl = "rtsp://192.168.0.50:8554/CH002.sdp"
rtpUrl = "rtsp://192.168.0.50:8554/CH002.sdp/track1"
inputFileName = "testInput.h264"
framesPerSec = 25
)
// flag values
const (
filterFixPTS = 0x0001
filterDropAudio = 0x0002
filterScale640 = 0x0004
filterScale320 = 0x0008
filterFixContinuity = 0x0010
filterEdgeDetection = 0x0020
filterMotionDetect = 0x0040
filterRepacket = 0x0080
dumpProgramInfo = 0x0100 // 256
dumpPacketStats = 0x0200 // 512
dumpPacketHeader = 0x0400 // 1024
dumpPacketPayload = 0x0800 // 2048
)
// globals
var (
sendClip = sendClipToRTP
packetsPerFrame = rtpPackets
clipCount int
expectCC int
dumpCC int
dumpPCRBase uint64
rtpSequenceNum uint16
conn net.Conn
ffmpegPath string
tempDir string
inputErrChan chan error
outputErrChan chan error
ringBuffer ringbuffer.RingBuffer
)
// command-line flags
var (
inputURL = flag.String("i", "", "Input RTSP URL")
outputURL = flag.String("o", "", "Output URL (HTTP, UDP or RTP)")
mode = flag.String("m", "r", "Mode: one of f,h,u,r or d")
flags = flag.Int("f", 0, "Flags: see readme for explanation")
frameRate = flag.Int("r", defaultFrameRate, "Input video frame rate (25fps by default)")
selectedPID = flag.Int("p", defaultPID, "Select packets with this packet ID (PID)")
)
func main() {
setUpDirs()
flag.Parse()
if *inputURL == "" {
log.Fatal("Input (-i) required\n")
}
switch *mode {
case "f":
sendClip = sendClipToFile
case "h":
sendClip = sendClipToHTTP
if *outputURL == "" {
*outputURL = defaultHTTPOutput
}
case "u":
sendClip = sendClipToUDP
packetsPerFrame = udpPackets
if *outputURL == "" {
*outputURL = defaultUDPOutput
}
case "r":
sendClip = sendClipToRTP
packetsPerFrame = rtpPackets
if *outputURL == "" {
*outputURL = defaultRTPOutput
}
case "d":
//sendClip = sendClipToStdout
default:
log.Fatalf("Invalid mode %s\n", *mode)
}
if *flags&filterFixContinuity != 0 && *flags&dumpProgramInfo != 0 {
log.Fatal("Cannot combine filterFixContinuity and dumpProgramInfo flags\n")
}
ringBuffer = ringbuffer.NewRingBuffer(bufferSize, mp2tPacketSize*mp2tMaxPackets)
inputErrChan = make(chan error, 10)
outputErrChan = make(chan error, 10)
go input(*inputURL, *outputURL)
go output(*outputURL)
for {
select {
default:
case err := <-inputErrChan:
fmt.Fprintln(os.Stderr, err)
fmt.Fprintln(os.Stderr, "Trying again in 10s")
time.Sleep(10 * time.Second)
go input(*inputURL, *outputURL)
case err := <-outputErrChan:
fmt.Fprintln(os.Stderr, err)
fmt.Fprintln(os.Stderr, "Attempting to write again!")
}
}
}
// setUpDirs sets directories based on the OS that Revid is running on
func setUpDirs() {
switch runtime.GOOS {
case "windows":
ffmpegPath = "C:/ffmpeg/ffmpeg"
tempDir = "tmp/"
case "darwin":
ffmpegPath = "/usr/local/bin/ffmpeg"
tempDir = "/tmp/"
default:
ffmpegPath = "/home/$USER/bin/ffmpeg"
tempDir = "/output/"
}
}
// input handles the reading from the specified input
func input(input string, output string) {
fmt.Printf("Reading video from %s\n", input)
// (re)initialize globals
clipCount = 0
expectCC = -1
dumpCC = -1
dumpPCRBase = 0
rtpSequenceNum = uint16(rand.Intn(1 << 15))
// for UDP and RTP only dial once
var err error
if strings.HasPrefix(output, "udp://") || strings.HasPrefix(output, "rtp://") {
conn, err = net.Dial("udp", output[6:])
if err != nil {
inputErrChan <- err
return
}
}
converter := tscreator.NewTsCreator(framesPerSec)
// Open the h264 file
file, err := os.Open(inputFileName)
if err != nil {
panic("Could not open file!")
return
}
stats, err := file.Stat()
if err != nil {
panic("Could not get file stats!")
}
buffer := make([]byte, stats.Size())
_, err = file.Read(buffer)
if err != nil {
panic("Could not read file!")
}
// Start parsing the h264 file and send nal access units to the converter
go tools.ParseH264Buffer(buffer,converter.NalInputChan)
go converter.Convert()
clipSize := 0
packetCount := 0
now := time.Now()
prevTime := now
fmt.Printf("Looping\n")
startPackets := [][]byte{
{71,64,17,16,0,66,240,65,0,1,193,0,0,255,1,255,0,1,252,128,48,72,46,1,6,70,70,109,112,101,103,37,115,116,114,101,97,109,101,100,32,98,121,32,116,104,101,32,71,101,111,86,105,115,105,111,110,32,82,116,115,112,32,83,101,114,118,101,114,99,176,214,195,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255},
{71,64,0,16,0,0,176,13,0,1,193,0,0,0,1,240,0,42,177,4,178,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255},
/*PMT*/{71,80,0,16,
/*Start of payload*/
0,2,176,18,0,1,193,0,0,0xE1,0x00,0xF0,0,0x1B,0xE1,0,0xF0,0,0x15,0xBD,0x4D,0x56,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255},
}
donePSI := false
for {
if clip, err := ringBuffer.Get(); err != nil {
inputErrChan <- err
return
} else {
ii := 0
for {
upperBound := clipSize + mp2tPacketSize
if ii < 3 && !donePSI {
packetByteSlice := startPackets[ii]
copy(clip[clipSize:upperBound],packetByteSlice)
ii++
} else {
donePSI = true
packet := <-converter.TsChan
packetByteSlice,err := packet.ToByteSlice()
if err != nil {
fmt.Println(err)
}
copy(clip[clipSize:upperBound],packetByteSlice)
}
//fmt.Println(clip[clipSize:upperBound])
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) {
clipCount++
if err := ringBuffer.DoneWriting(clipSize); err != nil {
inputErrChan <- err
return
}
clipSize = 0
packetCount = 0
prevTime = now
break
}
}
}
}
}
// output handles the writing to specified output
func output(output string) {
file, err := os.Create("output/saxonOut.ts")
if err != nil {
panic("Can't create output file!")
}
for {
if clip, err := ringBuffer.Read(); err == nil {
file.Write(clip)
/*
for err = sendClip(clip, output, conn); err != nil; {
outputErrChan <- err
//err = sendClip(clip, output, conn)
// TODO: figure out how to write to single file
}
*/
if err := ringBuffer.DoneReading(); err != nil {
outputErrChan <- err
}
}
}
}
// sendClipToFile writes a video clip to a /tmp file.
func sendClipToFile(clip []byte, _ string, _ net.Conn) error {
filename := fmt.Sprintf(tempDir+"vid%03d.ts", clipCount)
fmt.Printf("Writing %s (%d bytes)\n", filename, len(clip))
err := ioutil.WriteFile(filename, clip, 0644)
if err != nil {
return fmt.Errorf("Error writing file %s: %s", filename, err)
}
return nil
}
// sendClipToHTPP posts a video clip via HTTP, using a new TCP connection each time.
func sendClipToHTTP(clip []byte, output string, _ net.Conn) error {
timeout := time.Duration(httpTimeOut * time.Second)
client := http.Client{
Timeout: timeout,
}
hash := md5.Sum(clip)
url := output + strconv.Itoa(len(clip)) + "." + hex.EncodeToString(hash[:]) // NB: append size.digest to output
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)
if err == nil {
fmt.Printf("%s\n", body)
}
return err
}
// sendClipToUDP sends a video clip over UDP.
func sendClipToUDP(clip []byte, _ string, conn net.Conn) error {
size := udpPackets * mp2tPacketSize
fmt.Printf("Sending %d UDP packets of size %d (%d bytes)\n", len(clip)/size, size, len(clip))
for offset := 0; offset < len(clip); offset += size {
pkt := clip[offset : offset+size]
_, err := conn.Write(pkt)
if err != nil {
return fmt.Errorf("UDP write error %s. Is your player listening?", err)
}
}
return nil
}
// sendClipToRTP sends a video clip over RTP.
func sendClipToRTP(clip []byte, _ string, conn net.Conn) error {
size := rtpPackets * mp2tPacketSize
fmt.Printf("Sending %d RTP packets of size %d (%d bytes)\n",
len(clip)/size, size+rtpHeaderSize, len(clip))
pkt := make([]byte, rtpHeaderSize+rtpPackets*mp2tPacketSize)
for offset := 0; offset < len(clip); offset += size {
rtpEncapsulate(clip[offset:offset+size], pkt)
_, err := conn.Write(pkt)
if err != nil {
return fmt.Errorf("RTP write error %s. Is your player listening?", err)
}
}
return nil
}
// checkContinuityCounts checks that the continuity of the clip is correct
func checkContinuityCounts(clip []byte) error {
for offset := 0; offset < len(clip); offset += mp2tPacketSize {
dumpCC = -1
pkt := clip[offset : offset+mp2tPacketSize]
cc := int(pkt[3] & 0xf)
if dumpCC != -1 && cc != dumpCC {
return fmt.Errorf("Continuity count out of order. Expected %v, Got: %v.", dumpCC, cc)
}
dumpCC = (cc + 1) % 16
}
return nil
}
// sendClipToStdout dumps video stats to stdout.
func sendClipToStdout(clip []byte) error {
fmt.Printf("Dumping clip (%d bytes)\n", len(clip))
/*
if *flags&dumpProgramInfo != 0 {
return mp2tDumpProgram(clip)
}
*/
packetCount := 0
discontinuities := 0
var cc int
for offset := 0; offset < len(clip); offset += mp2tPacketSize {
packetCount++
pkt := clip[offset : offset+mp2tPacketSize]
pktPID, err := packet.Pid(pkt)
if err != nil {
return err
}
if pktPID != uint16(*selectedPID) {
continue
}
if *flags&(dumpPacketHeader|dumpPacketPayload) != 0 {
fmt.Printf("Packet #%d.%d\n", clipCount, packetCount)
}
hasPayload := pkt[3]&0x10 != 0
if !hasPayload {
continue // nothing to do
}
// extract interesting info from header
tei := pkt[1] & 0x80 >> 7
pusi := pkt[1] & 0x40 >> 6
tp := pkt[1] & 0x20 >> 5
tcs := pkt[3] & 0xc0 >> 6
afc := pkt[3] & 0x30 >> 4
cc = int(pkt[3] & 0xf)
di := pkt[5] & 0x80
if dumpCC != -1 && cc != dumpCC {
discontinuities++
fmt.Printf("Warning: Packet #%d.%d continuity counter out of order! Got %d, expected %d.\n",
clipCount, packetCount, cc, dumpCC)
}
dumpCC = (cc + 1) % 16
//if *flags&dumpPacketHeader != 0 {
fmt.Printf("\t\tTEI=%d, PUSI=%d, TP=%d, TSC=%d, AFC=%d, CC=%d\n", tei, pusi, tp, tcs, afc, cc)
//}
if afc == 3 {
// adaptation field, followed by payload
afl := adaptationfield.Length(pkt)
if adaptationfield.HasPCR(pkt) {
pcrBase, pcrExt, _ := mp2tGetPCR(pkt)
if *flags&dumpPacketHeader != 0 {
fmt.Printf("\t\tAFL=%d, PCRbase=%d, PCRext=%d, DI=%v\n", afl, pcrBase, pcrExt, di)
}
if pcrBase < dumpPCRBase {
fmt.Printf("Warning: PCRbase went backwards!\n")
}
dumpPCRBase = pcrBase
} else if *flags&dumpPacketHeader != 0 {
fmt.Printf("\t\tAFL=%d, DI=%v\n", afl, di)
}
}
if *flags&dumpPacketPayload != 0 {
fmt.Printf("\t\tPayload=%x\n", pkt)
}
}
if *flags&dumpPacketStats != 0 {
fmt.Printf("%d packets of size %d bytes (%d bytes, %d discontinuites)\n",
packetCount, packet.PacketSize, packetCount*packet.PacketSize, discontinuities)
}
return nil
}
// mp2tDumpProgram dumps MPEG-TS Program Association Table (PAT) and Program Map Tables (PMT).
func mp2tDumpProgram(clip []byte) error {
// NB: Comcast API requires a buffered reader
reader := bufio.NewReader(bytes.NewReader(clip))
_, err := packet.Sync(reader)
if err != nil {
return fmt.Errorf("Error reading sync byte: %s", err)
}
pat, err := psi.ReadPAT(reader)
if err != nil {
return fmt.Errorf("Error reading PAT: %s", err)
}
mp2tDumpPat(pat)
var pmts []psi.PMT
pm := pat.ProgramMap()
for pn, pid := range pm {
pmt, err := psi.ReadPMT(reader, pid)
if err != nil {
return fmt.Errorf("Error reading PMT: %s", err)
}
pmts = append(pmts, pmt)
mp2tDumpPmt(pn, pmt)
}
return nil
}
func mp2tDumpPat(pat psi.PAT) {
fmt.Printf("Pat\n")
fmt.Printf("\tPMT PIDs %v\n", pat.ProgramMap())
fmt.Printf("\tNumber of Programs %v\n", pat.NumPrograms())
}
func mp2tDumpPmt(pn uint16, pmt psi.PMT) {
// pn = program number
fmt.Printf("Program #%v PMT\n", pn)
fmt.Printf("\tPIDs %v\n", pmt.Pids())
fmt.Printf("\tElementary Streams")
for _, es := range pmt.ElementaryStreams() {
fmt.Printf("\t\tPID %v : StreamType %v\n", es.ElementaryPid(), es.StreamType())
for _, d := range es.Descriptors() {
fmt.Printf("\t\t\t%+v\n", d)
}
}
}
// Mp2tFixContinuity fixes discontinous MPEG-TS continuity counts (CC)
func mp2tFixContinuity(pkt []byte, pid uint16) bool {
hasPayload, err := packet.ContainsPayload(pkt)
if err != nil {
fmt.Printf("Warning: Packet bad.\n")
return false
}
if !hasPayload {
return false
}
if pktPID, _ := packet.Pid(pkt); pktPID != pid {
return false
}
fixed := false
// extract continuity counter from 2nd nibble of 4th byte of header
cc := int(pkt[3] & 0xf)
if expectCC == -1 {
expectCC = cc
} else if cc != expectCC {
fmt.Println("Have to fix")
pkt[3] = pkt[3]&0xf0 | byte(expectCC&0xf)
fixed = true
}
expectCC = (expectCC + 1) % 16
return fixed
}
// Mp2tGetPCR extracts the Program Clock Reference (PCR) from an MPEG-TS packet (if any)
func mp2tGetPCR(pkt []byte) (uint64, uint32, bool) {
if !adaptationfield.HasPCR(pkt) {
return 0, 0, false
}
pcrBytes, _ := adaptationfield.PCR(pkt) // 6 bytes
// first 33 bits are PCR base, next 6 bits are reserved, final 9 bits are PCR extension.
pcrBase := uint64(binary.BigEndian.Uint32(pcrBytes[:4]))<<1 | uint64(pcrBytes[4]&0x80>>7)
pcrExt := uint32(pcrBytes[4]&0x01)<<1 | uint32(pcrBytes[5])
return pcrBase, pcrExt, true
}
// rtpEncapsulate encapsulates MPEG-TS packets within an RTP header,
// setting the payload type accordingly (to 33) and incrementing the RTP sequence number.
func rtpEncapsulate(mp2tPacket []byte, pkt []byte) {
// RTP packet encapsulates the MP2T
// first 12 bytes is the header
// byte 0: version=2, padding=0, extension=0, cc=0
pkt[0] = 0x80 // version (2)
// byte 1: marker=0, pt = 33 (MP2T)
pkt[1] = 33
// bytes 2 & 3: sequence number
binary.BigEndian.PutUint16(pkt[2:4], rtpSequenceNum)
if rtpSequenceNum == ^uint16(0) {
rtpSequenceNum = 0
} else {
rtpSequenceNum++
}
// bytes 4,5,6&7: timestamp
timestamp := uint32(time.Now().UnixNano() / 1e6) // ms timestamp
binary.BigEndian.PutUint32(pkt[4:8], timestamp)
// bytes 8,9,10&11: SSRC
binary.BigEndian.PutUint32(pkt[8:12], rtpSSRC)
// payload follows
copy(pkt[rtpHeaderSize:rtpHeaderSize+rtpPackets*mp2tPacketSize], mp2tPacket)
}

123
revid/revid_test.go
View File

@ -30,124 +30,49 @@ package revid
import (
"testing"
)
/*******************************************************
Testing stuff related to connection i.e. rtsp, rtp, rtcp
********************************************************/
const (
rtpPort = 17300
rtcpPort = 17319
rtspUrl = "rtsp://192.168.0.50:8554/CH002.sdp"
rtpUrl = "rtsp://192.168.0.50:8554/CH002.sdp/track1"
"time"
)
/* Let's see if we can connect to an rtsp device then read an rtp stream,
and then convert the rtp packets to mpegts packets and output. */
func TestRTSP(t *testing.T) {
sess := rtsp.NewSession()
res, err := sess.Options(rtspUrl)
if err != nil {
t.Errorf("Shouldn't have got error: %v\n", err)
/*
* Testing with file input
*
*/
func TestFileInput(t *testing.T){
config := Config{
Input: file,
InputFileName: "testInput.h264",
Output: file,
OutputFileName: "output/TestFileAsInput.ts",
}
res, err = sess.Describe(rtspUrl)
revidInst, err := NewRevidInstance(config)
if err != nil {
log.Fatalln(err)
t.Errorf("Shouldn't have got error: %v\n", err)
t.Errorf("Should not have got error!")
}
p, err := rtsp.ParseSdp(&io.LimitedReader{R: res.Body, N: res.ContentLength})
if err != nil {
t.Errorf("Shouldn't have got error: %v\n", err)
}
log.Printf("%+v", p)
res, err = sess.Setup(rtpUrl, fmt.Sprintf("RTP/AVP;unicast;client_port=%d-%d", rtpPort, rtcpPort))
if err != nil {
t.Errorf("Shouldn't have got error: %v\n", err)
}
log.Println(res)
res, err = sess.Play(rtspUrl, res.Header.Get("Session"))
if err != nil {
t.Errorf("Shouldn't have got error: %v\n", err)
}
log.Println(res)
revidInst.Start()
time.Sleep(100*time.Second)
revidInst.Stop()
}
func TestRTP(t *testing.T) {
sess := rtsp.NewSession()
res, err := sess.Options(rtspUrl)
if err != nil {
t.Errorf("Shouldn't have got error: %v\n", err)
}
res, err = sess.Describe(rtspUrl)
if err != nil {
log.Fatalln(err)
t.Errorf("Shouldn't have got error: %v\n", err)
}
p, err := rtsp.ParseSdp(&io.LimitedReader{R: res.Body, N: res.ContentLength})
if err != nil {
t.Errorf("Shouldn't have got error: %v\n", err)
}
log.Printf("%+v", p)
res, err = sess.Setup(rtpUrl, fmt.Sprintf("RTP/AVP;unicast;client_port=%d-%d", rtpPort, rtcpPort))
if err != nil {
t.Errorf("Shouldn't have got error: %v\n", err)
}
log.Println(res)
res, err = sess.Play(rtspUrl, res.Header.Get("Session"))
if err != nil {
t.Errorf("Shouldn't have got error: %v\n", err)
}
log.Println(res)
// create udp connection for rtp stuff
rtpLaddr, err := net.ResolveUDPAddr("udp", "192.168.0.109:17300")
if err != nil {
t.Errorf("Local rtp addr not set! %v\n", err)
}
rtpAddr, err := net.ResolveUDPAddr("udp", "192.168.0.50:17300")
if err != nil {
t.Errorf("Resolving rtp address didn't work! %v\n", err)
}
rtpConn, err := net.DialUDP("udp", rtpLaddr, rtpAddr)
if err != nil {
t.Errorf("Conncection not established! %v\n", err)
}
// Create udp connection for rtcp stuff
rtcpLaddr, err := net.ResolveUDPAddr("udp", "192.168.0.109:17319")
if err != nil {
t.Errorf("Local RTCP address not resolved! %v\n", err)
}
rtcpAddr, err := net.ResolveUDPAddr("udp", "192.168.0.50:17301")
if err != nil {
t.Errorf("Remote RTCP address not resolved! %v\n", err)
}
rtcpConn, err := net.DialUDP("udp", rtcpLaddr, rtcpAddr)
if err != nil {
t.Errorf("Connection not established! %v\n", err)
}
// let's create a session that will store useful stuff from the connections
rtpSession := NewSession(rtpConn, rtcpConn)
time.Sleep(2 * time.Second)
select {
default:
t.Errorf("Should have got rtpPacket!")
case rtpPacket := <-rtpSession.RtpChan:
fmt.Printf("RTP packet: %v\n", rtpPacket)
}
}
/*
Testing use with raspivid
*/
/*
func TestRaspividInput(t *testing.T){
config := Config{
Input: raspivid,
Output: file,
OutputFileName: "output/TestRaspividOutput.ts",
}
revidInst, err := NewRevidInstance(config)
if err != nil {
t.Errorf("Should not have got an error!")
}
revidInst.Run()
time.Sleep(5*time.Second)
revidInst.Start()
time.Sleep(100*time.Second)
revidInst.Stop()
}
* */

246
revid/revidinstance.go
View File

@ -1,246 +0,0 @@
/*
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
// TODO: do error handling using logger
import (
"bufio"
"bytes"
"crypto/md5"
"encoding/binary"
"encoding/hex"
"flag"
"fmt"
"io"
"io/ioutil"
"log"
"math/rand"
"net"
"net/http"
"os"
"os/exec"
"runtime"
"strconv"
"strings"
"time"
"errors"
"log"
"/tools"
"ioutil"
"bitbucket.org/ausocean/av/ringbuffer"
"github.com/Comcast/gots/packet"
"github.com/Comcast/gots/packet/adaptationfield"
"github.com/Comcast/gots/psi"
)
// defaults and networking consts
const (
clipDuration = 1 // s
defaultPID = 256
defaultFrameRate = 25
defaultHTTPOutput = "http://localhost:8080?"
defaultUDPOutput = "udp://0.0.0.0:16384"
defaultRTPOutput = "rtp://0.0.0.0:16384"
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 = 1000 / clipDuration
bitrateOutputDelay = 60 // s
httpTimeOut = 5 // s
motionThreshold = "0.0025"
qscale = "3"
)
type RevidInst interface {
Run()
Stop()
ChangeState(newConfig Config) err
}
type revidInst struct {
expectCC int
dumpCC int
dumpPCRBase uint64
conn net.Conn
ffmpegPath string
tempDir string
ringBuffer ringbuffer.RingBuffer
Config Config
isRunning bool
Error *log.Logger
}
func NewRevidInstance(config Config)(r *revid, err error){
ringBuffer = ringbuffer.NewRingBuffer(bufferSize, mp2tPacketSize*mp2tMaxPackets)
r.Error = log.New(os.Stderr, "ERROR: ",log.Ldat|log.Ltime|log.Lshortfile)
r.expectCC = -1
r.dumpCC = -1
r.dumpPCRBase = 0
}
func(r *revidInst) Run(){
isRunning = true
go r.input()
go r.output()
}
func(r *revidInst) Stop(){
r.isRunning = false
}
// input handles the reading from the specified input
func (r *revid)input(input string, output string) {
// (re)initialize globals
r.expectCC = -1
r.dumpCC = -1
r.dumpPCRBase = 0
converter := tscreator.NewTsCreator(framesPerSec)
go converter.Convert()
h264Parser := h264.NewH264Parser(converter.NalInputChan)
go h264Parser.Parse()
var err error
var inputBuffer *Reader
switch(r.config.input){
case raspivid:
cmd := exec.Command("raspivid -o -")
stdout, _ := cmd.StdoutPipe()
err = cmd.Start()
inputReader = bufio.NewReader(stdout)
if err != nil {
r.Error.Println(err.Error())
return
}
default:
r.Error.Println("Input not valid!")
}
clipSize := 0
packetCount := 0
now := time.Now()
prevTime := now
for r.isRunning {
var h264Data []byte
if h264Data, err = ioutil.ReadAll(inputReader); err != nil {
r.Error.Println(err.Error())
}
h264Parser.SendInputData(h264Data)
if clip, err := ringBuffer.Get(); err != nil {
r.Error.Println(err.Error())
return
} else {
for {
upperBound := clipSize + mp2tPacketSize
// get ts packet from converter
clip = (<-converter.TsChan).ToByteSlice()
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) {
if err := ringBuffer.DoneWriting(clipSize); err != nil {
r.Error.Println(err.Error())
return
}
clipSize = 0
packetCount = 0
prevTime = now
break
}
}
}
}
}
// output handles the writing to specified output
func (r *revid)output(output string) {
for r.isRunning {
if clip, err := ringBuffer.Read(); err == nil {
fmt.Println(clip)
for err = sendClip(clip, output, conn); err != nil; {
r.Error.Println(err.Error())
err = sendClip(clip, output, conn)
}
if err := ringBuffer.DoneReading(); err != nil {
r.Error.Println(err.Error())
}
}
}
}
// sendClipToFile writes a video clip to a /tmp file.
func sendClipToFile(clip []byte, _ string, _ net.Conn) error {
return nil
}
// sendClipToHTPP posts a video clip via HTTP, using a new TCP connection each time.
func sendClipToHTTP(clip []byte, output string, _ net.Conn) error {
timeout := time.Duration(httpTimeOut * time.Second)
client := http.Client{
Timeout: timeout,
}
hash := md5.Sum(clip)
url := output + strconv.Itoa(len(clip)) + "." + hex.EncodeToString(hash[:]) // NB: append size.digest to output
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)
if err == nil {
fmt.Printf("%s\n", body)
}
return err
}
// sendClipToUDP sends a video clip over UDP.
func sendClipToUDP(clip []byte, _ string, conn net.Conn) error {
size := udpPackets * mp2tPacketSize
fmt.Printf("Sending %d UDP packets of size %d (%d bytes)\n", len(clip)/size, size, len(clip))
for offset := 0; offset < len(clip); offset += size {
pkt := clip[offset : offset+size]
_, err := conn.Write(pkt)
if err != nil {
return fmt.Errorf("UDP write error %s. Is your player listening?", err)
}
}
return nil
}

653
revid/revidv2.go
View File

@ -1,653 +0,0 @@
/*
NAME
revid - a testbed for re-muxing and re-directing video streams as MPEG-TS over various protocols.
DESCRIPTION
See Readme.md
AUTHOR
Alan Noble <anoble@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 main
import (
"bufio"
"bytes"
"crypto/md5"
"encoding/binary"
"encoding/hex"
"flag"
"fmt"
"io/ioutil"
"log"
"math/rand"
"net"
"net/http"
"os"
"runtime"
"strconv"
"strings"
"time"
"io"
"../packets"
"bitbucket.org/ausocean/av/ringbuffer"
"github.com/Comcast/gots/packet"
"github.com/Comcast/gots/packet/adaptationfield"
"github.com/Comcast/gots/psi"
"github.com/beatgammit/rtsp"
)
// defaults and networking consts
const (
clipDuration = 1 // s
defaultPID = 256
defaultFrameRate = 25
defaultHTTPOutput = "http://localhost:8080?"
defaultUDPOutput = "udp://0.0.0.0:16384"
defaultRTPOutput = "rtp://0.0.0.0:16384"
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 = 1000 / clipDuration
bitrateOutputDelay = 60 // s
httpTimeOut = 5 // s
motionThreshold = "0.0025"
qscale = "3"
rtpPort = 17300
rtcpPort = 17319
rtspUrl = "rtsp://192.168.0.50:8554/CH002.sdp"
rtpUrl = "rtsp://192.168.0.50:8554/CH002.sdp/track1"
)
// flag values
const (
filterFixPTS = 0x0001
filterDropAudio = 0x0002
filterScale640 = 0x0004
filterScale320 = 0x0008
filterFixContinuity = 0x0010
filterEdgeDetection = 0x0020
filterMotionDetect = 0x0040
filterRepacket = 0x0080
dumpProgramInfo = 0x0100 // 256
dumpPacketStats = 0x0200 // 512
dumpPacketHeader = 0x0400 // 1024
dumpPacketPayload = 0x0800 // 2048
)
// globals
var (
sendClip = sendClipToRTP
packetsPerFrame = rtpPackets
clipCount int
expectCC int
dumpCC int
dumpPCRBase uint64
rtpSequenceNum uint16
conn net.Conn
ffmpegPath string
tempDir string
inputErrChan chan error
outputErrChan chan error
ringBuffer ringbuffer.RingBuffer
)
// command-line flags
var (
inputURL = flag.String("i", "", "Input RTSP URL")
outputURL = flag.String("o", "", "Output URL (HTTP, UDP or RTP)")
mode = flag.String("m", "r", "Mode: one of f,h,u,r or d")
flags = flag.Int("f", 0, "Flags: see readme for explanation")
frameRate = flag.Int("r", defaultFrameRate, "Input video frame rate (25fps by default)")
selectedPID = flag.Int("p", defaultPID, "Select packets with this packet ID (PID)")
)
func main() {
setUpDirs()
flag.Parse()
if *inputURL == "" {
log.Fatal("Input (-i) required\n")
}
switch *mode {
case "f":
sendClip = sendClipToFile
case "h":
sendClip = sendClipToHTTP
if *outputURL == "" {
*outputURL = defaultHTTPOutput
}
case "u":
sendClip = sendClipToUDP
packetsPerFrame = udpPackets
if *outputURL == "" {
*outputURL = defaultUDPOutput
}
case "r":
sendClip = sendClipToRTP
packetsPerFrame = rtpPackets
if *outputURL == "" {
*outputURL = defaultRTPOutput
}
case "d":
//sendClip = sendClipToStdout
default:
log.Fatalf("Invalid mode %s\n", *mode)
}
if *flags&filterFixContinuity != 0 && *flags&dumpProgramInfo != 0 {
log.Fatal("Cannot combine filterFixContinuity and dumpProgramInfo flags\n")
}
ringBuffer = ringbuffer.NewRingBuffer(bufferSize, mp2tPacketSize*mp2tMaxPackets)
inputErrChan = make(chan error, 10)
outputErrChan = make(chan error, 10)
go input(*inputURL, *outputURL)
go output(*outputURL)
for {
select {
default:
case err := <-inputErrChan:
fmt.Fprintln(os.Stderr, err)
fmt.Fprintln(os.Stderr, "Trying again in 10s")
time.Sleep(10 * time.Second)
go input(*inputURL, *outputURL)
case err := <-outputErrChan:
fmt.Fprintln(os.Stderr, err)
fmt.Fprintln(os.Stderr, "Attempting to write again!")
}
}
}
// setUpDirs sets directories based on the OS that Revid is running on
func setUpDirs() {
switch runtime.GOOS {
case "windows":
ffmpegPath = "C:/ffmpeg/ffmpeg"
tempDir = "tmp/"
case "darwin":
ffmpegPath = "/usr/local/bin/ffmpeg"
tempDir = "/tmp/"
default:
ffmpegPath = "/home/$USER/bin/ffmpeg"
tempDir = "/tmp/"
}
}
// input handles the reading from the specified input
func input(input string, output string) {
fmt.Printf("Reading video from %s\n", input)
// (re)initialize globals
clipCount = 0
expectCC = -1
dumpCC = -1
dumpPCRBase = 0
rtpSequenceNum = uint16(rand.Intn(1 << 15))
// for UDP and RTP only dial once
var err error
if strings.HasPrefix(output, "udp://") || strings.HasPrefix(output, "rtp://") {
conn, err = net.Dial("udp", output[6:])
if err != nil {
inputErrChan <- err
return
}
}
sess := rtsp.NewSession()
res, err := sess.Options(rtspUrl)
if err != nil {
log.Fatalln(err)
}
fmt.Println("Options:")
fmt.Println(res)
res, err = sess.Describe(rtspUrl)
if err != nil {
log.Fatalln(err)
}
fmt.Println("Describe:")
fmt.Println(res)
p, err := rtsp.ParseSdp(&io.LimitedReader{R: res.Body, N: res.ContentLength})
if err != nil {
log.Fatalln(err)
}
log.Printf("%+v", p)
fmt.Println("Setting up!")
rtpPort, rtcpPort := 17300, 17319
res, err = sess.Setup(rtpUrl, fmt.Sprintf("RTP/AVP;unicast;client_port=%d-%d", rtpPort, rtcpPort))
if err != nil {
log.Fatalln(err)
}
log.Println(res)
fmt.Println("Playing !")
res, err = sess.Play(rtspUrl, res.Header.Get("Session"))
if err != nil {
log.Fatalln(err)
}
log.Println(res)
// create udp connection for rtp stuff
rtpLaddr, err := net.ResolveUDPAddr("udp", "192.168.0.109:17300")
if err != nil {
fmt.Println("Local rtp addr not set!")
}
rtpAddr, err := net.ResolveUDPAddr("udp", "192.168.0.50:17300")
if err != nil {
fmt.Println("Resolving rtp address didn't work!")
}
rtpConn, err := net.DialUDP("udp", rtpLaddr, rtpAddr)
if err != nil {
fmt.Println("Rtp dial didn't work!")
}
// Create udp connection for rtcp stuff
rtcpLaddr, err := net.ResolveUDPAddr("udp", "192.168.0.109:17319")
if err != nil {
fmt.Println("Local ")
}
rtcpAddr, err := net.ResolveUDPAddr("udp", "192.168.0.50:17301")
if err != nil {
fmt.Println("resolving rtcp address didn't work!")
}
rtcpConn, err := net.DialUDP("udp", rtcpLaddr, rtcpAddr)
if err != nil {
fmt.Println("Rtcp dial didnt't work!")
}
rtpSession := packets.NewSession(rtpConn,rtcpConn)
converter := packets.NewRtpToTsConverter()
go func(){
for{
select {
default:
case aPacket := <-rtpSession.RtpChan:
converter.InputChan<-aPacket
}
}
}()
go converter.Convert()
clipSize := 0
packetCount := 0
now := time.Now()
prevTime := now
fmt.Printf("Looping\n")
startPackets := [][]byte{
{71,64,17,16,0,66,240,65,0,1,193,0,0,255,1,255,0,1,252,128,48,72,46,1,6,70,70,109,112,101,103,37,115,116,114,101,97,109,101,100,32,98,121,32,116,104,101,32,71,101,111,86,105,115,105,111,110,32,82,116,115,112,32,83,101,114,118,101,114,99,176,214,195,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255},
{71,64,0,16,0,0,176,13,0,1,193,0,0,0,1,240,0,42,177,4,178,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255},
/*PMT*/{71,80,0,16,
/*Start of payload*/
0,2,176,18,0,1,193,0,0,255,255,240,0,27,225,0,240,0,193,91,65,224,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255,255},
}
donePSI := false
for {
if clip, err := ringBuffer.Get(); err != nil {
inputErrChan <- err
return
} else {
ii := 0
for {
upperBound := clipSize + mp2tPacketSize
if ii < 3 && !donePSI {
packetByteSlice := startPackets[ii]
copy(clip[clipSize:upperBound],packetByteSlice)
ii++
} else {
donePSI = true
packet := <-converter.TsChan
packetByteSlice := packet.ToByteSlice()
copy(clip[clipSize:upperBound],packetByteSlice)
}
//fmt.Println(clip[clipSize:upperBound])
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) {
clipCount++
if err := ringBuffer.DoneWriting(clipSize); err != nil {
inputErrChan <- err
return
}
clipSize = 0
packetCount = 0
prevTime = now
break
}
}
}
}
}
// output handles the writing to specified output
func output(output string) {
elapsedTime := time.Duration(0)
now := time.Now()
prevTime := now
for {
if clip, err := ringBuffer.Read(); err == nil {
now := time.Now()
for err = sendClip(clip, output, conn); err != nil; {
outputErrChan <- err
err = sendClip(clip, output, conn)
}
deltaTime := now.Sub(prevTime)
elapsedTime += deltaTime
if elapsedTime > bitrateOutputDelay*time.Second {
noOfBits := float64(len(clip)*8) / 1024.0 // convert bytes to kilobits
fmt.Printf("Bitrate: %d kbps\n", int64(noOfBits/float64(deltaTime/1e9)))
elapsedTime = time.Duration(0)
}
prevTime = now
if err := ringBuffer.DoneReading(); err != nil {
outputErrChan <- err
}
}
}
}
// sendClipToFile writes a video clip to a /tmp file.
func sendClipToFile(clip []byte, _ string, _ net.Conn) error {
filename := fmt.Sprintf(tempDir+"vid%03d.ts", clipCount)
fmt.Printf("Writing %s (%d bytes)\n", filename, len(clip))
err := ioutil.WriteFile(filename, clip, 0644)
if err != nil {
return fmt.Errorf("Error writing file %s: %s", filename, err)
}
return nil
}
// sendClipToHTPP posts a video clip via HTTP, using a new TCP connection each time.
func sendClipToHTTP(clip []byte, output string, _ net.Conn) error {
timeout := time.Duration(httpTimeOut * time.Second)
client := http.Client{
Timeout: timeout,
}
hash := md5.Sum(clip)
url := output + strconv.Itoa(len(clip)) + "." + hex.EncodeToString(hash[:]) // NB: append size.digest to output
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)
if err == nil {
fmt.Printf("%s\n", body)
}
return err
}
// sendClipToUDP sends a video clip over UDP.
func sendClipToUDP(clip []byte, _ string, conn net.Conn) error {
size := udpPackets * mp2tPacketSize
fmt.Printf("Sending %d UDP packets of size %d (%d bytes)\n", len(clip)/size, size, len(clip))
for offset := 0; offset < len(clip); offset += size {
pkt := clip[offset : offset+size]
_, err := conn.Write(pkt)
if err != nil {
return fmt.Errorf("UDP write error %s. Is your player listening?", err)
}
}
return nil
}
// sendClipToRTP sends a video clip over RTP.
func sendClipToRTP(clip []byte, _ string, conn net.Conn) error {
size := rtpPackets * mp2tPacketSize
fmt.Printf("Sending %d RTP packets of size %d (%d bytes)\n",
len(clip)/size, size+rtpHeaderSize, len(clip))
pkt := make([]byte, rtpHeaderSize+rtpPackets*mp2tPacketSize)
for offset := 0; offset < len(clip); offset += size {
rtpEncapsulate(clip[offset:offset+size], pkt)
_, err := conn.Write(pkt)
if err != nil {
return fmt.Errorf("RTP write error %s. Is your player listening?", err)
}
}
return nil
}
// checkContinuityCounts checks that the continuity of the clip is correct
func checkContinuityCounts(clip []byte) error {
for offset := 0; offset < len(clip); offset += mp2tPacketSize {
dumpCC = -1
pkt := clip[offset : offset+mp2tPacketSize]
cc := int(pkt[3] & 0xf)
if dumpCC != -1 && cc != dumpCC {
return fmt.Errorf("Continuity count out of order. Expected %v, Got: %v.", dumpCC, cc)
}
dumpCC = (cc + 1) % 16
}
return nil
}
// sendClipToStdout dumps video stats to stdout.
func sendClipToStdout(clip []byte) error {
fmt.Printf("Dumping clip (%d bytes)\n", len(clip))
/*
if *flags&dumpProgramInfo != 0 {
return mp2tDumpProgram(clip)
}
*/
packetCount := 0
discontinuities := 0
var cc int
for offset := 0; offset < len(clip); offset += mp2tPacketSize {
packetCount++
pkt := clip[offset : offset+mp2tPacketSize]
pktPID, err := packet.Pid(pkt)
if err != nil {
return err
}
if pktPID != uint16(*selectedPID) {
continue
}
if *flags&(dumpPacketHeader|dumpPacketPayload) != 0 {
fmt.Printf("Packet #%d.%d\n", clipCount, packetCount)
}
hasPayload := pkt[3]&0x10 != 0
if !hasPayload {
continue // nothing to do
}
// extract interesting info from header
tei := pkt[1] & 0x80 >> 7
pusi := pkt[1] & 0x40 >> 6
tp := pkt[1] & 0x20 >> 5
tcs := pkt[3] & 0xc0 >> 6
afc := pkt[3] & 0x30 >> 4
cc = int(pkt[3] & 0xf)
di := pkt[5] & 0x80
if dumpCC != -1 && cc != dumpCC {
discontinuities++
fmt.Printf("Warning: Packet #%d.%d continuity counter out of order! Got %d, expected %d.\n",
clipCount, packetCount, cc, dumpCC)
}
dumpCC = (cc + 1) % 16
//if *flags&dumpPacketHeader != 0 {
fmt.Printf("\t\tTEI=%d, PUSI=%d, TP=%d, TSC=%d, AFC=%d, CC=%d\n", tei, pusi, tp, tcs, afc, cc)
//}
if afc == 3 {
// adaptation field, followed by payload
afl := adaptationfield.Length(pkt)
if adaptationfield.HasPCR(pkt) {
pcrBase, pcrExt, _ := mp2tGetPCR(pkt)
if *flags&dumpPacketHeader != 0 {
fmt.Printf("\t\tAFL=%d, PCRbase=%d, PCRext=%d, DI=%v\n", afl, pcrBase, pcrExt, di)
}
if pcrBase < dumpPCRBase {
fmt.Printf("Warning: PCRbase went backwards!\n")
}
dumpPCRBase = pcrBase
} else if *flags&dumpPacketHeader != 0 {
fmt.Printf("\t\tAFL=%d, DI=%v\n", afl, di)
}
}
if *flags&dumpPacketPayload != 0 {
fmt.Printf("\t\tPayload=%x\n", pkt)
}
}
if *flags&dumpPacketStats != 0 {
fmt.Printf("%d packets of size %d bytes (%d bytes, %d discontinuites)\n",
packetCount, packet.PacketSize, packetCount*packet.PacketSize, discontinuities)
}
return nil
}
// mp2tDumpProgram dumps MPEG-TS Program Association Table (PAT) and Program Map Tables (PMT).
func mp2tDumpProgram(clip []byte) error {
// NB: Comcast API requires a buffered reader
reader := bufio.NewReader(bytes.NewReader(clip))
_, err := packet.Sync(reader)
if err != nil {
return fmt.Errorf("Error reading sync byte: %s", err)
}
pat, err := psi.ReadPAT(reader)
if err != nil {
return fmt.Errorf("Error reading PAT: %s", err)
}
mp2tDumpPat(pat)
var pmts []psi.PMT
pm := pat.ProgramMap()
for pn, pid := range pm {
pmt, err := psi.ReadPMT(reader, pid)
if err != nil {
return fmt.Errorf("Error reading PMT: %s", err)
}
pmts = append(pmts, pmt)
mp2tDumpPmt(pn, pmt)
}
return nil
}
func mp2tDumpPat(pat psi.PAT) {
fmt.Printf("Pat\n")
fmt.Printf("\tPMT PIDs %v\n", pat.ProgramMap())
fmt.Printf("\tNumber of Programs %v\n", pat.NumPrograms())
}
func mp2tDumpPmt(pn uint16, pmt psi.PMT) {
// pn = program number
fmt.Printf("Program #%v PMT\n", pn)
fmt.Printf("\tPIDs %v\n", pmt.Pids())
fmt.Printf("\tElementary Streams")
for _, es := range pmt.ElementaryStreams() {
fmt.Printf("\t\tPID %v : StreamType %v\n", es.ElementaryPid(), es.StreamType())
for _, d := range es.Descriptors() {
fmt.Printf("\t\t\t%+v\n", d)
}
}
}
// Mp2tFixContinuity fixes discontinous MPEG-TS continuity counts (CC)
func mp2tFixContinuity(pkt []byte, pid uint16) bool {
hasPayload, err := packet.ContainsPayload(pkt)
if err != nil {
fmt.Printf("Warning: Packet bad.\n")
return false
}
if !hasPayload {
return false
}
if pktPID, _ := packet.Pid(pkt); pktPID != pid {
return false
}
fixed := false
// extract continuity counter from 2nd nibble of 4th byte of header
cc := int(pkt[3] & 0xf)
if expectCC == -1 {
expectCC = cc
} else if cc != expectCC {
fmt.Println("Have to fix")
pkt[3] = pkt[3]&0xf0 | byte(expectCC&0xf)
fixed = true
}
expectCC = (expectCC + 1) % 16
return fixed
}
// Mp2tGetPCR extracts the Program Clock Reference (PCR) from an MPEG-TS packet (if any)
func mp2tGetPCR(pkt []byte) (uint64, uint32, bool) {
if !adaptationfield.HasPCR(pkt) {
return 0, 0, false
}
pcrBytes, _ := adaptationfield.PCR(pkt) // 6 bytes
// first 33 bits are PCR base, next 6 bits are reserved, final 9 bits are PCR extension.
pcrBase := uint64(binary.BigEndian.Uint32(pcrBytes[:4]))<<1 | uint64(pcrBytes[4]&0x80>>7)
pcrExt := uint32(pcrBytes[4]&0x01)<<1 | uint32(pcrBytes[5])
return pcrBase, pcrExt, true
}
// rtpEncapsulate encapsulates MPEG-TS packets within an RTP header,
// setting the payload type accordingly (to 33) and incrementing the RTP sequence number.
func rtpEncapsulate(mp2tPacket []byte, pkt []byte) {
// RTP packet encapsulates the MP2T
// first 12 bytes is the header
// byte 0: version=2, padding=0, extension=0, cc=0
pkt[0] = 0x80 // version (2)
// byte 1: marker=0, pt = 33 (MP2T)
pkt[1] = 33
// bytes 2 & 3: sequence number
binary.BigEndian.PutUint16(pkt[2:4], rtpSequenceNum)
if rtpSequenceNum == ^uint16(0) {
rtpSequenceNum = 0
} else {
rtpSequenceNum++
}
// bytes 4,5,6&7: timestamp
timestamp := uint32(time.Now().UnixNano() / 1e6) // ms timestamp
binary.BigEndian.PutUint32(pkt[4:8], timestamp)
// bytes 8,9,10&11: SSRC
binary.BigEndian.PutUint32(pkt[8:12], rtpSSRC)
// payload follows
copy(pkt[rtpHeaderSize:rtpHeaderSize+rtpPackets*mp2tPacketSize], mp2tPacket)
}

4
revid/test.bash
View File

@ -1,4 +0,0 @@
#!/bin/bash
echo Running Revid with input: rtsp://184.72.239.149/vod/mp4:BigBuckBunny_175k.mov
echo and output: rtp://0.0.0.0:1234
revid -i rtsp://184.72.239.149/vod/mp4:BigBuckBunny_175k.mov -m r -o rtp://0.0.0.0:1234

4
revid/test.bat
View File

@ -1,4 +0,0 @@
@echo off
echo Running Revid with input: rtsp://184.72.239.149/vod/mp4:BigBuckBunny_175k.mov
echo and output: rtp://0.0.0.0:1234
revid -i rtsp://184.72.239.149/vod/mp4:BigBuckBunny_175k.mov -m r -o rtp://0.0.0.0:1234

48
revid/tools/tsFuncs.go
View File

@ -1,48 +0,0 @@
package tools
// mp2tDumpProgram dumps MPEG-TS Program Association Table (PAT) and Program Map Tables (PMT).
func mp2tDumpProgram(clip []byte) error {
// NB: Comcast API requires a buffered reader
reader := bufio.NewReader(bytes.NewReader(clip))
_, err := packet.Sync(reader)
if err != nil {
return fmt.Errorf("Error reading sync byte: %s", err)
}
pat, err := psi.ReadPAT(reader)
if err != nil {
return fmt.Errorf("Error reading PAT: %s", err)
}
mp2tDumpPat(pat)
var pmts []psi.PMT
pm := pat.ProgramMap()
for pn, pid := range pm {
pmt, err := psi.ReadPMT(reader, pid)
if err != nil {
return fmt.Errorf("Error reading PMT: %s", err)
}
pmts = append(pmts, pmt)
mp2tDumpPmt(pn, pmt)
}
return nil
}
func mp2tDumpPat(pat psi.PAT) {
fmt.Printf("Pat\n")
fmt.Printf("\tPMT PIDs %v\n", pat.ProgramMap())
fmt.Printf("\tNumber of Programs %v\n", pat.NumPrograms())
}
func mp2tDumpPmt(pn uint16, pmt psi.PMT) {
// pn = program number
fmt.Printf("Program #%v PMT\n", pn)
fmt.Printf("\tPIDs %v\n", pmt.Pids())
fmt.Printf("\tElementary Streams")
for _, es := range pmt.ElementaryStreams() {
fmt.Printf("\t\tPID %v : StreamType %v\n", es.ElementaryPid(), es.StreamType())
for _, d := range es.Descriptors() {
fmt.Printf("\t\t\t%+v\n", d)
}
}
}

3
tools/helpers.go
View File

@ -31,8 +31,7 @@ package tools
import (
_"os"
_"fmt"
"reflect"
"../rtp"
"bitbucket.org/ausocean/av/rtp"
)
func BoolToByte(in bool) (out byte) {

72
tsgenerator/tsgenerator.go → tsgenerator/TsGenerator.go
View File

@ -37,11 +37,11 @@ import (
"../rtp"
)
type TsConverter interface {
Convert()
type TsGenerator interface {
Generate()
}
type tsCreator struct {
type tsGenerator struct {
TsChan <-chan *mpegts.MpegTsPacket
tsChan chan<- *mpegts.MpegTsPacket
InputChan chan<- rtp.RtpPacket
@ -57,49 +57,49 @@ type tsCreator struct {
isGenerating bool
}
func NewTsCreator(fps uint) (c *tsCreator) {
c = new(tsCreator)
func NewTsGenerator(fps uint) (g *tsGenerator) {
g = new(tsGenerator)
tsChan := make(chan *mpegts.MpegTsPacket, 100)
c.TsChan = tsChan
c.tsChan = tsChan
g.TsChan = tsChan
g.tsChan = tsChan
inputChan := make(chan rtp.RtpPacket, 100)
c.InputChan = inputChan
c.inputChan = inputChan
g.InputChan = inputChan
g.inputChan = inputChan
nalInputChan := make(chan []byte, 10000)
c.NalInputChan = nalInputChan
c.nalInputChan = nalInputChan
c.currentCC = 0
c.fps = fps
c.currentPcrTime = .0
c.currentPtsTime = .7
g.NalInputChan = nalInputChan
g.nalInputChan = nalInputChan
g.currentCC = 0
g.fps = fps
g.currentPcrTime = .0
g.currentPtsTime = .7
return
}
func (c* tsgenerator) genPts()(pts uint64){
pts = uint64(c.currentPtsTime * float64(90000))
c.currentPtsTime += 1.0/float64(c.fps)
func (g *tsGenerator) genPts()(pts uint64){
pts = uint64(g.currentPtsTime * float64(90000))
g.currentPtsTime += 1.0/float64(g.fps)
return
}
func (c* tsgenerator) genPcr()(pcr uint64){
pcr = uint64(c.currentPcrTime * float64(90000))
c.currentPcrTime += 1.0/float64(c.fps)
func (g *tsGenerator) genPcr()(pcr uint64){
pcr = uint64(g.currentPcrTime * float64(90000))
g.currentPcrTime += 1.0/float64(g.fps)
return
}
func (c *tsgenerator) Stop(){
isGenerating = false
func (g *tsGenerator) Stop(){
g.isGenerating = false
}
func (c *tsgenerator) Convert() {
isGenerating = true
func (g *tsGenerator) Generate() {
g.isGenerating = true
pesPktChan := make(chan []byte, 1000)
payloadByteChan := make(chan byte, 100000)
var rtpBuffer [](*rtp.RtpPacket)
for isGenerating {
for g.isGenerating {
select {
default:
case rtpPacket := <-c.inputChan:
case rtpPacket := <-g.inputChan:
rtpBuffer = append(rtpBuffer, &rtpPacket)
if len(rtpBuffer) > 2 {
// if there's something weird going on with sequence numbers then
@ -149,7 +149,7 @@ func (c *tsgenerator) Convert() {
buffer = append(buffer, rtpBuffer[0].Payload[2:]...)
if tools.GetEndBit(rtpBuffer[0]) == 1 {
rtpBuffer = rtpBuffer[1:]
c.NalInputChan <- buffer
g.NalInputChan <- buffer
break
}
rtpBuffer = rtpBuffer[1:]
@ -169,16 +169,16 @@ func (c *tsgenerator) Convert() {
buffer = append(buffer, rtpBuffer[0].Payload[0]&0xE0|rtpBuffer[0].Payload[1]&0x1F)
buffer = append(buffer, rtpBuffer[0].Payload[2:]...)
rtpBuffer = rtpBuffer[1:]
c.NalInputChan <- buffer
g.NalInputChan <- buffer
default:
}
}
}
case nalUnit := <-c.nalInputChan:
case nalUnit := <-g.nalInputChan:
pesPkt := pes.PESPacket{
StreamID: 0xE0,
PDI: byte(2),
PTS: c.genPts(),
PTS: g.genPts(),
Data: nalUnit,
HeaderLength: 5,
}
@ -193,19 +193,19 @@ func (c *tsgenerator) Convert() {
PUSI: pusi,
PID: 256,
RAI: pusi,
CC: c.currentCC,
CC: g.currentCC,
AFC: byte(3),
PCRF: pusi,
}
pkt.FillPayload(payloadByteChan)
if pusi {
pkt.PCR = c.genPcr()
pkt.PCR = g.genPcr()
pusi = false
}
if c.currentCC++; c.currentCC > 15 {
c.currentCC = 0
if g.currentCC++; g.currentCC > 15 {
g.currentCC = 0
}
c.tsChan <- &pkt
g.tsChan <- &pkt
}
}
}