From 5a71079ad0312a6a37aa935b872424c4ed1e2d28 Mon Sep 17 00:00:00 2001 From: KarlS Date: Wed, 8 Apr 2026 22:01:09 -0700 Subject: [PATCH] Add unit tests for image nodes --- synapse/client/__init__.py | 3 + synapse/client/nodes/__init__.py | 4 + synapse/client/nodes/image_sink.py | 38 ++++++++++ synapse/client/nodes/image_source.py | 51 +++++++++++++ synapse/tests/client/test_image_sink.py | 48 ++++++++++++ synapse/tests/client/test_image_source.py | 90 +++++++++++++++++++++++ 6 files changed, 234 insertions(+) create mode 100644 synapse/client/nodes/image_sink.py create mode 100644 synapse/client/nodes/image_source.py create mode 100644 synapse/tests/client/test_image_sink.py create mode 100644 synapse/tests/client/test_image_source.py diff --git a/synapse/client/__init__.py b/synapse/client/__init__.py index 7236d04..74f8b8f 100644 --- a/synapse/client/__init__.py +++ b/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 diff --git a/synapse/client/nodes/__init__.py b/synapse/client/nodes/__init__.py index 0722912..c439d13 100644 --- a/synapse/client/nodes/__init__.py +++ b/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, diff --git a/synapse/client/nodes/image_sink.py b/synapse/client/nodes/image_sink.py new file mode 100644 index 0000000..b2b65bd --- /dev/null +++ b/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, + ) diff --git a/synapse/client/nodes/image_source.py b/synapse/client/nodes/image_source.py new file mode 100644 index 0000000..eac8c81 --- /dev/null +++ b/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, + ) diff --git a/synapse/tests/client/test_image_sink.py b/synapse/tests/client/test_image_sink.py new file mode 100644 index 0000000..d66466a --- /dev/null +++ b/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") diff --git a/synapse/tests/client/test_image_source.py b/synapse/tests/client/test_image_source.py new file mode 100644 index 0000000..c3d1b2d --- /dev/null +++ b/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")