v1 CLARIOstar fluorescence: filter detection, filters by slot, z-profile of a well

pylabrobot/bmg_labtech/clariostar/__init__.py

@@ -13,6 +13,7 @@
   CLARIOstarFluorescenceSpectrumResult,
   CLARIOstarSimulationDriver,
   CLARIOstarStatus,
+  CLARIOstarZProfileResult,
   CommandRefusedError,
   DeviceConfiguration,
   Drawer,

pylabrobot/bmg_labtech/clariostar/driver/__init__.py

@@ -8,6 +8,7 @@
   CLARIOstarAbsorbanceSpectrumResult,
   CLARIOstarFluorescenceResult,
   CLARIOstarFluorescenceSpectrumResult,
+  CLARIOstarZProfileResult,
   Drawer,
   Fluorescence,
   FluorescenceConfiguration,

pylabrobot/bmg_labtech/clariostar/driver/features/__init__.py

@@ -8,6 +8,7 @@
 from .fluorescence import (
   CLARIOstarFluorescenceResult,
   CLARIOstarFluorescenceSpectrumResult,
+  CLARIOstarZProfileResult,
   Fluorescence,
   FluorescenceConfiguration,
 )

pylabrobot/bmg_labtech/clariostar/driver/features/measurement.py

@@ -427,14 +427,12 @@ def _get_plate_and_wells(
   # -- plate block ---------------------------------------------------------------------------------
 
   @staticmethod
-  def _encode_plate_block(plate: Plate, wells: Sequence[Well]) -> bytes:
-    """The 63-byte plate block: geometry in 0.01 mm (u16 BE), columns, rows, `00`, well mask.
+  def _encode_plate_geometry(plate: Plate) -> bytes:
+    """The plate's 14 bytes: size, first and last well centre in 0.01 mm (u16 BE), columns, rows.
 
-    Y is measured from the plate's back edge. The mask bit of a well is row * columns + column,
-    most significant bit first.
+    Y is measured from the plate's back edge.
     """
     items = plate.get_all_items()
-    columns, rows = plate.num_items_x, plate.num_items_y
 
     def center_from_back(well: Well):
       if well.location is None:
@@ -446,15 +444,27 @@ def center_from_back(well: Well):
     last_x, last_y = center_from_back(items[-1])
     geometry = (plate.get_size_x(), plate.get_size_y(), first_x, first_y, last_x, last_y)
     block = b"".join(round(value * 100).to_bytes(2, "big") for value in geometry)
-    block += bytes([columns, rows, 0x00])
+    return block + bytes([plate.num_items_x, plate.num_items_y])
+
+  @staticmethod
+  def _encode_well_mask(plate: Plate, wells: Sequence[Well]) -> bytes:
+    """The 48-byte well mask, a bit per well of `wells`.
 
+    The bit of a well is row * columns + column, most significant bit first.
+    """
+    items = plate.get_all_items()
+    columns, rows = plate.num_items_x, plate.num_items_y
     mask = bytearray(_WELL_MASK_SIZE)
     selected = {id(well) for well in wells}
     for index, well in enumerate(items):  # column-major: A1, B1, ..., A2, ...
       if id(well) in selected:
         bit = (index % rows) * columns + index // rows
         mask[bit // 8] |= 0x80 >> (bit % 8)
-    return block + bytes(mask)
+    return bytes(mask)
+
+  def _encode_plate_block(self, plate: Plate, wells: Sequence[Well]) -> bytes:
+    """The 63-byte plate block: `_encode_plate_geometry`, `00`, then `_encode_well_mask`."""
+    return self._encode_plate_geometry(plate) + b"\x00" + self._encode_well_mask(plate, wells)
 
   # -- scan and settling bytes ---------------------------------------------------------------------
 
@@ -631,6 +641,26 @@ async def _request_data(self, values: Optional[int]) -> List[bytes]:
       if received >= values:
         return replies
 
+  async def _drain_and_stop(self, what: str) -> None:
+    """Drain the reply still streaming, send STOP (`0b 00`) and drop what STOP left.
+
+    A refused STOP only warns.
+
+    Args:
+      what: what is stopped, for the log.
+    """
+    logger.warning("stopping the %s", what)
+    try:
+      # The reader sends a reply it has started to its end: drain it before STOP, and drop
+      # whatever STOP left after it, so the next command reads its own reply.
+      await self._driver._drain_rx()
+      await self._driver.send_command(
+        self._driver.CommandFamily.STOP, parameters=b"\x00", when_idle=False
+      )
+      await self._driver._purge_rx()
+    except (CommandRefusedError, FrameError) as e:  # e.g. the run ended meanwhile (214)
+      logger.warning("STOP not accepted: %s", e)
+
   async def _run_measurement(
     self,
     parameters: bytes,
@@ -692,15 +722,5 @@ async def _run_measurement(
           raise TimeoutError(f"the measurement took longer than {timeout} s")
         await asyncio.sleep(self._driver.default_poll_interval)
     except BaseException:  # a timeout, a cancelled call, a bad reply: stop the run first
-      logger.warning("stopping the measurement run")
-      try:
-        # The reader sends a reply it has started to its end: drain it before STOP, and drop
-        # whatever STOP left after it, so the next command reads its own reply.
-        await self._driver._drain_rx()
-        await self._driver.send_command(
-          self._driver.CommandFamily.STOP, parameters=b"\x00", when_idle=False
-        )
-        await self._driver._purge_rx()
-      except (CommandRefusedError, FrameError) as e:  # e.g. the run ended meanwhile (214)
-        logger.warning("STOP not accepted: %s", e)
+      await self._drain_and_stop("measurement run")
       raise

pylabrobot/bmg_labtech/clariostar/driver/master.py

@@ -112,8 +112,10 @@ class CommandFamily:
     RUN = 0x04
     REQUEST = 0x05
     TEMPERATURE_CONTROLLER = 0x06
+    FOCUS_WELL = 0x09
     STOP = 0x0B
     SET_PARAMETER = 0x0E
+    FILTER_SCAN = 0x24
     STATUS = 0x80
 
   class Command:
@@ -126,17 +128,23 @@ class Command:
     DRAWER_OPEN = 0x01
     # REQUEST
     DATA = 0x02
+    FOCUS_RESULT = 0x05
     PARAMETERS = 0x07
+    SPECTRAL_DATA = 0x11
+    FILTER_RESULT = 0x1B
     # SET_PARAMETER; seen only writing the two-byte test-run counter
     SET_PARAMETER_WORD = 0x0B
 
-  # Families whose payload is the family byte plus parameters, without a command byte.
+  # Families whose payload is the family byte plus parameters, without a command byte; FILTER_SCAN's
+  # first parameter is a mode bit field.
   _NO_COMMAND_FAMILIES = frozenset(
     {
       CommandFamily.RUN,
       CommandFamily.STOP,
       CommandFamily.STATUS,
       CommandFamily.TEMPERATURE_CONTROLLER,
+      CommandFamily.FOCUS_WELL,
+      CommandFamily.FILTER_SCAN,
     }
   )
 

pylabrobot/lib/fluorescence/channels.py

@@ -5,7 +5,7 @@
 """
 
 from dataclasses import dataclass
-from typing import List, Optional
+from typing import List, Optional, Union
 
 from pylabrobot.lib.optics.filters import DichroicFilter, OpticalFilter
 
@@ -14,7 +14,7 @@ def _check_excitation_or_emission(
   excitation_or_emission: str,
   wavelength: Optional[float],
   bandwidth: Optional[float],
-  optical_filter: Optional[OpticalFilter],
+  optical_filter: Optional[Union[OpticalFilter, int, str]],
 ) -> List[str]:
   """The violations of excitation or emission.
 
@@ -34,7 +34,7 @@ def _check_excitation_or_emission(
 
 
 def _check_dichroic(
-  dichroic_filter: Optional[DichroicFilter], split_wavelength: Optional[float]
+  dichroic_filter: Optional[Union[DichroicFilter, int, str]], split_wavelength: Optional[float]
 ) -> List[str]:
   """The violations of the dichroic: a fixed dichroic with a tunable split, a split not positive."""
   errors = []
@@ -60,23 +60,26 @@ class FluorescenceChannel:
   Attributes:
     excitation_wavelength: nm through a monochromator or a tunable source; None with a filter.
     excitation_bandwidth: nm about `excitation_wavelength`; None with a filter or the default.
-    excitation_filter: the excitation filter instead of a wavelength.
+    excitation_filter: the excitation filter instead of a wavelength: a filter, or its slot or name
+      on the device's slides.
     dichroic_split_wavelength: nm where a tunable dichroic splits excitation from emission; None
       otherwise.
-    dichroic_filter: a fixed dichroic instead of a split.
+    dichroic_filter: a fixed dichroic instead of a split: a filter, or its slot or name on the
+      device's slides.
     emission_wavelength: nm through a monochromator; None with a filter.
     emission_bandwidth: nm about `emission_wavelength`; None with a filter or the default.
-    emission_filter: the emission filter instead of a wavelength.
+    emission_filter: the emission filter instead of a wavelength: a filter, or its slot or name on
+      the device's slides.
   """
 
   excitation_wavelength: Optional[float] = None
   excitation_bandwidth: Optional[float] = None
-  excitation_filter: Optional[OpticalFilter] = None
+  excitation_filter: Optional[Union[OpticalFilter, int, str]] = None
   dichroic_split_wavelength: Optional[float] = None
-  dichroic_filter: Optional[DichroicFilter] = None
+  dichroic_filter: Optional[Union[DichroicFilter, int, str]] = None
   emission_wavelength: Optional[float] = None
   emission_bandwidth: Optional[float] = None
-  emission_filter: Optional[OpticalFilter] = None
+  emission_filter: Optional[Union[OpticalFilter, int, str]] = None
 
   def __post_init__(self) -> None:
     """Raise listing every component of the wrong combination or sign.

pylabrobot/lib/optics/holders.py

@@ -9,7 +9,7 @@
 F = TypeVar("F", bound=_FilterBase)
 
 
-class _FilterSlideBase(Generic[F]):
+class FilterSlide(Generic[F]):
   """The filters a device holds on one side, by slot.
 
   Attributes:
@@ -74,21 +74,21 @@ def __repr__(self) -> str:
     return f"{type(self).__name__}({self.maximum_slots} slots, {{{', '.join(held)}}})"
 
 
-class ExcitationFilterSlide(_FilterSlideBase[OpticalFilter]):
+class ExcitationFilterSlide(FilterSlide[OpticalFilter]):
   """The excitation filters a device holds, by slot."""
 
   _CATEGORY = "excitation"
   _FILTER_CLASS = OpticalFilter
 
 
-class EmissionFilterSlide(_FilterSlideBase[OpticalFilter]):
+class EmissionFilterSlide(FilterSlide[OpticalFilter]):
   """The emission filters a device holds, by slot."""
 
   _CATEGORY = "emission"
   _FILTER_CLASS = OpticalFilter
 
 
-class DichroicFilterSlide(_FilterSlideBase[DichroicFilter]):
+class DichroicFilterSlide(FilterSlide[DichroicFilter]):
   """The dichroic filters a device holds, by slot."""
 
   _CATEGORY = "dichroic"