From b54068e0a2a1fc9d708607fbdf081627c927fcbb Mon Sep 17 00:00:00 2001 From: Alan Noble Date: Thu, 21 Sep 2017 13:36:54 +0930 Subject: [PATCH] Improved error reporting and other changes to make more Go idiomatic. --- revid/revid.go | 197 ++++++++++++++++++++----------------------------- 1 file changed, 81 insertions(+), 116 deletions(-) diff --git a/revid/revid.go b/revid/revid.go index 9a36d763..21bc5f09 100644 --- a/revid/revid.go +++ b/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) }