kerberos-io/agent
1
0
Fork 0
mirror of https://github.com/kerberos-io/agent.git synced 2026-08-23 15:08:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

move to separate package + add recording method

Browse Source
This commit is contained in:
Thomas Quandalle
2022-06-21 17:06:27 +02:00
parent 3114745899
commit 02f02427bf
5 changed files with 142 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

111
machinery/src/capture/IPCamera.go Normal file
View File

@@ -0,0 +1,111 @@
package capture
import (
"strconv"
"time"
"github.com/kerberos-io/agent/machinery/src/log"
"github.com/kerberos-io/agent/machinery/src/models"
"github.com/kerberos-io/joy4/av/pubsub"
"github.com/kerberos-io/joy4/av"
"github.com/kerberos-io/joy4/av/avutil"
"github.com/kerberos-io/joy4/cgo/ffmpeg"
"github.com/kerberos-io/joy4/format"
)
func OpenRTSP(url string) (av.DemuxCloser, []av.CodecData, error) {
format.RegisterAll()
infile, err := avutil.Open(url)
if err == nil {
streams, errstreams := infile.Streams()
return infile, streams, errstreams
}
return nil, []av.CodecData{}, err
}
func GetVideoDecoder(streams []av.CodecData) *ffmpeg.VideoDecoder {
// Load video codec
var vstream av.VideoCodecData
for _, stream := range streams {
if stream.Type().IsAudio() {
//astream := stream.(av.AudioCodecData)
} else if stream.Type().IsVideo() {
vstream = stream.(av.VideoCodecData)
}
}
dec, _ := ffmpeg.NewVideoDecoder(vstream)
return dec
}
func HandleStream(infile av.DemuxCloser, queue *pubsub.Queue, communication *models.Communication) { //, wg *sync.WaitGroup) {
log.Log.Debug("HandleStream: started")
var err error
loop:
for {
// This will check if we need to stop the thread,
// because of a reconfiguration.
select {
case <-communication.HandleStream:
break loop
default:
}
var pkt av.Packet
if pkt, err = infile.ReadPacket(); err != nil { // sometimes this throws an end of file..
log.Log.Info(strconv.Itoa(len(pkt.Data)))
log.Log.Error(err.Error())
if err.Error() == "EOF" {
time.Sleep(30 * time.Second)
}
}
// Could be that a decode is throwing errors.
if len(pkt.Data) > 0 {
queue.WritePacket(pkt)
// This will check if we need to stop the thread,
// because of a reconfiguration.
select {
case <-communication.HandleStream:
break loop
default:
}
/*select {
case packetsBuffer <- pkt:
default:
}
select {
case webrtcPacketsRealtimeStream <- pkt:
default:
}*/
if pkt.IsKeyFrame {
// Increment packets, so we know the device
// is not blocking.
r := communication.PackageCounter.Load().(int64)
log.Log.Info("HandleStream: packet size " + strconv.Itoa(len(pkt.Data)))
communication.PackageCounter.Store((r + 1) % 1000)
/*select {
case packetsRealtime <- pkt:
default:
}
select {
case packetsRealtimeStream <- pkt:
default:
}*/
}
}
}
//wg.Done()
queue.Close()
log.Log.Debug("HandleStream: finished")
}

136
machinery/src/components/IPCamera.go → machinery/src/capture/main.go
View File

@@ -1,4 +1,4 @@
package components
package capture
import (
"os"
@@ -14,110 +14,12 @@ import (
"github.com/kerberos-io/joy4/format/mp4"
"github.com/kerberos-io/joy4/av"
"github.com/kerberos-io/joy4/av/avutil"
"github.com/kerberos-io/joy4/cgo/ffmpeg"
"github.com/kerberos-io/joy4/format"
)
func OpenRTSP(url string) (av.DemuxCloser, []av.CodecData, error) {
format.RegisterAll()
infile, err := avutil.Open(url)
if err == nil {
streams, errstreams := infile.Streams()
return infile, streams, errstreams
}
return nil, []av.CodecData{}, err
}
func GetVideoDecoder(streams []av.CodecData) *ffmpeg.VideoDecoder {
// Load video codec
var vstream av.VideoCodecData
for _, stream := range streams {
if stream.Type().IsAudio() {
//astream := stream.(av.AudioCodecData)
} else if stream.Type().IsVideo() {
vstream = stream.(av.VideoCodecData)
}
}
dec, _ := ffmpeg.NewVideoDecoder(vstream)
return dec
}
func HandleStream(infile av.DemuxCloser, queue *pubsub.Queue, communication *models.Communication) { //, wg *sync.WaitGroup) {
log.Log.Debug("HandleStream: started")
var err error
loop:
for {
// This will check if we need to stop the thread,
// because of a reconfiguration.
select {
case <-communication.HandleStream:
break loop
default:
}
var pkt av.Packet
if pkt, err = infile.ReadPacket(); err != nil { // sometimes this throws an end of file..
log.Log.Info(strconv.Itoa(len(pkt.Data)))
log.Log.Error(err.Error())
if err.Error() == "EOF" {
time.Sleep(30 * time.Second)
}
}
// Could be that a decode is throwing errors.
if len(pkt.Data) > 0 {
queue.WritePacket(pkt)
// This will check if we need to stop the thread,
// because of a reconfiguration.
select {
case <-communication.HandleStream:
break loop
default:
}
/*select {
case packetsBuffer <- pkt:
default:
}
select {
case webrtcPacketsRealtimeStream <- pkt:
default:
}*/
if pkt.IsKeyFrame {
// Increment packets, so we know the device
// is not blocking.
r := communication.PackageCounter.Load().(int64)
log.Log.Info("HandleStream: packet size " + strconv.Itoa(len(pkt.Data)))
communication.PackageCounter.Store((r + 1) % 1000)
/*select {
case packetsRealtime <- pkt:
default:
}
select {
case packetsRealtimeStream <- pkt:
default:
}*/
}
}
}
//wg.Done()
queue.Close()
log.Log.Debug("HandleStream: finished")
}
func RecordStream(recordingCursor *pubsub.QueueCursor, motion chan int64, devicename string, config *models.Config, streams []av.CodecData) {
log.Log.Debug("RecordStream: started")
func HandleRecordStream(recordingCursor *pubsub.QueueCursor, configuration *models.Configuration, communication *models.Communication, streams []av.CodecData) {
log.Log.Debug("HandleRecordStream: started")
config := configuration.Config
recordingPeriod := config.Capture.PostRecording // number of seconds to record.
maxRecordingPeriod := config.Capture.MaxLengthRecording // maximum number of seconds to record.
@@ -130,7 +32,7 @@ func RecordStream(recordingCursor *pubsub.QueueCursor, motion chan int64, device
if config.Capture.Continuous == "true" {
// Do not do anything!
log.Log.Info("Start continuous recording ")
log.Log.Info("HandleRecordStream: Start continuous recording ")
loc, _ := time.LoadLocation(config.Timezone)
now = time.Now().Unix()
@@ -165,7 +67,7 @@ func RecordStream(recordingCursor *pubsub.QueueCursor, motion chan int64, device
log.Log.Error(err.Error())
}
log.Log.Info("Recording finished: file save: " + name)
log.Log.Info("HandleRecordStream: Recording finished: file save: " + name)
file.Close()
// Check if need to convert to fragmented using bento
@@ -206,7 +108,7 @@ func RecordStream(recordingCursor *pubsub.QueueCursor, motion chan int64, device
(currentTimeInSeconds >= start2 && currentTimeInSeconds <= end2) {
} else {
log.Log.Debug("Disabled: no continuous recording at this moment. Not within specified time interval.")
log.Log.Debug("HandleRecordStream: Disabled: no continuous recording at this moment. Not within specified time interval.")
time.Sleep(5 * time.Second)
continue
}
@@ -237,8 +139,8 @@ func RecordStream(recordingCursor *pubsub.QueueCursor, motion chan int64, device
myMuxer = mp4.NewMuxer(file)
}
log.Log.Info("Recording starting: composing recording")
log.Log.Info("Recording starting: write header")
log.Log.Info("HandleRecordStream: composing recording")
log.Log.Info("HandleRecordStream: write header")
// Creating the file, might block sometimes.
if err := myMuxer.WriteHeader(streams); err != nil {
@@ -258,13 +160,13 @@ func RecordStream(recordingCursor *pubsub.QueueCursor, motion chan int64, device
} else {
log.Log.Info("Start motion based recording ")
log.Log.Info("HandleRecordStream: Start motion based recording ")
var myMuxer *mp4.Muxer
var file *os.File
var err error
for _ = range motion {
for _ = range communication.HandleMotion {
now = time.Now().Unix()
timestamp = now
@@ -284,7 +186,7 @@ func RecordStream(recordingCursor *pubsub.QueueCursor, motion chan int64, device
fullName := "./data/recordings/" + name
// Running...
log.Log.Info("Recording started")
log.Log.Info("HandleRecordStream: Recording started")
file, err = os.Create(fullName)
if err == nil {
myMuxer = mp4.NewMuxer(file)
@@ -292,8 +194,8 @@ func RecordStream(recordingCursor *pubsub.QueueCursor, motion chan int64, device
start := false
log.Log.Info("Recording starting: composing recording")
log.Log.Info("Recording starting: write header")
log.Log.Info("HandleRecordStream: composing recording")
log.Log.Info("HandleRecordStream: write header")
// Creating the file, might block sometimes.
if err := myMuxer.WriteHeader(streams); err != nil {
log.Log.Error(err.Error())
@@ -311,16 +213,16 @@ func RecordStream(recordingCursor *pubsub.QueueCursor, motion chan int64, device
now := time.Now().Unix()
select {
case <-motion:
case <-communication.HandleMotion:
timestamp = now
log.Log.Info("Recording expanding: motion detected while recording. Expanding recording.")
log.Log.Info("HandleRecordStream: motion detected while recording. Expanding recording.")
default:
}
if timestamp+recordingPeriod-now <= 0 || now-startRecording >= maxRecordingPeriod {
break
}
if pkt.IsKeyFrame {
log.Log.Info("Recording writing: write frames")
log.Log.Info("HandleRecordStream: write frames")
start = true
}
if start {
@@ -332,7 +234,7 @@ func RecordStream(recordingCursor *pubsub.QueueCursor, motion chan int64, device
// This will write the trailer as well.
myMuxer.WriteTrailer()
log.Log.Info("Recording finished: file save: " + name)
log.Log.Info("HandleRecordStream: file save: " + name)
file.Close()
myMuxer = nil
runtime.GC()
@@ -349,5 +251,5 @@ func RecordStream(recordingCursor *pubsub.QueueCursor, motion chan int64, device
}
}
log.Log.Debug("RecordStream: finished")
log.Log.Debug("HandleRecordStream: finished")
}

14
machinery/src/components/Kerberos.go
View File

@@ -6,6 +6,7 @@ import (
"sync/atomic"
"time"
"github.com/kerberos-io/agent/machinery/src/capture"
"github.com/kerberos-io/agent/machinery/src/cloud"
"github.com/kerberos-io/agent/machinery/src/computervision"
"github.com/kerberos-io/agent/machinery/src/log"
@@ -22,7 +23,7 @@ func Bootstrap(configuration *models.Configuration, communication *models.Commun
packageCounter.Store(int64(0))
communication.PackageCounter = &packageCounter
communication.HandleStream = make(chan string, 1)
communication.HandleMotion = make(chan string, 1)
communication.HandleMotion = make(chan int64, 1)
communication.HandleUpload = make(chan string, 1)
communication.HandleHeartBeat = make(chan string, 1)
communication.HandleLiveSD = make(chan int64, 1)
@@ -48,10 +49,11 @@ func RunAgent(configuration *models.Configuration, communication *models.Communi
config := configuration.Config
// Currently only support H264 encoded cameras, this will change.
// Establishing the camera connection
log.Log.Info("RunAgent: opening RTSP stream")
rtspUrl := config.Capture.IPCamera.RTSP
infile, streams, err := OpenRTSP(rtspUrl)
infile, streams, err := capture.OpenRTSP(rtspUrl)
//var decoder *ffmpeg.VideoDecoder
var queue *pubsub.Queue
@@ -62,7 +64,7 @@ func RunAgent(configuration *models.Configuration, communication *models.Communi
// At some routines we will need to decode the image.
// Make sure its properly locked as we only have a single decoder.
var decoderMutex sync.Mutex
decoder := GetVideoDecoder(streams)
decoder := capture.GetVideoDecoder(streams)
// Create a packet queue, which is filled by the HandleStream routing
// and consumed by all other routines: motion, livestream, etc.
@@ -77,7 +79,7 @@ func RunAgent(configuration *models.Configuration, communication *models.Communi
go cloud.HandleHeartBeat(configuration, communication)
// Handle the camera stream
go HandleStream(infile, queue, communication) //, &wg)
go capture.HandleStream(infile, queue, communication) //, &wg)
// Handle processing of motion
motionCursor := queue.Oldest()
@@ -87,6 +89,10 @@ func RunAgent(configuration *models.Configuration, communication *models.Communi
livestreamCursor := queue.Oldest()
go cloud.HandleLiveStreamSD(livestreamCursor, configuration, communication, mqttClient, decoder, &decoderMutex)
// Handle recording
recordingCursor := queue.Oldest()
go capture.HandleRecordStream(recordingCursor, configuration, communication, streams)
//-----------
// This will go into a blocking state, once this channel is triggered
// the agent will cleanup and restart.

2
machinery/src/computervision/main.go
View File

@@ -234,7 +234,7 @@ func ProcessMotion(motionCursor *pubsub.QueueCursor, configuration *models.Confi
if detectMotion && FindMotion(matArray, coordinatesToCheck) {
mqttClient.Publish("kerberos/"+key+"/device/"+config.Key+"/motion", 2, false, "motion")
fmt.Println(key)
communication.HandleMotion <- strconv.FormatInt(time.Now().Unix(), 10)
communication.HandleMotion <- time.Now().Unix()
}
}

2
machinery/src/models/Communication.go
View File

@@ -10,7 +10,7 @@ type Communication struct {
PackageCounter *atomic.Value
HandleBootstrap chan string
HandleStream chan string
HandleMotion chan string
HandleMotion chan int64
HandleUpload chan string
HandleHeartBeat chan string
HandleLiveSD chan int64