use low-res in SD streaming

machinery/src/components/Kerberos.go

@@ -107,6 +107,7 @@ func RunAgent(configuration *models.Configuration, communication *models.Communi
 		// At some routines we will need to decode the image.
 		// Make sure its properly locked as we only have a single decoder.
 		var decoderMutex sync.Mutex
+		var subDecoderMutex sync.Mutex
 		decoder := capture.GetVideoDecoder(streams)
 
 		var subDecoder *ffmpeg.VideoDecoder
@@ -117,6 +118,7 @@ func RunAgent(configuration *models.Configuration, communication *models.Communi
 		communication.Decoder = decoder
 		communication.SubDecoder = subDecoder
 		communication.DecoderMutex = &decoderMutex
+		communication.SubDecoderMutex = &subDecoderMutex
 
 		// Create a packet queue, which is filled by the HandleStream routing
 		// and consumed by all other routines: motion, livestream, etc.
@@ -162,20 +164,20 @@ func RunAgent(configuration *models.Configuration, communication *models.Communi
 
 		if subStreamEnabled {
 			motionCursor := subQueue.Latest()
-			go computervision.ProcessMotion(motionCursor, configuration, communication, mqttClient, subDecoder, &decoderMutex)
+			go computervision.ProcessMotion(motionCursor, configuration, communication, mqttClient, subDecoder, &subDecoderMutex)
 		} else {
 			motionCursor := queue.Latest()
 			go computervision.ProcessMotion(motionCursor, configuration, communication, mqttClient, decoder, &decoderMutex)
 		}
 
 		// Handle livestream SD (low resolution over MQTT)
-		//if subStreamEnabled {
-		//	livestreamCursor := subQueue.Latest()
-		//	cloud.HandleLiveStreamSD(livestreamCursor, configuration, communication, mqttClient, subDecoder, &decoderMutex)
-		//} else {
-		livestreamCursor := queue.Latest()
-		go cloud.HandleLiveStreamSD(livestreamCursor, configuration, communication, mqttClient, decoder, &decoderMutex)
-		//}
+		if subStreamEnabled {
+			livestreamCursor := subQueue.Latest()
+			go cloud.HandleLiveStreamSD(livestreamCursor, configuration, communication, mqttClient, subDecoder, &subDecoderMutex)
+		} else {
+			livestreamCursor := queue.Latest()
+			go cloud.HandleLiveStreamSD(livestreamCursor, configuration, communication, mqttClient, decoder, &decoderMutex)
+		}
 
 		// Handle livestream HD (high resolution over WEBRTC)
 		communication.HandleLiveHDHandshake = make(chan models.SDPPayload, 1)

machinery/src/computervision/main.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"bytes"
 	"image"
+	"image/jpeg"
 	"io/ioutil"
 	"os"
 	"sort"
@@ -16,7 +17,8 @@ import (
 	"github.com/kerberos-io/agent/machinery/src/log"
 	"github.com/kerberos-io/agent/machinery/src/models"
 	"github.com/kerberos-io/joy4/av/pubsub"
-	"github.com/whorfin/go-libjpeg/jpeg"
+
+	//"github.com/whorfin/go-libjpeg/jpeg"
 
 	geo "github.com/kellydunn/golang-geo"
 	"github.com/kerberos-io/joy4/av"
@@ -138,7 +140,8 @@ func ProcessMotion(motionCursor *pubsub.QueueCursor, configuration *models.Confi
 						t := strconv.FormatInt(time.Now().Unix(), 10)
 						f, err := os.Create("./data/snapshots/" + t + ".jpg")
 						if err == nil {
-							jpeg.Encode(f, &rgbImage.Image, &jpeg.EncoderOptions{Quality: 70})
+							jpeg.Encode(f, &rgbImage.Image, &jpeg.Options{Quality: 15})
+							//jpeg.Encode(f, &rgbImage.Image, &jpeg.EncoderOptions{Quality: 70, })
 							f.Close()
 						}
 					}
@@ -230,7 +233,8 @@ func GetRawImage(pkt av.Packet, dec *ffmpeg.VideoDecoder, decoderMutex *sync.Mut
 func ImageToBytes(img image.Image) ([]byte, error) {
 	buffer := new(bytes.Buffer)
 	w := bufio.NewWriter(buffer)
-	err := jpeg.Encode(w, img, &jpeg.EncoderOptions{Quality: 70})
+	//err := jpeg.Encode(w, img, &jpeg.EncoderOptions{Quality: 70})
+	err := jpeg.Encode(w, img, &jpeg.Options{Quality: 15})
 	return buffer.Bytes(), err
 }
 

machinery/src/models/Communication.go

@@ -29,6 +29,7 @@ type Communication struct {
 	IsConfiguring         *abool.AtomicBool
 	Queue                 *pubsub.Queue
 	DecoderMutex          *sync.Mutex
+	SubDecoderMutex       *sync.Mutex
 	Decoder               *ffmpeg.VideoDecoder
 	SubDecoder            *ffmpeg.VideoDecoder
 }