python-codecs: major refactor to support hw acceleration and on deman…

…d color space conversion

plugins/python-codecs/.vscode/settings.json

@@ -1,7 +1,7 @@
 
 {
     // docker installation
-    // "scrypted.debugHost": "koushik-ubuntu",
+    // "scrypted.debugHost": "192.168.2.156",
     // "scrypted.serverRoot": "/server",
 
     // windows installation

plugins/python-codecs/package.json

@@ -1,6 +1,6 @@
 {
    "name": "@scrypted/python-codecs",
-   "version": "0.1.58",
+   "version": "0.1.61",
    "description": "Python Codecs for Scrypted",
    "keywords": [
       "scrypted",

plugins/python-codecs/src/generator_common.py

@@ -12,3 +12,13 @@ def createVideoFrame(image) -> scrypted_sdk.VideoFrame:
         'timestamp': time.time() * 1000,
         'flush': flush,
     }
+
+async def createImageMediaObject(image: scrypted_sdk.Image):
+    ret = await scrypted_sdk.mediaManager.createMediaObject(image, scrypted_sdk.ScryptedMimeTypes.Image.value, {
+        'format': None,
+        'width': image.width,
+        'height': image.height,
+        'toBuffer': lambda options = None: image.toBuffer(options),
+        'toImage': lambda options = None: image.toImage(options),
+    })
+    return ret

plugins/python-codecs/src/gst_generator.py

@@ -12,36 +12,54 @@
     GObject.threads_init()
     Gst.init(None)
 except:
-    pass
+    Gst = None
 
-async def createPipelineIterator(pipeline: str):
+async def createPipelineIterator(pipeline: str, gst = None):
     loop = asyncio.get_running_loop()
-    pipeline = '{pipeline} ! queue leaky=downstream max-size-buffers=0 ! appsink name=appsink emit-signals=true sync=false max-buffers=-1 drop=true'.format(pipeline=pipeline)
+    pipeline = '{pipeline} ! appsink name=appsink emit-signals=true sync=false max-buffers=-1 drop=true'.format(pipeline=pipeline)
     print(pipeline)
-    gst = Gst.parse_launch(pipeline)
-    bus = gst.get_bus()
-
-    def on_bus_message(bus, message):
-        t = str(message.type)
-        # print(t)
-        if t == str(Gst.MessageType.EOS):
-            print('EOS: Stream ended.')
-            finish()
-        elif t == str(Gst.MessageType.WARNING):
-            err, debug = message.parse_warning()
-            print('Warning: %s: %s\n' % (err, debug))
-            print('Ending stream due to warning. If this camera is causing errors, switch to the libav decoder.');
-            finish();
-        elif t == str(Gst.MessageType.ERROR):
-            err, debug = message.parse_error()
-            print('Error: %s: %s\n' % (err, debug))
-            finish()
+    finished = concurrent.futures.Future()
 
-    def stopGst():
-        bus.remove_signal_watch()
-        bus.disconnect(watchId)
-        gst.set_state(Gst.State.NULL)
+    newGst = not gst
+    if gst:
+        bin = Gst.parse_bin_from_description(pipeline, False)
+        gst.add(bin)
+        gst = bin
+
+        def stopGst():
+            gst.set_state(Gst.State.NULL)
+
+    else:
+        gst = Gst.parse_launch(pipeline)
+
+        def on_bus_message(bus, message):
+            t = str(message.type)
+            # print(t)
+            if t == str(Gst.MessageType.EOS):
+                print('EOS: Stream ended.')
+                finish()
+            elif t == str(Gst.MessageType.WARNING):
+                err, debug = message.parse_warning()
+                print('Warning: %s: %s\n' % (err, debug))
+                print('Ending stream due to warning. If this camera is causing errors, switch to the libav decoder.');
+                finish()
+            elif t == str(Gst.MessageType.ERROR):
+                err, debug = message.parse_error()
+                print('Error: %s: %s\n' % (err, debug))
+                finish()
+
+        bus = gst.get_bus()
+        watchId = bus.connect('message', on_bus_message)
+        bus.add_signal_watch()
+
+        def stopGst():
+            bus.remove_signal_watch()
+            bus.disconnect(watchId)
+            gst.set_state(Gst.State.NULL)
 
+    finished.add_done_callback(lambda _: threading.Thread(target=stopGst, name="StopGst").start())
+
+    hasFinished = False
     def finish():
         nonlocal hasFinished
         hasFinished = True
@@ -50,12 +68,6 @@ def finish():
         if not finished.done():
             finished.set_result(None)
 
-    watchId = bus.connect('message', on_bus_message)
-    bus.add_signal_watch()
-
-    finished = concurrent.futures.Future()
-    finished.add_done_callback(lambda _: threading.Thread(target=stopGst, name="StopGst").start())
-    hasFinished = False
 
     appsink = gst.get_by_name('appsink')
     yieldQueue = Queue()
@@ -98,10 +110,33 @@ def on_new_sample(sink):
 
 def mainThread():
     async def asyncMain():
-        gst, gen = createPipelineIterator('rtspsrc location=rtsp://localhost:59668/18cc179a814fd5b3 ! rtph264depay ! h264parse ! vtdec_hw ! videoconvert ! video/x-raw')
+        gst, gen = await createPipelineIterator('rtspsrc location=rtsp://localhost:63876/674e895e04ddfd15 ! rtph264depay ! h264parse ! vtdec_hw ! video/x-raw(memory:GLMemory)')
         i = 0
+        first = True
         async for sample in gen():
-            print('sample')
+            import time
+            print(time.time())
+            if first:
+                first = False
+
+            for i in range(1, 10):
+                caps = sample.get_caps()
+                p = "appsrc name=appsrc emit-signals=True is-live=True \
+                    caps={caps} ! videocrop left=0 top=0 right=10 bottom=10 ! gldownload".format(caps = caps.to_string().replace(' ', ''))
+                # p = "appsrc name=appsrc emit-signals=True is-live=True \
+                #     caps={caps} ! gldownload !\
+                #     videoconvert ! videoscale name=videoscale ! video/x-raw,format=RGB,width=640,height=480".format(caps = caps.to_string().replace(' ', ''))
+                gst2, gen2 = await createPipelineIterator(p)
+                appsrc = gst2.get_by_name('appsrc')
+                vs = gst2.get_by_name('videoscale')
+                g2 = gen2()
+
+                buffer = sample.get_buffer()
+                appsrc.emit("push-buffer", buffer)
+                s2 = await g2.__anext__()
+                print(time.time())
+                await g2.aclose()
+
             i = i + 1
             if i == 10:
                 break
@@ -111,6 +146,8 @@ async def asyncMain():
     loop.run_forever()
 
 if __name__ == "__main__":
+    test = 334
+    foo = f"{test}"
     threading.Thread(target = mainThread).start()
     mainLoop = GLib.MainLoop()
     mainLoop.run()