feat(api): Update absorbance reader driver namespace for byonoy_devices library version 2024.9.0

api/src/opentrons/drivers/absorbance_reader/abstract.py

@@ -1,6 +1,7 @@
 from abc import ABC, abstractmethod
-from typing import Dict, List, Tuple
+from typing import Dict, List, Optional, Tuple
 from opentrons.drivers.types import (
+    ABSMeasurementMode,
     AbsorbanceReaderLidStatus,
     AbsorbanceReaderDeviceState,
     AbsorbanceReaderPlatePresence,
@@ -32,11 +33,18 @@ async def get_available_wavelengths(self) -> List[int]:
         ...
 
     @abstractmethod
-    async def get_single_measurement(self, wavelength: int) -> List[float]:
+    async def initialize_measurement(
+        self,
+        wavelengths: List[int],
+        mode: ABSMeasurementMode = ABSMeasurementMode.SINGLE,
+        reference_wavelength: Optional[int] = None,
+    ) -> None:
+        """Initialize measurement for the device in single or multi mode for the given wavelengths"""
         ...
 
     @abstractmethod
-    async def initialize_measurement(self, wavelength: int) -> None:
+    async def get_measurement(self) -> List[List[float]]:
+        """Gets an absorbance reading with the current config."""
         ...
 
     @abstractmethod

api/src/opentrons/drivers/absorbance_reader/async_byonoy.py

@@ -3,26 +3,28 @@
 import re
 from concurrent.futures.thread import ThreadPoolExecutor
 from functools import partial
-from typing import Optional, List, Dict, Tuple
+from typing import Any, Optional, List, Dict, Tuple
 
 from .hid_protocol import (
     AbsorbanceHidInterface as AbsProtocol,
     ErrorCodeNames,
     DeviceStateNames,
     SlotStateNames,
+    MeasurementConfig,
 )
 from opentrons.drivers.types import (
     AbsorbanceReaderLidStatus,
     AbsorbanceReaderPlatePresence,
     AbsorbanceReaderDeviceState,
+    ABSMeasurementMode,
 )
 from opentrons.drivers.rpi_drivers.types import USBPort
 from opentrons.hardware_control.modules.errors import AbsorbanceReaderDisconnectedError
 
 
 SN_PARSER = re.compile(r'ATTRS{serial}=="(?P<serial>.+?)"')
 VERSION_PARSER = re.compile(r"Absorbance (?P<version>V\d+\.\d+\.\d+)")
-SERIAL_PARSER = re.compile(r"SN: (?P<serial>BYO[A-Z]{3}[0-9]{5})")
+SERIAL_PARSER = re.compile(r"(?P<serial>BYO[A-Z]{3}[0-9]{5})")
 
 
 class AsyncByonoy:
@@ -72,13 +74,13 @@ async def create(
         loop = loop or asyncio.get_running_loop()
         executor = ThreadPoolExecutor(max_workers=1)
 
-        import pybyonoy_device_library as byonoy  # type: ignore[import-not-found]
+        import byonoy_devices as byonoy  # type: ignore[import-not-found]
 
         interface: AbsProtocol = byonoy
 
         device_sn = cls.serial_number_from_port(usb_port.name)
         found: List[AbsProtocol.Device] = await loop.run_in_executor(
-            executor=executor, func=byonoy.byonoy_available_devices
+            executor=executor, func=byonoy.available_devices
         )
         device = cls.match_device_with_sn(device_sn, found)
 
@@ -110,7 +112,7 @@ def __init__(
         self._loop = loop
         self._supported_wavelengths: Optional[list[int]] = None
         self._device_handle: Optional[int] = None
-        self._current_config: Optional[AbsProtocol.MeasurementConfig] = None
+        self._current_config: Optional[MeasurementConfig] = None
 
     async def open(self) -> bool:
         """
@@ -121,7 +123,7 @@ async def open(self) -> bool:
 
         err, device_handle = await self._loop.run_in_executor(
             executor=self._executor,
-            func=partial(self._interface.byonoy_open_device, self._device),
+            func=partial(self._interface.open_device, self._device),
         )
         self._raise_if_error(err.name, f"Error opening device: {err}")
         self._device_handle = device_handle
@@ -132,7 +134,7 @@ async def close(self) -> None:
         handle = self._verify_device_handle()
         await self._loop.run_in_executor(
             executor=self._executor,
-            func=partial(self._interface.byonoy_free_device, handle),
+            func=partial(self._interface.free_device, handle),
         )
         self._device_handle = None
 
@@ -143,15 +145,15 @@ async def is_open(self) -> bool:
         handle = self._verify_device_handle()
         return await self._loop.run_in_executor(
             executor=self._executor,
-            func=partial(self._interface.byonoy_device_open, handle),
+            func=partial(self._interface.device_open, handle),
         )
 
     async def get_device_information(self) -> Dict[str, str]:
         """Get serial number and version info."""
         handle = self._verify_device_handle()
         err, device_info = await self._loop.run_in_executor(
             executor=self._executor,
-            func=partial(self._interface.byonoy_get_device_information, handle),
+            func=partial(self._interface.get_device_information, handle),
         )
         self._raise_if_error(err.name, f"Error getting device information: {err}")
         serial_match = SERIAL_PARSER.match(device_info.sn)
@@ -170,7 +172,7 @@ async def get_device_status(self) -> AbsorbanceReaderDeviceState:
         handle = self._verify_device_handle()
         err, status = await self._loop.run_in_executor(
             executor=self._executor,
-            func=partial(self._interface.byonoy_get_device_status, handle),
+            func=partial(self._interface.get_device_status, handle),
         )
         self._raise_if_error(err.name, f"Error getting device status: {err}")
         return self.convert_device_state(status.name)
@@ -182,11 +184,9 @@ async def update_firmware(self, firmware_file_path: str) -> Tuple[bool, str]:
             return False, f"Firmware file not found: {firmware_file_path}"
         err = await self._loop.run_in_executor(
             executor=self._executor,
-            func=partial(
-                self._interface.byonoy_update_device, handle, firmware_file_path
-            ),
+            func=partial(self._interface.update_device, handle, firmware_file_path),
         )
-        if err.name != "BYONOY_ERROR_NO_ERROR":
+        if err.name != "NO_ERROR":
             return False, f"Byonoy update failed with error: {err}"
         return True, ""
 
@@ -195,7 +195,7 @@ async def get_device_uptime(self) -> int:
         handle = self._verify_device_handle()
         err, uptime = await self._loop.run_in_executor(
             executor=self._executor,
-            func=partial(self._interface.byonoy_get_device_uptime, handle),
+            func=partial(self._interface.get_device_uptime, handle),
         )
         self._raise_if_error(err.name, "Error getting device uptime: ")
         return uptime
@@ -205,7 +205,7 @@ async def get_lid_status(self) -> AbsorbanceReaderLidStatus:
         handle = self._verify_device_handle()
         err, lid_info = await self._loop.run_in_executor(
             executor=self._executor,
-            func=partial(self._interface.byonoy_get_device_parts_aligned, handle),
+            func=partial(self._interface.get_device_parts_aligned, handle),
         )
         self._raise_if_error(err.name, f"Error getting lid status: {err}")
         return (
@@ -217,79 +217,93 @@ async def get_supported_wavelengths(self) -> list[int]:
         handle = self._verify_device_handle()
         err, wavelengths = await self._loop.run_in_executor(
             executor=self._executor,
-            func=partial(
-                self._interface.byonoy_abs96_get_available_wavelengths, handle
-            ),
+            func=partial(self._interface.abs96_get_available_wavelengths, handle),
         )
         self._raise_if_error(err.name, "Error getting available wavelengths: ")
         self._supported_wavelengths = wavelengths
         return wavelengths
 
-    async def get_single_measurement(self, wavelength: int) -> List[float]:
-        """Get a single measurement based on the current configuration."""
+    async def get_measurement(self) -> List[List[float]]:
+        """Get a measurement based on the current configuration."""
         handle = self._verify_device_handle()
         assert (
-            self._current_config
-            and self._current_config.sample_wavelength == wavelength
-        )
+            self._current_config is not None
+        ), "Cannot get measurement without initializing."
+        measure_func: Any = self._interface.abs96_single_measure
+        if isinstance(self._current_config, AbsProtocol.MultiMeasurementConfig):
+            measure_func = self._interface.abs96_multiple_measure
         err, measurements = await self._loop.run_in_executor(
             executor=self._executor,
             func=partial(
-                self._interface.byonoy_abs96_single_measure,
+                measure_func,
                 handle,
                 self._current_config,
             ),
         )
-        self._raise_if_error(err.name, f"Error getting single measurement: {err}")
-        return measurements
+        self._raise_if_error(err.name, f"Error getting measurement: {err}")
+        return measurements if isinstance(measurements[0], List) else [measurements]  # type: ignore
 
     async def get_plate_presence(self) -> AbsorbanceReaderPlatePresence:
         """Get the state of the plate for the reader."""
         handle = self._verify_device_handle()
         err, presence = await self._loop.run_in_executor(
             executor=self._executor,
-            func=partial(self._interface.byonoy_get_device_slot_status, handle),
+            func=partial(self._interface.get_device_slot_status, handle),
         )
         self._raise_if_error(err.name, f"Error getting slot status: {err}")
         return self.convert_plate_presence(presence.name)
 
     def _get_supported_wavelengths(self) -> List[int]:
         handle = self._verify_device_handle()
         wavelengths: List[int]
-        err, wavelengths = self._interface.byonoy_abs96_get_available_wavelengths(
-            handle
-        )
+        err, wavelengths = self._interface.abs96_get_available_wavelengths(handle)
         self._raise_if_error(err.name, f"Error getting available wavelengths: {err}")
         self._supported_wavelengths = wavelengths
         return wavelengths
 
-    def _initialize_measurement(self, conf: AbsProtocol.MeasurementConfig) -> None:
+    def _initialize_measurement(self, conf: MeasurementConfig) -> None:
         handle = self._verify_device_handle()
-        err = self._interface.byonoy_abs96_initialize_single_measurement(handle, conf)
+        if isinstance(conf, AbsProtocol.SingleMeasurementConfig):
+            err = self._interface.abs96_initialize_single_measurement(handle, conf)
+        else:
+            err = self._interface.abs96_initialize_multiple_measurement(handle, conf)
         self._raise_if_error(err.name, f"Error initializing measurement: {err}")
         self._current_config = conf
 
-    def _set_sample_wavelength(self, wavelength: int) -> AbsProtocol.MeasurementConfig:
+    def _initialize(
+        self,
+        mode: ABSMeasurementMode,
+        wavelengths: List[int],
+        reference_wavelength: Optional[int] = None,
+    ) -> None:
         if not self._supported_wavelengths:
             self._get_supported_wavelengths()
         assert self._supported_wavelengths
-        if wavelength in self._supported_wavelengths:
-            conf = self._interface.ByonoyAbs96SingleMeasurementConfig()
-            conf.sample_wavelength = wavelength
-            return conf
+        conf: MeasurementConfig
+        if set(wavelengths).issubset(self._supported_wavelengths):
+            if mode == ABSMeasurementMode.SINGLE:
+                conf = self._interface.Abs96SingleMeasurementConfig()
+                conf.sample_wavelength = wavelengths[0] or 0
+                conf.reference_wavelength = reference_wavelength or 0
+            else:
+                conf = self._interface.Abs96MultipleMeasurementConfig()
+                conf.sample_wavelengths = wavelengths
         else:
             raise ValueError(
-                f"Unsupported wavelength: {wavelength}, expected: {self._supported_wavelengths}"
+                f"Unsupported wavelength: {wavelengths}, expected: {self._supported_wavelengths}"
             )
-
-    def _initialize(self, wavelength: int) -> None:
-        conf = self._set_sample_wavelength(wavelength)
         self._initialize_measurement(conf)
 
-    async def initialize(self, wavelength: int) -> None:
-        """Initialize the device so we can start reading samples from it."""
+    async def initialize(
+        self,
+        mode: ABSMeasurementMode,
+        wavelengths: List[int],
+        reference_wavelength: Optional[int] = None,
+    ) -> None:
+        """initialize the device so we can start reading samples from it."""
         await self._loop.run_in_executor(
-            executor=self._executor, func=partial(self._initialize, wavelength)
+            executor=self._executor,
+            func=partial(self._initialize, mode, wavelengths, reference_wavelength),
         )
 
     def _verify_device_handle(self) -> int:
@@ -304,12 +318,12 @@ def _raise_if_error(
         msg: str = "Error occurred: ",
     ) -> None:
         if err_name in [
-            "BYONOY_ERROR_DEVICE_CLOSED",
-            "BYONOY_ERROR_DEVICE_COMMUNICATION_FAILURE",
-            "BYONOY_ERROR_UNSUPPORTED_OPERATION",
+            "DEVICE_CLOSED",
+            "DEVICE_COMMUNICATION_FAILURE",
+            "UNSUPPORTED_OPERATION",
         ]:
             raise AbsorbanceReaderDisconnectedError(self._device.sn)
-        if err_name != "BYONOY_ERROR_NO_ERROR":
+        if err_name != "NO_ERROR":
             raise RuntimeError(msg, err_name)
 
     @staticmethod

api/src/opentrons/drivers/absorbance_reader/driver.py

@@ -4,11 +4,14 @@
 from typing import Dict, Optional, List, Tuple, TYPE_CHECKING
 
 from opentrons.drivers.types import (
+    ABSMeasurementMode,
     AbsorbanceReaderLidStatus,
     AbsorbanceReaderDeviceState,
     AbsorbanceReaderPlatePresence,
 )
-from opentrons.drivers.absorbance_reader.abstract import AbstractAbsorbanceReaderDriver
+from opentrons.drivers.absorbance_reader.abstract import (
+    AbstractAbsorbanceReaderDriver,
+)
 from opentrons.drivers.rpi_drivers.types import USBPort
 
 if TYPE_CHECKING:
@@ -60,12 +63,16 @@ async def get_lid_status(self) -> AbsorbanceReaderLidStatus:
     async def get_available_wavelengths(self) -> List[int]:
         return await self._connection.get_supported_wavelengths()
 
-    async def get_single_measurement(self, wavelength: int) -> List[float]:
-        # TODO (cb, 08-02-2024): The list of measurements for 96 wells is rotated 180 degrees (well A1 is where well H12 should be) this must be corrected
-        return await self._connection.get_single_measurement(wavelength)
+    async def initialize_measurement(
+        self,
+        wavelengths: List[int],
+        mode: ABSMeasurementMode = ABSMeasurementMode.SINGLE,
+        reference_wavelength: Optional[int] = None,
+    ) -> None:
+        await self._connection.initialize(mode, wavelengths, reference_wavelength)
 
-    async def initialize_measurement(self, wavelength: int) -> None:
-        await self._connection.initialize(wavelength)
+    async def get_measurement(self) -> List[List[float]]:
+        return await self._connection.get_measurement()
 
     async def get_status(self) -> AbsorbanceReaderDeviceState:
         return await self._connection.get_device_status()