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

Updated logging

This commit is contained in:
Unknown 2018-01-24 12:10:03 +10:30
parent bd9ff08c12
commit 0b28a304fd
1 changed files with 46 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

95
revid/RevidInstance.go
View File

@ -72,6 +72,13 @@ const (
bitrateTime = 60 bitrateTime = 60
) )
const (
Error = "Error"
Warning = "Warning"
Info = "Info"
Debug = "Debug"
)
const ( const (
Raspivid = 0 Raspivid = 0
Rtp = 1 Rtp = 1
@ -106,8 +113,6 @@ type RevidInst interface {
} }
type revidInst struct { type revidInst struct {
expectCC int
dumpCC int
dumpPCRBase uint64 dumpPCRBase uint64
conn net.Conn conn net.Conn
ffmpegPath string ffmpegPath string
@ -115,7 +120,6 @@ type revidInst struct {
ringBuffer ringbuffer.RingBuffer ringBuffer ringbuffer.RingBuffer
config Config config Config
isRunning bool isRunning bool
Error *log.Logger
outputFile *os.File outputFile *os.File
inputFile *os.File inputFile *os.File
generator tsgenerator.TsGenerator generator tsgenerator.TsGenerator
@ -129,9 +133,6 @@ func (r *revidInst) GetConfigRef() *Config{
func NewRevidInstance(config Config) (r *revidInst, err error) { func NewRevidInstance(config Config) (r *revidInst, err error) {
r = new(revidInst) r = new(revidInst)
r.ringBuffer = ringbuffer.NewRingBuffer(bufferSize, mp2tPacketSize*mp2tMaxPackets) 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.dumpPCRBase = 0
r.ChangeState(config) r.ChangeState(config)
switch r.config.Output { switch r.config.Output {
@ -156,8 +157,8 @@ func NewRevidInstance(config Config) (r *revidInst, err error) {
go r.h264Parser.Parse() go r.h264Parser.Parse()
go r.input() go r.input()
go r.generator.Generate() go r.generator.Generate()
r.ErrOut("New revid instance created! config is:") r.Log(Info, "New revid instance created! config is:")
r.ErrOut(fmt.Sprintf("%v",r.config)) r.Log(Info, fmt.Sprintf("%v",r.config))
return return
} }
@ -167,8 +168,8 @@ func (r *revidInst) ChangeState(newconfig Config) error {
return nil return nil
} }
func (r *revidInst) ErrOut(m string){ func (r *revidInst) Log(type, m string){
r.config.Logger.Log("Debug",m) r.config.Logger.Log(type,m)
} }
func (r *revidInst)IsRunning() bool { func (r *revidInst)IsRunning() bool {
@ -177,58 +178,53 @@ func (r *revidInst)IsRunning() bool {
func (r *revidInst) Start() { func (r *revidInst) Start() {
if r.isRunning { if r.isRunning {
r.ErrOut("Start() has been called but revid already running!") r.Log(Warning,"Start() has been called but revid already running!")
return return
} }
r.ErrOut("Starting Revid!") r.Log(Debug,"Starting Revid!")
var h264Data []byte
switch r.config.Input { switch r.config.Input {
case Raspivid: case Raspivid:
r.ErrOut("Starting raspivid!") r.Log(Debug,"Starting raspivid!")
cmd = exec.Command("raspivid", "-o", "-", "-n", "-t", "0", "-b", cmd = exec.Command("raspivid", "-o", "-", "-n", "-t", "0", "-b",
r.config.Bitrate,"-qp", r.config.Quantization, "-w", r.config.Width, "-h", r.config.Height, "-fps", r.config.FrameRate, "-ih", "-g", "100") r.config.Bitrate,"-qp", r.config.Quantization, "-w", r.config.Width, "-h", r.config.Height, "-fps", r.config.FrameRate, "-ih", "-g", "100")
stdout, _ := cmd.StdoutPipe() stdout, _ := cmd.StdoutPipe()
err := cmd.Start() err := cmd.Start()
inputReader = bufio.NewReader(stdout) inputReader = bufio.NewReader(stdout)
if err != nil { if err != nil {
r.ErrOut(err.Error()) r.Log(Error,err.Error())
return return
} }
case File: r.isRunning = true
default:
r.ErrOut("Input not valid!")
}
r.isRunning = true
var h264Data []byte
switch r.config.Input {
case Raspivid:
go func() { go func() {
r.ErrOut("Reading camera data!") r.Log(Debug, "Reading camera data!")
for r.isRunning { for r.isRunning {
h264Data = make([]byte, 1) h264Data = make([]byte, 1)
_, err := io.ReadFull(inputReader, h264Data) _, err := io.ReadFull(inputReader, h264Data)
if err != nil { if err != nil {
if err.Error() == "EOF" && r.isRunning { switch{
r.ErrOut("No data from camera!") case err.Error() == "EOF" && r.isRunning:
r.Log(Error, "No data from camera!")
time.Sleep(5*time.Second) time.Sleep(5*time.Second)
} else { case r.isRunning:
r.ErrOut(err.Error()) r.Log(Error, err.Error())
} }
} else { } else {
r.h264Parser.InputByteChan <- h264Data[0] r.h264Parser.InputByteChan <- h264Data[0]
} }
} }
r.ErrOut("Out of reading routine") r.Log(Debug, "Out of reading routine!")
}() }()
case File: case File:
stats, err := r.inputFile.Stat() stats, err := r.inputFile.Stat()
if err != nil { if err != nil {
r.ErrOut("Could not get file stats!") r.Log(Error, "Could not get input file stats!")
return return
} }
h264Data = make([]byte, stats.Size()) h264Data = make([]byte, stats.Size())
_, err = r.inputFile.Read(h264Data) _, err = r.inputFile.Read(h264Data)
if err != nil { if err != nil {
r.ErrOut(err.Error()) r.Log(Error, err.Error())
} }
for i := range h264Data { for i := range h264Data {
r.h264Parser.InputByteChan <- h264Data[i] r.h264Parser.InputByteChan <- h264Data[i]
@ -239,7 +235,7 @@ func (r *revidInst) Start() {
func (r *revidInst) Stop() { func (r *revidInst) Stop() {
if r.isRunning { if r.isRunning {
r.ErrOut("Stopping revid!") r.Log(Debug,"Stopping revid!")
r.isRunning = false r.isRunning = false
cmd.Process.Kill() cmd.Process.Kill()
} }
@ -252,7 +248,8 @@ func (r *revidInst) input() {
prevTime := now prevTime := now
for { for {
if clip, err := r.ringBuffer.Get(); err != nil { if clip, err := r.ringBuffer.Get(); err != nil {
r.ErrOut(err.Error()) r.Log(Error,err.Error())
r.Log(Warning,"Clearing tsPkt chan!")
tsPktChanLen := len(r.generator.GetTsOutputChan()) tsPktChanLen := len(r.generator.GetTsOutputChan())
for i := 0; i < tsPktChanLen; i++ { for i := 0; i < tsPktChanLen; i++ {
<-(r.generator.GetTsOutputChan()) <-(r.generator.GetTsOutputChan())
@ -263,7 +260,7 @@ func (r *revidInst) input() {
tsPacket := <-(r.generator.GetTsOutputChan()) tsPacket := <-(r.generator.GetTsOutputChan())
byteSlice, err := tsPacket.ToByteSlice() byteSlice, err := tsPacket.ToByteSlice()
if err != nil { if err != nil {
r.ErrOut(err.Error()) r.Log(Error,err.Error())
} }
upperBound := clipSize + mp2tPacketSize upperBound := clipSize + mp2tPacketSize
copy(clip[clipSize:upperBound], byteSlice) copy(clip[clipSize:upperBound], byteSlice)
@ -274,8 +271,8 @@ func (r *revidInst) input() {
if (packetCount == mp2tMaxPackets) || if (packetCount == mp2tMaxPackets) ||
(now.Sub(prevTime) > clipDuration*time.Second && packetCount%packetsPerFrame == 0) { (now.Sub(prevTime) > clipDuration*time.Second && packetCount%packetsPerFrame == 0) {
if err := r.ringBuffer.DoneWriting(clipSize); err != nil { if err := r.ringBuffer.DoneWriting(clipSize); err != nil {
r.ErrOut(err.Error()) r.Log(Error,err.Error())
r.ErrOut("Dropping that clip!") r.Log(Warning,"Dropping clip!")
} }
clipSize = 0 clipSize = 0
packetCount = 0 packetCount = 0
@ -293,36 +290,36 @@ func (r *revidInst) output() {
bytes := 0 bytes := 0
delay := 0 delay := 0
for r.isRunning { for r.isRunning {
if r.ringBuffer.GetNoOfElements() < 2 { switch{
case r.ringBuffer.GetNoOfElements() < 2:
delay++ delay++
time.Sleep(time.Duration(delay)*time.Millisecond) time.Sleep(time.Duration(delay)*time.Millisecond)
} else { case delay > 10:
if delay > 10 { delay -= 10
delay -= 10
}
} }
if clip, err := r.ringBuffer.Read(); err == nil { if clip, err := r.ringBuffer.Read(); err == nil {
r.ErrOut(fmt.Sprintf("Delay is: %v\n", delay)) r.Log(Debug,fmt.Sprintf("Delay is: %v\n", delay))
r.ErrOut(fmt.Sprintf("Ring buffer size: %v\n", r.ringBuffer.GetNoOfElements()))
switch r.config.Output { switch r.config.Output {
case File: case File:
r.outputFile.Write(clip) r.outputFile.Write(clip)
case HttpOut: case HttpOut:
bytes += len(clip) bytes += len(clip)
for err := r.sendClipToHTTP(clip, r.config.HttpAddress); err != nil; { for err := r.sendClipToHTTP(clip, r.config.HttpAddress); err != nil; {
r.ErrOut("Post failed trying again!") r.Log(Error,err.Error())
r.Log(Warning,"Post failed trying again!")
err = r.sendClipToHTTP(clip, r.config.HttpAddress) err = r.sendClipToHTTP(clip, r.config.HttpAddress)
} }
default: default:
r.ErrOut("No output defined!") r.Log(Error,"No output defined!")
} }
if err := r.ringBuffer.DoneReading(); err != nil { if err := r.ringBuffer.DoneReading(); err != nil {
r.ErrOut(err.Error()) r.Log(Error,err.Error())
} }
now = time.Now() now = time.Now()
deltaTime := now.Sub(prevTime) deltaTime := now.Sub(prevTime)
if deltaTime > time.Duration(bitrateTime)*time.Second { if deltaTime > time.Duration(bitrateTime)*time.Second {
fmt.Printf("Bitrate: %v bytes/s\n", int64(float64(bytes) / float64(deltaTime/1e9))) r.Log(Info,fmt.Sprintf("Bitrate: %v bits/s\n", int64(float64(bytes*8) / float64(deltaTime/1e9))))
r.Log(Info,fmt.Sprintf("Ring buffer size: %v\n", r.ringBuffer.GetNoOfElements()))
prevTime = now prevTime = now
bytes = 0 bytes = 0
} }
@ -337,7 +334,7 @@ func (r *revidInst)sendClipToHTTP(clip []byte, output string) error {
Timeout: timeout, Timeout: timeout,
} }
url := output + strconv.Itoa(len(clip)) url := output + strconv.Itoa(len(clip))
r.ErrOut(fmt.Sprintf("Posting %s (%d bytes)\n", url, len(clip))) r.Log(Debug,fmt.Sprintf("Posting %s (%d bytes)\n", url, len(clip)))
resp, err := client.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 != nil { if err != nil {
return fmt.Errorf("Error posting to %s: %s", output, err) return fmt.Errorf("Error posting to %s: %s", output, err)
@ -345,9 +342,9 @@ func (r *revidInst)sendClipToHTTP(clip []byte, output string) error {
defer resp.Body.Close() defer resp.Body.Close()
body, err := ioutil.ReadAll(resp.Body) body, err := ioutil.ReadAll(resp.Body)
if err == nil { if err == nil {
r.ErrOut(fmt.Sprintf("%s\n", body)) r.Log(Debug,fmt.Sprintf("%s\n", body))
} else { } else {
r.ErrOut(err.Error()) r.Log(Error,err.Error())
} }
return nil return nil
} }