Update Cloud.go

machinery/src/capture/main.go

@@ -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

machinery/src/cloud/Cloud.go

@@ -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, macs, ips, "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.")
 		}
 	}
 }

machinery/src/components/Config.go

@@ -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

machinery/src/models/Config.go

@@ -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"`