Add MPFAsyncVideoCapture (#100)

Author: brosenberg42

detection/api/CMakeLists.txt

@@ -52,6 +52,9 @@ set(SOURCE_FILES
         include/MPFVideoCapture.h
         src/MPFVideoCapture.cpp
 
+        include/MPFAsyncVideoCapture.h
+        src/MPFAsyncVideoCapture.cpp
+
         include/frame_transformers/SearchRegion.h
         src/frame_transformers/SearchRegion.cpp
 

detection/api/include/BlockingQueue.h

@@ -51,66 +51,128 @@ class BlockingQueue {
 
 public:
     explicit BlockingQueue(int max_size=-1)
-            : max_size_(max_size), halt_(false)
+            : max_size_(max_size)
+            , halt_(false)
+            , adding_complete_(false)
     {
     }
 
+    /**
+     * Adds a copy of item to the queue. Blocks if queue is already full.
+     * @param item object to be copied in to queue
+     */
     void push(const T& item) {
         auto lock = acquire_lock();
-        await_free_space(lock);
+        wait_until_can_add(lock);
         queue_.push(item);
         cond_.notify_all();
     }
 
+    /**
+     * Moves item in to the queue. Blocks if queue is already full.
+     * @param item object to be moved in to queue
+     */
     void push(T&& item) {
         auto lock = acquire_lock();
-        await_free_space(lock);
+        wait_until_can_add(lock);
         queue_.push(std::move(item));
         cond_.notify_all();
     }
 
+    /**
+     * Constructs an instance of T in-place in the queue. Blocks if queue is already full.
+     * @param args Arguments to be passed to T's constructor
+     */
     template <typename... Args>
     void emplace(Args&&... args) {
         auto lock = acquire_lock();
-        await_free_space(lock);
+        wait_until_can_add(lock);
         queue_.emplace(std::forward<Args>(args)...);
         cond_.notify_all();
     }
 
+
+    /**
+     * Moves an item out of the queue and returns it. Blocks if the queue is empty.
+     * @return The object from the front of the queue
+     */
     T pop() {
         auto lock = acquire_lock();
-        await_non_empty(lock);
+        wait_until_can_remove(lock);
         T result = std::move(queue_.front());
         queue_.pop();
         cond_.notify_all();
         return result;
     }
 
+    /**
+     * Indicates that both producers and consumers should stop processing even if there are items
+     * in the queue. Any future adds or removes will cause a QueueHaltedException to be thrown.
+     * This can be used to prevent one side from blocking indefinitely when the other side fails.
+     */
     void halt() {
         auto lock = acquire_lock();
         halt_ = true;
         cond_.notify_all();
     }
 
+    /**
+     * Indicates that no more items will be added to the queue. Consumers should finish processing
+     * items already in the queue. Any attempt to add items after this is called will cause a
+     * QueueHaltedException to be thrown. This can be used to prevent consumers from blocking
+     * indefinitely when a producer finishes.
+     *
+     * Loosely based on C#'s BlockingCollection<T>.CompleteAdding method
+     * (https://docs.microsoft.com/en-us/dotnet/api/system.collections.concurrent.blockingcollection-1.completeadding?view=net-5.0#System_Collections_Concurrent_BlockingCollection_1_CompleteAdding).
+     */
+    void complete_adding() {
+        auto lock = acquire_lock();
+        adding_complete_ = true;
+        cond_.notify_all();
+    }
+
+    /**
+     * @return true if adding complete. Queue may or may not be empty.
+     */
+    bool is_adding_complete() {
+        auto lock = acquire_lock();
+        return halt_ || adding_complete_;
+    }
+
+    /**
+     * The queue is "completed" if it has been halted or if a producer called complete_adding()
+     * and the remaining items have all been removed from the queue.
+     * @return true if items should not be added or removed from queue.
+     */
+    bool is_completed() {
+        auto lock = acquire_lock();
+        return halt_ || (adding_complete_ && queue_.empty());
+    }
+
+
 private:
     int max_size_;
     std::queue<T> queue_;
     std::mutex mutex_;
     std::condition_variable cond_;
     bool halt_;
+    bool adding_complete_;
+
 
-    void await_non_empty(std::unique_lock<std::mutex> &lock) {
-        cond_.wait(lock, [this] { return (halt_ || !queue_.empty()); });
-        if (halt_) {
+    void wait_until_can_remove(std::unique_lock<std::mutex> &lock) {
+        cond_.wait(lock, [this] { return halt_ || adding_complete_ || !queue_.empty(); });
+        if (halt_ || (queue_.empty() && adding_complete_)) {
             throw QueueHaltedException();
         }
     }
 
-    void await_free_space(std::unique_lock<std::mutex> &lock) {
+    void wait_until_can_add(std::unique_lock<std::mutex> &lock) {
         if (max_size_ > 0) {
-            cond_.wait(lock, [this] { return (halt_ || queue_.size() < max_size_); });
+            cond_.wait(lock, [this] {
+                return halt_ || adding_complete_ || queue_.size() < max_size_;
+            });
         }
-        if (halt_) {
+        if (halt_ || adding_complete_) {
             throw QueueHaltedException();
         }
     }

detection/api/include/MPFAsyncVideoCapture.h

@@ -0,0 +1,115 @@
+/******************************************************************************
+ * NOTICE                                                                     *
+ *                                                                            *
+ * This software (or technical data) was produced for the U.S. Government     *
+ * under contract, and is subject to the Rights in Data-General Clause        *
+ * 52.227-14, Alt. IV (DEC 2007).                                             *
+ *                                                                            *
+ * Copyright 2021 The MITRE Corporation. All Rights Reserved.                 *
+ ******************************************************************************/
+
+/******************************************************************************
+ * Copyright 2021 The MITRE Corporation                                       *
+ *                                                                            *
+ * Licensed under the Apache License, Version 2.0 (the "License");            *
+ * you may not use this file except in compliance with the License.           *
+ * You may obtain a copy of the License at                                    *
+ *                                                                            *
+ *    http://www.apache.org/licenses/LICENSE-2.0                              *
+ *                                                                            *
+ * Unless required by applicable law or agreed to in writing, software        *
+ * distributed under the License is distributed on an "AS IS" BASIS,          *
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.   *
+ * See the License for the specific language governing permissions and        *
+ * limitations under the License.                                             *
+ ******************************************************************************/
+
+#ifndef OPENMPF_CPP_COMPONENT_SDK_MPFASYNCVIDEOCAPTURE_H
+#define OPENMPF_CPP_COMPONENT_SDK_MPFASYNCVIDEOCAPTURE_H
+
+#include <future>
+#include <string>
+
+#include <opencv2/core.hpp>
+
+#include "MPFDetectionComponent.h"
+#include "MPFVideoCapture.h"
+#include "BlockingQueue.h"
+
+namespace MPF { namespace COMPONENT {
+
+    struct MPFFrame {
+    public:
+        // Include index field because we can't query MPFVideoCapture to get the current frame
+        // position. This is because MPFVideoCapture will likely be a few frames ahead by the time
+        // it is queried.
+        int index;
+        cv::Mat data;
+
+        explicit MPFFrame(int index = -1, cv::Mat data = {});
+
+        bool isValid() const;
+
+        explicit operator bool() const;
+    };
+
+
+    /**
+     * Reads frames from a regular MPFVideoCapture on a background thread. Frames are buffered
+     * in a fixed sized BlockingQueue. This class intentionally exposes a subset of the
+     * MPFVideoCapture functionality. Functionality that would impossible or difficult to write in
+     * a thread-safe manner has been omitted. Frame transformers, frame filters, and feed forward
+     * are all supported.
+     */
+    class MPFAsyncVideoCapture {
+    public:
+        explicit MPFAsyncVideoCapture(const MPFVideoJob &videoJob,
+                                      bool enableFrameTransformers=true,
+                                      bool enableFrameFiltering=true);
+
+        explicit MPFAsyncVideoCapture(std::string videoPath, int frameQueueSize=4);
+
+        ~MPFAsyncVideoCapture();
+
+        MPFFrame Read();
+
+        void ReverseTransform(MPFVideoTrack &videoTrack) const;
+
+        /**
+         * @return An object that can do the reverse transform even after MPFAsyncVideoCapture has
+         *         been destroyed
+         */
+        ReverseTransformer GetReverseTransformer() const;
+
+        int GetFrameCount() const;
+
+        double GetFrameRate() const;
+
+        cv::Size GetFrameSize() const;
+
+        cv::Size GetOriginalFrameSize() const;
+
+
+    private:
+        BlockingQueue<MPFFrame> frameQueue_;
+
+        // Fields for properties of the video that don't change as it is being read.
+        // We can't just query the underlying video capture on the fly because it is being used by
+        // the frameReader thread. These fields get set prior to handing the video capture
+        // over to the frameReader thread.
+        int frameCount_;
+        double frameRate_;
+        cv::Size frameSize_;
+        cv::Size originalFrameSize_;
+
+        ReverseTransformer reverseTransformer_;
+
+        std::shared_future<void> doneReadingFuture_;
+
+        MPFAsyncVideoCapture(MPFVideoCapture&& videoCapture, int frameQueueSize);
+    };
+}}
+
+
+
+#endif //OPENMPF_CPP_COMPONENT_SDK_MPFASYNCVIDEOCAPTURE_H