remote stream

Author: eloquentarduino

library.properties

@@ -1,5 +1,5 @@
 name=EloquentEsp32cam
-version=2.7.10
+version=2.7.11
 author=Simone Salerno <support@eloquentarduino.com>
 maintainer=Simone Salerno <support@eloquentarduino.com>
 sentence=Use your Esp32-cam like an expert

src/eloquent_esp32cam/mqtt-stream.h

@@ -0,0 +1,152 @@
+#pragma once
+
+#ifndef MQTT_MAX_PACKET_SIZE
+#define MQTT_MAX_PACKET_SIZE 30000
+#endif
+
+#include <WiFi.h>
+#include <MQTT.h>
+#include "./extra/exception.h"
+
+using Eloquent::Error::Exception;
+
+typedef struct {
+    uint8_t *buf;
+    uint32_t len;
+} mqtt_frame_t;
+
+namespace Eloquent {
+    namespace Esp32cam {
+        /**
+         * Stream camera frames over MQTT
+         */
+        class MqttStream {
+        public:
+            Exception exception;
+
+            /**
+             *
+             */
+            MqttStream() :
+                exception("MqttStream"),
+                _throttle(100) {
+
+            }
+
+            /**
+             * Don't send frames too often
+             * @param ms
+             */
+            void throttle(size_t ms) {
+                _throttle = ms;
+            }
+
+            /**
+             *
+             * @return
+             */
+            Exception& begin(String clientID) {
+                _clientID = clientID;
+                _mqttClient.begin("mqtt.eloquentarduino.com", 1883, _wifiClient);
+
+                if (WiFi.status() != WL_CONNECTED)
+                    return exception.set("WiFi not connected");
+
+                if (!_mqttClient.connect(_clientID.c_str(), _clientID.c_str(), "password")) {
+                    ESP_LOGE("MqttStream", "Can't connect to MQTT broker. Are you sure your CLIENT_ID is valid?");
+                    return exception.set("Can't connect to MQTT broker");
+                }
+
+                _queue = xQueueCreate(1, sizeof(mqtt_frame_t));
+
+                xTaskCreate([](void *args) {
+                    MqttStream *self = (MqttStream*) args;
+                    mqtt_frame_t frame;
+                    String topic = String("/") + self->_clientID + "/stream";
+
+                    while (true) {
+                        // await for new frame
+                        if (xQueueReceive(self->_queue, &frame, 10) != pdPASS)
+                            continue;
+
+                        if (WiFi.status() != WL_CONNECTED) {
+                            ESP_LOGE("MqttStream", "WiFi not connected");
+                            delay(500);
+                            continue;
+                        }
+
+                        if (!self->_mqttClient.connected()) {
+                            ESP_LOGE("MqttStream", "Can't connect to MQTT broker. Are you sure your CLIENT_ID is valid?");
+                            delay(10);
+                            continue;
+                        }
+
+                        // publish
+                        if (frame.len > 0) {
+                            if (frame.len > MQTT_MAX_PACKET_SIZE) {
+                                ESP_LOGE("MqttStream", "Frame too large (%d bytes, max %d allowed)", frame.len, MQTT_MAX_PACKET_SIZE);
+                                continue;
+                            }
+
+                            if (!self->_mqttClient.publish(topic.c_str(), (const char*) frame.buf, frame.len))
+                                ESP_LOGE("MqttStream", "Can't publish frame");
+                        }
+
+//                        self->_mqttClient.loop();
+                    }
+                }, "MqttStream", 50000, this, 0, NULL);
+
+                return exception.clear();
+            }
+
+            /**
+             *
+             * @tparam Camera
+             * @param camera
+             * @return
+             */
+            template<typename Camera>
+            Exception& queue(Camera& camera) {
+                if (_queue == NULL)
+                    return exception.set("Queue is not initialized. Did you call begin()?");
+
+                const size_t now = millis();
+
+                // throttle
+                if (now > _queuedAt && now - _queuedAt < _throttle)
+                    return exception.clear();
+
+                // discard old frame, if not consumed yet
+                if (uxQueueSpacesAvailable(_queue) > 0)
+                    xQueueReset(_queue);
+
+                if (!camera.hasFrame())
+                    return exception.set("Cannot queue empty frame");
+
+                mqtt_frame_t frame = {
+                        .buf = camera.frame->buf,
+                        .len = camera.frame->len
+                };
+
+                if (xQueueSendToFront(_queue, (void *) &frame, 0) != pdTRUE)
+                    return exception.set("Cannot queue frame");
+
+                _queuedAt = millis();
+                return exception.clear();
+            }
+
+        protected:
+            size_t _queuedAt;
+            size_t _throttle;
+            String _clientID;
+            String _location;
+            WiFiClient _wifiClient;
+            MQTTClient _mqttClient;
+            QueueHandle_t _queue;
+        };
+    }
+}
+
+namespace eloq {
+    static Eloquent::Esp32cam::MqttStream mqttStream;
+}

src/eloquent_esp32cam/remote-stream.h

@@ -0,0 +1,99 @@
+#pragma once
+
+#include <WiFi.h>
+#include "./extra/exception.h"
+
+namespace Eloquent {
+    class RemoteStream {
+    public:
+        Exception exception;
+
+        /**
+         *
+         */
+        RemoteStream() :
+            exception("RemoteStream"),
+            _throttle(100),
+            _sentAt(0) {
+
+        }
+
+        /**
+         * Don't send frames too often
+         * @param ms
+         */
+        void throttle(size_t ms) {
+            _throttle = ms;
+        }
+
+        /**
+         *
+         * @param clientID
+         * @return
+         */
+        Exception& begin(String clientID) {
+            _clientID = clientID;
+
+            if (WiFi.status() != WL_CONNECTED)
+                return exception.set("WiFi not connected");
+
+            if (!_wifiClient.connect(getServerName().c_str(), 10300))
+                return exception.set("Can't connect to streaming server");
+
+            return exception.clear();
+        }
+
+        /**
+         *
+         * @param frame
+         * @return
+         */
+        Exception& send(camera_fb_t *frame) {
+            const size_t now = millis();
+
+            if (now > _sentAt && now - _sentAt < _throttle)
+                return exception.clear();
+
+            if (!_wifiClient.connected())
+                _wifiClient.connect(getServerName().c_str(), 10300);
+
+            if (!_wifiClient.connected())
+                return exception.set("Can't connect to streaming server");
+
+            _wifiClient.print("##SOF##");
+            _wifiClient.print(_clientID);
+            _wifiClient.print("##DATA##");
+            _wifiClient.write(frame->buf, frame->len);
+            _wifiClient.print("##EOF##");
+            _wifiClient.flush();
+
+            return exception.clear();
+        }
+
+        /**
+         *
+         * @return
+         */
+        String address() {
+            return String("http://") + getServerName() + "/esp32-cam-stream?client_id=" + _clientID;
+        }
+
+    protected:
+        size_t _throttle;
+        size_t _sentAt;
+        String _clientID;
+        WiFiClient _wifiClient;
+
+        /**
+         *
+         * @return
+         */
+        String getServerName() {
+            return "mqtt.eloquentarduino.com";
+        }
+    };
+}
+
+namespace eloq {
+    static Eloquent::RemoteStream remoteStream;
+}