From ecadf7a4dba0b1faa3acf7277057272939866365 Mon Sep 17 00:00:00 2001 From: Cedric Verstraeten Date: Tue, 11 Jun 2024 22:47:01 +0200 Subject: [PATCH] add realtime processing endpoint --- .gitignore | 1 + README.md | 2 + machinery/data/config/config.json | 4 +- machinery/src/cloud/Cloud.go | 79 +++++++++++++ machinery/src/components/Kerberos.go | 9 ++ machinery/src/config/main.go | 8 ++ machinery/src/models/Config.go | 68 ++++++------ ui/package.json | 1 - ui/public/locales/en/translation.json | 7 +- ui/src/pages/Settings/Settings.jsx | 153 ++++++++++++++++++-------- 10 files changed, 250 insertions(+), 82 deletions(-) diff --git a/.gitignore b/.gitignore index 70b23fc..1e4aa56 100644 --- a/.gitignore +++ b/.gitignore @@ -11,4 +11,5 @@ machinery/data/snapshots machinery/test* machinery/init-dev.sh machinery/.env +machinery/vendor deployments/docker/private-docker-compose.yaml \ No newline at end of file diff --git a/README.md b/README.md index 5f4b7dd..4ccc2b5 100644 --- a/README.md +++ b/README.md @@ -221,6 +221,8 @@ Next to attaching the configuration file, it is also possible to override the co | `AGENT_MQTT_URI` | A MQTT broker endpoint that is used for bi-directional communication (live view, onvif, etc) | "tcp://mqtt.kerberos.io:1883" | | `AGENT_MQTT_USERNAME` | Username of the MQTT broker. | "" | | `AGENT_MQTT_PASSWORD` | Password of the MQTT broker. | "" | +| `AGENT_REALTIME_PROCESSING` | If `AGENT_REALTIME_PROCESSING` set to `true`, the agent will send key frames to the topic | "" | +| `AGENT_REALTIME_PROCESSING_TOPIC` | The topic to which keyframes will be send in base64 encoded format. | "" | | `AGENT_STUN_URI` | When using WebRTC, you'll need to provide a STUN server. | "stun:turn.kerberos.io:8443" | | `AGENT_FORCE_TURN` | Force using a TURN server, by generating relay candidates only. | "false" | | `AGENT_TURN_URI` | When using WebRTC, you'll need to provide a TURN server. | "turn:turn.kerberos.io:8443" | diff --git a/machinery/data/config/config.json b/machinery/data/config/config.json index e171fa9..bdde575 100644 --- a/machinery/data/config/config.json +++ b/machinery/data/config/config.json @@ -114,5 +114,7 @@ "hub_private_key": "", "hub_site": "", "condition_uri": "", - "encryption": {} + "encryption": {}, + "realtimeprocessing": "true", + "realtimeprocessing_topic": "" } diff --git a/machinery/src/cloud/Cloud.go b/machinery/src/cloud/Cloud.go index bedae2d..3c8c06c 100644 --- a/machinery/src/cloud/Cloud.go +++ b/machinery/src/cloud/Cloud.go @@ -328,6 +328,12 @@ loop: if err != nil { log.Log.Debug("cloud.HandleHeartBeat(): error while creating pull point subscription: " + err.Error()) } + + /*outputs, err := onvif.GetRelayOutputs(device) + fmt.Println(outputs) + if err != nil { + log.Log.Debug("cloud.HandleHeartBeat(): error while getting relay outputs: " + err.Error()) + }*/ } } else { @@ -694,6 +700,79 @@ func HandleLiveStreamHD(livestreamCursor *packets.QueueCursor, configuration *mo } } +func HandleRealtimeProcessing(processingCursor *packets.QueueCursor, configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, rtspClient capture.RTSPClient) { + + log.Log.Debug("cloud.RealtimeProcessing(): started") + + config := configuration.Config + + // If offline made is enabled, we will stop the thread. + if config.Offline == "true" { + log.Log.Debug("cloud.RealtimeProcessing(): stopping as Offline is enabled.") + } else { + + // Check if we need to enable the realtime processing + if config.RealtimeProcessing == "true" { + + hubKey := "" + if config.Cloud == "s3" && config.S3 != nil && config.S3.Publickey != "" { + hubKey = config.S3.Publickey + } else if config.Cloud == "kstorage" && config.KStorage != nil && config.KStorage.CloudKey != "" { + hubKey = config.KStorage.CloudKey + } + // This is the new way ;) + if config.HubKey != "" { + hubKey = config.HubKey + } + + // We will publish the keyframes to the MQTT topic. + realtimeProcessingTopic := "kerberos/keyframes/" + hubKey + if config.RealtimeProcessingTopic != "" { + realtimeProcessingTopic = config.RealtimeProcessingTopic + } + + var cursorError error + var pkt packets.Packet + + for cursorError == nil { + pkt, cursorError = processingCursor.ReadPacket() + if len(pkt.Data) == 0 || !pkt.IsKeyFrame { + continue + } + + log.Log.Info("cloud.RealtimeProcessing(): Sending base64 encoded images to MQTT.") + 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-keyframe", + DeviceId: configuration.Config.Key, + Value: valueMap, + }, + } + payload, err := models.PackageMQTTMessage(configuration, message) + if err == nil { + + mqttClient.Publish(realtimeProcessingTopic, 0, false, payload) + } else { + log.Log.Info("cloud.RealtimeProcessing(): something went wrong while sending acknowledge config to hub: " + string(payload)) + } + } + } + + } else { + log.Log.Debug("cloud.RealtimeProcessing(): stopping as Liveview is disabled.") + } + } + + log.Log.Debug("cloud.HandleLiveStreamSD(): finished") +} + // VerifyHub godoc // @Router /api/hub/verify [post] // @ID verify-hub diff --git a/machinery/src/components/Kerberos.go b/machinery/src/components/Kerberos.go index da0c563..f6eec60 100644 --- a/machinery/src/components/Kerberos.go +++ b/machinery/src/components/Kerberos.go @@ -276,6 +276,15 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu go computervision.ProcessMotion(motionCursor, configuration, communication, mqttClient, rtspClient) } + // Handle realtime processing if enabled. + if subStreamEnabled { + realtimeProcessingCursor := subQueue.Latest() + go cloud.HandleRealtimeProcessing(realtimeProcessingCursor, configuration, communication, mqttClient, rtspClient) + } else { + realtimeProcessingCursor := queue.Latest() + go cloud.HandleRealtimeProcessing(realtimeProcessingCursor, configuration, communication, mqttClient, rtspClient) + } + // Handle Upload to cloud provider (Kerberos Hub, Kerberos Vault and others) go cloud.HandleUpload(configDirectory, configuration, communication) diff --git a/machinery/src/config/main.go b/machinery/src/config/main.go index bd5f25d..41041fa 100644 --- a/machinery/src/config/main.go +++ b/machinery/src/config/main.go @@ -382,6 +382,14 @@ func OverrideWithEnvironmentVariables(configuration *models.Configuration) { configuration.Config.MQTTPassword = value break + /* Real-time streaming of keyframes to a MQTT topic */ + case "AGENT_REALTIME_PROCESSING": + configuration.Config.RealtimeProcessing = value + break + case "AGENT_REALTIME_PROCESSING_TOPIC": + configuration.Config.RealtimeProcessingTopic = value + break + /* WebRTC settings for live-streaming (remote) */ case "AGENT_STUN_URI": configuration.Config.STUNURI = value diff --git a/machinery/src/models/Config.go b/machinery/src/models/Config.go index e72a4fe..4b063a9 100644 --- a/machinery/src/models/Config.go +++ b/machinery/src/models/Config.go @@ -12,39 +12,41 @@ type Configuration struct { // Config is the highlevel struct which contains all the configuration of // your Kerberos Open Source instance. type Config struct { - Type string `json:"type"` - Key string `json:"key"` - Name string `json:"name"` - FriendlyName string `json:"friendly_name"` - Time string `json:"time" bson:"time"` - Offline string `json:"offline"` - AutoClean string `json:"auto_clean"` - RemoveAfterUpload string `json:"remove_after_upload"` - MaxDirectorySize int64 `json:"max_directory_size"` - Timezone string `json:"timezone"` - Capture Capture `json:"capture"` - Timetable []*Timetable `json:"timetable"` - Region *Region `json:"region"` - Cloud string `json:"cloud" bson:"cloud"` - S3 *S3 `json:"s3,omitempty" bson:"s3,omitempty"` - KStorage *KStorage `json:"kstorage,omitempty" bson:"kstorage,omitempty"` - Dropbox *Dropbox `json:"dropbox,omitempty" bson:"dropbox,omitempty"` - MQTTURI string `json:"mqtturi" bson:"mqtturi,omitempty"` - MQTTUsername string `json:"mqtt_username" bson:"mqtt_username"` - MQTTPassword string `json:"mqtt_password" bson:"mqtt_password"` - STUNURI string `json:"stunuri" bson:"stunuri"` - ForceTurn string `json:"turn_force" bson:"turn_force"` - TURNURI string `json:"turnuri" bson:"turnuri"` - TURNUsername string `json:"turn_username" bson:"turn_username"` - TURNPassword string `json:"turn_password" bson:"turn_password"` - HeartbeatURI string `json:"heartbeaturi" bson:"heartbeaturi"` /*obsolete*/ - HubEncryption string `json:"hub_encryption" bson:"hub_encryption"` - HubURI string `json:"hub_uri" bson:"hub_uri"` - HubKey string `json:"hub_key" bson:"hub_key"` - HubPrivateKey string `json:"hub_private_key" bson:"hub_private_key"` - HubSite string `json:"hub_site" bson:"hub_site"` - ConditionURI string `json:"condition_uri" bson:"condition_uri"` - Encryption *Encryption `json:"encryption,omitempty" bson:"encryption,omitempty"` + Type string `json:"type"` + Key string `json:"key"` + Name string `json:"name"` + FriendlyName string `json:"friendly_name"` + Time string `json:"time" bson:"time"` + Offline string `json:"offline"` + AutoClean string `json:"auto_clean"` + RemoveAfterUpload string `json:"remove_after_upload"` + MaxDirectorySize int64 `json:"max_directory_size"` + Timezone string `json:"timezone"` + Capture Capture `json:"capture"` + Timetable []*Timetable `json:"timetable"` + Region *Region `json:"region"` + Cloud string `json:"cloud" bson:"cloud"` + S3 *S3 `json:"s3,omitempty" bson:"s3,omitempty"` + KStorage *KStorage `json:"kstorage,omitempty" bson:"kstorage,omitempty"` + Dropbox *Dropbox `json:"dropbox,omitempty" bson:"dropbox,omitempty"` + MQTTURI string `json:"mqtturi" bson:"mqtturi,omitempty"` + MQTTUsername string `json:"mqtt_username" bson:"mqtt_username"` + MQTTPassword string `json:"mqtt_password" bson:"mqtt_password"` + STUNURI string `json:"stunuri" bson:"stunuri"` + ForceTurn string `json:"turn_force" bson:"turn_force"` + TURNURI string `json:"turnuri" bson:"turnuri"` + TURNUsername string `json:"turn_username" bson:"turn_username"` + TURNPassword string `json:"turn_password" bson:"turn_password"` + HeartbeatURI string `json:"heartbeaturi" bson:"heartbeaturi"` /*obsolete*/ + HubEncryption string `json:"hub_encryption" bson:"hub_encryption"` + HubURI string `json:"hub_uri" bson:"hub_uri"` + HubKey string `json:"hub_key" bson:"hub_key"` + HubPrivateKey string `json:"hub_private_key" bson:"hub_private_key"` + HubSite string `json:"hub_site" bson:"hub_site"` + ConditionURI string `json:"condition_uri" bson:"condition_uri"` + Encryption *Encryption `json:"encryption,omitempty" bson:"encryption,omitempty"` + RealtimeProcessing string `json:"realtimeprocessing,omitempty" bson:"realtimeprocessing,omitempty"` + RealtimeProcessingTopic string `json:"realtimeprocessing_topic" bson:"realtimeprocessing_topic"` } // Capture defines which camera type (Id) you are using (IP, USB or Raspberry Pi camera), diff --git a/ui/package.json b/ui/package.json index 673254c..3b3c09c 100644 --- a/ui/package.json +++ b/ui/package.json @@ -1,7 +1,6 @@ { "name": "agent-ui", "version": "0.1.0", - "private": false, "dependencies": { "@giantmachines/redux-websocket": "^1.5.1", "@kerberos-io/ui": "^1.76.0", diff --git a/ui/public/locales/en/translation.json b/ui/public/locales/en/translation.json index aeb1059..f7587be 100644 --- a/ui/public/locales/en/translation.json +++ b/ui/public/locales/en/translation.json @@ -160,7 +160,12 @@ "description2_mqtt": "to the Kerberos Agent, to achieve for example livestreaming or ONVIF (PTZ) capabilities.", "mqtt_brokeruri": "Broker Uri", "mqtt_username": "Username", - "mqtt_password": "Password" + "mqtt_password": "Password", + "realtimeprocessing": "Realtime Processing", + "description_realtimeprocessing": "By enabling realtime processing, you will receive realtime video keyframes through the MQTT connection specified above.", + "realtimeprocessing_topic": "Topic to publish", + "realtimeprocessing_enabled": "Enable realtime processing", + "description_realtimeprocessing_enabled": "Send realtime video keyframes through MQTT." }, "conditions": { "timeofinterest": "Time Of Interest", diff --git a/ui/src/pages/Settings/Settings.jsx b/ui/src/pages/Settings/Settings.jsx index ac8494e..0eccb85 100644 --- a/ui/src/pages/Settings/Settings.jsx +++ b/ui/src/pages/Settings/Settings.jsx @@ -1123,6 +1123,87 @@ class Settings extends React.Component { )} + {/* STUN/TURN block */} + {showStreamingSection && config.offline !== 'true' && ( + + +

{t('settings.streaming.stun_turn_forward')}

+ + +

{t('settings.streaming.stun_turn_description_forward')}

+ +
+ + this.onUpdateToggle( + 'capture', + 'forwardwebrtc', + event, + config.capture + ) + } + /> +
+ {t('settings.streaming.stun_turn_webrtc')} +

+ {t('settings.streaming.stun_turn_description_webrtc')} +

+
+
+ +
+ + this.onUpdateToggle( + 'capture', + 'transcodingwebrtc', + event, + config.capture + ) + } + /> +
+ {t('settings.streaming.stun_turn_transcode')} +

+ {t( + 'settings.streaming.stun_turn_description_transcode' + )} +

+
+
+ + {config.capture.transcodingwebrtc === 'true' && ( + + this.onUpdateNumberField( + 'capture', + 'transcodingresolution', + value, + config.capture + ) + } + /> + )} + + +