mirror of
https://github.com/kerberos-io/agent.git
synced 2026-08-26 15:38:32 +00:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c59d511ea3 | ||
|
|
6f8745dc3a | ||
|
|
65d3d649b9 |
@@ -272,6 +272,9 @@ func HandleRecordStream(queue *pubsub.Queue, configuration *models.Configuration
|
||||
var file *os.File
|
||||
var err error
|
||||
|
||||
var lastDuration time.Duration
|
||||
var lastRecordingTime int64
|
||||
|
||||
for motion := range communication.HandleMotion {
|
||||
|
||||
timestamp = time.Now().Unix()
|
||||
@@ -281,7 +284,16 @@ func HandleRecordStream(queue *pubsub.Queue, configuration *models.Configuration
|
||||
// If we have prerecording we will substract the number of seconds.
|
||||
// Taking into account FPS = GOP size (Keyfram interval)
|
||||
if config.Capture.PreRecording > 0 {
|
||||
startRecording = startRecording - int64(config.Capture.PreRecording) + 1
|
||||
|
||||
// Might be that recordings are coming short after each other.
|
||||
// Therefore we do some math with the current time and the last recording time.
|
||||
|
||||
timeBetweenNowAndLastRecording := startRecording - lastRecordingTime
|
||||
if timeBetweenNowAndLastRecording > int64(config.Capture.PreRecording) {
|
||||
startRecording = startRecording - int64(config.Capture.PreRecording) + 1
|
||||
} else {
|
||||
startRecording = startRecording - timeBetweenNowAndLastRecording
|
||||
}
|
||||
}
|
||||
|
||||
// timestamp_microseconds_instanceName_regionCoordinates_numberOfChanges_token
|
||||
@@ -351,7 +363,7 @@ func HandleRecordStream(queue *pubsub.Queue, configuration *models.Configuration
|
||||
log.Log.Info("HandleRecordStream: closing recording (timestamp: " + strconv.FormatInt(timestamp, 10) + ", recordingPeriod: " + strconv.FormatInt(recordingPeriod, 10) + ", now: " + strconv.FormatInt(now, 10) + ", startRecording: " + strconv.FormatInt(startRecording, 10) + ", maxRecordingPeriod: " + strconv.FormatInt(maxRecordingPeriod, 10))
|
||||
break
|
||||
}
|
||||
if pkt.IsKeyFrame && !start {
|
||||
if pkt.IsKeyFrame && !start && pkt.Time >= lastDuration {
|
||||
log.Log.Info("HandleRecordStream: write frames")
|
||||
start = true
|
||||
}
|
||||
@@ -378,6 +390,9 @@ func HandleRecordStream(queue *pubsub.Queue, configuration *models.Configuration
|
||||
myMuxer.WriteTrailerWithPacket(nextPkt)
|
||||
log.Log.Info("HandleRecordStream: file save: " + name)
|
||||
|
||||
lastDuration = pkt.Time
|
||||
lastRecordingTime = time.Now().Unix()
|
||||
|
||||
// Cleanup muxer
|
||||
myMuxer.Close()
|
||||
myMuxer = nil
|
||||
|
||||
@@ -221,6 +221,10 @@ func HandleHeartBeat(configuration *models.Configuration, communication *models.
|
||||
isEnterprise = true
|
||||
}
|
||||
|
||||
// Congert to string
|
||||
macs, _ := json.Marshal(system.MACs)
|
||||
ips, _ := json.Marshal(system.IPs)
|
||||
|
||||
var object = fmt.Sprintf(`{
|
||||
"key" : "%s",
|
||||
"version" : "3.0.0",
|
||||
@@ -235,8 +239,8 @@ func HandleHeartBeat(configuration *models.Configuration, communication *models.
|
||||
"totalMemory" : "%d",
|
||||
"usedMemory" : "%d",
|
||||
"freeMemory" : "%d",
|
||||
"macs" : "%v",
|
||||
"ips" : "%v",
|
||||
"macs" : %s,
|
||||
"ips" : %s,
|
||||
"board" : "",
|
||||
"disk1size" : "%s",
|
||||
"disk3size" : "%s",
|
||||
@@ -250,7 +254,7 @@ func HandleHeartBeat(configuration *models.Configuration, communication *models.
|
||||
"docker" : true,
|
||||
"kios" : false,
|
||||
"raspberrypi" : false
|
||||
}`, config.Key, system.Version, system.CPUId, username, key, name, isEnterprise, system.Hostname, system.Architecture, system.TotalMemory, system.UsedMemory, system.FreeMemory, system.MACs, system.IPs, "0", "0", "0", uptimeString, config.HubSite, onvifEnabled)
|
||||
}`, config.Key, system.Version, system.CPUId, username, key, name, isEnterprise, system.Hostname, system.Architecture, system.TotalMemory, system.UsedMemory, system.FreeMemory, ips, macs, "0", "0", "0", uptimeString, config.HubSite, onvifEnabled)
|
||||
|
||||
var jsonStr = []byte(object)
|
||||
buffy := bytes.NewBuffer(jsonStr)
|
||||
@@ -308,51 +312,59 @@ func HandleLiveStreamSD(livestreamCursor *pubsub.QueueCursor, configuration *mod
|
||||
|
||||
config := configuration.Config
|
||||
|
||||
// If offline made is enabled, we will stop the thread.
|
||||
if config.Offline == "true" {
|
||||
log.Log.Debug("HandleLiveStreamSD: stopping as Offline is enabled.")
|
||||
} else {
|
||||
|
||||
// Allocate frame
|
||||
frame := ffmpeg.AllocVideoFrame()
|
||||
// Check if we need to enable the live stream
|
||||
if config.Capture.Liveview != "false" {
|
||||
|
||||
key := ""
|
||||
if config.Cloud == "s3" && config.S3 != nil && config.S3.Publickey != "" {
|
||||
key = config.S3.Publickey
|
||||
} else if config.Cloud == "kstorage" && config.KStorage != nil && config.KStorage.CloudKey != "" {
|
||||
key = config.KStorage.CloudKey
|
||||
}
|
||||
// This is the new way ;)
|
||||
if config.HubKey != "" {
|
||||
key = config.HubKey
|
||||
}
|
||||
// Allocate frame
|
||||
frame := ffmpeg.AllocVideoFrame()
|
||||
|
||||
topic := "kerberos/" + key + "/device/" + config.Key + "/live"
|
||||
|
||||
lastLivestreamRequest := int64(0)
|
||||
|
||||
var cursorError error
|
||||
var pkt av.Packet
|
||||
|
||||
for cursorError == nil {
|
||||
pkt, cursorError = livestreamCursor.ReadPacket()
|
||||
if len(pkt.Data) == 0 || !pkt.IsKeyFrame {
|
||||
continue
|
||||
key := ""
|
||||
if config.Cloud == "s3" && config.S3 != nil && config.S3.Publickey != "" {
|
||||
key = config.S3.Publickey
|
||||
} else if config.Cloud == "kstorage" && config.KStorage != nil && config.KStorage.CloudKey != "" {
|
||||
key = config.KStorage.CloudKey
|
||||
}
|
||||
now := time.Now().Unix()
|
||||
select {
|
||||
case <-communication.HandleLiveSD:
|
||||
lastLivestreamRequest = now
|
||||
default:
|
||||
// This is the new way ;)
|
||||
if config.HubKey != "" {
|
||||
key = config.HubKey
|
||||
}
|
||||
if now-lastLivestreamRequest > 3 {
|
||||
continue
|
||||
}
|
||||
log.Log.Info("HandleLiveStreamSD: Sending base64 encoded images to MQTT.")
|
||||
sendImage(frame, topic, mqttClient, pkt, decoder, decoderMutex)
|
||||
}
|
||||
|
||||
// Cleanup the frame.
|
||||
frame.Free()
|
||||
topic := "kerberos/" + key + "/device/" + config.Key + "/live"
|
||||
|
||||
lastLivestreamRequest := int64(0)
|
||||
|
||||
var cursorError error
|
||||
var pkt av.Packet
|
||||
|
||||
for cursorError == nil {
|
||||
pkt, cursorError = livestreamCursor.ReadPacket()
|
||||
if len(pkt.Data) == 0 || !pkt.IsKeyFrame {
|
||||
continue
|
||||
}
|
||||
now := time.Now().Unix()
|
||||
select {
|
||||
case <-communication.HandleLiveSD:
|
||||
lastLivestreamRequest = now
|
||||
default:
|
||||
}
|
||||
if now-lastLivestreamRequest > 3 {
|
||||
continue
|
||||
}
|
||||
log.Log.Info("HandleLiveStreamSD: Sending base64 encoded images to MQTT.")
|
||||
sendImage(frame, topic, mqttClient, pkt, decoder, decoderMutex)
|
||||
}
|
||||
|
||||
// Cleanup the frame.
|
||||
frame.Free()
|
||||
|
||||
} else {
|
||||
log.Log.Debug("HandleLiveStreamSD: stopping as Liveview is disabled.")
|
||||
}
|
||||
}
|
||||
|
||||
log.Log.Debug("HandleLiveStreamSD: finished")
|
||||
@@ -375,34 +387,41 @@ func HandleLiveStreamHD(livestreamCursor *pubsub.QueueCursor, configuration *mod
|
||||
log.Log.Debug("HandleLiveStreamHD: stopping as Offline is enabled.")
|
||||
} else {
|
||||
|
||||
// Should create a track here.
|
||||
track := webrtc.NewVideoTrack()
|
||||
go webrtc.WriteToTrack(livestreamCursor, configuration, communication, mqttClient, track, codecs, decoder, decoderMutex)
|
||||
// Check if we need to enable the live stream
|
||||
if config.Capture.Liveview != "false" {
|
||||
|
||||
// Should create a track here.
|
||||
track := webrtc.NewVideoTrack()
|
||||
go webrtc.WriteToTrack(livestreamCursor, configuration, communication, mqttClient, track, codecs, decoder, decoderMutex)
|
||||
|
||||
if config.Capture.ForwardWebRTC == "true" {
|
||||
// We get a request with an offer, but we'll forward it.
|
||||
for m := range communication.HandleLiveHDHandshake {
|
||||
// Forward SDP
|
||||
m.CloudKey = config.Key
|
||||
request, err := json.Marshal(m)
|
||||
if err == nil {
|
||||
mqttClient.Publish("kerberos/webrtc/request", 2, false, request)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
log.Log.Info("HandleLiveStreamHD: Waiting for peer connections.")
|
||||
for handshake := range communication.HandleLiveHDHandshake {
|
||||
log.Log.Info("HandleLiveStreamHD: setting up a peer connection.")
|
||||
key := config.Key + "/" + handshake.Cuuid
|
||||
webrtc.CandidatesMutex.Lock()
|
||||
_, ok := webrtc.CandidateArrays[key]
|
||||
if !ok {
|
||||
webrtc.CandidateArrays[key] = make(chan string, 30)
|
||||
}
|
||||
webrtc.CandidatesMutex.Unlock()
|
||||
webrtc.InitializeWebRTCConnection(configuration, communication, mqttClient, track, handshake, webrtc.CandidateArrays[key])
|
||||
|
||||
if config.Capture.ForwardWebRTC == "true" {
|
||||
// We get a request with an offer, but we'll forward it.
|
||||
for m := range communication.HandleLiveHDHandshake {
|
||||
// Forward SDP
|
||||
m.CloudKey = config.Key
|
||||
request, err := json.Marshal(m)
|
||||
if err == nil {
|
||||
mqttClient.Publish("kerberos/webrtc/request", 2, false, request)
|
||||
}
|
||||
}
|
||||
|
||||
} else {
|
||||
log.Log.Info("HandleLiveStreamHD: Waiting for peer connections.")
|
||||
for handshake := range communication.HandleLiveHDHandshake {
|
||||
log.Log.Info("HandleLiveStreamHD: setting up a peer connection.")
|
||||
key := config.Key + "/" + handshake.Cuuid
|
||||
webrtc.CandidatesMutex.Lock()
|
||||
_, ok := webrtc.CandidateArrays[key]
|
||||
if !ok {
|
||||
webrtc.CandidateArrays[key] = make(chan string, 30)
|
||||
}
|
||||
webrtc.CandidatesMutex.Unlock()
|
||||
webrtc.InitializeWebRTCConnection(configuration, communication, mqttClient, track, handshake, webrtc.CandidateArrays[key])
|
||||
|
||||
}
|
||||
log.Log.Debug("HandleLiveStreamHD: stopping as Liveview is disabled.")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -244,6 +244,9 @@ func OverrideWithEnvironmentVariables(configuration *models.Configuration) {
|
||||
case "AGENT_CAPTURE_CONTINUOUS":
|
||||
configuration.Config.Capture.Continuous = value
|
||||
break
|
||||
case "AGENT_CAPTURE_LIVEVIEW":
|
||||
configuration.Config.Capture.Liveview = value
|
||||
break
|
||||
case "AGENT_CAPTURE_MOTION":
|
||||
configuration.Config.Capture.Motion = value
|
||||
break
|
||||
|
||||
@@ -53,6 +53,7 @@ type Capture struct {
|
||||
Continuous string `json:"continuous,omitempty"`
|
||||
Snapshots string `json:"snapshots,omitempty"`
|
||||
Motion string `json:"motion,omitempty"`
|
||||
Liveview string `json:"liveview,omitempty"`
|
||||
PostRecording int64 `json:"postrecording"`
|
||||
PreRecording int64 `json:"prerecording"`
|
||||
MaxLengthRecording int64 `json:"maxlengthrecording"`
|
||||
|
||||
Reference in New Issue
Block a user