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

Update revid to incorperate use of RingBuffer

This commit is contained in:
Saxon Milton 2017-12-02 15:57:10 +10:30
parent e3d229d579
commit 0f028b799c
4 changed files with 147 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
revid/Readme.md
View File

@ -96,7 +96,7 @@ Alan Noble <anoble@gmail.com>
# License
revid is Copyright (C) 2017 Alan Noble.
ringBuffer 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

169
revid/revid.go
View File

@ -31,7 +31,9 @@ package main
import (
"bufio"
"bytes"
"crypto/md5"
"encoding/binary"
"encoding/hex"
"flag"
"fmt"
"io"
@ -47,6 +49,8 @@ import (
"strings"
"time"
"../utilities"
"github.com/Comcast/gots/packet"
"github.com/Comcast/gots/packet/adaptationfield"
"github.com/Comcast/gots/psi"
@ -65,8 +69,10 @@ const (
rtpPackets = 7 // # of RTP packets per ethernet frame (7 is the max)
rtpHeaderSize = 12
rtpSSRC = 1 // any value will do
bitsInByte = 8
bufferSize = 1000
bitrateOutputDelay = 60 // s
httpTimeOut = 5 // s
clipDuration = 1 // s
)
// flag values
@ -91,14 +97,18 @@ var (
dumpCC int
dumpPCRBase uint64
rtpSequenceNum uint16
conn net.Conn
ffmpegPath string
tempDir string
inputErrChan chan error
outputErrChan chan error
ringBuffer utilities.RingBuffer
)
// command-line flags
var (
input = flag.String("i", "", "Input RTSP URL")
output = flag.String("o", "", "Output URL (HTTP, UDP or RTP)")
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)")
@ -109,7 +119,7 @@ func main() {
setUpDirs()
flag.Parse()
if *input == "" {
if *inputURL == "" {
log.Fatal("Input (-i) required\n")
}
@ -118,20 +128,20 @@ func main() {
sendClip = sendClipToFile
case "h":
sendClip = sendClipToHTTP
if *output == "" {
*output = defaultHTTPOutput
if *outputURL == "" {
*outputURL = defaultHTTPOutput
}
case "u":
sendClip = sendClipToUDP
packetsPerFrame = udpPackets
if *output == "" {
*output = defaultUDPOutput
if *outputURL == "" {
*outputURL = defaultUDPOutput
}
case "r":
sendClip = sendClipToRTP
packetsPerFrame = rtpPackets
if *output == "" {
*output = defaultRTPOutput
if *outputURL == "" {
*outputURL = defaultRTPOutput
}
case "d":
sendClip = sendClipToStdout
@ -143,28 +153,47 @@ func main() {
log.Fatal("Cannot combine filterFixContinuity and dumpProgramInfo flags\n")
}
ringBuffer.Make(bufferSize, mp2tPacketSize*mp2tMaxPackets)
inputErrChan = make(chan error, 10)
outputErrChan = make(chan error, 10)
go input(*inputURL, *outputURL)
go output(*outputURL)
// Sit in here once we execute the go routines and handle errors that may
// appear in the error channels
for {
err := readWriteVideo(*input, *output)
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!")
}
}
}
// set up directories based on the OS we're running on
// setUpDirs sets directories based on the OS that Revid is running on
func setUpDirs() {
if runtime.GOOS == "windows" {
switch runtime.GOOS {
case "windows":
ffmpegPath = "C:/ffmpeg/ffmpeg"
tempDir = "tmp/"
} else {
case "darwin":
ffmpegPath = "/usr/local/bin/ffmpeg"
tempDir = "/tmp/"
default:
ffmpegPath = "/usr/bin/ffmpeg"
tempDir = "/tmp/"
}
}
// readWriteVideo reads video from an RTSP stream (specified by the input URL) and
// rewrites the video in various formats and/or different protocols (HTTP, UDP or RTP).
func readWriteVideo(input string, output string) error {
// input handles the reading from the specified input
func input(input string, output string) {
fmt.Printf("Reading video from %s\n", input)
args := []string{
@ -200,11 +229,13 @@ func readWriteVideo(input string, output string) error {
cmd := exec.Command(ffmpegPath, args...)
stdout, err := cmd.StdoutPipe()
if err != nil {
return fmt.Errorf("Error creating pipe: %s", err)
inputErrChan <- err
return
}
err = cmd.Start()
if err != nil {
return fmt.Errorf("Error starting pipe: %s", err)
inputErrChan <- err
return
}
// (re)initialize globals
@ -215,30 +246,31 @@ func readWriteVideo(input string, output string) error {
rtpSequenceNum = uint16(rand.Intn(1 << 15))
// for UDP and RTP only dial once
var conn net.Conn
if strings.HasPrefix(output, "udp://") || strings.HasPrefix(output, "rtp://") {
conn, err = net.Dial("udp", output[6:])
if err != nil {
return fmt.Errorf("Error dialing %s: %s", output, err)
inputErrChan <- err
return
}
defer conn.Close()
}
br := bufio.NewReader(stdout)
pkt := make([]byte, mp2tPacketSize)
clip := make([]byte, mp2tMaxPackets*mp2tPacketSize)
clipSize := 0
packetCount := 0
now := time.Now()
prevTime := now
fmt.Printf("Looping\n")
elapsedTime := time.Duration(0)
for {
if clip, err := ringBuffer.Get(); err != nil {
inputErrChan <- err
return
} else {
for {
_, err := io.ReadFull(br, pkt)
if err != nil {
return fmt.Errorf("Error reading from ffmpeg: %s", err)
inputErrChan <- err
return
}
if *flags&filterFixContinuity != 0 && mp2tFixContinuity(pkt, packetCount, uint16(*selectedPID)) {
fmt.Printf("Packet #%d.%d fixed\n", clipCount, packetCount)
@ -246,30 +278,51 @@ func readWriteVideo(input string, output string) error {
copy(clip[clipSize:], pkt)
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) > 1*time.Second && packetCount%packetsPerFrame == 0) {
(now.Sub(prevTime) > clipDuration*time.Second && packetCount%packetsPerFrame == 0) {
clipCount++
if err = sendClip(clip[:clipSize], output, conn); err != nil {
return err
}
// Calculate bitrate and Output
deltaTime := now.Sub(prevTime)
elapsedTime += deltaTime
if elapsedTime > bitrateOutputDelay*time.Nanosecond {
noOfBits := float64(clipSize * bitsInByte) / 1024.0 // to kbits
fmt.Printf("Bitrate: %d kbps\n", int64(noOfBits/float64(deltaTime/1e9)))
elapsedTime = time.Duration(0)
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, clipSize, _ := ringBuffer.Read(); clip != nil {
now := time.Now()
err := sendClip(clip[:clipSize], output, conn)
for err != nil {
outputErrChan <- err
err = sendClip(clip[:clipSize], output, conn)
}
deltaTime := now.Sub(prevTime)
elapsedTime += deltaTime
if elapsedTime > bitrateOutputDelay*time.Second {
noOfBits := float64(len(clip[:clipSize])*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
}
}
}
return nil
}
// sendClipToFile writes a video clip to a /tmp file.
@ -285,9 +338,21 @@ func sendClipToFile(clip []byte, _ string, _ net.Conn) error {
// sendClipToHTPP posts a video clip via HTTP, using a new TCP connection each time.
func sendClipToHTTP(clip []byte, output string, _ net.Conn) error {
url := output + strconv.Itoa(len(clip)) // NB: append the size to the output
if err := checkContinuityCounts(clip); err != nil {
return err
}
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 := http.Post(url, "video/mp2t", bytes.NewReader(clip)) // lighter than NewBuffer
resp, err := client.Post(url, "video/mp2t", bytes.NewReader(clip)) // lighter than NewBuffer
if err := checkContinuityCounts(clip); err != nil {
return err
}
if err != nil {
return fmt.Errorf("Error posting to %s: %s", output, err)
}
@ -315,6 +380,9 @@ func sendClipToUDP(clip []byte, _ string, conn net.Conn) error {
// sendClipToRTP sends a video clip over RTP.
func sendClipToRTP(clip []byte, _ string, conn net.Conn) error {
if err := checkContinuityCounts(clip); err != nil {
return err
}
size := rtpPackets * mp2tPacketSize
fmt.Printf("Sending %d RTP packets of size %d (%d bytes)\n",
len(clip)/size, size+rtpHeaderSize, len(clip))
@ -329,6 +397,21 @@ func sendClipToRTP(clip []byte, _ string, conn net.Conn) error {
return nil
}
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, _ string, _ net.Conn) error {
fmt.Printf("Dumping clip (%d bytes)\n", len(clip))

4
revid/test.bash Normal file
View File

@ -0,0 +1,4 @@
#!/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 Normal file
View File

@ -0,0 +1,4 @@
@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