Camera driver

from ados.sdk.drivers.camera import (
    CameraDriver,
    CameraCandidate,
    CameraCapabilities,
    CameraSession,
    FrameBuffer,
)

class MyCameraDriver(CameraDriver):
    async def discover(self) -> list[CameraCandidate]:
        ...

    async def open(self, candidate, config) -> CameraSession:
        ...

    async def close(self, session) -> None:
        ...

    def capabilities(self, session) -> CameraCapabilities:
        ...

    async def frame_iterator(self, session):
        # async generator yielding FrameBuffer
        ...

    async def set_param(self, session, param, value) -> None:
        ...

CameraCandidate(
    driver_id="com.example.my-camera",
    device_id="usb-1.2.3",
    label="MyCorp ThermalCam on USB 1.2.3",
    bus="usb",
    vid_pid=(0x1e4e, 0x0100),
)

CameraCapabilities(
    radiometric=True,
    bit_depth=14,
    width=160,
    height=120,
    fps=8.7,
    pixel_format="Y16",
    streaming_protocol="uvc",
    color_spaces=[],
    has_audio=False,
)

FrameBuffer(
    timestamp_ns=ns,
    sequence=seq,
    width=160,
    height=120,
    pixel_format="Y16",
    data=memoryview(raw_bytes),
    radiometric_k=memoryview(temp_K_matrix),
    metadata={"shutter_open": True},
)

agent:
  permissions:
    - id: sensor.camera.register   # required, gates registration
    - id: hardware.usb.uvc         # for a USB UVC camera (the FLIR Lepton reference path)
    - id: hardware.camera.csi      # for a MIPI CSI camera on the SBC connector
    - id: hardware.usb             # for a raw USB device behind a vendor SDK

class Plugin:
    id = "com.example.my-camera"
    version = "1.0.0"

    async def on_start(self, ctx):
        self.driver = MyCameraDriver()
        await ctx.peripheral_manager.register_camera_driver(self.driver)