Merge pull request #208 from kerberos-io/feature/increase-chunk-size

feature/increase-chunk-size

machinery/data/config/config.json

@@ -26,6 +26,7 @@
 		"recording": "true",
 		"snapshots": "true",
 		"liveview": "true",
+		"liveview_chunking": "false",
 		"motion": "true",
 		"postrecording": 20,
 		"prerecording": 10,

machinery/src/capture/gortsplib.go

@@ -478,6 +478,7 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 				Time:            pts2,
 				TimeLegacy:      pts,
 				CompositionTime: pts2,
+				CurrentTime:     time.Now().UnixMilli(),
 				Idx:             g.AudioG711Index,
 				IsVideo:         false,
 				IsAudio:         true,
@@ -519,6 +520,7 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 				Time:            pts2,
 				TimeLegacy:      pts,
 				CompositionTime: pts2,
+				CurrentTime:     time.Now().UnixMilli(),
 				Idx:             g.AudioG711Index,
 				IsVideo:         false,
 				IsAudio:         true,
@@ -631,19 +633,13 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 					return
 				}
 
-				// Extract DTS from RTP packets
-				//dts2, err := dtsExtractor.Extract(filteredAU, pts2)
-				//if err != nil {
-				// log.Log.Error("capture.golibrtsp.Start(): " + err.Error())
-				// return
-				//}
-
 				pkt := packets.Packet{
 					IsKeyFrame:      idrPresent,
 					Packet:          rtppkt,
 					Data:            enc,
 					Time:            pts2,
 					TimeLegacy:      pts,
+					CurrentTime:     time.Now().UnixMilli(),
 					CompositionTime: pts2,
 					Idx:             g.VideoH264Index,
 					IsVideo:         true,
@@ -788,6 +784,7 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 					Data:            enc,
 					Time:            pts2,
 					TimeLegacy:      pts,
+					CurrentTime:     time.Now().UnixMilli(),
 					CompositionTime: pts2,
 					Idx:             g.VideoH265Index,
 					IsVideo:         true,

machinery/src/capture/main.go

@@ -68,10 +68,16 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 		postRecording := config.Capture.PostRecording * 1000           // number of seconds to record.
 		maxRecordingPeriod := config.Capture.MaxLengthRecording * 1000 // maximum number of seconds to record.
 
-		// Synchronise the last synced time
-		now := time.Now().UnixMilli()
-		startRecording := now
-		timestamp := now
+		// We will calculate the maxRecordingPeriod based on the preRecording and postRecording values.
+		if maxRecordingPeriod == 0 {
+			// If maxRecordingPeriod is not set, we will use the preRecording and postRecording values
+			maxRecordingPeriod = preRecording + postRecording
+		}
+
+		if maxRecordingPeriod < preRecording+postRecording {
+			log.Log.Error("capture.main.HandleRecordStream(): maxRecordingPeriod is less than preRecording + postRecording, this is not allowed. Setting maxRecordingPeriod to preRecording + postRecording.")
+			maxRecordingPeriod = preRecording + postRecording
+		}
 
 		if config.FriendlyName != "" {
 			config.Name = config.FriendlyName
@@ -105,8 +111,6 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 			// Do not do anything!
 			log.Log.Info("capture.main.HandleRecordStream(continuous): start recording")
 
-			now = time.Now().Unix()
-			timestamp = now
 			start := false
 
 			// If continuous record the full length
@@ -114,6 +118,8 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 			// Recording file name
 			fullName := ""
 
+			var startRecording int64 = 0 // start recording timestamp in milliseconds
+
 			// Get as much packets we need.
 			var cursorError error
 			var pkt packets.Packet
@@ -132,7 +138,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 				now := time.Now().UnixMilli()
 
 				if start && // If already recording and current frame is a keyframe and we should stop recording
-					nextPkt.IsKeyFrame && (timestamp+postRecording-now <= 0 || now-startRecording >= maxRecordingPeriod-1000) {
+					nextPkt.IsKeyFrame && (startRecording+postRecording-now <= 0 || now-startRecording > maxRecordingPeriod-500) {
 
 					pts := convertPTS(pkt.TimeLegacy)
 					if pkt.IsVideo {
@@ -236,7 +242,6 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 					}
 
 					start = true
-					timestamp = now
 
 					// timestamp_microseconds_instanceName_regionCoordinates_numberOfChanges_token
 					// 1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4
@@ -247,7 +252,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 					// - Number of changes
 					// - Token
 
-					startRecording = time.Now().UnixMilli()
+					startRecording = pkt.CurrentTime
 					startRecordingSeconds := startRecording / 1000            // convert to seconds
 					startRecordingMilliseconds := startRecording % 1000       // convert to milliseconds
 					s := strconv.FormatInt(startRecordingSeconds, 10) + "_" + // start timestamp in seconds
@@ -409,16 +414,12 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 
 			log.Log.Info("capture.main.HandleRecordStream(motiondetection): Start motion based recording ")
 
-			var lastDuration int64 = 0      // last duration in milliseconds
-			var lastRecordingTime int64 = 0 // last recording time in milliseconds
+			var lastRecordingTime int64 = 0 // last recording timestamp in milliseconds
 			var displayTime int64 = 0       // display time in milliseconds
 
 			var videoTrack uint32
 			var audioTrack uint32
 
-			streams, _ := rtspClient.GetVideoStreams()
-			videoStream := streams[0] // We will use the first video stream, as we only expect one video stream.
-
 			for motion := range communication.HandleMotion {
 
 				// Get as much packets we need.
@@ -427,43 +428,30 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 				var nextPkt packets.Packet
 				recordingCursor := queue.Oldest() // Start from the latest packet in the queue)
 
-				timestamp = time.Now().UnixMilli()
-				startRecording = time.Now().UnixMilli() // we mark the current time when the record started.
+				now := time.Now().UnixMilli()
+				motionTimestamp := now
 
-				// If we have prerecording we will substract the number of seconds.
-				// Taking into account FPS = GOP size (Keyfram interval)
-				var preRecordingDelta int64 = 0
-				if preRecording > 0 {
+				start := false
 
-					fps := videoStream.FPS
-					queueSize := queue.GetSize()
+				if cursorError == nil {
+					pkt, cursorError = recordingCursor.ReadPacket()
+				}
 
-					// Based on the GOP size and FPS we can calculate the pre-recording time.
-					// It might be that the queue size is 0, in that case we will not calculate the pre-recording time.
-					queuedAvailablePreRecording := preRecording
-					if queueSize > 0 && fps > 0 {
-						queuedAvailablePreRecording = int64(queueSize) / int64(fps) * 1000 // convert to milliseconds
-					}
-					timeBetweenNowAndLastRecording := startRecording - lastRecordingTime
-					if lastRecordingTime == 0 {
-						timeBetweenNowAndLastRecording = 0
-					}
+				displayTime = pkt.CurrentTime
+				startRecording := pkt.CurrentTime
 
-					// Might be that recordings are coming short after each other.
-					// Therefore we do some math with the current time and the last recording time.
-					if timeBetweenNowAndLastRecording >= preRecording {
-						displayTime = startRecording - preRecording + 1000
-						preRecordingDelta = preRecording
-					} else if timeBetweenNowAndLastRecording < preRecording {
-						// If the time between now and the last recording is less than the pre-recording time,
-						// we will use the pre-recording time.
-						if lastRecordingTime == 0 && queuedAvailablePreRecording < preRecording {
-							displayTime = startRecording - queuedAvailablePreRecording
-						} else {
-							preRecordingDelta = timeBetweenNowAndLastRecording
-							displayTime = startRecording - preRecordingDelta
-						}
-					}
+				// We have more packets in the queue (which might still be older than where we close the previous recording).
+				// In that case we will use the last recording time to determine the start time of the recording, otherwise
+				// we will have duplicate frames in the recording.
+				if startRecording < lastRecordingTime {
+					displayTime = lastRecordingTime
+					startRecording = lastRecordingTime
+				}
+
+				// If startRecording is 0, we will continue as it might be we are in a state of restarting the agent.
+				if startRecording == 0 {
+					log.Log.Info("capture.main.HandleRecordStream(motiondetection): startRecording is 0, we will continue as it might be we are in a state of restarting the agent.")
+					continue
 				}
 
 				// timestamp_microseconds_instanceName_regionCoordinates_numberOfChanges_token
@@ -527,12 +515,6 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 					log.Log.Debug("capture.main.HandleRecordStream(continuous): no AAC audio codec detected, skipping audio track.")
 				}
 
-				start := false
-
-				if cursorError == nil {
-					pkt, cursorError = recordingCursor.ReadPacket()
-				}
-
 				for cursorError == nil {
 
 					nextPkt, cursorError = recordingCursor.ReadPacket()
@@ -540,24 +522,23 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 						log.Log.Error("capture.main.HandleRecordStream(motiondetection): " + cursorError.Error())
 					}
 
-					now := time.Now().UnixMilli()
-
+					now = time.Now().UnixMilli()
 					select {
 					case motion := <-communication.HandleMotion:
-						timestamp = now
+						motionTimestamp = now
 						log.Log.Info("capture.main.HandleRecordStream(motiondetection): motion detected while recording. Expanding recording.")
 						numberOfChanges := motion.NumberOfChanges
 						log.Log.Info("capture.main.HandleRecordStream(motiondetection): Received message with recording data, detected changes to save: " + strconv.Itoa(numberOfChanges))
 					default:
 					}
 
-					if (timestamp+postRecording+(preRecording-preRecordingDelta)-now < 0 || now-startRecording > maxRecordingPeriod-preRecordingDelta) && nextPkt.IsKeyFrame {
-						log.Log.Info("capture.main.HandleRecordStream(motiondetection): timestamp+postRecording-now < 0  - " + strconv.FormatInt(timestamp+postRecording-now, 10) + " < 0")
-						log.Log.Info("capture.main.HandleRecordStream(motiondetection): now-startRecording > maxRecordingPeriod-1000 - " + strconv.FormatInt(now-startRecording, 10) + " > " + strconv.FormatInt(maxRecordingPeriod-1000, 10))
-						log.Log.Info("capture.main.HandleRecordStream(motiondetection): closing recording (timestamp: " + strconv.FormatInt(timestamp, 10) + ", postRecording: " + strconv.FormatInt(postRecording, 10) + ", now: " + strconv.FormatInt(now, 10) + ", startRecording: " + strconv.FormatInt(startRecording, 10) + ", maxRecordingPeriod: " + strconv.FormatInt(maxRecordingPeriod, 10))
+					if (motionTimestamp+postRecording-now < 0 || now-startRecording > maxRecordingPeriod-500) && nextPkt.IsKeyFrame {
+						log.Log.Info("capture.main.HandleRecordStream(motiondetection): timestamp+postRecording-now < 0  - " + strconv.FormatInt(motionTimestamp+postRecording-now, 10) + " < 0")
+						log.Log.Info("capture.main.HandleRecordStream(motiondetection): now-startRecording > maxRecordingPeriod-500 - " + strconv.FormatInt(now-startRecording, 10) + " > " + strconv.FormatInt(maxRecordingPeriod-500, 10))
+						log.Log.Info("capture.main.HandleRecordStream(motiondetection): closing recording (timestamp: " + strconv.FormatInt(motionTimestamp, 10) + ", postRecording: " + strconv.FormatInt(postRecording, 10) + ", now: " + strconv.FormatInt(now, 10) + ", startRecording: " + strconv.FormatInt(startRecording, 10) + ", maxRecordingPeriod: " + strconv.FormatInt(maxRecordingPeriod, 10))
 						break
 					}
-					if pkt.IsKeyFrame && !start && (pkt.Time >= lastDuration || pkt.Time == 0) {
+					if pkt.IsKeyFrame && !start && pkt.CurrentTime >= startRecording {
 						// We start the recording if we have a keyframe and the last duration is 0 or less than the current packet time.
 						// It could be start we start from the beginning of the recording.
 						log.Log.Debug("capture.main.HandleRecordStream(motiondetection): write frames")
@@ -590,8 +571,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 
 				// Update the last duration and last recording time.
 				// This is used to determine if we need to start a new recording.
-				lastDuration = pkt.Time
-				lastRecordingTime = time.Now().UnixMilli()
+				lastRecordingTime = pkt.CurrentTime
 
 				// This will close the recording and write the last packet.
 				mp4Video.Close(&config)

machinery/src/cloud/Cloud.go

@@ -672,6 +672,7 @@ func HandleLiveStreamSD(livestreamCursor *packets.QueueCursor, configuration *mo
 		// Check if we need to enable the live stream
 		if config.Capture.Liveview != "false" {
 
+			deviceId := config.Key
 			hubKey := ""
 			if config.Cloud == "s3" && config.S3 != nil && config.S3.Publickey != "" {
 				hubKey = config.S3.Publickey
@@ -706,24 +707,77 @@ func HandleLiveStreamSD(livestreamCursor *packets.QueueCursor, configuration *mo
 				img, err := rtspClient.DecodePacket(pkt)
 				if err == nil {
 					bytes, _ := utils.ImageToBytes(&img)
-					encoded := base64.StdEncoding.EncodeToString(bytes)
 
-					valueMap := make(map[string]interface{})
-					valueMap["image"] = encoded
-					message := models.Message{
-						Payload: models.Payload{
-							Action:   "receive-sd-stream",
-							DeviceId: configuration.Config.Key,
-							Value:    valueMap,
-						},
-					}
-					payload, err := models.PackageMQTTMessage(configuration, message)
-					if err == nil {
-						mqttClient.Publish("kerberos/hub/"+hubKey, 0, false, payload)
+					chunking := config.Capture.LiveviewChunking
+
+					if chunking == "true" {
+
+						// Split encoded image into chunks of 2kb
+						// This is to prevent the MQTT message to be too large.
+						// By default, bytes are not encoded to base64 here; you are splitting the raw JPEG/PNG bytes.
+						// However, in MQTT and web contexts, binary data may not be handled well, so base64 is often used.
+						// To avoid base64 encoding, just send the raw []byte chunks as you do here.
+						// If you want to avoid base64, make sure the receiver can handle binary payloads.
+
+						chunkSize := 25 * 1024 // 25KB chunks
+						var chunks [][]byte
+						for i := 0; i < len(bytes); i += chunkSize {
+							end := i + chunkSize
+							if end > len(bytes) {
+								end = len(bytes)
+							}
+							chunk := bytes[i:end]
+							chunks = append(chunks, chunk)
+						}
+
+						log.Log.Infof("cloud.HandleLiveStreamSD(): Sending %d chunks of size %d bytes.", len(chunks), chunkSize)
+
+						timestamp := time.Now().Unix()
+						for i, chunk := range chunks {
+							valueMap := make(map[string]interface{})
+							valueMap["id"] = timestamp
+							valueMap["chunk"] = chunk
+							valueMap["chunkIndex"] = i
+							valueMap["chunkSize"] = chunkSize
+							valueMap["chunkCount"] = len(chunks)
+							message := models.Message{
+								Payload: models.Payload{
+									Version:  "v1.0.0",
+									Action:   "receive-sd-stream",
+									DeviceId: deviceId,
+									Value:    valueMap,
+								},
+							}
+							payload, err := models.PackageMQTTMessage(configuration, message)
+							if err == nil {
+								mqttClient.Publish("kerberos/hub/"+hubKey+"/"+deviceId, 1, false, payload)
+								log.Log.Infof("cloud.HandleLiveStreamSD(): sent chunk %d/%d to MQTT topic kerberos/hub/%s/%s", i+1, len(chunks), hubKey, deviceId)
+								time.Sleep(33 * time.Millisecond) // Sleep to avoid flooding the MQTT broker with messages
+							} else {
+								log.Log.Info("cloud.HandleLiveStreamSD(): something went wrong while sending acknowledge config to hub: " + string(payload))
+							}
+						}
 					} else {
-						log.Log.Info("cloud.HandleLiveStreamSD(): something went wrong while sending acknowledge config to hub: " + string(payload))
+
+						valueMap := make(map[string]interface{})
+						valueMap["image"] = bytes
+						message := models.Message{
+							Payload: models.Payload{
+								Action:   "receive-sd-stream",
+								DeviceId: configuration.Config.Key,
+								Value:    valueMap,
+							},
+						}
+						payload, err := models.PackageMQTTMessage(configuration, message)
+						if err == nil {
+							mqttClient.Publish("kerberos/hub/"+hubKey, 0, false, payload)
+						} else {
+							log.Log.Info("cloud.HandleLiveStreamSD(): something went wrong while sending acknowledge config to hub: " + string(payload))
+						}
+
 					}
 				}
+				time.Sleep(1000 * time.Millisecond) // Sleep to avoid flooding the MQTT broker with messages
 			}
 
 		} else {

machinery/src/config/main.go

@@ -392,6 +392,11 @@ func OverrideWithEnvironmentVariables(configuration *models.Configuration) {
 				configuration.Config.MQTTPassword = value
 				break
 
+			/* MQTT chunking of low-resolution images into multiple messages */
+			case "AGENT_CAPTURE_LIVEVIEW_CHUNKING":
+				configuration.Config.Capture.LiveviewChunking = value
+				break
+
 			/* Real-time streaming of keyframes to a MQTT topic */
 			case "AGENT_REALTIME_PROCESSING":
 				configuration.Config.RealtimeProcessing = value

machinery/src/log/main.go

@@ -118,6 +118,16 @@ func (self *Logging) Info(sentence string) {
 	}
 }
 
+func (self *Logging) Infof(format string, args ...interface{}) {
+	switch self.Logger {
+	case "go-logging":
+		gologging.Infof(format, args...)
+	case "logrus":
+		logrus.Infof(format, args...)
+	default:
+	}
+}
+
 func (self *Logging) Warning(sentence string) {
 	switch self.Logger {
 	case "go-logging":

machinery/src/models/Config.go

@@ -62,6 +62,7 @@ type Capture struct {
 	Snapshots             string      `json:"snapshots,omitempty"`
 	Motion                string      `json:"motion,omitempty"`
 	Liveview              string      `json:"liveview,omitempty"`
+	LiveviewChunking      string      `json:"liveview_chunking,omitempty" bson:"liveview_chunking,omitempty"`
 	Continuous            string      `json:"continuous,omitempty"`
 	PostRecording         int64       `json:"postrecording"`
 	PreRecording          int64       `json:"prerecording"`

machinery/src/models/MQTT.go

@@ -132,6 +132,7 @@ type Message struct {
 // The payload structure which is used to send over
 // and receive messages from the MQTT broker
 type Payload struct {
+	Version        string                 `json:"version"` // Version of the message, e.g. "1.0"
 	Action         string                 `json:"action"`
 	DeviceId       string                 `json:"device_id"`
 	Signature      string                 `json:"signature"`

machinery/src/packets/packet.go

@@ -17,6 +17,7 @@ type Packet struct {
 	CompositionTime int64  // packet presentation time minus decode time for H264 B-Frame
 	Time            int64  // packet decode time
 	TimeLegacy      time.Duration
+	CurrentTime     int64  // current time in milliseconds (UNIX timestamp)
 	Data            []byte // packet data
 	Gopsize         int    // size of the GOP
 }

machinery/src/utils/main.go

@@ -23,7 +23,7 @@ import (
 	"github.com/kerberos-io/agent/machinery/src/models"
 )
 
-const VERSION = "3.3.5"
+const VERSION = "3.5.0"
 
 const letterBytes = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ"
 

machinery/src/webrtc/main.go

@@ -281,7 +281,8 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 				candateBinary, err := json.Marshal(candateJSON)
 				if err == nil {
 					valueMap["candidate"] = string(candateBinary)
-					valueMap["sdp"] = []byte(base64.StdEncoding.EncodeToString([]byte(answer.SDP)))
+					// SDP is not needed to be send..
+					//valueMap["sdp"] = []byte(base64.StdEncoding.EncodeToString([]byte(answer.SDP)))
 					valueMap["session_id"] = handshake.SessionID
 				} else {
 					log.Log.Info("webrtc.main.InitializeWebRTCConnection(): something went wrong while marshalling candidate: " + err.Error())