Proof-of-concept: Add network support (IPv4Transport)

src/seabreeze/backends.py

@@ -61,6 +61,9 @@ def use(
         if "pyusb_backend" in kwargs:
             pyusb_backend = kwargs.pop("pyusb_backend")
             BackendConfig.api_kwargs["pyusb_backend"] = pyusb_backend
+        if "network_adapter" in kwargs:
+            network_adapter = kwargs.pop("network_adapter")
+            BackendConfig.api_kwargs["network_adapter"] = network_adapter
     if kwargs:
         raise TypeError(
             f"unknown keyword arguments {set(kwargs)!r} for backend {backend!r}"

src/seabreeze/pyseabreeze/api.py

@@ -17,6 +17,8 @@
 from seabreeze.pyseabreeze.devices import SeaBreezeDevice
 from seabreeze.pyseabreeze.devices import _model_class_registry
 from seabreeze.pyseabreeze.transport import DeviceIdentity
+from seabreeze.pyseabreeze.transport import IPv4Transport
+from seabreeze.pyseabreeze.transport import IPv4TransportHandle
 from seabreeze.pyseabreeze.transport import USBTransport
 from seabreeze.pyseabreeze.transport import USBTransportDeviceInUse
 from seabreeze.pyseabreeze.transport import USBTransportError
@@ -35,20 +37,24 @@
 ] = weakref.WeakValueDictionary()
 
 
-def _seabreeze_device_factory(device: USBTransportHandle) -> SeaBreezeDevice:
+def _seabreeze_device_factory(
+    device: USBTransportHandle | IPv4TransportHandle,
+) -> SeaBreezeDevice:
     """return existing instances instead of creating temporary ones
 
     Parameters
     ----------
-    device : USBTransportHandle
+    device : USBTransportHandle | IPv4TransportHandle
 
     Returns
     -------
     dev : SeaBreezeDevice
     """
     global _seabreeze_device_instance_registry
-    if not isinstance(device, USBTransportHandle):
-        raise TypeError("needs to be instance of USBTransportHandle")
+    if not isinstance(device, (USBTransportHandle, IPv4TransportHandle)):
+        raise TypeError(
+            f"needs to be instance of USBTransportHandle or IPv4TransportHandle and not '{type(device)}'"
+        )
     ident = device.identity
     try:
         return _seabreeze_device_instance_registry[ident]
@@ -95,8 +101,9 @@ def add_ipv4_device_location(
         self, device_type: str, ip_address: str, port: int
     ) -> None:
         """add ipv4 device location"""
-        # IPV4Transport.register_device(device_type, ip_address, port)
-        raise NotImplementedError("ipv4 communication not implemented for pyseabreeze")
+        IPv4Transport.register_model(
+            device_type, ipv4_address=ip_address, ipv4_port=port
+        )
 
     def list_devices(self) -> list[_SeaBreezeDevice]:
         """returns available SeaBreezeDevices
@@ -127,6 +134,18 @@ def list_devices(self) -> list[_SeaBreezeDevice]:
                 else:
                     dev.close()
             devices.append(dev)  # type: ignore
+        for ipv4_dev in IPv4Transport.list_devices(**self._kwargs):
+            # get the correct communication interface
+            dev = _seabreeze_device_factory(ipv4_dev)
+            if not dev.is_open:
+                # opening the device will populate its serial number
+                try:
+                    dev.open()
+                except RuntimeError:
+                    raise
+                else:
+                    dev.close()
+            devices.append(dev)  # type: ignore
         return devices
 
     # note: to be fully consistent with cseabreeze this shouldn't be a staticmethod

src/seabreeze/pyseabreeze/devices.py

@@ -20,6 +20,7 @@
 from seabreeze.pyseabreeze.protocol import OBP2Protocol
 from seabreeze.pyseabreeze.protocol import OBPProtocol
 from seabreeze.pyseabreeze.protocol import OOIProtocol
+from seabreeze.pyseabreeze.transport import IPv4Transport
 from seabreeze.pyseabreeze.transport import USBTransport
 from seabreeze.pyseabreeze.types import PySeaBreezeTransport
 from seabreeze.types import SeaBreezeFeatureAccessor
@@ -312,7 +313,11 @@ def __new__(cls: type[DT], raw_device: Any = None) -> SeaBreezeDevice:
             raise SeaBreezeError(
                 "Don't instantiate SeaBreezeDevice directly. Use `SeabreezeAPI.list_devices()`."
             )
-        for transport in {USBTransport}:
+        transports: list[type[PySeaBreezeTransport[Any]]] = [
+            IPv4Transport,
+            USBTransport,
+        ]
+        for transport in transports:
             supported_model = transport.supported_model(raw_device)
             if supported_model is not None:
                 break
@@ -365,7 +370,7 @@ def __repr__(self) -> str:
         return f"<SeaBreezeDevice {self.model}:{self.serial_number}>"
 
     def open(self) -> None:
-        """open the spectrometer usb connection
+        """open the spectrometer connection
 
         Returns
         -------
@@ -1146,13 +1151,17 @@ class HDX(SeaBreezeDevice):
     model_name = "HDX"
 
     # communication config
-    transport = (USBTransport,)
+    transport = (
+        IPv4Transport,
+        USBTransport,
+    )
     usb_vendor_id = 0x2457
     usb_product_id = 0x2003
     usb_endpoint_map = EndPointMap(
         ep_out=0x01, lowspeed_in=0x81, highspeed_in=0x82, highspeed_in2=0x86
     )
     usb_protocol = OBPProtocol
+    ipv4_protocol = OBPProtocol
 
     # spectrometer config
     dark_pixel_indices = DarkPixelIndices.from_ranges()
@@ -1171,6 +1180,7 @@ class HDX(SeaBreezeDevice):
         sbf.spectrometer.SeaBreezeSpectrometerFeatureHDX,
         sbf.rawusb.SeaBreezeRawUSBBusAccessFeature,
         sbf.nonlinearity.NonlinearityCoefficientsFeatureOBP,
+        sbf.multicast.SeaBreezeMulticastFeatureOBP,
     )
 
 

src/seabreeze/pyseabreeze/features/multicast.py

@@ -1,4 +1,8 @@
+import struct
+from typing import List
+
 from seabreeze.pyseabreeze.features._base import SeaBreezeFeature
+from seabreeze.pyseabreeze.protocol import OBPProtocol
 
 
 # Definition
@@ -16,3 +20,41 @@ def set_multicast_enable_state(
         self, interface_index: int, enable_state: bool
     ) -> None:
         raise NotImplementedError("implement in derived class")
+
+    def get_multicast_group_address(self, interface_index: int) -> List[int]:
+        raise NotImplementedError("implement in derived class")
+
+    def get_multicast_group_port(self, interface_index: int) -> int:
+        raise NotImplementedError("implement in derived class")
+
+    def get_multicast_ttl(self, interface_index: int) -> int:
+        raise NotImplementedError("implement in derived class")
+
+
+# OBP implementation
+# ==================
+#
+class SeaBreezeMulticastFeatureOBP(SeaBreezeMulticastFeature):
+    _required_protocol_cls = OBPProtocol
+
+    def get_multicast_enable_state(self, interface_index: int) -> bool:
+        ret = self.protocol.query(0x00000A80, int(interface_index))
+        return bool(struct.unpack("<B", ret))
+
+    def set_multicast_enable_state(
+        self, interface_index: int, enable_state: bool
+    ) -> None:
+        raise NotImplementedError("missing from HDX FW documentation")
+
+    def get_multicast_group_address(self, interface_index: int) -> List[int]:
+        ret = self.protocol.query(0x00000A81, int(interface_index))
+        ip = struct.unpack("<BBBB", ret)
+        return [int(ip[0]), int(ip[1]), int(ip[2]), int(ip[3])]
+
+    def get_multicast_group_port(self, interface_index: int) -> int:
+        ret = self.protocol.query(0x00000A82, int(interface_index))
+        return int(struct.unpack("<H", ret)[0])
+
+    def get_multicast_ttl(self, interface_index: int) -> int:
+        ret = self.protocol.query(0x00000A83, int(interface_index))
+        return int(struct.unpack("<B", ret)[0])

src/seabreeze/pyseabreeze/protocol.py

@@ -229,6 +229,13 @@ class OBPProtocol(PySeaBreezeProtocol):
             0x00420004: "",  # GET_TE_TEMPERATURE
             0x00420010: "<B",  # SET_TE_ENABLE
             0x00420011: "<f",  # SET_TE_SETPOINT
+            0x00000A80: "<B",  # GET_MC_ENABLED
+            0x00000A81: "<B",  # GET_MC_GROUP_ADDR
+            0x00000A82: "<B",  # GET_MC_GROUP_PORT
+            0x00000A83: "<B",  # GET_MC_TTL
+            0xE11: "",  # GET_DEV_STR
+            0xE00: "",  # GET_VID
+            0xE01: "",  # GET_PID
         }.items()
     }  # add more here if you implement new features