1
0
mirror of https://github.com/opencv/opencv.git synced 2026-07-29 15:23:05 +04:00

feat: G-API: Custom stream sources in Python (#27276)

Implemented:

- A C++ proxy class PythonCustomStreamSource that implements the
  IStreamSource interface. This class acts as a bridge between
  G-API’s internal streaming engine and user-defined Python
  objects. Internally, it stores a reference to a Python object
  (PyObject*) and is responsible for: calling the Python object’s
  pull() method to retrieve the next frame, calling the descr_of()
  method to obtain the frame format description, acquiring and
  releasing the Python GIL as needed, converting the returned
  numpy.ndarray into cv::Mat, and handling any exceptions or
  conversion errors with proper diagnostics.

- A Python-facing factory function, cv.gapi.wip.make_py_src(),
  which takes a Python object as an argument and wraps it into a
  cv::Ptr<IStreamSource>. Internally, this function constructs a
  PythonCustomStreamSource instance and passes the Python object to
  it. This design allows Python users to define any class that
  implements two methods: pull() and descr_of(). No subclassing or
  special decorators are required on the Python side. The user
  simply needs to implement the expected interface.

Co-authored-by: Leonor Francisco <leonor.francisco@tecnico.ulisboa.pt>
This commit is contained in:
Rita Melo
2025-06-05 18:14:56 +01:00
parent 868fc5c581
commit 439b6af379
5 changed files with 272 additions and 0 deletions
@@ -539,7 +539,38 @@ try:
self.assertEqual(0.0, cv.norm(convertNV12p2BGR(expected1), actual1, cv.NORM_INF))
self.assertEqual(0.0, cv.norm(convertNV12p2BGR(expected2), actual2, cv.NORM_INF))
def test_python_custom_stream_source(self):
class MySource:
def __init__(self):
self.count = 0
def pull(self):
if self.count >= 3:
return None
self.count += 1
return np.ones((10, 10, 3), np.uint8) * self.count
def descr_of(self):
return np.zeros((10, 10, 3), np.uint8)
g_in = cv.GMat()
g_out = cv.gapi.copy(g_in)
c = cv.GComputation(g_in, g_out)
comp = c.compileStreaming()
src = cv.gapi.wip.make_py_src(MySource())
comp.setSource([src])
comp.start()
frames = []
while True:
has_frame, frame = comp.pull()
if not has_frame:
break
frames.append(frame)
self.assertEqual(len(frames), 3)
except unittest.SkipTest as e: