Merge pull request #210 from kerberos-io/feature/minimize-sd-view-image

feature/minimize-sd-view-image

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/go.mod

@@ -26,6 +26,7 @@ require (
 	github.com/kerberos-io/joy4 v1.0.64
 	github.com/kerberos-io/onvif v1.0.0
 	github.com/minio/minio-go/v6 v6.0.57
+	github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646
 	github.com/op/go-logging v0.0.0-20160315200505-970db520ece7
 	github.com/pion/interceptor v0.1.40
 	github.com/pion/rtp v1.8.19
@@ -41,6 +42,7 @@ require (
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.36.0
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.36.0
 	go.opentelemetry.io/otel/sdk v1.36.0
+	go.opentelemetry.io/otel/trace v1.36.0
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1
 )
 
@@ -118,7 +120,6 @@ require (
 	github.com/ziutek/mymysql v1.5.4 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
 	go.opentelemetry.io/otel/metric v1.36.0 // indirect
-	go.opentelemetry.io/otel/trace v1.36.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.6.0 // indirect
 	golang.org/x/arch v0.16.0 // indirect
 	golang.org/x/crypto v0.38.0 // indirect

machinery/go.sum

@@ -847,6 +847,8 @@ github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9G
 github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
 github.com/montanaflynn/stats v0.7.1 h1:etflOAAHORrCC44V+aR6Ftzort912ZU+YLiSTuV8eaE=
 github.com/montanaflynn/stats v0.7.1/go.mod h1:etXPPgVO6n31NxCd9KQUMvCM+ve0ruNzt6R8Bnaayow=
+github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646 h1:zYyBkD/k9seD2A7fsi6Oo2LfFZAehjjQMERAvZLEDnQ=
+github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646/go.mod h1:jpp1/29i3P1S/RLdc7JQKbRpFeM1dOBd8T9ki5s+AY8=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
 github.com/nxadm/tail v1.4.11 h1:8feyoE3OzPrcshW5/MJ4sGESc5cqmGkGCWlco4l0bqY=
 github.com/nxadm/tail v1.4.11/go.mod h1:OTaG3NK980DZzxbRq6lEuzgU+mug70nY11sMd4JXXHc=

machinery/src/capture/main.go

@@ -448,6 +448,12 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 					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
 				// 1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4
 				// - Timestamp
@@ -751,7 +757,8 @@ func Base64Image(captureDevice *Capture, communication *models.Communication) st
 				var img image.YCbCr
 				img, err = (*rtspClient).DecodePacket(pkt)
 				if err == nil {
-					bytes, _ := utils.ImageToBytes(&img)
+					imageResized, _ := utils.ResizeImage(&img, 100000)
+					bytes, _ := utils.ImageToBytes(imageResized)
 					encodedImage = base64.StdEncoding.EncodeToString(bytes)
 					break
 				} else {

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
@@ -705,25 +706,79 @@ func HandleLiveStreamSD(livestreamCursor *packets.QueueCursor, configuration *mo
 				log.Log.Info("cloud.HandleLiveStreamSD(): Sending base64 encoded images to MQTT.")
 				img, err := rtspClient.DecodePacket(pkt)
 				if err == nil {
-					bytes, _ := utils.ImageToBytes(&img)
-					encoded := base64.StdEncoding.EncodeToString(bytes)
+					imageResized, _ := utils.ResizeImage(&img, 100000)
+					bytes, _ := utils.ImageToBytes(imageResized)
 
-					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 {
@@ -810,7 +865,8 @@ func HandleRealtimeProcessing(processingCursor *packets.QueueCursor, configurati
 				log.Log.Info("cloud.RealtimeProcessing(): Sending base64 encoded images to MQTT.")
 				img, err := rtspClient.DecodePacket(pkt)
 				if err == nil {
-					bytes, _ := utils.ImageToBytes(&img)
+					imageResized, _ := utils.ResizeImage(&img, 100000)
+					bytes, _ := utils.ImageToBytes(imageResized)
 					encoded := base64.StdEncoding.EncodeToString(bytes)
 
 					valueMap := make(map[string]interface{})

machinery/src/components/Kerberos.go

@@ -698,7 +698,8 @@ func GetSnapshotRaw(c *gin.Context, captureDevice *capture.Capture, configuratio
 	image := capture.JpegImage(captureDevice, communication)
 
 	// encode image to jpeg
-	bytes, _ := utils.ImageToBytes(&image)
+	imageResized, _ := utils.ResizeImage(&image, 100000)
+	bytes, _ := utils.ImageToBytes(imageResized)
 
 	// Return image/jpeg
 	c.Data(200, "image/jpeg", bytes)

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/routers/websocket/main.go

@@ -159,7 +159,8 @@ logreader:
 				var img image.YCbCr
 				img, err = (*rtspClient).DecodePacket(pkt)
 				if err == nil {
-					bytes, _ := utils.ImageToBytes(&img)
+					imageResized, _ := utils.ResizeImage(&img, 100000)
+					bytes, _ := utils.ImageToBytes(imageResized)
 					encodedImage = base64.StdEncoding.EncodeToString(bytes)
 				} else {
 					continue

machinery/src/utils/main.go

@@ -21,9 +21,11 @@ import (
 	"github.com/kerberos-io/agent/machinery/src/encryption"
 	"github.com/kerberos-io/agent/machinery/src/log"
 	"github.com/kerberos-io/agent/machinery/src/models"
+
+	"github.com/nfnt/resize"
 )
 
-const VERSION = "3.3.5"
+const VERSION = "3.5.0"
 
 const letterBytes = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ"
 
@@ -401,9 +403,21 @@ func Decrypt(directoryOrFile string, symmetricKey []byte) {
 	}
 }
 
-func ImageToBytes(img image.Image) ([]byte, error) {
+func ImageToBytes(img *image.Image) ([]byte, error) {
 	buffer := new(bytes.Buffer)
 	w := bufio.NewWriter(buffer)
-	err := jpeg.Encode(w, img, &jpeg.Options{Quality: 15})
+	err := jpeg.Encode(w, *img, &jpeg.Options{Quality: 35})
+	log.Log.Debug("ImageToBytes() - buffer size: " + strconv.Itoa(buffer.Len()))
 	return buffer.Bytes(), err
 }
+
+func ResizeImage(img image.Image, maxSize uint64) (*image.Image, error) {
+	if img == nil {
+		return nil, errors.New("image is nil")
+	}
+
+	// resize to width 640 using Lanczos resampling
+	// and preserve aspect ratio
+	m := resize.Resize(640, 0, img, resize.Lanczos3)
+	return &m, nil
+}

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())

ui/yarn.lock

@@ -1715,10 +1715,10 @@
     "@jridgewell/resolve-uri" "^3.0.3"
     "@jridgewell/sourcemap-codec" "^1.4.10"
 
-"@kerberos-io/ui@^1.71.0":
-  version "1.71.0"
-  resolved "https://registry.yarnpkg.com/@kerberos-io/ui/-/ui-1.71.0.tgz#06914c94e8b0982068d2099acf8158917a511bfc"
-  integrity sha512-pHCTn/iQTcQEPoCK82eJHGRn6BgzW3wgV4C+mNqdKOtLTquxL+vh7molEgC66tl3DGf7HyjSNa8LuoxYbt9TEg==
+"@kerberos-io/ui@^1.76.0":
+  version "1.77.0"
+  resolved "https://registry.yarnpkg.com/@kerberos-io/ui/-/ui-1.77.0.tgz#b748b2a9abf793ff2a9ba64ee41f84debc0ca9dc"
+  integrity sha512-CHh4jeLKwrYvJRL5PM3UEN4p2k1fqwMKgSF2U6IR4v0fE2FwPc/2Ry4zGk6pvLDFHbDpR9jUkHX+iNphvStoyQ==
   dependencies:
     "@emotion/react" "^11.10.4"
     "@emotion/styled" "^11.10.4"