mirror of https://bitbucket.org/ausocean/av.git
Improved error reporting and other changes to make more Go idiomatic.
This commit is contained in:
Alan Noble
parent
d0e796f47d
commit
b54068e0a2
197
revid/revid.go
197
revid/revid.go
|
|
@ -40,6 +40,7 @@ import (
|
|||
"math/rand"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/exec"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
|
@ -50,32 +51,23 @@ import (
|
|||
"github.com/Comcast/gots/psi"
|
||||
)
|
||||
|
||||
// program modes
|
||||
const (
|
||||
File = iota
|
||||
HTTP
|
||||
UDP
|
||||
RTP
|
||||
Dump
|
||||
)
|
||||
|
||||
// defaults and networking consts
|
||||
const (
|
||||
defaultPid = 256
|
||||
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 // # 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
|
||||
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
|
||||
ffmpegPath = "/usr/bin/ffmpeg"
|
||||
)
|
||||
|
||||
// flags
|
||||
// flag values
|
||||
const (
|
||||
filterFixPTS = 0x0001
|
||||
filterDropAudio = 0x0002
|
||||
|
|
@ -91,27 +83,25 @@ const (
|
|||
// globals
|
||||
var (
|
||||
sendClip = sendClipToRTP
|
||||
packetsPerFrame = RTPPackets
|
||||
flags *int
|
||||
frameRate *int
|
||||
selectedPid *int
|
||||
packetsPerFrame = rtpPackets
|
||||
clipCount int
|
||||
expectCC int
|
||||
dumpCC int
|
||||
dumpPcrBase uint64
|
||||
dumpPCRBase uint64
|
||||
RTPSequenceNum uint16
|
||||
)
|
||||
|
||||
func main() {
|
||||
var (
|
||||
input = flag.String("i", "", "Input RTSP URL")
|
||||
output = 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)")
|
||||
// command-line flags
|
||||
var (
|
||||
input = flag.String("i", "", "Input RTSP URL")
|
||||
output = 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() {
|
||||
flag.Parse()
|
||||
|
||||
if *input == "" {
|
||||
|
|
@ -128,20 +118,20 @@ func main() {
|
|||
}
|
||||
case "u":
|
||||
sendClip = sendClipToUDP
|
||||
packetsPerFrame = UDPPackets
|
||||
packetsPerFrame = udpPackets
|
||||
if *output == "" {
|
||||
*output = defaultUDPOutput
|
||||
}
|
||||
case "r":
|
||||
sendClip = sendClipToRTP
|
||||
packetsPerFrame = RTPPackets
|
||||
packetsPerFrame = rtpPackets
|
||||
if *output == "" {
|
||||
*output = defaultRTPOutput
|
||||
}
|
||||
case "d":
|
||||
sendClip = sendClipToStdout
|
||||
default:
|
||||
log.Fatal("Invalid mode %s\n", *mode)
|
||||
log.Fatalf("Invalid mode %s\n", *mode)
|
||||
}
|
||||
|
||||
if *flags&filterFixContinuity != 0 && *flags&dumpProgramInfo != 0 {
|
||||
|
|
@ -149,9 +139,10 @@ func main() {
|
|||
}
|
||||
|
||||
for {
|
||||
_ = readWriteVideo(*input, *output)
|
||||
fmt.Printf("Trying again in 10s\n")
|
||||
time.Sleep(1e10 * time.Nanosecond)
|
||||
err := readWriteVideo(*input, *output)
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
fmt.Fprintln(os.Stderr, "Trying again in 10s")
|
||||
time.Sleep(10 * time.Second)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -160,78 +151,59 @@ func main() {
|
|||
func readWriteVideo(input string, output string) error {
|
||||
fmt.Printf("Reading video from %s\n", input)
|
||||
|
||||
var videoArg, audioArg [2]string
|
||||
args := []string{
|
||||
"-r", strconv.Itoa(*frameRate),
|
||||
"-i", input,
|
||||
}
|
||||
|
||||
if *flags&(filterFixPTS|filterScale640|filterScale320) == 0 {
|
||||
videoArg[0] = "-vcodec"
|
||||
videoArg[1] = "copy"
|
||||
args = append(args, "-vcodec", "copy")
|
||||
} else {
|
||||
videoArg[0] = "-vf"
|
||||
videoArg[1] = ""
|
||||
vfArg := []string{}
|
||||
|
||||
if *flags&filterFixPTS != 0 {
|
||||
videoArg[1] += "setpts='PTS-STARTPTS'" // start counting PTS from zero
|
||||
vfArg = append(vfArg, "setpts='PTS-STARTPTS'") // start counting PTS from zero
|
||||
}
|
||||
if *flags&filterScale640 != 0 {
|
||||
videoArg[1] += ", scale=640:352"
|
||||
vfArg = append(vfArg, "scale=640:352")
|
||||
} else if *flags&filterScale320 != 0 {
|
||||
videoArg[1] += ", scale=320:176"
|
||||
vfArg = append(vfArg, "scale=320:176")
|
||||
}
|
||||
args = append(args, "-vf", strings.Join(vfArg, ","))
|
||||
|
||||
}
|
||||
|
||||
if *flags&filterDropAudio == 0 {
|
||||
audioArg[0] = "-acodec"
|
||||
audioArg[1] = "copy"
|
||||
args = append(args, "-acodec", "copy")
|
||||
} else {
|
||||
audioArg[0] = "-an"
|
||||
audioArg[1] = ""
|
||||
}
|
||||
|
||||
fmt.Printf("Executing: %s -r %d -i %s %s \"%s\" %s %s -f mpegts -\n",
|
||||
ffmpegPath, *frameRate, input, videoArg[0], videoArg[1], audioArg[0], audioArg[1])
|
||||
|
||||
var cmd *exec.Cmd
|
||||
if audioArg[1] == "" {
|
||||
cmd = exec.Command(ffmpegPath,
|
||||
"-r", strconv.Itoa(*frameRate),
|
||||
"-i", input,
|
||||
videoArg[0], videoArg[1],
|
||||
audioArg[0],
|
||||
"-f", "mpegts", "-")
|
||||
|
||||
} else {
|
||||
cmd = exec.Command(ffmpegPath,
|
||||
"-r", strconv.Itoa(*frameRate),
|
||||
"-i", input,
|
||||
videoArg[0], videoArg[1],
|
||||
audioArg[0], audioArg[1],
|
||||
"-f", "mpegts", "-")
|
||||
args = append(args, "-an")
|
||||
}
|
||||
args = append(args, "-f", "mpegts", "-")
|
||||
|
||||
fmt.Printf("Executing: %s %s\n", ffmpegPath, strings.Join(args, " "))
|
||||
cmd := exec.Command(ffmpegPath, args...)
|
||||
stdout, err := cmd.StdoutPipe()
|
||||
if err != nil {
|
||||
fmt.Printf("Error creating pipe: %s\n", err)
|
||||
return err
|
||||
return fmt.Errorf("Error creating pipe: %s", err)
|
||||
}
|
||||
err = cmd.Start()
|
||||
if err != nil {
|
||||
fmt.Printf("Error starting pipe: %s\n", err)
|
||||
return err
|
||||
return fmt.Errorf("Error starting pipe: %s", err)
|
||||
}
|
||||
|
||||
// (re)initialize globals
|
||||
clipCount = 0
|
||||
expectCC = -1
|
||||
dumpCC = -1
|
||||
dumpPcrBase = 0
|
||||
dumpPCRBase = 0
|
||||
RTPSequenceNum = uint16(rand.Intn(1 << 15))
|
||||
|
||||
// for UDP and RTP only dial once
|
||||
var conn net.Conn
|
||||
if strings.Index(output, "udp://") == 0 || strings.Index(output, "rtp://") == 0 {
|
||||
if strings.HasPrefix(output, "udp://") || strings.HasPrefix(output, "rtp://") {
|
||||
conn, err = net.Dial("udp", output[6:])
|
||||
if err != nil {
|
||||
fmt.Printf("Error dialing %s: %s\n", output, err)
|
||||
return err
|
||||
return fmt.Errorf("Error dialing %s: %s", output, err)
|
||||
}
|
||||
defer conn.Close()
|
||||
}
|
||||
|
|
@ -241,17 +213,17 @@ func readWriteVideo(input string, output string) error {
|
|||
clip := make([]byte, mp2tMaxPackets*mp2tPacketSize)
|
||||
clipSize := 0
|
||||
packetCount := 0
|
||||
prevTime := time.Now()
|
||||
now := time.Now()
|
||||
prevTime := now
|
||||
fmt.Printf("Looping\n")
|
||||
|
||||
for {
|
||||
_, err := io.ReadFull(br, pkt)
|
||||
if err != nil {
|
||||
fmt.Printf("Error reading from ffmpeg: %s\n", err)
|
||||
return err
|
||||
return fmt.Errorf("Error reading from ffmpeg: %s", err)
|
||||
|
||||
}
|
||||
if *flags&filterFixContinuity != 0 && mp2tFixContinuity(pkt, packetCount, (uint16)(*selectedPid)) {
|
||||
if *flags&filterFixContinuity != 0 && mp2tFixContinuity(pkt, packetCount, uint16(*selectedPID)) {
|
||||
fmt.Printf("Packet #%d.%d fixed\n", clipCount, packetCount)
|
||||
}
|
||||
copy(clip[clipSize:], pkt)
|
||||
|
|
@ -259,15 +231,16 @@ func readWriteVideo(input string, output string) error {
|
|||
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) ||
|
||||
(time.Now().Sub(prevTime) > 1*time.Second && packetCount%packetsPerFrame == 0) {
|
||||
(now.Sub(prevTime) > 1*time.Second && packetCount%packetsPerFrame == 0) {
|
||||
clipCount++
|
||||
if err = sendClip(clip[:clipSize], output, conn); err != nil {
|
||||
return err
|
||||
}
|
||||
clipSize = 0
|
||||
packetCount = 0
|
||||
prevTime = time.Now()
|
||||
prevTime = now
|
||||
}
|
||||
}
|
||||
return nil
|
||||
|
|
@ -279,40 +252,36 @@ func sendClipToFile(clip []byte, _ string, _ net.Conn) error {
|
|||
fmt.Printf("Writing %s (%d bytes)\n", filename, len(clip))
|
||||
err := ioutil.WriteFile(filename, clip, 0644)
|
||||
if err != nil {
|
||||
fmt.Printf("Error writing file %s: %s\n", filename, err)
|
||||
return err
|
||||
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 {
|
||||
URL := output + strconv.Itoa(len(clip)) // NB: append the size to the output
|
||||
fmt.Printf("Posting %s (%d bytes)\n", URL, len(clip))
|
||||
resp, err := http.Post(URL, "video/mp2t", bytes.NewReader(clip)) // lighter than NewBuffer
|
||||
url := output + strconv.Itoa(len(clip)) // NB: append the size to the output
|
||||
fmt.Printf("Posting %s (%d bytes)\n", url, len(clip))
|
||||
resp, err := http.Post(url, "video/mp2t", bytes.NewReader(clip)) // lighter than NewBuffer
|
||||
if err != nil {
|
||||
fmt.Printf("Error posting to %s: %s\n", output, err)
|
||||
return err
|
||||
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(string(body) + "\n")
|
||||
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
|
||||
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 {
|
||||
fmt.Printf("UDP write error %s. Is your player listening?\n", err)
|
||||
return err
|
||||
|
||||
return fmt.Errorf("UDP write error %s. Is your player listening?", err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
|
|
@ -320,16 +289,15 @@ 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 {
|
||||
size := RTPPackets * mp2tPacketSize
|
||||
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)
|
||||
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 {
|
||||
fmt.Printf("RTP write error %s. Is your player listening?\n", err)
|
||||
return err
|
||||
return fmt.Errorf("RTP write error %s. Is your player listening?", err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
|
|
@ -351,11 +319,11 @@ func sendClipToStdout(clip []byte, _ string, _ net.Conn) error {
|
|||
packetCount++
|
||||
pkt := clip[offset : offset+mp2tPacketSize]
|
||||
|
||||
pktPid, err := packet.Pid(pkt)
|
||||
pktPID, err := packet.Pid(pkt)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if pktPid != (uint16)(*selectedPid) {
|
||||
if pktPID != uint16(*selectedPID) {
|
||||
continue
|
||||
}
|
||||
|
||||
|
|
@ -391,14 +359,14 @@ func sendClipToStdout(clip []byte, _ string, _ net.Conn) error {
|
|||
// adaptation field, followed by payload
|
||||
afl := adaptationfield.Length(pkt)
|
||||
if adaptationfield.HasPCR(pkt) {
|
||||
pcrBase, pcrExt, _ := mp2tGetPcr(pkt)
|
||||
pcrBase, pcrExt, _ := mp2tGetPCR(pkt)
|
||||
if *flags&dumpPacketHeader != 0 {
|
||||
fmt.Printf("\t\tAFL=%d, PCRbase=%d, PCRext=%d\n", afl, pcrBase, pcrExt)
|
||||
}
|
||||
if pcrBase < dumpPcrBase {
|
||||
if pcrBase < dumpPCRBase {
|
||||
fmt.Printf("Warning: PCRbase went backwards!\n")
|
||||
}
|
||||
dumpPcrBase = pcrBase
|
||||
dumpPCRBase = pcrBase
|
||||
} else if *flags&dumpPacketHeader != 0 {
|
||||
fmt.Printf("\t\tAFL=%d\n", afl)
|
||||
}
|
||||
|
|
@ -422,13 +390,11 @@ func mp2tDumpProgram(clip []byte) error {
|
|||
|
||||
_, err := packet.Sync(reader)
|
||||
if err != nil {
|
||||
fmt.Println("Warning: Bad MPEG-TS sync byte")
|
||||
return err
|
||||
return fmt.Errorf("Error reading sync byte: %s", err)
|
||||
}
|
||||
pat, err := psi.ReadPAT(reader)
|
||||
if err != nil {
|
||||
fmt.Printf("ReadPAT error: %s\n", err)
|
||||
return err
|
||||
return fmt.Errorf("Error reading PAT: %s", err)
|
||||
}
|
||||
mp2tDumpPat(pat)
|
||||
|
||||
|
|
@ -437,8 +403,7 @@ func mp2tDumpProgram(clip []byte) error {
|
|||
for pn, pid := range pm {
|
||||
pmt, err := psi.ReadPMT(reader, pid)
|
||||
if err != nil {
|
||||
fmt.Printf("ReadPMT error: %s\n", err)
|
||||
return err
|
||||
return fmt.Errorf("Error reading PMT: %s", err)
|
||||
}
|
||||
pmts = append(pmts, pmt)
|
||||
mp2tDumpPmt(pn, pmt)
|
||||
|
|
@ -458,7 +423,7 @@ func mp2tDumpPmt(pn uint16, pmt psi.PMT) {
|
|||
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())
|
||||
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)
|
||||
}
|
||||
|
|
@ -475,7 +440,7 @@ func mp2tFixContinuity(pkt []byte, packetCount int, pid uint16) bool {
|
|||
if !hasPayload {
|
||||
return false
|
||||
}
|
||||
if pktPid, _ := packet.Pid(pkt); pktPid != pid {
|
||||
if pktPID, _ := packet.Pid(pkt); pktPID != pid {
|
||||
return false
|
||||
}
|
||||
fixed := false
|
||||
|
|
@ -491,8 +456,8 @@ func mp2tFixContinuity(pkt []byte, packetCount int, pid uint16) bool {
|
|||
return fixed
|
||||
}
|
||||
|
||||
// Mp2tGetPcr extracts the Program Clock Reference (PCR) from an MPEG-TS packet (if any)
|
||||
func mp2tGetPcr(pkt []byte) (uint64, uint32, bool) {
|
||||
// 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
|
||||
}
|
||||
|
|
@ -523,8 +488,8 @@ func RTPEncapsulate(mp2tPacket []byte, pkt []byte) {
|
|||
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)
|
||||
binary.BigEndian.PutUint32(pkt[8:12], rtpSSRC)
|
||||
|
||||
// payload follows
|
||||
copy(pkt[RTPHeaderSize:RTPHeaderSize+RTPPackets*mp2tPacketSize], mp2tPacket)
|
||||
copy(pkt[rtpHeaderSize:rtpHeaderSize+rtpPackets*mp2tPacketSize], mp2tPacket)
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue