Merge pull request #22037 from xiong-jie-y:py_gapi_add_state_kernel

Add stateful kernel to python G-API

modules/gapi/include/opencv2/gapi/python/python.hpp

@@ -31,19 +31,22 @@ struct GPythonContext
     const cv::GArgs      &ins;
     const cv::GMetaArgs  &in_metas;
     const cv::GTypesInfo &out_info;
+
+    cv::optional<cv::GArg> m_state;
 };
 
 using Impl = std::function<cv::GRunArgs(const GPythonContext&)>;
+using Setup = std::function<cv::GArg(const GMetaArgs&, const GArgs&)>;
 
 class GAPI_EXPORTS GPythonKernel
 {
 public:
     GPythonKernel() = default;
-    GPythonKernel(Impl run);
+    GPythonKernel(Impl run, Setup setup);
 
-    cv::GRunArgs operator()(const GPythonContext& ctx);
-private:
-    Impl m_run;
+    Impl  run;
+    Setup setup       = nullptr;
+    bool  is_stateful = false;
 };
 
 class GAPI_EXPORTS GPythonFunctor : public cv::gapi::GFunctor
@@ -51,7 +54,8 @@ class GAPI_EXPORTS GPythonFunctor : public cv::gapi::GFunctor
 public:
     using Meta = cv::GKernel::M;
 
-    GPythonFunctor(const char* id, const Meta &meta, const Impl& impl);
+    GPythonFunctor(const char* id, const Meta& meta, const Impl& impl,
+                   const Setup& setup = nullptr);
 
     GKernelImpl    impl()    const override;
     gapi::GBackend backend() const override;

modules/gapi/misc/python/pyopencv_gapi.hpp

@@ -660,7 +660,8 @@ static cv::GRunArgs run_py_kernel(cv::detail::PyObjectHolder kernel,
         // NB: Doesn't increase reference counter (false),
         // because PyObject already have ownership.
         // In case exception decrement reference counter.
-        cv::detail::PyObjectHolder args(PyTuple_New(ins.size()), false);
+        cv::detail::PyObjectHolder args(
+                PyTuple_New(ctx.m_state.has_value() ? ins.size() + 1 : ins.size()), false);
         for (size_t i = 0; i < ins.size(); ++i)
         {
             // NB: If meta is monostate then object isn't associated with G-TYPE.
@@ -690,6 +691,12 @@ static cv::GRunArgs run_py_kernel(cv::detail::PyObjectHolder kernel,
             }
             ++in_idx;
         }
+
+        if (ctx.m_state.has_value())
+        {
+            PyTuple_SetItem(args.get(), ins.size(), pyopencv_from(ctx.m_state.value()));
+        }
+
         // NB: Doesn't increase reference counter (false).
         // In case PyObject_CallObject return NULL, do nothing in destructor.
         cv::detail::PyObjectHolder result(
@@ -736,6 +743,86 @@ static cv::GRunArgs run_py_kernel(cv::detail::PyObjectHolder kernel,
     return outs;
 }
 
+static void unpackMetasToTuple(const cv::GMetaArgs&        meta,
+                               const cv::GArgs&            gargs,
+                               cv::detail::PyObjectHolder& tuple)
+{
+    size_t idx = 0;
+    for (auto&& m : meta)
+    {
+        switch (m.index())
+        {
+            case cv::GMetaArg::index_of<cv::GMatDesc>():
+                PyTuple_SetItem(tuple.get(), idx, pyopencv_from(cv::util::get<cv::GMatDesc>(m)));
+                break;
+            case cv::GMetaArg::index_of<cv::GScalarDesc>():
+                PyTuple_SetItem(tuple.get(), idx,
+                        pyopencv_from(cv::util::get<cv::GScalarDesc>(m)));
+                break;
+            case cv::GMetaArg::index_of<cv::GArrayDesc>():
+                PyTuple_SetItem(tuple.get(), idx,
+                        pyopencv_from(cv::util::get<cv::GArrayDesc>(m)));
+                break;
+            case cv::GMetaArg::index_of<cv::GOpaqueDesc>():
+                PyTuple_SetItem(tuple.get(), idx,
+                        pyopencv_from(cv::util::get<cv::GOpaqueDesc>(m)));
+                break;
+            case cv::GMetaArg::index_of<cv::util::monostate>():
+                PyTuple_SetItem(tuple.get(), idx, pyopencv_from(gargs[idx]));
+                break;
+            case cv::GMetaArg::index_of<cv::GFrameDesc>():
+                util::throw_error(
+                        std::logic_error("GFrame isn't supported for custom operation"));
+                break;
+        }
+        ++idx;
+    }
+}
+
+static cv::GArg setup_py(cv::detail::PyObjectHolder setup,
+                         const cv::GMetaArgs&       meta,
+                         const cv::GArgs&           gargs)
+{
+    PyGILState_STATE gstate;
+    gstate = PyGILState_Ensure();
+
+    cv::GArg out;
+
+    try
+    {
+        // NB: Doesn't increase reference counter (false),
+        // because PyObject already have ownership.
+        // In case exception decrement reference counter.
+        cv::detail::PyObjectHolder args(PyTuple_New(meta.size()), false);
+        unpackMetasToTuple(meta, gargs, args);
+        // NB: Take an onwership because this state is "Python" type so it will be wrapped as-is
+        // into cv::GArg and stored in GPythonBackend. Object without ownership can't
+        // be dealocated outside this function.
+        cv::detail::PyObjectHolder result(PyObject_CallObject(setup.get(), args.get()), true);
+
+        if (PyErr_Occurred())
+        {
+            PyErr_PrintEx(0);
+            PyErr_Clear();
+            throw std::logic_error("Python kernel failed with error!");
+        }
+        // NB: In fact it's impossible situation, because errors were handled above.
+        GAPI_Assert(result.get() && "Python kernel returned NULL!");
+
+        if (!pyopencv_to(result.get(), out, ArgInfo("arg", false)))
+        {
+            util::throw_error(std::logic_error("Unsupported output meta type"));
+        }
+    }
+    catch (...)
+    {
+        PyGILState_Release(gstate);
+        throw;
+    }
+    PyGILState_Release(gstate);
+    return out;
+}
+
 static GMetaArg get_meta_arg(PyObject* obj)
 {
     cv::GMetaArg arg;
@@ -774,32 +861,7 @@ static GMetaArgs run_py_meta(cv::detail::PyObjectHolder out_meta,
         // because PyObject already have ownership.
         // In case exception decrement reference counter.
         cv::detail::PyObjectHolder args(PyTuple_New(meta.size()), false);
-        size_t idx = 0;
-        for (auto&& m : meta)
-        {
-            switch (m.index())
-            {
-                case cv::GMetaArg::index_of<cv::GMatDesc>():
-                    PyTuple_SetItem(args.get(), idx, pyopencv_from(cv::util::get<cv::GMatDesc>(m)));
-                    break;
-                case cv::GMetaArg::index_of<cv::GScalarDesc>():
-                    PyTuple_SetItem(args.get(), idx, pyopencv_from(cv::util::get<cv::GScalarDesc>(m)));
-                    break;
-                case cv::GMetaArg::index_of<cv::GArrayDesc>():
-                    PyTuple_SetItem(args.get(), idx, pyopencv_from(cv::util::get<cv::GArrayDesc>(m)));
-                    break;
-                case cv::GMetaArg::index_of<cv::GOpaqueDesc>():
-                    PyTuple_SetItem(args.get(), idx, pyopencv_from(cv::util::get<cv::GOpaqueDesc>(m)));
-                    break;
-                case cv::GMetaArg::index_of<cv::util::monostate>():
-                    PyTuple_SetItem(args.get(), idx, pyopencv_from(gargs[idx]));
-                    break;
-                case cv::GMetaArg::index_of<cv::GFrameDesc>():
-                    util::throw_error(std::logic_error("GFrame isn't supported for custom operation"));
-                    break;
-            }
-            ++idx;
-        }
+        unpackMetasToTuple(meta, gargs, args);
         // NB: Doesn't increase reference counter (false).
         // In case PyObject_CallObject return NULL, do nothing in destructor.
         cv::detail::PyObjectHolder result(
@@ -860,6 +922,10 @@ static PyObject* pyopencv_cv_gapi_kernels(PyObject* , PyObject* py_args, PyObjec
                     "Python kernel should contain run, please use cv.gapi.kernel to define kernel");
             return NULL;
         }
+        PyObject* setup = nullptr;
+        if (PyObject_HasAttrString(user_kernel, "setup")) {
+            setup  = PyObject_GetAttrString(user_kernel, "setup");
+        }
 
         std::string id;
         if (!pyopencv_to(id_obj, id, ArgInfo("id", false)))
@@ -869,11 +935,23 @@ static PyObject* pyopencv_cv_gapi_kernels(PyObject* , PyObject* py_args, PyObjec
         }
 
         using namespace std::placeholders;
-        gapi::python::GPythonFunctor f(id.c_str(),
-                std::bind(run_py_meta  , cv::detail::PyObjectHolder{out_meta}, _1, _2),
-                std::bind(run_py_kernel, cv::detail::PyObjectHolder{run}    , _1));
+
+        if (setup)
+        {
+            gapi::python::GPythonFunctor f(
+                id.c_str(), std::bind(run_py_meta, cv::detail::PyObjectHolder{out_meta}, _1, _2),
+                std::bind(run_py_kernel, cv::detail::PyObjectHolder{run}, _1),
+                std::bind(setup_py, cv::detail::PyObjectHolder{setup}, _1, _2));
             pkg.include(f);
         }
+        else
+        {
+            gapi::python::GPythonFunctor f(
+                id.c_str(), std::bind(run_py_meta, cv::detail::PyObjectHolder{out_meta}, _1, _2),
+                std::bind(run_py_kernel, cv::detail::PyObjectHolder{run}, _1));
+            pkg.include(f);
+        }
+    }
     return pyopencv_from(pkg);
 }
 

modules/gapi/misc/python/test/test_gapi_sample_pipelines.py

@@ -432,7 +432,7 @@ try:
             with self.assertRaises(Exception): create_op([cv.GMat, int], [cv.GMat]).on(cv.GMat())
 
 
-        def test_stateful_kernel(self):
+        def test_state_in_class(self):
             @cv.gapi.op('custom.sum', in_types=[cv.GArray.Int], out_types=[cv.GOpaque.Int])
             class GSum:
                 @staticmethod

modules/gapi/misc/python/test/test_gapi_stateful_kernel.py

@@ -0,0 +1,142 @@
+#!/usr/bin/env python
+
+import numpy as np
+import cv2 as cv
+import os
+import sys
+import unittest
+
+from tests_common import NewOpenCVTests
+
+
+try:
+
+    if sys.version_info[:2] < (3, 0):
+        raise unittest.SkipTest('Python 2.x is not supported')
+
+
+    class CounterState:
+        def __init__(self):
+            self.counter = 0
+
+
+    @cv.gapi.op('stateful_counter',
+                in_types=[cv.GOpaque.Int],
+                out_types=[cv.GOpaque.Int])
+    class GStatefulCounter:
+        """Accumulate state counter on every call"""
+
+        @staticmethod
+        def outMeta(desc):
+            return cv.empty_gopaque_desc()
+
+
+    @cv.gapi.kernel(GStatefulCounter)
+    class GStatefulCounterImpl:
+        """Implementation for GStatefulCounter operation."""
+
+        @staticmethod
+        def setup(desc):
+            return CounterState()
+
+        @staticmethod
+        def run(value, state):
+            state.counter += value
+            return state.counter
+
+
+    class gapi_sample_pipelines(NewOpenCVTests):
+        def test_stateful_kernel_single_instance(self):
+            g_in  = cv.GOpaque.Int()
+            g_out = GStatefulCounter.on(g_in)
+            comp  = cv.GComputation(cv.GIn(g_in), cv.GOut(g_out))
+            pkg   = cv.gapi.kernels(GStatefulCounterImpl)
+
+            nums = [i for i in range(10)]
+            acc = 0
+            for v in nums:
+                acc = comp.apply(cv.gin(v), args=cv.gapi.compile_args(pkg))
+
+            self.assertEqual(sum(nums), acc)
+
+
+        def test_stateful_kernel_multiple_instances(self):
+            # NB: Every counter has his own independent state.
+            g_in   = cv.GOpaque.Int()
+            g_out0 = GStatefulCounter.on(g_in)
+            g_out1 = GStatefulCounter.on(g_in)
+            comp   = cv.GComputation(cv.GIn(g_in), cv.GOut(g_out0, g_out1))
+            pkg    = cv.gapi.kernels(GStatefulCounterImpl)
+
+            nums = [i for i in range(10)]
+            acc0 = acc1 = 0
+            for v in nums:
+                acc0, acc1 = comp.apply(cv.gin(v), args=cv.gapi.compile_args(pkg))
+
+            ref = sum(nums)
+            self.assertEqual(ref, acc0)
+            self.assertEqual(ref, acc1)
+
+
+        def test_stateful_throw_setup(self):
+            @cv.gapi.kernel(GStatefulCounter)
+            class GThrowStatefulCounterImpl:
+                """Implementation for GStatefulCounter operation
+                   that throw exception in setup method"""
+
+                @staticmethod
+                def setup(desc):
+                    raise Exception('Throw from setup method')
+
+                @staticmethod
+                def run(value, state):
+                    raise Exception('Unreachable')
+
+            g_in  = cv.GOpaque.Int()
+            g_out = GStatefulCounter.on(g_in)
+            comp  = cv.GComputation(cv.GIn(g_in), cv.GOut(g_out))
+            pkg   = cv.gapi.kernels(GThrowStatefulCounterImpl)
+
+            with self.assertRaises(Exception): comp.apply(cv.gin(42),
+                                                          args=cv.gapi.compile_args(pkg))
+
+
+        def test_stateful_reset(self):
+            g_in  = cv.GOpaque.Int()
+            g_out = GStatefulCounter.on(g_in)
+            comp  = cv.GComputation(cv.GIn(g_in), cv.GOut(g_out))
+            pkg   = cv.gapi.kernels(GStatefulCounterImpl)
+
+            cc = comp.compileStreaming(args=cv.gapi.compile_args(pkg))
+
+            cc.setSource(cv.gin(1))
+            cc.start()
+            for i in range(1, 10):
+                _, actual = cc.pull()
+                self.assertEqual(i, actual)
+            cc.stop()
+
+            cc.setSource(cv.gin(2))
+            cc.start()
+            for i in range(2, 10, 2):
+                _, actual = cc.pull()
+                self.assertEqual(i, actual)
+            cc.stop()
+
+
+except unittest.SkipTest as e:
+
+    message = str(e)
+
+    class TestSkip(unittest.TestCase):
+        def setUp(self):
+            self.skipTest('Skip tests: ' + message)
+
+        def test_skip():
+            pass
+
+    pass
+
+
+if __name__ == '__main__':
+    NewOpenCVTests.bootstrap()

modules/gapi/src/backends/python/gpythonbackend.cpp

@@ -6,26 +6,25 @@
 
 #include <ade/util/zip_range.hpp> // zip_range, indexed
 
+#include "compiler/gmodel.hpp"
+#include <opencv2/gapi/garg.hpp>
 #include <opencv2/gapi/util/throw.hpp> // throw_error
 #include <opencv2/gapi/python/python.hpp>
 
 #include "api/gbackend_priv.hpp"
 #include "backends/common/gbackend.hpp"
 
-cv::gapi::python::GPythonKernel::GPythonKernel(cv::gapi::python::Impl run)
-    : m_run(run)
+cv::gapi::python::GPythonKernel::GPythonKernel(cv::gapi::python::Impl  runf,
+                                               cv::gapi::python::Setup setupf)
+    : run(runf), setup(setupf), is_stateful(setup != nullptr)
 {
 }
 
-cv::GRunArgs cv::gapi::python::GPythonKernel::operator()(const cv::gapi::python::GPythonContext& ctx)
-{
-    return m_run(ctx);
-}
-
 cv::gapi::python::GPythonFunctor::GPythonFunctor(const char* id,
-                                                 const cv::gapi::python::GPythonFunctor::Meta &meta,
-                                                 const cv::gapi::python::Impl& impl)
-    : gapi::GFunctor(id), impl_{GPythonKernel{impl}, meta}
+                                                 const cv::gapi::python::GPythonFunctor::Meta& meta,
+                                                 const cv::gapi::python::Impl& impl,
+                                                 const cv::gapi::python::Setup& setup)
+    : gapi::GFunctor(id), impl_{GPythonKernel{impl, setup}, meta}
 {
 }
 
@@ -68,6 +67,7 @@ class GPythonExecutable final: public cv::gimpl::GIslandExecutable
     virtual cv::RMat allocate(const cv::GMatDesc&) const override { return {}; }
 
     virtual bool canReshape() const override { return true; }
+    virtual void handleNewStream() override;
     virtual void reshape(ade::Graph&, const cv::GCompileArgs&) override {
         // Do nothing here
     }
@@ -80,6 +80,7 @@ public:
     cv::gimpl::GModel::ConstGraph m_gm;
     cv::gapi::python::GPythonKernel m_kernel;
     ade::NodeHandle m_op;
+    cv::GArg m_node_state;
 
     cv::GTypesInfo m_out_info;
     cv::GMetaArgs  m_in_metas;
@@ -153,6 +154,15 @@ static void writeBack(cv::GRunArg& arg, cv::GRunArgP& out)
     }
 }
 
+void GPythonExecutable::handleNewStream()
+{
+    if (!m_kernel.is_stateful)
+        return;
+
+    m_node_state = m_kernel.setup(cv::gimpl::GModel::collectInputMeta(m_gm, m_op),
+                                  m_gm.metadata(m_op).get<cv::gimpl::Op>().args);
+}
+
 void GPythonExecutable::run(std::vector<InObj>  &&input_objs,
                             std::vector<OutObj> &&output_objs)
 {
@@ -165,9 +175,15 @@ void GPythonExecutable::run(std::vector<InObj>  &&input_objs,
                          std::back_inserter(inputs),
                          std::bind(&packArg, std::ref(m_res), _1));
 
+    cv::gapi::python::GPythonContext ctx{inputs, m_in_metas, m_out_info, /*state*/{}};
 
-    cv::gapi::python::GPythonContext ctx{inputs, m_in_metas, m_out_info};
-    auto outs = m_kernel(ctx);
+    // NB: For stateful kernel add state to its execution context
+    if (m_kernel.is_stateful)
+    {
+        ctx.m_state = cv::optional<cv::GArg>(m_node_state);
+    }
+
+    auto outs = m_kernel.run(ctx);
 
     for (auto&& it : ade::util::zip(outs, output_objs))
     {
@@ -225,6 +241,12 @@ GPythonExecutable::GPythonExecutable(const ade::Graph& g,
     m_op = *it;
     m_kernel = cag.metadata(m_op).get<PythonUnit>().kernel;
 
+    // If kernel is stateful then prepare storage for its state.
+    if (m_kernel.is_stateful)
+    {
+        m_node_state = cv::GArg{ };
+    }
+
     // Ensure this the only op in the graph
     if (std::any_of(it+1, nodes.end(), is_op))
     {