From e9fb26d624ed03ee09ec48dd274cc2126a64ca40 Mon Sep 17 00:00:00 2001 From: richardsonjack Date: Thu, 7 Dec 2017 11:54:15 +1030 Subject: [PATCH] Add 3 different options: no motion, edge detect, motion detect --- revid/revid.go | 228 +++++++++++++++++++++++++++++++++++++------------ 1 file changed, 175 insertions(+), 53 deletions(-) diff --git a/revid/revid.go b/revid/revid.go index 721399c6..a6e60fa3 100644 --- a/revid/revid.go +++ b/revid/revid.go @@ -36,9 +36,11 @@ import ( "encoding/hex" "flag" "fmt" + "image" "io" "io/ioutil" "log" + "math" "math/rand" "net" "net/http" @@ -58,21 +60,22 @@ import ( // 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 // # first multiple of 7 and 8 greater than 2000 + 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 + bufferSize = 1000/clipDuration bitrateOutputDelay = 60 // s httpTimeOut = 5 // s - clipDuration = 1 // s + motionThreshold = 0 ) // flag values @@ -82,6 +85,8 @@ const ( filterScale640 = 0x0004 filterScale320 = 0x0008 filterFixContinuity = 0x0010 + filterEdgeDetection = 0x0020 + filterMotionDetect = 0x0040 dumpProgramInfo = 0x0100 // 256 dumpPacketStats = 0x0200 // 512 dumpPacketHeader = 0x0400 // 1024 @@ -168,7 +173,7 @@ func main() { case err := <-inputErrChan: fmt.Fprintln(os.Stderr, err) fmt.Fprintln(os.Stderr, "Trying again in 10s") - time.Sleep(10 * time.Second) + //time.Sleep(10 * time.Second) go input(*inputURL, *outputURL) case err := <-outputErrChan: fmt.Fprintln(os.Stderr, err) @@ -195,49 +200,7 @@ func setUpDirs() { // input handles the reading from the specified input func input(input string, output string) { fmt.Printf("Reading video from %s\n", input) - - args := []string{ - "-r", strconv.Itoa(*frameRate), - "-i", input, - } - - if *flags&(filterFixPTS|filterScale640|filterScale320) == 0 { - args = append(args, "-vcodec", "copy") - } else { - vfArg := []string{} - - if *flags&filterFixPTS != 0 { - vfArg = append(vfArg, "setpts='PTS-STARTPTS'") // start counting PTS from zero - } - if *flags&filterScale640 != 0 { - vfArg = append(vfArg, "scale=640:352") - } else if *flags&filterScale320 != 0 { - vfArg = append(vfArg, "scale=320:176") - } - args = append(args, "-vf", strings.Join(vfArg, ",")) - - } - - if *flags&filterDropAudio == 0 { - args = append(args, "-acodec", "copy") - } else { - 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 { - inputErrChan <- err - return - } - err = cmd.Start() - if err != nil { - inputErrChan <- err - return - } - + // (re)initialize globals clipCount = 0 expectCC = -1 @@ -245,6 +208,8 @@ func input(input string, output string) { dumpPCRBase = 0 rtpSequenceNum = uint16(rand.Intn(1 << 15)) + var err error + // for UDP and RTP only dial once if strings.HasPrefix(output, "udp://") || strings.HasPrefix(output, "rtp://") { conn, err = net.Dial("udp", output[6:]) @@ -254,12 +219,115 @@ func input(input string, output string) { } } - br := bufio.NewReader(stdout) + var br *bufio.Reader + + if *flags&filterEdgeDetection != 0{ + args := []string{ + "-r", "25", + "-i", input, + "-pix_fmt", "gray", + "-vf", "edgedetect", + "-f", "mpegts", "-", + } + + fmt.Printf("Executing: %s %s\n", ffmpegPath, strings.Join(args, " ")) + cmd := exec.Command(ffmpegPath, args...) + stdout, _ := cmd.StdoutPipe() + + err := cmd.Start() + + if err != nil { + inputErrChan <- err + return + } + + br = bufio.NewReader(stdout) + }else{ + + + + if *flags&filterMotionDetect != 0{ + args := []string{ + "-i", input, + "-vf", "\"select=gt(scene,0.0001),setpts=N/(25*TB),fps=fps=25\"", + "-f", "mpegts", "-", + } + + fmt.Printf("Executing: %s %s\n", ffmpegPath, strings.Join(args, " ")) + cmd := exec.Command(ffmpegPath, args...) + stdout, _ := cmd.StdoutPipe() + + err := cmd.Start() + + if err != nil { + inputErrChan <- err + return + } + + br = bufio.NewReader(stdout) + + }else{ + args := []string{ + "-r", strconv.Itoa(*frameRate), + "-i", input, + } + + if *flags&(filterFixPTS|filterScale640|filterScale320) == 0 { + args = append(args, "-vcodec", "copy") + } else { + vfArg := []string{} + + if *flags&filterFixPTS != 0 { + vfArg = append(vfArg, "setpts='PTS-STARTPTS'") // start counting PTS from zero + } + if *flags&filterScale640 != 0 { + vfArg = append(vfArg, "scale=640:352") + } else if *flags&filterScale320 != 0 { + vfArg = append(vfArg, "scale=320:176") + } + args = append(args, "-vf", strings.Join(vfArg, ",")) + + } + + if *flags&filterDropAudio == 0 { + args = append(args, "-acodec", "copy") + } else { + 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 { + inputErrChan <- err + return + } + err = cmd.Start() + if err != nil { + inputErrChan <- err + return + } + + + + br = bufio.NewReader(stdout) + } + + } + + clipStore(br,input) + +} + +func clipStore(br *bufio.Reader, input string){ + clipSize := 0 packetCount := 0 now := time.Now() prevTime := now fmt.Printf("Looping\n") + for { if clip, err := ringBuffer.Get(); err != nil { inputErrChan <- err @@ -268,15 +336,18 @@ func input(input string, output string) { for { upperBound := clipSize + mp2tPacketSize _, err := io.ReadFull(br, clip[clipSize:upperBound]) + if err != nil { inputErrChan <- err return } - if *flags&filterFixContinuity != 0 && mp2tFixContinuity(clip[clipSize:upperBound], packetCount, uint16(*selectedPID)) { + + if *flags&filterFixContinuity != 0 && mp2tFixContinuity(clip[clipSize:upperBound], uint16(*selectedPID)) { fmt.Printf("Packet #%d.%d fixed\n", clipCount, packetCount) } 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) || @@ -296,6 +367,42 @@ func input(input string, output string) { } } +func printGray(img *image.Gray) { + shade := []string{" ", "█"} + for i, p := range img.Pix { + x := 0 + if p > 20 { + x = 1 + } + fmt.Print(shade[x]) + if (i+1)%640 == 0 { + fmt.Print("\n") + } + } +} + +func checkSlice(a,b *image.Gray){ + for i , _ := range a.Pix{ + fmt.Printf("%v - %v = %v\n", a.Pix[i], b.Pix[i], a.Pix[i]-b.Pix[i]) + + } +} + + +func FastCompare(img1, img2 *image.Gray) (float64, error) { + if img1.Bounds() != img2.Bounds() { + return 0, fmt.Errorf("image bounds not equal: %+v, %+v", img1.Bounds(), img2.Bounds()) + } + + accumError := float64(0) + + for i := 0; i < len(img1.Pix); i+= 20 { + accumError += float64((img1.Pix[i] - img2.Pix[i]) * (img1.Pix[i] - img2.Pix[i])) + } + + return math.Sqrt(accumError), nil +} + // output handles the writing to specified output func output(output string) { elapsedTime := time.Duration(0) @@ -304,7 +411,9 @@ func output(output string) { for { if clip, err := ringBuffer.Read(); err == nil { now := time.Now() + err := sendClip(clip, output, conn) + sendClipToStdout(clip, output, conn) for err != nil { outputErrChan <- err err = sendClip(clip, output, conn) @@ -440,6 +549,7 @@ func sendClipToStdout(clip []byte, _ string, _ net.Conn) error { 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++ @@ -458,14 +568,14 @@ func sendClipToStdout(clip []byte, _ string, _ net.Conn) error { if adaptationfield.HasPCR(pkt) { pcrBase, pcrExt, _ := mp2tGetPCR(pkt) if *flags&dumpPacketHeader != 0 { - fmt.Printf("\t\tAFL=%d, PCRbase=%d, PCRext=%d\n", afl, pcrBase, pcrExt) + 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\n", afl) + fmt.Printf("\t\tAFL=%d, DI=%v\n", afl, di) } } if *flags&dumpPacketPayload != 0 { @@ -528,10 +638,11 @@ func mp2tDumpPmt(pn uint16, pmt psi.PMT) { } // Mp2tFixContinuity fixes discontinous MPEG-TS continuity counts (CC) -func mp2tFixContinuity(pkt []byte, packetCount int, pid uint16) bool { +func mp2tFixContinuity(pkt []byte, pid uint16) bool { + hasPayload, err := packet.ContainsPayload(pkt) if err != nil { - fmt.Printf("Warning: Packet #%d.%d bad.\n", clipCount, packetCount) + fmt.Printf("Warning: Packet bad.\n") return false } if !hasPayload { @@ -546,6 +657,7 @@ func mp2tFixContinuity(pkt []byte, packetCount int, pid uint16) bool { if expectCC == -1 { expectCC = cc } else if cc != expectCC { + fmt.Println("Have to fix") pkt[3] = pkt[3]&0xf0 | byte(expectCC&0xf) fixed = true } @@ -553,6 +665,16 @@ func mp2tFixContinuity(pkt []byte, packetCount int, pid uint16) bool { return fixed } +func resetContinuityCount(pkt []byte) (bool){ + + if afc := pkt[3] & 0x30 >> 4; afc == 3{ + pkt[5] = pkt[5] | byte(1 << 7) + } + + return true + +} + // 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) {