update api framework (#8)

* update api framework

* fix linter

* fix typo

* add the manifest and install/uninstall scripts

entry.sh

@@ -3,6 +3,6 @@ DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
 
 $DIR/bootstrap.sh $DIR $DIR/venv
 
-$DIR/venv/bin/python $DIR/src/main.py $@ --port 50051
+$DIR/venv/bin/python $DIR/src/main.py
 
 exit 0

install.sh

@@ -0,0 +1,8 @@
+#! /bin/bash
+
+set -uxeo pipefail
+
+DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
+
+rm -f ~/manifest.json
+ln -s "$DIR/manifest.json" ~/manifest.json

manifest.json

@@ -0,0 +1,13 @@
+{
+    "services": {
+        "camera-app": {
+            "type": "app_third_party",
+            "python": false,
+            "args": [
+                "entry.sh"
+            ],
+            "http_gui_port": 0,
+            "display_name": "Kivy Camera App"
+        }
+    }
+}

setup.cfg

@@ -23,7 +23,7 @@ install_requires =
     wheel
     kivy
     farm_ng_amiga
-    PyTurboJPEG
+    kornia-rs
 tests_require =
     pytest
 

src/main.py

@@ -11,21 +11,25 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+from __future__ import annotations
+
 import argparse
 import asyncio
+import logging
 import os
-from typing import List
+from pathlib import Path
+from typing import Literal
 
-import grpc
-from farm_ng.oak import oak_pb2
-from farm_ng.oak.camera_client import OakCameraClient
-from farm_ng.service import service_pb2
-from farm_ng.service.service_client import ClientConfig
-from turbojpeg import TurboJPEG
+from farm_ng.core.event_client import EventClient
+from farm_ng.core.event_service_pb2 import EventServiceConfig
+from farm_ng.core.event_service_pb2 import EventServiceConfigList
+from farm_ng.core.event_service_pb2 import SubscribeRequest
+from farm_ng.core.events_file_reader import proto_from_json_file
+from farm_ng.core.uri_pb2 import Uri
+from kornia_rs import ImageDecoder
 
 os.environ["KIVY_NO_ARGS"] = "1"
 
-
 from kivy.config import Config  # noreorder # noqa: E402
 
 Config.set("graphics", "resizable", False)
@@ -40,15 +44,20 @@
 from kivy.graphics.texture import Texture  # noqa: E402
 
 
+logger = logging.getLogger("amiga.apps.camera")
+
+
 class CameraApp(App):
-    def __init__(self, address: str, port: int, stream_every_n: int) -> None:
+
+    STREAM_NAMES = ["rgb", "disparity", "left", "right"]
+
+    def __init__(self, service_config: EventServiceConfig, stream_every_n: int) -> None:
         super().__init__()
-        self.address = address
-        self.port = port
+        self.service_config = service_config
         self.stream_every_n = stream_every_n
 
-        self.image_decoder = TurboJPEG()
-        self.tasks: List[asyncio.Task] = []
+        self.image_decoder = ImageDecoder()
+        self.image_subscription_tasks: list[asyncio.Task] = []
 
     def build(self):
         return Builder.load_file("res/main.kv")
@@ -62,98 +71,89 @@ async def run_wrapper():
             # we don't actually need to set asyncio as the lib because it is
             # the default, but it doesn't hurt to be explicit
             await self.async_run(async_lib="asyncio")
-            for task in self.tasks:
+            for task in self.image_subscription_tasks:
                 task.cancel()
 
-        # configure the camera client
-        config = ClientConfig(address=self.address, port=self.port)
-        client = OakCameraClient(config)
-
-        # Stream camera frames
-        self.tasks.append(asyncio.ensure_future(self.stream_camera(client)))
+        # stream camera frames
+        self.image_subscription_tasks: list[asyncio.Task] = [
+            asyncio.create_task(self.stream_camera(view_name))
+            for view_name in self.STREAM_NAMES
+        ]
 
-        return await asyncio.gather(run_wrapper(), *self.tasks)
+        return await asyncio.gather(run_wrapper(), *self.image_subscription_tasks)
 
-    async def stream_camera(self, client: OakCameraClient) -> None:
-        """This task listens to the camera client's stream and populates the tabbed panel with all 4 image streams
-        from the oak camera."""
+    async def stream_camera(
+        self, view_name: Literal["rgb", "disparity", "left", "right"] = "rgb"
+    ) -> None:
+        """Subscribes to the camera service and populates the tabbed panel with all 4 image streams."""
         while self.root is None:
             await asyncio.sleep(0.01)
 
-        response_stream = None
-
-        while True:
-            # check the state of the service
-            state = await client.get_state()
-
-            if state.value not in [
-                service_pb2.ServiceState.IDLE,
-                service_pb2.ServiceState.RUNNING,
-            ]:
-                # Cancel existing stream, if it exists
-                if response_stream is not None:
-                    response_stream.cancel()
-                    response_stream = None
-                print("Camera service is not streaming or ready to stream")
-                await asyncio.sleep(0.1)
-                continue
-
-            # Create the stream
-            if response_stream is None:
-                response_stream = client.stream_frames(every_n=self.stream_every_n)
-
+        async for _, message in EventClient(self.service_config).subscribe(
+            SubscribeRequest(
+                uri=Uri(path=f"/{view_name}"), every_n=self.stream_every_n
+            ),
+            decode=True,
+        ):
             try:
-                # try/except so app doesn't crash on killed service
-                response: oak_pb2.StreamFramesReply = await response_stream.read()
-                assert response and response != grpc.aio.EOF, "End of stream"
+                img = self.image_decoder.decode(message.image_data)
             except Exception as e:
-                print(e)
-                response_stream.cancel()
-                response_stream = None
+                logger.exception(f"Error decoding image: {e}")
                 continue
 
-            # get the sync frame
-            frame: oak_pb2.OakSyncFrame = response.frame
-
-            # get image and show
-            for view_name in ["rgb", "disparity", "left", "right"]:
-                # Skip if view_name was not included in frame
-                try:
-                    # Decode the image and render it in the correct kivy texture
-                    img = self.image_decoder.decode(
-                        getattr(frame, view_name).image_data
-                    )
-                    texture = Texture.create(
-                        size=(img.shape[1], img.shape[0]), icolorfmt="bgr"
-                    )
-                    texture.flip_vertical()
-                    texture.blit_buffer(
-                        img.tobytes(),
-                        colorfmt="bgr",
-                        bufferfmt="ubyte",
-                        mipmap_generation=False,
-                    )
-                    self.root.ids[view_name].texture = texture
-
-                except Exception as e:
-                    print(e)
+            # create the opengl texture and set it to the image
+            texture = Texture.create(size=(img.shape[1], img.shape[0]), icolorfmt="rgb")
+            texture.flip_vertical()
+            texture.blit_buffer(
+                bytes(img.data),
+                colorfmt="rgb",
+                bufferfmt="ubyte",
+                mipmap_generation=False,
+            )
+            self.root.ids[view_name].texture = texture
+
+
+def find_config_by_name(
+    service_configs: EventServiceConfigList, name: str
+) -> EventServiceConfig | None:
+    """Utility function to find a service config by name.
+
+    Args:
+        service_configs: List of service configs
+        name: Name of the service to find
+    """
+    for config in service_configs.configs:
+        if config.name == name:
+            return config
+    return None
 
 
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(prog="amiga-camera-app")
-    parser.add_argument("--port", type=int, required=True, help="The camera port.")
     parser.add_argument(
-        "--address", type=str, default="localhost", help="The camera address"
+        "--service-config", type=Path, default="/opt/farmng/config.json"
     )
+    parser.add_argument("--camera-name", type=str, default="oak1")
     parser.add_argument(
         "--stream-every-n", type=int, default=1, help="Streaming frequency"
     )
     args = parser.parse_args()
 
+    # config with all the configs
+    service_config_list: EventServiceConfigList = proto_from_json_file(
+        args.service_config, EventServiceConfigList()
+    )
+
+    # filter out services to pass to the events client manager
+    oak_service_config = find_config_by_name(service_config_list, args.camera_name)
+    if oak_service_config is None:
+        raise RuntimeError(f"Could not find service config for {args.camera_name}")
+
     loop = asyncio.get_event_loop()
+
     try:
         loop.run_until_complete(
-            CameraApp(args.address, args.port, args.stream_every_n).app_func()
+            CameraApp(oak_service_config, args.stream_every_n).app_func()
         )
     except asyncio.CancelledError:
         pass

uninstall.sh

@@ -0,0 +1,8 @@
+#! /bin/bash
+
+set -uxeo pipefail
+
+DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
+
+rm -f ~/manifest.json
+rm -f ~/.config/systemd/user/*.service