Add unit tests for image nodes

Author: karls-science

synapse/client/__init__.py

@@ -6,10 +6,13 @@
 
 from synapse.client.channel import Channel
 from synapse.client.signal_config import SignalConfig, ElectrodeConfig, PixelConfig
+from synapse.api.datatype_pb2 import PixelFormat
 
 from synapse.client.nodes.broadband_source import BroadbandSource
 from synapse.client.nodes.disk_writer import DiskWriter
 from synapse.client.nodes.electrical_stimulation import ElectricalStimulation
+from synapse.client.nodes.image_sink import ImageSink
+from synapse.client.nodes.image_source import ImageSource
 from synapse.client.nodes.optical_stimulation import OpticalStimulation
 from synapse.client.nodes.spike_source import SpikeSource
 from synapse.client.nodes.spike_binner import SpikeBinner

synapse/client/nodes/__init__.py

@@ -1,5 +1,7 @@
 from synapse.client.nodes.broadband_source import BroadbandSource
 from synapse.client.nodes.electrical_stimulation import ElectricalStimulation
+from synapse.client.nodes.image_sink import ImageSink
+from synapse.client.nodes.image_source import ImageSource
 from synapse.client.nodes.optical_stimulation import OpticalStimulation
 from synapse.client.nodes.spectral_filter import SpectralFilter
 from synapse.client.nodes.spike_binner import SpikeBinner
@@ -14,6 +16,8 @@
     NodeType.kBroadbandSource: BroadbandSource,
     NodeType.kDiskWriter: DiskWriter,
     NodeType.kElectricalStimulation: ElectricalStimulation,
+    NodeType.kImageSink: ImageSink,
+    NodeType.kImageSource: ImageSource,
     NodeType.kOpticalStimulation: OpticalStimulation,
     NodeType.kSpectralFilter: SpectralFilter,
     NodeType.kSpikeBinner: SpikeBinner,

synapse/client/nodes/image_sink.py

@@ -0,0 +1,38 @@
+from typing import Optional
+
+from synapse.api.node_pb2 import NodeConfig, NodeType
+from synapse.api.nodes.image_sink_pb2 import ImageSinkConfig
+from synapse.client.node import Node
+
+
+class ImageSink(Node):
+    type = NodeType.kImageSink
+
+    def __init__(
+        self,
+        peripheral_id: int,
+        frame_rate_hz: int,
+    ):
+        self.peripheral_id: int = peripheral_id
+        self.frame_rate_hz: int = frame_rate_hz
+
+    def _to_proto(self):
+        n = NodeConfig()
+        p = ImageSinkConfig(
+            peripheral_id=self.peripheral_id,
+            frame_rate_hz=self.frame_rate_hz,
+        )
+        n.image_sink.CopyFrom(p)
+        return n
+
+    @staticmethod
+    def _from_proto(proto: Optional[ImageSinkConfig]):
+        if not proto:
+            raise ValueError("parameter 'proto' is missing")
+        if not isinstance(proto, ImageSinkConfig):
+            raise ValueError("proto is not of type ImageSinkConfig")
+
+        return ImageSink(
+            peripheral_id=proto.peripheral_id,
+            frame_rate_hz=proto.frame_rate_hz,
+        )

synapse/client/nodes/image_source.py

@@ -0,0 +1,51 @@
+from typing import Optional
+
+from synapse.api.datatype_pb2 import PixelFormat
+from synapse.api.node_pb2 import NodeConfig, NodeType
+from synapse.api.nodes.image_source_pb2 import ImageSourceConfig
+from synapse.client.node import Node
+
+
+class ImageSource(Node):
+    type = NodeType.kImageSource
+
+    def __init__(
+        self,
+        peripheral_id: int,
+        width: int,
+        height: int,
+        format: PixelFormat,
+        frame_rate_hz: int,
+    ):
+        self.peripheral_id: int = peripheral_id
+        self.width: int = width
+        self.height: int = height
+        self.format: PixelFormat = format
+        self.frame_rate_hz: int = frame_rate_hz
+
+    def _to_proto(self):
+        n = NodeConfig()
+        p = ImageSourceConfig(
+            peripheral_id=self.peripheral_id,
+            width=self.width,
+            height=self.height,
+            format=self.format,
+            frame_rate_hz=self.frame_rate_hz,
+        )
+        n.image_source.CopyFrom(p)
+        return n
+
+    @staticmethod
+    def _from_proto(proto: Optional[ImageSourceConfig]):
+        if not proto:
+            raise ValueError("parameter 'proto' is missing")
+        if not isinstance(proto, ImageSourceConfig):
+            raise ValueError("proto is not of type ImageSourceConfig")
+
+        return ImageSource(
+            peripheral_id=proto.peripheral_id,
+            width=proto.width,
+            height=proto.height,
+            format=proto.format,
+            frame_rate_hz=proto.frame_rate_hz,
+        )

synapse/tests/client/test_image_sink.py

@@ -0,0 +1,48 @@
+import pytest
+from synapse.api.nodes.image_sink_pb2 import ImageSinkConfig
+from synapse.client.nodes.image_sink import ImageSink
+
+
+def test_image_sink_to_proto():
+    sink = ImageSink(
+        peripheral_id=1,
+        frame_rate_hz=30,
+    )
+
+    proto = sink._to_proto()
+
+    assert proto.image_sink.peripheral_id == 1
+    assert proto.image_sink.frame_rate_hz == 30
+
+
+def test_image_sink_from_proto():
+    proto = ImageSinkConfig(
+        peripheral_id=2,
+        frame_rate_hz=60,
+    )
+
+    sink = ImageSink._from_proto(proto)
+
+    assert sink.peripheral_id == 2
+    assert sink.frame_rate_hz == 60
+
+
+def test_image_sink_from_proto_roundtrip():
+    original = ImageSink(
+        peripheral_id=5,
+        frame_rate_hz=24,
+    )
+
+    proto = original._to_proto()
+    restored = ImageSink._from_proto(proto.image_sink)
+
+    assert restored.peripheral_id == original.peripheral_id
+    assert restored.frame_rate_hz == original.frame_rate_hz
+
+
+def test_image_sink_from_invalid_proto():
+    with pytest.raises(ValueError, match="missing"):
+        ImageSink._from_proto(None)
+
+    with pytest.raises(ValueError, match="not of type"):
+        ImageSink._from_proto("invalid")

synapse/tests/client/test_image_source.py

@@ -0,0 +1,90 @@
+import pytest
+from synapse.api.datatype_pb2 import PixelFormat
+from synapse.api.nodes.image_source_pb2 import ImageSourceConfig
+from synapse.client.nodes.image_source import ImageSource
+
+
+def test_image_source_to_proto():
+    source = ImageSource(
+        peripheral_id=1,
+        width=1920,
+        height=1080,
+        format=PixelFormat.kRGB888,
+        frame_rate_hz=30,
+    )
+
+    proto = source._to_proto()
+
+    assert proto.image_source.peripheral_id == 1
+    assert proto.image_source.width == 1920
+    assert proto.image_source.height == 1080
+    assert proto.image_source.format == PixelFormat.kRGB888
+    assert proto.image_source.frame_rate_hz == 30
+
+
+def test_image_source_to_proto_pixel_formats():
+    for fmt in [
+        PixelFormat.kPixelFormatUnknown,
+        PixelFormat.kYUV420_888,
+        PixelFormat.kRGB888,
+        PixelFormat.kRGBA8888,
+        PixelFormat.kGrayscale8,
+        PixelFormat.kRAW10,
+        PixelFormat.kRAW16,
+        PixelFormat.kNV12,
+        PixelFormat.kNV21,
+    ]:
+        source = ImageSource(
+            peripheral_id=0,
+            width=640,
+            height=480,
+            format=fmt,
+            frame_rate_hz=60,
+        )
+        proto = source._to_proto()
+        assert proto.image_source.format == fmt
+
+
+def test_image_source_from_proto():
+    proto = ImageSourceConfig(
+        peripheral_id=2,
+        width=3840,
+        height=2160,
+        format=PixelFormat.kNV21,
+        frame_rate_hz=15,
+    )
+
+    source = ImageSource._from_proto(proto)
+
+    assert source.peripheral_id == 2
+    assert source.width == 3840
+    assert source.height == 2160
+    assert source.format == PixelFormat.kNV21
+    assert source.frame_rate_hz == 15
+
+
+def test_image_source_from_proto_roundtrip():
+    original = ImageSource(
+        peripheral_id=3,
+        width=1280,
+        height=720,
+        format=PixelFormat.kGrayscale8,
+        frame_rate_hz=120,
+    )
+
+    proto = original._to_proto()
+    restored = ImageSource._from_proto(proto.image_source)
+
+    assert restored.peripheral_id == original.peripheral_id
+    assert restored.width == original.width
+    assert restored.height == original.height
+    assert restored.format == original.format
+    assert restored.frame_rate_hz == original.frame_rate_hz
+
+
+def test_image_source_from_invalid_proto():
+    with pytest.raises(ValueError, match="missing"):
+        ImageSource._from_proto(None)
+
+    with pytest.raises(ValueError, match="not of type"):
+        ImageSource._from_proto("invalid")