Merge pull request #25 from zauberzeug/active-cameras

Unify camera providers

.github/workflows/pytest.yml

@@ -20,6 +20,7 @@ jobs:
         run: |
           poetry config virtualenvs.create false --local
           poetry install
+          sudo apt-get update && sudo apt-get install -y v4l-utils
       - name: test
         run: pytest
       - name: examples

.vscode/settings.json

@@ -17,6 +17,7 @@
     "--disable=R0914", // Too many local variables
     "--disable=R0915", // Too many statements
     "--disable=R1702", // Too many nested blocks
+    "--disable=R1705", // Unnecessary "else" after "return"
     "--disable=W0102", // Dangerous default value as argument
     "--disable=W0511", // TODO
     "--disable=W0718", // Catching too general exception

docs/src/dev_log_file.py

@@ -8,7 +8,7 @@
 from rosys.driving import Odometer, Steerer, joystick
 from rosys.hardware import RobotSimulation, WheelsSimulation
 
-PATH = Path('~/.rosys')
+PATH = Path('~/.rosys').expanduser()
 PATH.mkdir(parents=True, exist_ok=True)
 
 logging.config.dictConfig({

docs/src/example_camera_control.py

@@ -0,0 +1,86 @@
+#!/usr/bin/env python3
+import logging
+
+from nicegui import ui
+
+import rosys.vision
+
+
+def add_card(camera: rosys.vision.Camera, container: ui.element) -> None:
+    uid = camera.id
+    if uid not in streams:
+        with container:
+            camera_card = ui.card().tight()
+            camera_cards[uid] = camera_card
+            print(f'adding card for {uid}')
+            with camera_grid:
+                with camera_card:
+                    streams[uid] = ui.interactive_image()
+                    ui.label(uid).classes('m-2')
+                    with ui.row():
+                        ui.switch('stream').bind_value(camera, 'streaming')
+                        ui.button('capture', on_click=camera.capture_image)
+                        ui.button('disconnect', on_click=camera.disconnect).bind_enabled_from(camera, 'is_connected')
+                    if isinstance(camera, rosys.vision.ConfigurableCamera):
+                        create_camera_settings_panel(camera)
+
+    streams[uid].set_source(camera.get_latest_image_url())
+
+
+def create_camera_settings_panel(camera: rosys.vision.ConfigurableCamera) -> None:
+    camera_parameters = camera.get_capabilities()
+    parameter_names = [parameter.name for parameter in camera_parameters]
+    with ui.expansion('Settings').classes('w-full').bind_enabled_from(camera, 'is_connected'):
+        if isinstance(camera, rosys.vision.RtspCamera):
+            ui.label('URL') \
+                .bind_text_from(camera, 'url', backward=lambda x: x or 'URL not available')
+        if 'fps' in parameter_names:
+            with ui.card(), ui.row():
+                ui.label('FPS:')
+                ui.select(options=list(range(1, 31)), on_change=lambda e: camera.set_parameters({'fps': e.value})) \
+                    .bind_value_from(camera, 'parameters', backward=lambda params: params['fps'])
+        if 'jovision_profile' in parameter_names:
+            with ui.card():
+                ui.switch('High Quality',
+                          on_change=lambda e: camera.set_parameters({'jovision_profile': 0 if e.value else 1}))
+        if 'exposure' in parameter_names:
+            with ui.card(), ui.row():
+                ui.label('Exposure:')
+                ui.select(options=list(range(0, 255)), on_change=lambda e: camera.set_parameters({'exposure': e.value})) \
+                    .bind_value_from(camera, 'parameters', backward=lambda params: params['exposure'])
+        if 'auto_exposure' in parameter_names:
+            with ui.card():
+                ui.switch('Auto Exposure', on_change=lambda e: camera.set_parameters({'auto_exposure': e.value})) \
+                    .bind_value_from(camera, 'parameters', backward=lambda params: params['auto_exposure'])
+        if 'color' in parameter_names:
+            with ui.card(), ui.row():
+                ui.label('Color:')
+                with ui.button(icon='colorize'):
+                    ui.color_picker(on_pick=lambda e: camera.set_parameters({'color': e.color}))
+
+
+def update_camera_cards() -> None:
+    providers: list[rosys.vision.CameraProvider] = [
+        rtsp_camera_provider,
+        usb_camera_provider,
+        simulated_camera_provider,
+    ]
+    for provider in providers:
+        for camera in provider.cameras.values():
+            add_card(camera, camera_grid)
+
+
+logging.basicConfig(level=logging.INFO)
+streams: dict[str, ui.interactive_image] = {}
+camera_cards: dict[str, ui.card] = {}
+camera_grid = ui.row()
+
+rtsp_camera_provider = rosys.vision.RtspCameraProvider()
+usb_camera_provider = rosys.vision.UsbCameraProvider()
+simulated_camera_provider = rosys.vision.SimulatedCameraProvider()
+
+ui.timer(0.1, update_camera_cards)
+
+simulated_camera_provider.add_cameras(1)
+
+ui.run(title='RoSys', port=8080)

docs/src/example_camera_streams.py

@@ -3,7 +3,7 @@
 
 import rosys
 
-camera_provider = rosys.vision.RtspCameraProviderHardware()
+camera_provider = rosys.vision.RtspCameraProvider()
 
 
 def refresh() -> None:

docs/src/example_cameras_images.py

@@ -1,20 +1,11 @@
 #!/usr/bin/env python3
 from nicegui import ui
 
-from rosys.vision import UsbCameraProviderSimulation, camera_provider
+from rosys.vision import ImageSize, SimulatedCamera
 
-camera_provider = UsbCameraProviderSimulation()
-camera_provider.add_camera(camera_provider.create_calibrated('test_cam', width=800, height=600))
+camera = SimulatedCamera(id='test_cam', resolution=ImageSize(width=800, height=600))
 
-
-def refresh() -> None:
-    for uid, camera in camera_provider.cameras.items():
-        if uid not in feeds:
-            feeds[uid] = ui.interactive_image()
-        feeds[uid].set_source(camera_provider.get_latest_image_url(camera))
-
-
-feeds: dict[str, ui.interactive_image] = {}
-ui.timer(0.3, refresh)
+image = ui.interactive_image()
+ui.timer(0.3, lambda: image.set_source(camera.get_latest_image_url()))
 
 ui.run(title='RoSys')

docs/src/example_cameras_remote.py

@@ -9,14 +9,14 @@
     can = rosys.hardware.CanHardware(robot_brain)
     wheels = rosys.hardware.WheelsHardware(robot_brain, can=can)
     robot = rosys.hardware.RobotHardware([can, wheels], robot_brain)
-    camera_provider = rosys.vision.UsbCameraProviderHardware()
+    camera_provider = rosys.vision.UsbCameraProvider()
 else:
     wheels = rosys.hardware.WheelsSimulation()
     robot = rosys.hardware.RobotSimulation([wheels])
-    camera_provider = rosys.vision.UsbCameraProviderSimulation()
-    camera_provider.restore = lambda _: None  # NOTE: disable persistence
-    test_cam = camera_provider.create_calibrated('test_cam', width=800, height=600)
-    rosys.on_startup(lambda: camera_provider.add_camera(test_cam))
+    rosys.vision.SimulatedCameraProvider.USE_PERSISTENCE = False
+    camera_provider = rosys.vision.SimulatedCameraProvider()
+    camera = rosys.vision.SimulatedCamera(id='test_cam', resolution=rosys.vision.ImageSize(width=800, height=600))
+    rosys.on_startup(lambda: camera_provider.add_camera(camera))
 steerer = rosys.driving.Steerer(wheels)
 odometer = rosys.driving.Odometer(wheels)
 
@@ -25,7 +25,7 @@ async def add_main_camera(camera: rosys.vision.Camera) -> None:
     camera_card.clear()  # remove "seeking camera" label
     with camera_card:
         main_cam = ui.interactive_image()
-        ui.timer(0.1, lambda: main_cam.set_source(camera_provider.get_latest_image_url(camera)))
+        ui.timer(0.1, lambda: main_cam.set_source(camera.get_latest_image_url()))
 
 camera_provider.CAMERA_ADDED.register_ui(add_main_camera)
 

pyproject.toml

@@ -45,6 +45,7 @@ imgsize = "^2.1"
 netifaces = "^0.11.0"
 httpx = "^0.24.0"
 matplotlib = "^3.7.2"
+pyudev = ">=0.21.0"
 
 [tool.poetry.dev-dependencies]
 autopep8 = "^1.5.5"

rosys/driving/odometer.py

@@ -88,10 +88,10 @@ def _compute_odometry_frame(local_pose: Pose, global_pose: Pose) -> Pose:
         frame.time = global_pose.time
         return frame
 
-    def reset(self):
+    def reset(self) -> None:
         self.prediction = Pose()
         self.detection = None
         self.current_velocity = None
-        self.last_movement: float = 0
-        self.history: list[Pose] = []
+        self.last_movement = 0
+        self.history.clear()
         self.odometry_frame = Pose()

rosys/run.py

@@ -115,7 +115,7 @@ def popen() -> str:
     if is_stopping():
         return ''
     try:
-        return await asyncio.wait_for(io_bound(popen), timeout)
+        return await asyncio.wait_for(io_bound(popen), timeout) or ''
     except asyncio.TimeoutError:
         log.warning(f'Command "{command}" timed out after {timeout} seconds.')
         return ''

rosys/vision/__init__.py

@@ -1,17 +1,14 @@
 from .calibration import Calibration, Extrinsics, Intrinsics
-from .camera import Camera
+from .camera import CalibratableCamera, Camera, ConfigurableCamera, TransformableCamera
 from .camera_objects_ import CameraObjects as camera_objects
 from .camera_projector import CameraProjector
 from .camera_provider import CameraProvider
-from .combining_camera_provider import CombiningCameraProvider
 from .detections import BoxDetection, Detection, Detections, PointDetection
 from .detector import Autoupload, Detector
 from .detector_hardware import DetectorHardware
 from .detector_simulation import DetectorSimulation, SimulatedObject
 from .image import Image, ImageSize
 from .multi_camera_provider import MultiCameraProvider
-from .rtsp_camera import RtspCamera
-from .rtsp_camera_provider_hardware import RtspCameraProviderHardware
-from .usb_camera import UsbCamera
-from .usb_camera_provider_hardware import UsbCameraProviderHardware
-from .usb_camera_provider_simulation import UsbCameraProviderSimulation
+from .rtsp_camera import RtspCamera, RtspCameraProvider
+from .simulated_camera import SimulatedCamera, SimulatedCameraProvider
+from .usb_camera import UsbCamera, UsbCameraProvider

rosys/vision/calibration.py

@@ -18,6 +18,14 @@ class Intrinsics:
     rotation: Rotation
     size: ImageSize
 
+    @staticmethod
+    def create_default(width: int = 800, height: int = 600, *, focal_length: float = 570) -> Intrinsics:
+        size = ImageSize(width=width, height=height)
+        K: list[list[float]] = [[focal_length, 0, size.width / 2], [0, focal_length, size.height / 2], [0, 0, 1]]
+        D: list[float] = [0] * 5
+        rotation = Rotation.zero()
+        return Intrinsics(matrix=K, distortion=D, rotation=rotation, size=size)
+
 
 log = logging.getLogger('rosys.world.calibration')
 

rosys/vision/camera/__init__.py

@@ -0,0 +1,11 @@
+from .calibratable_camera import CalibratableCamera
+from .camera import Camera
+from .configurable_camera import ConfigurableCamera
+from .transformable_camera import TransformableCamera
+
+__all__ = [
+    'CalibratableCamera',
+    'Camera',
+    'ConfigurableCamera',
+    'TransformableCamera',
+]

rosys/vision/camera/calibratable_camera.py

@@ -0,0 +1,38 @@
+from typing import Optional, Self
+
+import numpy as np
+
+from ...geometry import Rotation
+from ..calibration import Calibration, Extrinsics, Intrinsics
+from .camera import Camera
+
+
+class CalibratableCamera(Camera):
+
+    def __init__(self, **kwargs) -> None:
+        super().__init__(**kwargs)
+        self.calibration: Optional[Calibration] = None
+        self.focal_length: Optional[float] = None
+
+    @property
+    def is_calibrated(self) -> bool:
+        return self.calibration is not None
+
+    @classmethod
+    def create_calibrated(cls, *,
+                          width: int = 800, height: int = 600,
+                          x: float = 0.0, y: float = 0.0, z: float = 1.0,
+                          roll: float = np.pi, pitch: float = 0.0, yaw: float = 0.0,
+                          **kwargs) -> Self:
+        camera = cls(**kwargs)
+        camera.set_perfect_calibration(width=width, height=height, x=x, y=y, z=z, roll=roll, pitch=pitch, yaw=yaw)
+        return camera
+
+    def set_perfect_calibration(self, *,
+                                width=800, height=600,
+                                x: float = 0.0, y: float = 0.0, z: float = 1.0,
+                                roll: float = np.pi, pitch: float = 0.0, yaw: float = 0.0) -> None:
+        self.calibration = Calibration(
+            intrinsics=Intrinsics.create_default(width, height),
+            extrinsics=Extrinsics(rotation=Rotation.from_euler(roll, pitch, yaw), translation=[x, y, z]),
+        )