Merge pull request #21719 from TolyaTalamanov:at/pipeline_modeling_tool-skip-frames-for-nodes

TolyaTalamanov · dmatveev · web-flow · commit 9dd8e4df7fc9 · 2022-03-25T08:22:45.000Z
[G-API] Pipeline modeling tool: Skip frames

* Add skip feature

* Refactoring

* Fix warning

* Put more comments

* Fix comments to review

* Agregate common params into structure

* Fix warning

* Clean up &amp; add test

* Add assert

* Fix warning on Mac

* Update modules/gapi/samples/pipeline_modeling_tool.cpp

Co-authored-by: Dmitry Matveev &lt;dmitry.matveev@intel.com&gt;
diff --git a/modules/gapi/samples/pipeline_modeling_tool.cpp b/modules/gapi/samples/pipeline_modeling_tool.cpp
@@ -173,6 +173,50 @@ static PLMode strToPLMode(const std::string& mode_str) {
     }
 }
 
+template <>
+CallParams read<CallParams>(const cv::FileNode& fn) {
+    auto name =
+        check_and_read<std::string>(fn, "name", "node");
+    // FIXME: Impossible to read size_t due OpenCV limitations.
+    auto call_every_nth_opt = readOpt<int>(fn["call_every_nth"]);
+    auto call_every_nth = call_every_nth_opt.value_or(1);
+    if (call_every_nth <= 0) {
+        throw std::logic_error(
+                name + " call_every_nth must be greater than zero\n"
+                "Current call_every_nth: " + std::to_string(call_every_nth));
+    }
+    return CallParams{std::move(name), static_cast<size_t>(call_every_nth)};
+}
+
+template <>
+InferParams read<InferParams>(const cv::FileNode& fn) {
+    auto name =
+        check_and_read<std::string>(fn, "name", "node");
+
+    InferParams params;
+    params.path          = read<ModelPath>(fn);
+    params.device        = check_and_read<std::string>(fn, "device", name);
+    params.input_layers  = readList<std::string>(fn, "input_layers", name);
+    params.output_layers = readList<std::string>(fn, "output_layers", name);
+
+    return params;
+}
+
+template <>
+DummyParams read<DummyParams>(const cv::FileNode& fn) {
+    auto name =
+        check_and_read<std::string>(fn, "name", "node");
+
+    DummyParams params;
+    params.time = check_and_read<double>(fn, "time", name);
+    if (params.time < 0) {
+        throw std::logic_error(name + " time must be positive");
+    }
+    params.output = check_and_read<OutputDescr>(fn, "output", name);
+
+    return params;
+}
+
 static std::vector<std::string> parseExecList(const std::string& exec_list) {
     std::vector<std::string> pl_types;
     std::stringstream ss(exec_list);
@@ -316,31 +360,17 @@ int main(int argc, char* argv[]) {
             if (!nodes_fn.isSeq()) {
                 throw std::logic_error("nodes in " + name + " must be a sequence");
             }
+
             for (auto node_fn : nodes_fn) {
-                auto node_name =
-                    check_and_read<std::string>(node_fn, "name", "node");
+                auto call_params = read<CallParams>(node_fn);
                 auto node_type =
                     check_and_read<std::string>(node_fn, "type", "node");
                 if (node_type == "Dummy") {
-                    auto time =
-                        check_and_read<double>(node_fn, "time", node_name);
-                    if (time < 0) {
-                        throw std::logic_error(node_name + " time must be positive");
-                    }
-                    auto output =
-                        check_and_read<OutputDescr>(node_fn, "output", node_name);
-                    builder.addDummy(node_name, time, output);
+                    builder.addDummy(call_params, read<DummyParams>(node_fn));
                 } else if (node_type == "Infer") {
-                    InferParams params;
-                    params.path   = read<ModelPath>(node_fn);
-                    params.device =
-                        check_and_read<std::string>(node_fn, "device", node_name);
-                    params.input_layers =
-                        readList<std::string>(node_fn, "input_layers", node_name);
-                    params.output_layers =
-                        readList<std::string>(node_fn, "output_layers", node_name);
-                    params.config = config;
-                    builder.addInfer(node_name, params);
+                    auto infer_params = read<InferParams>(node_fn);
+                    infer_params.config = config;
+                    builder.addInfer(call_params, infer_params);
                 } else {
                     throw std::logic_error("Unsupported node type: " + node_type);
                 }
diff --git a/modules/gapi/samples/pipeline_modeling_tool/pipeline_builder.hpp b/modules/gapi/samples/pipeline_modeling_tool/pipeline_builder.hpp
@@ -23,11 +23,16 @@ struct Edge {
     P dst;
 };
 
+struct CallParams {
+    std::string name;
+    size_t      call_every_nth;
+};
+
 struct CallNode {
     using F = std::function<void(const cv::GProtoArgs&, cv::GProtoArgs&)>;
 
-    std::string name;
-    F           run;
+    CallParams params;
+    F          run;
 };
 
 struct DataNode {
@@ -44,6 +49,80 @@ struct Node {
     Kind kind;
 };
 
+struct SubGraphCall {
+    G_API_OP(GSubGraph,
+             <cv::GMat(cv::GMat, cv::GComputation, cv::GCompileArgs, size_t)>,
+             "custom.subgraph") {
+        static cv::GMatDesc outMeta(const cv::GMatDesc& in,
+                                    cv::GComputation    comp,
+                                    cv::GCompileArgs    compile_args,
+                                    const size_t        call_every_nth) {
+            GAPI_Assert(call_every_nth > 0);
+            auto out_metas =
+                comp.compile(in, std::move(compile_args)).outMetas();
+            GAPI_Assert(out_metas.size() == 1u);
+            GAPI_Assert(cv::util::holds_alternative<cv::GMatDesc>(out_metas[0]));
+            return cv::util::get<cv::GMatDesc>(out_metas[0]);
+        }
+
+    };
+
+    struct SubGraphState {
+        cv::Mat       last_result;
+        cv::GCompiled cc;
+        int           call_counter = 0;
+    };
+
+    GAPI_OCV_KERNEL_ST(SubGraphImpl, GSubGraph, SubGraphState) {
+            static void setup(const cv::GMatDesc&             in,
+                              cv::GComputation                comp,
+                              cv::GCompileArgs                compile_args,
+                              const size_t                    /*call_every_nth*/,
+                              std::shared_ptr<SubGraphState>& state,
+                              const cv::GCompileArgs&         /*args*/) {
+                state.reset(new SubGraphState{});
+                state->cc = comp.compile(in, std::move(compile_args));
+                auto out_desc =
+                    cv::util::get<cv::GMatDesc>(state->cc.outMetas()[0]);
+                utils::createNDMat(state->last_result,
+                                   out_desc.dims,
+                                   out_desc.depth);
+            }
+
+            static void run(const cv::Mat&   in,
+                            cv::GComputation /*comp*/,
+                            cv::GCompileArgs /*compile_args*/,
+                            const size_t     call_every_nth,
+                            cv::Mat&         out,
+                            SubGraphState&   state) {
+                // NB: Make a call on the first iteration and skip the furthers.
+                if (state.call_counter == 0) {
+                    state.cc(in, state.last_result);
+                }
+                state.last_result.copyTo(out);
+                state.call_counter = (state.call_counter + 1) % call_every_nth;
+            }
+    };
+
+    void operator()(const cv::GProtoArgs& inputs, cv::GProtoArgs& outputs);
+
+    size_t numInputs()  const { return 1; }
+    size_t numOutputs() const { return 1; }
+
+    cv::GComputation comp;
+    cv::GCompileArgs compile_args;
+    size_t           call_every_nth;
+};
+
+void SubGraphCall::operator()(const cv::GProtoArgs& inputs,
+                                    cv::GProtoArgs& outputs) {
+    GAPI_Assert(inputs.size() == 1u);
+    GAPI_Assert(cv::util::holds_alternative<cv::GMat>(inputs[0]));
+    GAPI_Assert(outputs.empty());
+    auto in = cv::util::get<cv::GMat>(inputs[0]);
+    outputs.emplace_back(GSubGraph::on(in, comp, compile_args, call_every_nth));
+}
+
 struct DummyCall {
     G_API_OP(GDummy,
              <cv::GMat(cv::GMat, double, OutputDescr)>,
@@ -166,6 +245,11 @@ struct ImportPath {
 
 using ModelPath = cv::util::variant<ImportPath, LoadPath>;
 
+struct DummyParams {
+    double      time;
+    OutputDescr output;
+};
+
 struct InferParams {
     std::string name;
     ModelPath   path;
@@ -178,11 +262,11 @@ struct InferParams {
 class PipelineBuilder {
 public:
     PipelineBuilder();
-    void addDummy(const std::string& name,
-                  const double       time,
-                  const OutputDescr& output);
+    void addDummy(const CallParams&  call_params,
+                  const DummyParams& dummy_params);
 
-    void addInfer(const std::string& name, const InferParams& params);
+    void addInfer(const CallParams&  call_params,
+                  const InferParams& infer_params);
 
     void setSource(const std::string&           name,
                    std::shared_ptr<DummySource> src);
@@ -197,8 +281,8 @@ class PipelineBuilder {
 
 private:
     template <typename CallT>
-    void addCall(const std::string& name,
-                 CallT&&            call);
+    void addCall(const CallParams& call_params,
+                 CallT&&           call);
 
     Pipeline::Ptr construct();
 
@@ -226,20 +310,21 @@ class PipelineBuilder {
 
 PipelineBuilder::PipelineBuilder() : m_state(new State{}) { };
 
-void PipelineBuilder::addDummy(const std::string&  name,
-                               const double        time,
-                               const OutputDescr&  output) {
+void PipelineBuilder::addDummy(const CallParams&  call_params,
+                               const DummyParams& dummy_params) {
     m_state->kernels.include<DummyCall::GCPUDummy>();
-    addCall(name, DummyCall{time, output});
+    addCall(call_params,
+            DummyCall{dummy_params.time, dummy_params.output});
 }
 
 template <typename CallT>
-void PipelineBuilder::addCall(const std::string& name,
-                              CallT&&            call) {
+void PipelineBuilder::addCall(const CallParams& call_params,
+                              CallT&&           call) {
 
     size_t num_inputs  = call.numInputs();
     size_t num_outputs = call.numOutputs();
-    Node::Ptr call_node(new Node{{},{},Node::Kind{CallNode{name, std::move(call)}}});
+    Node::Ptr call_node(new Node{{},{},Node::Kind{CallNode{call_params,
+                                                           std::move(call)}}});
     // NB: Create placeholders for inputs.
     call_node->in_nodes.resize(num_inputs);
     // NB: Create outputs with empty data.
@@ -249,36 +334,39 @@ void PipelineBuilder::addCall(const std::string& name,
                                                    Node::Kind{DataNode{}}});
     }
 
-    auto it = m_state->calls_map.find(name);
+    auto it = m_state->calls_map.find(call_params.name);
     if (it != m_state->calls_map.end()) {
-        throw std::logic_error("Node: " + name + " already exists!");
+        throw std::logic_error("Node: " + call_params.name + " already exists!");
     }
-    m_state->calls_map.emplace(name, call_node);
+    m_state->calls_map.emplace(call_params.name, call_node);
     m_state->all_calls.emplace_back(call_node);
 }
 
-void PipelineBuilder::addInfer(const std::string& name,
-                               const InferParams& params) {
+void PipelineBuilder::addInfer(const CallParams&  call_params,
+                               const InferParams& infer_params) {
     // NB: No default ctor for Params.
     std::unique_ptr<cv::gapi::ie::Params<cv::gapi::Generic>> pp;
-    if (cv::util::holds_alternative<LoadPath>(params.path)) {
-       auto load_path = cv::util::get<LoadPath>(params.path);
-       pp.reset(new cv::gapi::ie::Params<cv::gapi::Generic>(name,
+    if (cv::util::holds_alternative<LoadPath>(infer_params.path)) {
+       auto load_path = cv::util::get<LoadPath>(infer_params.path);
+       pp.reset(new cv::gapi::ie::Params<cv::gapi::Generic>(call_params.name,
                                                             load_path.xml,
                                                             load_path.bin,
-                                                            params.device));
+                                                            infer_params.device));
     } else {
-        GAPI_Assert(cv::util::holds_alternative<ImportPath>(params.path));
-        auto import_path = cv::util::get<ImportPath>(params.path);
-        pp.reset(new cv::gapi::ie::Params<cv::gapi::Generic>(name,
+        GAPI_Assert(cv::util::holds_alternative<ImportPath>(infer_params.path));
+        auto import_path = cv::util::get<ImportPath>(infer_params.path);
+        pp.reset(new cv::gapi::ie::Params<cv::gapi::Generic>(call_params.name,
                                                              import_path.blob,
-                                                             params.device));
+                                                             infer_params.device));
     }
 
-    pp->pluginConfig(params.config);
+    pp->pluginConfig(infer_params.config);
     m_state->networks += cv::gapi::networks(*pp);
 
-    addCall(name, InferCall{name, params.input_layers, params.output_layers});
+    addCall(call_params,
+            InferCall{call_params.name,
+                      infer_params.input_layers,
+                      infer_params.output_layers});
 }
 
 void PipelineBuilder::addEdge(const Edge& edge) {
@@ -318,7 +406,7 @@ void PipelineBuilder::setSource(const std::string&           name,
                                 std::shared_ptr<DummySource> src) {
     GAPI_Assert(!m_state->src && "Only single source pipelines are supported!");
     m_state->src = src;
-    addCall(name, SourceCall{});
+    addCall(CallParams{name, 1u/*call_every_nth*/}, SourceCall{});
 }
 
 void PipelineBuilder::setMode(PLMode mode) {
@@ -405,7 +493,7 @@ Pipeline::Ptr PipelineBuilder::construct() {
             if (in_data_node.expired()) {
                 const auto& call = cv::util::get<CallNode>(call_node->kind);
                 throw std::logic_error(
-                        "Node: " + call.name + " in Pipeline: " + m_state->name +
+                        "Node: " + call.params.name + " in Pipeline: " + m_state->name +
                         " has dangling input by in port: " + std::to_string(i));
             }
         }
@@ -424,8 +512,14 @@ Pipeline::Ptr PipelineBuilder::construct() {
             sorted_calls.push_back(n);
         }
     }
+
+    m_state->kernels.include<SubGraphCall::SubGraphImpl>();
+    m_state->compile_args.emplace_back(m_state->networks);
+    m_state->compile_args.emplace_back(m_state->kernels);
+
     // (2). Go through every call node.
     for (auto call_node : sorted_calls) {
+        auto& call = cv::util::get<CallNode>(call_node->kind);
         cv::GProtoArgs outputs;
         cv::GProtoArgs inputs;
         for (size_t i = 0; i < call_node->in_nodes.size(); ++i) {
@@ -437,8 +531,37 @@ Pipeline::Ptr PipelineBuilder::construct() {
             // (3). Extract proto input from every input node.
             inputs.push_back(in_data.arg.value());
         }
+        // NB: If node shouldn't be called on each iterations,
+        // it should be wrapped into subgraph which is able to skip calling.
+        if (call.params.call_every_nth != 1u) {
+            // FIXME: Limitation of the subgraph operation (<GMat(GMat)>).
+            // G-API doesn't support dynamic number of inputs/outputs.
+            if (inputs.size() > 1u) {
+                throw std::logic_error(
+                        "skip_frame_nth is supported only for single input subgraphs\n"
+                        "Current subgraph has " + std::to_string(inputs.size()) + " inputs");
+            }
+
+            if (outputs.size() > 1u) {
+                throw std::logic_error(
+                        "skip_frame_nth is supported only for single output subgraphs\n"
+                        "Current subgraph has " + std::to_string(inputs.size()) + " outputs");
+            }
+            // FIXME: Should be generalized.
+            // Now every subgraph contains only single node
+            // which has single input/output.
+            GAPI_Assert(cv::util::holds_alternative<cv::GMat>(inputs[0]));
+            cv::GProtoArgs subgr_inputs{cv::GProtoArg{cv::GMat()}};
+            cv::GProtoArgs subgr_outputs;
+            call.run(subgr_inputs, subgr_outputs);
+            auto comp = cv::GComputation(cv::GProtoInputArgs{subgr_inputs},
+                                         cv::GProtoOutputArgs{subgr_outputs});
+            call = CallNode{CallParams{call.params.name, 1u/*call_every_nth*/},
+                            SubGraphCall{std::move(comp),
+                                         m_state->compile_args,
+                                         call.params.call_every_nth}};
+        }
         // (4). Run call and get outputs.
-        auto call = cv::util::get<CallNode>(call_node->kind);
         call.run(inputs, outputs);
         // (5) If call node doesn't have inputs
         // it means that it's input producer node (Source).
@@ -460,9 +583,6 @@ Pipeline::Ptr PipelineBuilder::construct() {
         }
     }
 
-    m_state->compile_args.emplace_back(m_state->networks);
-    m_state->compile_args.emplace_back(m_state->kernels);
-
     if (m_state->mode == PLMode::STREAMING) {
         GAPI_Assert(graph_inputs.size() == 1);
         GAPI_Assert(cv::util::holds_alternative<cv::GMat>(graph_inputs[0]));
diff --git a/modules/gapi/samples/pipeline_modeling_tool/test_pipeline_modeling_tool.py b/modules/gapi/samples/pipeline_modeling_tool/test_pipeline_modeling_tool.py
