Capture interpipe with openCV

[tomklar]

Hello RedgeRun Team,

i try to capture interpipes with openCV in Python. In the following example code you can see how i try to get this work. I use videotestsrc. To check if interpipes are working the videotestsrc first is displayed for 5 seconds in a xvimagesink. After that i try to capture the appsink of a interpipe in cv.VideoCapture().

``` Example Code 1 ```

```python #!/usr/bin/env python3 import cv2 as cv import numpy as np import time from pygstc.gstc import * # Create PipelineEntity object to manage each pipeline class PipelineEntity(object): def __init__(self, client, name, description): self._name = name self._description = description self._client = client print("Creating pipeline: " + self._name) self._client.pipeline_create(self._name, self._description) def play(self): print("Playing pipeline: " + self._name) self._client.pipeline_play(self._name) def stop(self): print("Stopping pipeline: " + self._name) self._client.pipeline_stop(self._name) def delete(self): print("Deleting pipeline: " + self._name) self._client.pipeline_delete(self._name) def eos(self): print("Sending EOS to pipeline: " + self._name) self._client.event_eos(self._name) def set_file_location(self, location): print("Setting " + self._name + " pipeline recording/snapshot location to " + location); filesink_name = "filesink_" + self._name; self._client.element_set(self._name, filesink_name, 'location', location); def listen_to(self, sink): print(self._name + " pipeline listening to " + sink); self._client.element_set(self._name, self._name + '_src', 'listen-to', sink); pipelines_base = [] pipelines_video_rec = [] pipelines_video_enc = [] pipelines_snap = [] # Create GstD Python client client = GstdClient() camera1 = PipelineEntity(client, 'camera1', 'videotestsrc ! queue ! interpipesink name=camera1 sync=true') pipelines_base.append(camera1) camera1_empty_pipe = PipelineEntity(client, 'camera1_empty_pipe', 'interpipesrc listen-to=camera1 ! videoconvert ! interpipesink name=camera1_empty_pipe') pipelines_base.append(camera1_empty_pipe) # Create display pipeline display_pipe = PipelineEntity(client, 'display_pipe', 'interpipesrc listen-to="camera1_empty_pipe" ! xvimagesink') pipelines_base.append(display_pipe) #Create appsink pipeline appsink_pipe = PipelineEntity(client, 'appsink_pipe', 'interpipesrc listen-to="camera1_empty_pipe" ! appsink') # Play base pipelines # appsink_pipe is not started here with base pipelines for pipeline in pipelines_base: pipeline.play() print("show display_pipe for 5 seconds in xvimagsink") time.sleep(5) print("go on and try to show appsink_pipe with openCV") print("") #try to start appsink_pipe without PipelineEntity class play = client.pipeline_play('appsink_pipe') def preprocess(play): print("start capturing interpipe with cv.VideoCapture ") print(" ") # init video capture with appsink_pipe vod = cv.VideoCapture(play, cv.CAP_GSTREAMER) # read the first frame ret, frame = vod.read() # as long as frames are successfully read while ret: cv.imshow('Display Interpipe appsink', frame) k = cv.waitKey(1) # user Esc if k == 27: break # continue to next frame ret, frame = vod.read() # release the capture & destroy all windows vod.release() cv.destroyAllWindows() pipelines_base.append(appsink_pipe) for pipeline in pipelines_base: pipeline.stop() for pipeline in pipelines_base: pipeline.delete() if __name__ == '__main__': preprocess(play) ```

Try to run this results the following errors

``` Console output errors for "Example Code 1" ```

``` Creating pipeline: camera1 Creating pipeline: camera1_empty_pipe Creating pipeline: display_pipe Creating pipeline: appsink_pipe Playing pipeline: camera1 Playing pipeline: camera1_empty_pipe Playing pipeline: display_pipe show display_pipe for 5 seconds in xvimagsink go on and try to show appsink_pipe with openCV start capturing interpipe with cv.VideoCapture [ WARN:0] global /home/thomas/opencv/modules/videoio/src/cap_gstreamer.cpp (711) open OpenCV | GStreamer warning: Error opening bin: empty pipeline not allowed [ WARN:0] global /home/thomas/opencv/modules/videoio/src/cap_gstreamer.cpp (480) isPipelinePlaying OpenCV | GStreamer warning: GStreamer: pipeline have not been created Stopping pipeline: camera1 Stopping pipeline: camera1_empty_pipe Stopping pipeline: display_pipe Stopping pipeline: appsink_pipe Deleting pipeline: camera1 Deleting pipeline: camera1_empty_pipe Deleting pipeline: display_pipe Deleting pipeline: appsink_pipe ```

I also tried to have the pipeline started with the PipelineEntity class directly out of cv.VideoCapture()

#Create appsink pipeline
appsink_pipe = PipelineEntity(client, 'appsink_pipe', 'interpipesrc listen-to="camera1_empty_pipe" ! appsink')

# init video capture with appsink_pipe
    vod = cv.VideoCapture(appsink_pipe, cv.CAP_GSTREAMER)

``` Example Code 2 (changed) ```

```python #!/usr/bin/env python3 import cv2 as cv import numpy as np import time from pygstc.gstc import * # Create PipelineEntity object to manage each pipeline class PipelineEntity(object): def __init__(self, client, name, description): self._name = name self._description = description self._client = client print("Creating pipeline: " + self._name) self._client.pipeline_create(self._name, self._description) def play(self): print("Playing pipeline: " + self._name) self._client.pipeline_play(self._name) def stop(self): print("Stopping pipeline: " + self._name) self._client.pipeline_stop(self._name) def delete(self): print("Deleting pipeline: " + self._name) self._client.pipeline_delete(self._name) def eos(self): print("Sending EOS to pipeline: " + self._name) self._client.event_eos(self._name) def set_file_location(self, location): print("Setting " + self._name + " pipeline recording/snapshot location to " + location); filesink_name = "filesink_" + self._name; self._client.element_set(self._name, filesink_name, 'location', location); def listen_to(self, sink): print(self._name + " pipeline listening to " + sink); self._client.element_set(self._name, self._name + '_src', 'listen-to', sink); pipelines_base = [] pipelines_video_rec = [] pipelines_video_enc = [] pipelines_snap = [] # Create GstD Python client client = GstdClient() camera1 = PipelineEntity(client, 'camera1', 'videotestsrc ! queue ! interpipesink name=camera1 sync=true') pipelines_base.append(camera1) camera1_empty_pipe = PipelineEntity(client, 'camera1_empty_pipe', 'interpipesrc listen-to=camera1 ! videoconvert ! interpipesink name=camera1_empty_pipe') pipelines_base.append(camera1_empty_pipe) # Create display pipeline display_pipe = PipelineEntity(client, 'display_pipe', 'interpipesrc listen-to="camera1_empty_pipe" ! xvimagesink') pipelines_base.append(display_pipe) #Create appsink pipeline appsink_pipe = PipelineEntity(client, 'appsink_pipe', 'interpipesrc listen-to="camera1_empty_pipe" ! appsink') # Play base pipelines # appsink_pipe is not started here with base pipelines for pipeline in pipelines_base: pipeline.play() print("show display_pipe for 5 seconds in xvimagsink") time.sleep(5) print("go on and try to show appsink_pipe with openCV") print("") #try to start appsink_pipe without PipelineEntity class #play = client.pipeline_play('appsink_pipe') def preprocess(appsink_pipe): print("start capturing interpipe with cv.VideoCapture ") print(" ") # init video capture with appsink_pipe vod = cv.VideoCapture(appsink_pipe, cv.CAP_GSTREAMER) # read the first frame ret, frame = vod.read() # as long as frames are successfully read while ret: cv.imshow('Display Interpipe appsink', frame) k = cv.waitKey(1) # user Esc if k == 27: break # continue to next frame ret, frame = vod.read() # release the capture & destroy all windows vod.release() cv.destroyAllWindows() pipelines_base.append(appsink_pipe) for pipeline in pipelines_base: pipeline.stop() for pipeline in pipelines_base: pipeline.delete() if __name__ == '__main__': preprocess(appsink_pipe) ```

This also doesn't work and brings the following console output.

``` Console output errors for "Example Code 2 (changed)" ```

``` Creating pipeline: camera1 Creating pipeline: camera1_empty_pipe Creating pipeline: display_pipe Creating pipeline: appsink_pipe Playing pipeline: camera1 Playing pipeline: camera1_empty_pipe Playing pipeline: display_pipe show display_pipe for 5 seconds in xvimagsink go on and try to show appsink_pipe with openCV start capturing interpipe with cv.VideoCapture Traceback (most recent call last): File "/home/thomas/cam/gtc-2020-demo/python-example/Test5_cleanup1.py", line 115, in preprocess(appsink_pipe) File "/home/thomas/cam/gtc-2020-demo/python-example/Test5_cleanup1.py", line 84, in preprocess vod = cv.VideoCapture(appsink_pipe, cv.CAP_GSTREAMER) TypeError: an integer is required (got type PipelineEntity) ```

If someone wants to try this, you have to stop and delete the pipelines after running the changed python script. I use the following shell script to so this.

``` shell script to stop and delete pipelines ```

``` #!/bin/bash gstd-client pipeline_stop camera1 gstd-client pipeline_stop camera1_empty_pipe gstd-client pipeline_stop display_pipe gstd-client pipeline_stop appsink_pipe gstd-client pipeline_delete camera1 gstd-client pipeline_delete camera1_empty_pipe gstd-client pipeline_delete display_pipe gstd-client pipeline_delete appsink_pipe ```

This example codes are based on the gtc-2020-demo code from RidgeRun.

I hope that there is a solution for this and i am just trying it the wrong way. It would be great to get this work with your help. Last but not least, I would like to wish you all a Merry Christmas and a Happy New Year 2021!

[tomklar]

Hello,

so i was reading in the cap_gstreamer.cpp from openCV. It is not made to work with gstd commands.

Is there any solution to capture the interpipe in openCV? Maybe with TCPsink->TCPsrc or v4l2sink or Gstreamer Tee?

[drewjgray]

I would be interested in this too!

[antoine-em]

Hi,

I am trying something similar here but when I try to read the stream from the interpipe my code execution (in Python) is blocked on the cv2.VideoCapture(video_source, cv2.CAP_GSTREAMER) instruction.

I have tried the same pipeline using the a share memory pipe instead of the interpipe and it is working fine (ie. my pipeline is not blocked). But as my pipelines are provisioned using gstd and the interpipes I would rather prefer to pipe my OpenCV appsink using directly the interpipe sink.

Here is my pipeline definition

Pipeline to create the video source test with some interpipe sink

# video_source
videotestsrc ! x264enc ! interpipesink name=videotest

Then I am using with OpenCV this interpipesrc

interpipesrc listen-to=videotest ! h264parse ! avdec_h264 ! videoconvert ! appsink

Is there any configuration I have missed here or it is "just" not compatible with OpenCV ?

Thank you for your feedback

[michaelgruner]

Hello all, yes this is possible. Allow me some time to modify the script above.

[michaelgruner]

Oops, my bad, I did not notice that you were using Gstd as well. No this is not possible, since the pipelines live in Gstd (a separate process), and Interpipes need to live in the same process in order to communicate with each other.

The concept of using Interpipes in a cv.VideoCapture does work, but the pipelines would need to be created in the same process.

[antoine-em]

@michaelgruner

The concept of using Interpipes in a cv.VideoCapture does work, but the pipelines would need to be created in the same process.

Thank you very much for your quick update. I understand then I will use the share memory element for the time being.

[michaelgruner]

Having said that, here's a modified version of the script shared by @tomklar that combines Interpipes and OpenCV VideoCapture.

#!/usr/bin/env python3
import cv2 as cv
import numpy as np
import time

import gi
gi.require_version('Gst', '1.0')
from gi.repository import Gst

# Create PipelineEntity object to manage each pipeline
class PipelineEntity(object):
    def __init__(self, name, description):
        self._name = name
        self._description = description
        print("Creating pipeline: " + self._name)
        self._pipe = Gst.parse_launch(self._description)

    def play(self):
        print("Playing pipeline: " + self._name)
        self._pipe.set_state(Gst.State.PLAYING)

    def stop(self):
        print("Stopping pipeline: " + self._name)
        self._pipe.set_state(Gst.State.NULL)

    def delete(self):
        print("Deleting pipeline: " + self._name)
        self.stop()
        self._pipe = None

    def eos(self):
        print("Sending EOS to pipeline: " + self._name)
        self._pipe.send_event(Gst.Event.new_eos())

    def set_file_location(self, location):
        print("Setting " + self._name + " pipeline recording/snapshot location to " + location);
        filesink_name = "filesink_" + self._name;
        self._pipe.get_by_name(filesink_name).set_property ('location', location);

    def listen_to(self, sink):
        print(self._name + " pipeline listening to " + sink);
        self._pipe.get_by_name(self._name + '_src').set_property ('listen-to', sink);

pipelines_base = []
pipelines_video_rec = []
pipelines_video_enc = []
pipelines_snap = []

Gst.init()

camera1 = PipelineEntity('camera1', 'videotestsrc ! video/x-raw,format=BGR ! queue ! interpipesink name=camera1 sync=true')
pipelines_base.append(camera1)

# Play base pipelines
# appsink_pipe is not started here with base pipelines
for pipeline in pipelines_base:
    pipeline.play()

def preprocess(listen_to):
    print("start capturing interpipe with cv.VideoCapture ")
    print(" ")

    # init video capture with appsink_pipe
    appsink_pipe = 'interpipesrc listen-to="%s" format=time ! appsink'%listen_to
    vod = cv.VideoCapture(appsink_pipe, cv.CAP_GSTREAMER)

    # read the first frame
    ret, frame = vod.read()

    # as long as frames are successfully read
    while ret:
        cv.imshow('Display Interpipe appsink', frame)

        k = cv.waitKey(1)
        # user Esc
        if k == 27:
            break

        # continue to next frame
        ret, frame = vod.read()

    # release the capture & destroy all windows
    vod.release()
    cv.destroyAllWindows()

    for pipeline in pipelines_base:
        pipeline.stop()
    for pipeline in pipelines_base:
        pipeline.delete()

if __name__ == '__main__':
    preprocess(camera1._name)

[michaelgruner]

Closing, feel free to re-open if you need anything.