add logging + friendly_name might be empty

machinery/data/config/config.json

@@ -1,20 +1,25 @@
-
 {
 	"type": "",
-	"key": "",
-	"name": "agent",
+	"key": "okkey",
+	"name": "rtsp2",
 	"time": "false",
 	"offline": "false",
-	"auto_clean": "true",
+	"auto_clean": "100",
 	"remove_after_upload": "true",
-	"max_directory_size": 100,
-	"timezone": "Africa/Ceuta",
+	"max_directory_size": 20,
+	"timezone": "Europe/Brussels",
 	"capture": {
 		"name": "",
 		"ipcamera": {
-			"rtsp": "",
-			"sub_rtsp": "",
-			"fps": ""
+			"width": 640,
+			"height": 480,
+			"fps": "",
+			"rtsp": "rtsp://fake.kerberos.io/stream",
+			"sub_rtsp": "rtsp://kellyvd:cedricve123@192.168.1.68:554/Streaming/Channels/102?transportmode=multicast\u0026profile=Profile_1",
+			"onvif": "true",
+			"onvif_xaddr": "192.168.1.68",
+			"onvif_username": "kellyvd",
+			"onvif_password": "cedricve123"
 		},
 		"usbcamera": {
 			"device": ""
@@ -22,27 +27,27 @@
 		"raspicamera": {
 			"device": ""
 		},
-		"continuous": "false",
 		"recording": "true",
 		"snapshots": "true",
-		"liveview": "true",
 		"motion": "true",
-		"postrecording": 20,
-		"prerecording": 10,
-		"maxlengthrecording": 30,
+		"liveview": "true",
+		"continuous": "false",
+		"postrecording": 5,
+		"prerecording": 5,
+		"maxlengthrecording": 10,
 		"transcodingwebrtc": "",
 		"transcodingresolution": 0,
 		"forwardwebrtc": "",
 		"fragmented": "false",
-		"fragmentedduration": 8,
+		"fragmentedduration": 10,
 		"pixelChangeThreshold": 150
 	},
 	"timetable": [
 		{
 			"start1": 0,
-			"end1": 43199,
-			"start2": 43200,
-			"end2": 86400
+			"end1": 0,
+			"start2": 0,
+			"end2": 0
 		},
 		{
 			"start1": 0,
@@ -89,15 +94,106 @@
 			"x2": 800,
 			"y2": 640
 		},
-		"polygon": []
+		"polygon": [
+			{
+				"id": "1694456798869",
+				"coordinates": [
+					{
+						"x": 458.8176559163439,
+						"y": 420.6597577161105
+					},
+					{
+						"x": 541.8959724597637,
+						"y": 422.62130444776193
+					},
+					{
+						"x": 540.2578560606811,
+						"y": 446.03551153214977
+					},
+					{
+						"x": 459.51092569845395,
+						"y": 445.07041692205803
+					}
+				]
+			},
+			{
+				"id": "1694455955284",
+				"coordinates": [
+					{
+						"x": 82.63849814838939,
+						"y": 122.83665165528404
+					},
+					{
+						"x": 259.18591194149286,
+						"y": 122.83665165528404
+					},
+					{
+						"x": 259.18591194149286,
+						"y": 232.42716889666335
+					},
+					{
+						"x": 82.63849814838939,
+						"y": 232.42716889666335
+					}
+				]
+			},
+			{
+				"id": "1694455953115",
+				"coordinates": [
+					{
+						"x": 200.7130606007035,
+						"y": 171.25495957161723
+					},
+					{
+						"x": 445.73676749725524,
+						"y": 171.25495957161723
+					},
+					{
+						"x": 445.73676749725524,
+						"y": 297.5911664681689
+					},
+					{
+						"x": 200.7130606007035,
+						"y": 297.5911664681689
+					}
+				]
+			},
+			{
+				"id": "0",
+				"coordinates": [
+					{
+						"x": 351.8431271737717,
+						"y": 65.62379883848743
+					},
+					{
+						"x": 609.6534108617149,
+						"y": 65.62379883848743
+					},
+					{
+						"x": 609.6534108617149,
+						"y": 246.50145841295546
+					},
+					{
+						"x": 351.8431271737717,
+						"y": 246.50145841295546
+					}
+				]
+			}
+		]
 	},
 	"cloud": "s3",
 	"s3": {
 		"proxyuri": "http://proxy.kerberos.io",
 		"bucket": "kerberosaccept",
-		"region": "eu-west-1"
+		"region": "eu-west1",
+		"username": "cedricve"
+	},
+	"kstorage": {
+		"uri": "https://api.vault.kerberos.io",
+		"access_key": "pDCQve0XL1DXwy2x",
+		"secret_access_key": "GTpus90EKrKYuplqL3g!2oiQ@s",
+		"directory": "cedricve"
 	},
-	"kstorage": {},
 	"dropbox": {},
 	"mqtturi": "tcp://mqtt.kerberos.io:1883",
 	"mqtt_username": "",
@@ -107,9 +203,9 @@
 	"turn_username": "username1",
 	"turn_password": "password1",
 	"heartbeaturi": "",
-	"hub_uri": "https://api.cloud.kerberos.io",
-	"hub_key": "",
-	"hub_private_key": "",
+	"hub_uri": "http://localhost:8081",
+	"hub_key": "AKIA5V6EGLUWXDU4JBEI",
+	"hub_private_key": "DIOXToTpAlYpa4atf7aWtxHd9xEw/Fk4XioL0tek",
 	"hub_site": "",
 	"condition_uri": ""
-}
+}

machinery/main.go

@@ -9,6 +9,8 @@ import (
 	"github.com/kerberos-io/agent/machinery/src/components"
 	"github.com/kerberos-io/agent/machinery/src/log"
 	"github.com/kerberos-io/agent/machinery/src/models"
+
+	configService "github.com/kerberos-io/agent/machinery/src/config"
 	"github.com/kerberos-io/agent/machinery/src/routers"
 	"github.com/kerberos-io/agent/machinery/src/utils"
 	"gopkg.in/DataDog/dd-trace-go.v1/ddtrace/tracer"
@@ -92,10 +94,10 @@ func main() {
 			configuration.Port = port
 
 			// Open this configuration either from Kerberos Agent or Kerberos Factory.
-			components.OpenConfig(configDirectory, &configuration)
+			configService.OpenConfig(configDirectory, &configuration)
 
 			// We will override the configuration with the environment variables
-			components.OverrideWithEnvironmentVariables(&configuration)
+			configService.OverrideWithEnvironmentVariables(&configuration)
 
 			// Printing final configuration
 			utils.PrintConfiguration(&configuration)
@@ -113,7 +115,7 @@ func main() {
 			if configuration.Config.Key == "" {
 				key := utils.RandStringBytesMaskImpr(30)
 				configuration.Config.Key = key
-				err := components.StoreConfig(configDirectory, configuration.Config)
+				err := configService.StoreConfig(configDirectory, configuration.Config)
 				if err == nil {
 					log.Log.Info("Main: updated unique key for agent to: " + key)
 				} else {

machinery/src/cloud/Cloud.go

@@ -302,10 +302,27 @@ loop:
 								onvifPresetsList, err = json.Marshal(presets)
 								if err != nil {
 									log.Log.Error("HandleHeartBeat: error while marshalling presets: " + err.Error())
+									onvifPresetsList = []byte("[]")
 								}
+							} else {
+								if err != nil {
+									log.Log.Error("HandleHeartBeat: error while getting presets: " + err.Error())
+								} else {
+									log.Log.Debug("HandleHeartBeat: no presets found.")
+								}
+								onvifPresetsList = []byte("[]")
 							}
+						} else {
+							log.Log.Error("HandleHeartBeat: error while getting PTZ configurations: " + err.Error())
+							onvifPresetsList = []byte("[]")
 						}
+					} else {
+						log.Log.Error("HandleHeartBeat: error while connecting to ONVIF device: " + err.Error())
+						onvifPresetsList = []byte("[]")
 					}
+				} else {
+					log.Log.Debug("HandleHeartBeat: ONVIF is not enabled.")
+					onvifPresetsList = []byte("[]")
 				}
 
 				// Check if the agent is running inside a cluster (Kerberos Factory) or as

machinery/src/components/Kerberos.go

@@ -14,6 +14,7 @@ import (
 	"github.com/kerberos-io/agent/machinery/src/capture"
 	"github.com/kerberos-io/agent/machinery/src/cloud"
 	"github.com/kerberos-io/agent/machinery/src/computervision"
+	configService "github.com/kerberos-io/agent/machinery/src/config"
 	"github.com/kerberos-io/agent/machinery/src/log"
 	"github.com/kerberos-io/agent/machinery/src/models"
 	"github.com/kerberos-io/agent/machinery/src/onvif"
@@ -72,7 +73,7 @@ func Bootstrap(configDirectory string, configuration *models.Configuration, comm
 
 	// We'll create a MQTT handler, which will be used to communicate with Kerberos Hub.
 	// Configure a MQTT client which helps for a bi-directional communication
-	mqttClient := routers.ConfigureMQTT(configuration, communication)
+	mqttClient := routers.ConfigureMQTT(configDirectory, configuration, communication)
 
 	// Run the agent and fire up all the other
 	// goroutines which do image capture, motion detection, onvif, etc.
@@ -87,15 +88,15 @@ func Bootstrap(configDirectory string, configuration *models.Configuration, comm
 
 		if status == "not started" {
 			// We will re open the configuration, might have changed :O!
-			OpenConfig(configDirectory, configuration)
+			configService.OpenConfig(configDirectory, configuration)
 			// We will override the configuration with the environment variables
-			OverrideWithEnvironmentVariables(configuration)
+			configService.OverrideWithEnvironmentVariables(configuration)
 		}
 
 		// Reset the MQTT client, might have provided new information, so we need to reconnect.
 		if routers.HasMQTTClientModified(configuration) {
 			routers.DisconnectMQTT(mqttClient, &configuration.Config)
-			mqttClient = routers.ConfigureMQTT(configuration, communication)
+			mqttClient = routers.ConfigureMQTT(configDirectory, configuration, communication)
 		}
 
 		// We will create a new cancelable context, which will be used to cancel and restart.
@@ -134,6 +135,10 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu
 	width := videoStream.(av.VideoCodecData).Width()
 	height := videoStream.(av.VideoCodecData).Height()
 
+	// Set config values as well
+	configuration.Config.Capture.IPCamera.Width = width
+	configuration.Config.Capture.IPCamera.Height = height
+
 	var queue *pubsub.Queue
 	var subQueue *pubsub.Queue
 
@@ -162,6 +167,13 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu
 				time.Sleep(time.Second * 3)
 				return status
 			}
+
+			width := videoStream.(av.VideoCodecData).Width()
+			height := videoStream.(av.VideoCodecData).Height()
+
+			// Set config values as well
+			configuration.Config.Capture.IPCamera.Width = width
+			configuration.Config.Capture.IPCamera.Height = height
 		}
 
 		if cameraSettings.RTSP != rtspUrl || cameraSettings.SubRTSP != subRtspUrl || cameraSettings.Width != width || cameraSettings.Height != height || cameraSettings.Num != num || cameraSettings.Denum != denum || cameraSettings.Codec != videoStream.(av.VideoCodecData).Type() {
@@ -190,6 +202,7 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu
 			cameraSettings.Denum = denum
 			cameraSettings.Codec = videoStream.(av.VideoCodecData).Type()
 			cameraSettings.Initialized = true
+
 		} else {
 			log.Log.Info("RunAgent: camera settings did not change, keeping decoder")
 		}
@@ -285,10 +298,10 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu
 		(*communication.CancelContext)()
 
 		// We will re open the configuration, might have changed :O!
-		OpenConfig(configDirectory, configuration)
+		configService.OpenConfig(configDirectory, configuration)
 
 		// We will override the configuration with the environment variables
-		OverrideWithEnvironmentVariables(configuration)
+		configService.OverrideWithEnvironmentVariables(configuration)
 
 		// Here we are cleaning up everything!
 		if configuration.Config.Offline != "true" {

machinery/src/config/main.go

@@ -1,4 +1,4 @@
-package components
+package config
 
 import (
 	"context"
@@ -210,7 +210,7 @@ func OverrideWithEnvironmentVariables(configuration *models.Configuration) {
 				configuration.Config.Key = value
 				break
 			case "AGENT_NAME":
-				configuration.Config.Name = value
+				configuration.Config.FriendlyName = value
 				break
 			case "AGENT_TIMEZONE":
 				configuration.Config.Timezone = value

machinery/src/models/Config.go

@@ -70,9 +70,11 @@ type Capture struct {
 // IPCamera configuration, such as the RTSP url of the IPCamera and the FPS.
 // Also includes ONVIF integration
 type IPCamera struct {
+	Width         int    `json:"width"`
+	Height        int    `json:"height"`
+	FPS           string `json:"fps"`
 	RTSP          string `json:"rtsp"`
 	SubRTSP       string `json:"sub_rtsp"`
-	FPS           string `json:"fps"`
 	ONVIF         string `json:"onvif,omitempty" bson:"onvif"`
 	ONVIFXAddr    string `json:"onvif_xaddr,omitempty" bson:"onvif_xaddr"`
 	ONVIFUsername string `json:"onvif_username,omitempty" bson:"onvif_username"`

machinery/src/routers/http/Routes.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/kerberos-io/agent/machinery/src/cloud"
 	"github.com/kerberos-io/agent/machinery/src/components"
+	configService "github.com/kerberos-io/agent/machinery/src/config"
 	"github.com/kerberos-io/agent/machinery/src/log"
 	"github.com/kerberos-io/agent/machinery/src/models"
 	"github.com/kerberos-io/agent/machinery/src/utils"
@@ -40,7 +41,7 @@ func AddRoutes(r *gin.Engine, authMiddleware *jwt.GinJWTMiddleware, configDirect
 		var config models.Config
 		err := c.BindJSON(&config)
 		if err == nil {
-			err := components.SaveConfig(configDirectory, config, configuration, communication)
+			err := configService.SaveConfig(configDirectory, config, configuration, communication)
 			if err == nil {
 				c.JSON(200, gin.H{
 					"data": "☄ Reconfiguring",
@@ -165,7 +166,7 @@ func AddRoutes(r *gin.Engine, authMiddleware *jwt.GinJWTMiddleware, configDirect
 			var config models.Config
 			err := c.BindJSON(&config)
 			if err == nil {
-				err := components.SaveConfig(configDirectory, config, configuration, communication)
+				err := configService.SaveConfig(configDirectory, config, configuration, communication)
 				if err == nil {
 					c.JSON(200, gin.H{
 						"data": "☄ Reconfiguring",
@@ -215,7 +216,7 @@ func AddRoutes(r *gin.Engine, authMiddleware *jwt.GinJWTMiddleware, configDirect
 				// We will only send an image once per second.
 				time.Sleep(time.Second * 1)
 				log.Log.Info("AddRoutes (/stream): reading from MJPEG stream")
-				img, err := components.GetImageFromFilePath(configDirectory)
+				img, err := configService.GetImageFromFilePath(configDirectory)
 				return img, err
 			}
 			h := components.StartMotionJPEG(imageFunction, 80)

machinery/src/routers/mqtt/main.go

@@ -8,6 +8,8 @@ import (
 	"time"
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"github.com/gofrs/uuid"
+	configService "github.com/kerberos-io/agent/machinery/src/config"
 	"github.com/kerberos-io/agent/machinery/src/log"
 	"github.com/kerberos-io/agent/machinery/src/models"
 	"github.com/kerberos-io/agent/machinery/src/onvif"
@@ -55,12 +57,22 @@ func HasMQTTClientModified(configuration *models.Configuration) bool {
 }
 
 func PackageMQTTMessage(msg Message) ([]byte, error) {
-	// We'll generate an unique id, and encrypt it using the private key.
-	msg.Mid = "0123456789+1"
+	// Create a Version 4 UUID.
+	u2, err := uuid.NewV4()
+	if err != nil {
+		log.Log.Error("failed to generate UUID: " + err.Error())
+	}
+
+	// We'll generate an unique id, and encrypt / decrypt it using the private key if available.
+	msg.Mid = u2.String()
 	msg.Timestamp = time.Now().Unix()
+
+	// At the moment we don't do the encryption part, but we'll implement it
+	// once the legacy methods (subscriptions are moved).
 	msg.Encrypted = false
 	msg.PublicKey = ""
 	msg.Fingerprint = ""
+
 	payload, err := json.Marshal(msg)
 	return payload, err
 }
@@ -98,7 +110,7 @@ func PackageMQTTMessage(msg Message) ([]byte, error) {
 // - {devicekey}/{sessionid}/answer: once a WebRTC request is received through (kerberos/register), we'll draft an answer and send it back to the remote WebRTC client.
 // - kerberos/{hubkey}/device/{devicekey}/motion: a motion signal
 
-func ConfigureMQTT(configuration *models.Configuration, communication *models.Communication) mqtt.Client {
+func ConfigureMQTT(configDirectory string, configuration *models.Configuration, communication *models.Communication) mqtt.Client {
 
 	config := configuration.Config
 
@@ -174,8 +186,9 @@ func ConfigureMQTT(configuration *models.Configuration, communication *models.Co
 				log.Log.Info("ConfigureMQTT: " + mqttClientID + " connected to " + mqttURL)
 
 				// Create a susbcription for listen and reply
-				MQTTListenerHandler(c, hubKey, configuration, communication)
+				MQTTListenerHandler(c, hubKey, configDirectory, configuration, communication)
 
+				// Legacy methods below -> should be converted to the above method.
 				// Create a subscription to know if send out a livestream or not.
 				MQTTListenerHandleLiveSD(c, hubKey, configuration, communication)
 
@@ -207,7 +220,7 @@ func ConfigureMQTT(configuration *models.Configuration, communication *models.Co
 	return nil
 }
 
-func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configuration *models.Configuration, communication *models.Communication) {
+func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configDirectory string, configuration *models.Configuration, communication *models.Communication) {
 	if hubKey == "" {
 		log.Log.Info("MQTTListenerHandler: no hub key provided, not subscribing to kerberos/hub/{hubkey}")
 	} else {
@@ -225,6 +238,7 @@ func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configuration *m
 
 			var message Message
 			json.Unmarshal(msg.Payload(), &message)
+
 			if message.Mid != "" && message.Timestamp != 0 {
 				// Messages might be encrypted, if so we'll
 				// need to decrypt them.
@@ -242,6 +256,8 @@ func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configuration *m
 					// Not relevant for this device, so we'll ignore it.
 				} else {
 					// We'll find out which message we received, and act accordingly.
+					log.Log.Info("MQTTListenerHandler: received message with action: " + payload.Action)
+
 					switch payload.Action {
 					case "record":
 						HandleRecording(mqttClient, hubKey, payload, configuration, communication)
@@ -249,6 +265,10 @@ func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configuration *m
 						HandleGetPTZPosition(mqttClient, hubKey, payload, configuration, communication)
 					case "update-ptz-position":
 						HandleUpdatePTZPosition(mqttClient, hubKey, payload, configuration, communication)
+					case "request-config":
+						HandleRequestConfig(mqttClient, hubKey, payload, configuration, communication)
+					case "update-config":
+						HandleUpdateConfig(mqttClient, hubKey, payload, configDirectory, configuration, communication)
 					}
 				}
 			}
@@ -336,10 +356,95 @@ func HandleUpdatePTZPosition(mqttClient mqtt.Client, hubKey string, payload Payl
 	}
 }
 
+// We received a request config request, we'll fetch the current config and send it back.
+type RequestConfigPayload struct {
+	Timestamp int64 `json:"timestamp"` // timestamp of the preset request.
+}
+
+func HandleRequestConfig(mqttClient mqtt.Client, hubKey string, payload Payload, configuration *models.Configuration, communication *models.Communication) {
+	value := payload.Value
+
+	// Convert map[string]interface{} to RequestConfigPayload
+	jsonData, _ := json.Marshal(value)
+	var configPayload RequestConfigPayload
+	json.Unmarshal(jsonData, &configPayload)
+
+	if configPayload.Timestamp != 0 {
+		// Get Config from the device
+
+		key := configuration.Config.Key
+		name := configuration.Config.Name
+
+		if key != "" && name != "" {
+
+			var configMap map[string]interface{}
+			inrec, _ := json.Marshal(configuration.Config)
+			json.Unmarshal(inrec, &configMap)
+
+			message := Message{
+				Payload: Payload{
+					Action:   "receive-config",
+					DeviceId: configuration.Config.Key,
+					Value:    configMap,
+				},
+			}
+			payload, err := PackageMQTTMessage(message)
+			if err == nil {
+				mqttClient.Publish("kerberos/hub/"+hubKey, 0, false, payload)
+			} else {
+				log.Log.Info("HandleRequestConfig: something went wrong while sending config to hub: " + string(payload))
+			}
+
+		} else {
+			log.Log.Info("HandleRequestConfig: no config available")
+		}
+
+		log.Log.Info("HandleRequestConfig: Received a request for the config")
+	}
+}
+
+// We received a update config request, we'll update the current config and send a confirmation back.
+type UpdateConfigPayload struct {
+	Timestamp int64         `json:"timestamp"` // timestamp of the preset request.
+	Config    models.Config `json:"config"`
+}
+
+func HandleUpdateConfig(mqttClient mqtt.Client, hubKey string, payload Payload, configDirectory string, configuration *models.Configuration, communication *models.Communication) {
+	value := payload.Value
+
+	// Convert map[string]interface{} to UpdateConfigPayload
+	jsonData, _ := json.Marshal(value)
+	var configPayload UpdateConfigPayload
+	json.Unmarshal(jsonData, &configPayload)
+
+	if configPayload.Timestamp != 0 {
+
+		config := configPayload.Config
+		err := configService.SaveConfig(configDirectory, config, configuration, communication)
+		if err == nil {
+			log.Log.Info("HandleUpdateConfig: Config updated")
+
+			message := Message{
+				Payload: Payload{
+					Action:   "acknowledge-update-config",
+					DeviceId: configuration.Config.Key,
+				},
+			}
+			payload, err := PackageMQTTMessage(message)
+			if err == nil {
+				mqttClient.Publish("kerberos/hub/"+hubKey, 0, false, payload)
+			} else {
+				log.Log.Info("HandleRequestConfig: something went wrong while sending acknowledge config to hub: " + string(payload))
+			}
+		} else {
+			log.Log.Info("HandleUpdateConfig: Config update failed")
+		}
+	}
+}
+
 func DisconnectMQTT(mqttClient mqtt.Client, config *models.Config) {
 	if mqttClient != nil {
 		// Cleanup all subscriptions
-
 		// New methods
 		mqttClient.Unsubscribe("kerberos/agent/" + PREV_HubKey)