Feature: Live stream in application (#66)

* Add live stream functionality in app

* Remove unnecessary newlines

* Restructure start_live_stream method

* Rename variables and add temp fix comment

Author: osode

examples/g3pycontroller/__main__.py

@@ -2,9 +2,13 @@
 import logging
 from typing import List, Optional, Set, Tuple, cast
 
+import numpy as np
 from eventkinds import AppEventKind, ControlEventKind
 from kivy.app import App
+from kivy.clock import Clock
 from kivy.core.window import Window
+from kivy.graphics import Rectangle
+from kivy.graphics.texture import Texture
 from kivy.lang.builder import Builder
 from kivy.properties import BooleanProperty
 from kivy.uix.label import Label
@@ -47,7 +51,6 @@
             text: "Connect"
             on_press: app.send_app_event(AppEventKind.ENTER_CONTROL_SESSION)
 
-
 <UserMessagePopup>:
     size_hint: None, None
     size: 400, 200
@@ -107,8 +110,19 @@
 
 <LiveScreen>:
     BoxLayout:
-        Label:
-            text: "Here you can see your glasses in action."
+        Widget:
+            id: display
+            size_hint_x: 0.8
+            size_hint_y: 1
+        BoxLayout:
+            orientation: "vertical"
+            size_hint_x: 0.2
+            Button:
+                text: "Start"
+                on_press: app.send_control_event(ControlEventKind.START_LIVE)
+            Button:
+                text: "Stop"
+                on_press: app.send_control_event(ControlEventKind.STOP_LIVE)
 
 <SelectableList>:
     viewclass: 'SelectableLabel'
@@ -242,7 +256,8 @@ def set_recording_status(self, is_recording: bool) -> None:
 
 
 class LiveScreen(Screen):
-    pass
+    def clear(self, *args):
+        self.ids.display.canvas.clear()
 
 
 class UserMessagePopup(Popup):
@@ -256,6 +271,8 @@ def __init__(self, **kwargs):
         self.tasks: Set[asyncio.Task] = set()
         self.app_events: asyncio.Queue[AppEventKind] = asyncio.Queue()
         self.control_events: asyncio.Queue[ControlEventKind] = asyncio.Queue()
+        self.live_stream_task: Optional[asyncio.Task] = None
+        self.read_frames_task: Optional[asyncio.Task] = None
         self.add_widget(DiscoveryScreen(name="discovery"))
         self.add_widget(ControlScreen(name="control"))
 
@@ -391,8 +408,81 @@ async def handle_control_event(self, g3: Glasses3, event: ControlEventKind) -> N
                 await g3.recorder.stop()
             case ControlEventKind.DELETE_RECORDING:
                 await self.delete_selected_recording(g3)
+            case ControlEventKind.START_LIVE:
+                self.start_live_stream(g3)
+            case ControlEventKind.STOP_LIVE:
+                await self.stop_live_stream()
         self.get_screen("control").set_task_running_status(False)
 
+    def start_live_stream(self, g3: Glasses3) -> None:
+        async def live_stream():
+            async with g3.stream_rtsp() as streams:
+                async with streams.scene_camera.decode() as decoded_stream:
+                    live_screen = self.get_screen("control").ids.sm.get_screen("live")
+                    Window.bind(on_resize=live_screen.clear)
+                    self.latest_frame = await decoded_stream.get()
+                    self.read_frames_task = self.create_task(
+                        update_frame(decoded_stream, streams), name="update_frame"
+                    )
+                    self.scheduling_delay = 1 / 25
+                    Clock.schedule_once(draw_frame, self.scheduling_delay)
+                    await self.read_frames_task
+
+        async def update_frame(decoded_stream, streams):
+            while True:
+                self.latest_frame = await decoded_stream.get()
+                logging.debug(streams.scene_camera.stats)
+
+        def draw_frame(dt):
+            """NOTE: The scheduling if frame redraw would ideally be made with Clock.schedule_interval at 25Hz, but due to performance issues a dynamic scheduling delay is implemented as a temporary fix."""
+            display = self.get_screen("control").ids.sm.get_screen("live").ids.display
+            update_scheduling_delay(dt)
+            if self.read_frames_task is not None:
+                if not self.read_frames_task.done():
+                    Clock.schedule_once(draw_frame, self.scheduling_delay)
+            with display.canvas:
+                image = np.flip(self.latest_frame.to_ndarray(format="bgr24"), 0)
+                texture = Texture.create(
+                    size=(image.shape[1], image.shape[0]), colorfmt="bgr"
+                )
+                image = np.reshape(image, -1)
+                texture.blit_buffer(image, colorfmt="bgr", bufferfmt="ubyte")
+                Rectangle(
+                    texture=texture,
+                    pos=(0, (display.top - display.width * 9 / 16) / 2),
+                    size=(display.width, display.width * 9 / 16),
+                )
+
+        def update_scheduling_delay(last_delay):
+            if last_delay > self.scheduling_delay:
+                self.scheduling_delay = last_delay
+            elif last_delay < self.scheduling_delay:
+                self.scheduling_delay -= (self.scheduling_delay - last_delay) / 2
+
+        def live_stream_task_running() -> bool:
+            if self.live_stream_task is not None:
+                return not self.live_stream_task.done()
+            else:
+                return False
+
+        if live_stream_task_running():
+            logging.info("Task not started: live_stream_task already running.")
+        else:
+            self.live_stream_task = self.create_task(
+                live_stream(), name="live_stream_task"
+            )
+
+    async def stop_live_stream(self) -> None:
+        if self.read_frames_task is not None:
+            if not self.read_frames_task.cancelled():
+                await self.cancel_task(self.read_frames_task)
+        if self.live_stream_task is not None:
+            if not self.live_stream_task.cancelled():
+                await self.cancel_task(self.live_stream_task)
+        live_screen = self.get_screen("control").ids.sm.get_screen("live")
+        Window.unbind(on_resize=live_screen.clear)
+        live_screen.clear()
+
     async def delete_selected_recording(self, g3: Glasses3) -> None:
         selected = (
             self.get_screen("control")

examples/g3pycontroller/eventkinds.py

@@ -5,6 +5,8 @@ class ControlEventKind(Enum):
     START_RECORDING = auto()
     STOP_RECORDING = auto()
     DELETE_RECORDING = auto()
+    START_LIVE = auto()
+    STOP_LIVE = auto()
 
 
 class AppEventKind(Enum):