v1 CLARIOstar: simulated fluorescence and luminescence reads

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

@@ -1469,6 +1469,61 @@ def _get_results(
 
   # -- measuring -----------------------------------------------------------------------------------
 
+  def _get_run_parameters(
+    self,
+    fluorescence_channels: Sequence[FluorescenceChannel],
+    gains: Sequence[int],
+    focal_height: Optional[float],
+    measurement: MeasurementSettings,
+    shake: Optional[ShakeSettings],
+    wells: Optional[List[Well]],
+    *,
+    timeout: Optional[float],
+    optic_position: OpticPosition,
+    flying_mode: bool,
+    enhanced_dynamic_range: bool,
+  ) -> Tuple[Plate, List[Well], bytes]:
+    """The plate, the wells and the RUN parameters of a read, every argument checked.
+
+    Raises:
+      RuntimeError: if no plate is in the plate holder.
+      ValueError: if an argument is out of range (`_check_arguments`), a matrix scan has several
+        wells, or the settings cannot read the wells of the plate (`_get_plate_and_wells`).
+    """
+    self._check_arguments(
+      fluorescence_channels,
+      gains,
+      focal_height,
+      measurement,
+      timeout=timeout,
+      optic_position=optic_position,
+      flying_mode=flying_mode,
+      enhanced_dynamic_range=enhanced_dynamic_range,
+    )
+    plate, measured = self._get_plate_and_wells(measurement, shake, wells)
+    if focal_height is None:
+      focal_height = self._get_default_focal_height(measured)
+    if measurement.well_scan == "matrix" and len(measured) > 1:
+      # No fluorescence matrix of several wells is verified: its reply layout is unknown.
+      raise ValueError(f"a fluorescence matrix scan reads one well, got {len(measured)}")
+    parameters = self._compute_run_parameters(
+      plate,
+      measured,
+      measurement,
+      shake,
+      focal_height=focal_height,
+      kind=self._DISCRETE,
+      count=len(fluorescence_channels),
+      encoded_fluorescence_channels=[
+        self._encode_fluorescence_channel(fluorescence_channel, gain)
+        for fluorescence_channel, gain in zip(fluorescence_channels, gains)
+      ],
+      optics=self._OPTIC_POSITIONS[optic_position],
+      flying_mode=flying_mode,
+      enhanced_dynamic_range=enhanced_dynamic_range,
+    )
+    return plate, measured, parameters
+
   def _get_fluorescence_channels_and_gains(
     self,
     fluorescence_channels: Union[FluorescenceChannel, Sequence[FluorescenceChannel]],
@@ -1579,38 +1634,18 @@ async def measure(
     if optic_position is None:
       optic_position = self._default_optic_position
     filled, gains = self._get_fluorescence_channels_and_gains(fluorescence_channels, gain)
-    self._check_arguments(
+    plate, wells, parameters = self._get_run_parameters(
       filled,
       gains,
       focal_height,
       measurement,
+      shake_settings,
+      wells,
       timeout=timeout,
       optic_position=optic_position,
       flying_mode=flying_mode,
       enhanced_dynamic_range=enhanced_dynamic_range,
     )
-    plate, wells = self._get_plate_and_wells(measurement, shake_settings, wells)
-    if focal_height is None:
-      focal_height = self._get_default_focal_height(wells)
-    if measurement.well_scan == "matrix" and len(wells) > 1:
-      # No fluorescence matrix of several wells is verified: its reply layout is unknown.
-      raise ValueError(f"a fluorescence matrix scan reads one well, got {len(wells)}")
-    parameters = self._compute_run_parameters(
-      plate,
-      wells,
-      measurement,
-      shake_settings,
-      focal_height=focal_height,
-      kind=self._DISCRETE,
-      count=len(filled),
-      encoded_fluorescence_channels=[
-        self._encode_fluorescence_channel(fluorescence_channel, gain)
-        for fluorescence_channel, gain in zip(filled, gains)
-      ],
-      optics=self._OPTIC_POSITIONS[optic_position],
-      flying_mode=flying_mode,
-      enhanced_dynamic_range=enhanced_dynamic_range,
-    )
     timeout = self.default_read_timeout if timeout is None else timeout
     logger.info(
       "fluorescence: %s, %d of %d wells of %s, %s scan, %s optic",

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

@@ -14,6 +14,7 @@
 from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
 from unittest.mock import patch
 
+from pylabrobot.bmg_labtech.clariostar.device import CLARIOstarPlus
 from pylabrobot.bmg_labtech.clariostar.driver.configuration_tests import decoded
 from pylabrobot.bmg_labtech.clariostar.driver.features import fluorescence
 from pylabrobot.bmg_labtech.clariostar.driver.features import measurement as measurement_module
@@ -43,6 +44,7 @@
   load_session_replies,
   make_driver,
 )
+from pylabrobot.bmg_labtech.clariostar.driver.simulator import CLARIOstarSimulationFluorescence
 from pylabrobot.lib.fluorescence.channels import FluorescenceChannel
 from pylabrobot.lib.fluorescence.results import FluorescenceSpectrumResult
 from pylabrobot.lib.optics.filters import DichroicFilter, OpticalFilter
@@ -1642,5 +1644,86 @@ async def test_out_of_range_z_scans_send_nothing(self):
     self.assertEqual(io.written, [])
 
 
+async def simulated_reader():
+  """A set-up simulated CLARIOstar Plus with a Corning 96 plate in its drawer; the fluorescence
+  and the plate."""
+  reader = CLARIOstarPlus(simulation=True)
+  await reader.setup()
+  plate = cor_96_wellplate_360uL_Fb("plate")
+  await reader.drawer.open()
+  reader.plate_holder.assign_child_resource(plate)
+  await reader.drawer.close()
+  feature = reader.fluorescence
+  assert isinstance(feature, CLARIOstarSimulationFluorescence)
+  return feature, plate
+
+
+class SimulatedFluorescenceTests(unittest.IsolatedAsyncioTestCase):
+  async def asyncSetUp(self):
+    self.fluorescence, self.plate = await simulated_reader()
+    self.column_1 = self.plate.get_items("A1:H1")
+    self.identifiers = [f"{row}1" for row in "ABCDEFGH"]
+
+  async def test_a_read_is_checked_and_reads_within_the_detectors_range(self):
+    mcherry = replace(BASELINE, excitation_wavelength=570, emission_wavelength=620)
+    fitc_result, mcherry_result = await self.fluorescence.measure(
+      [BASELINE, mcherry], gain=[1000, 1500], wells=self.column_1
+    )
+    self.assertEqual((fitc_result.gain, mcherry_result.gain), (1000, 1500))
+    self.assertEqual(mcherry_result.emission_wavelength, 620)
+    for result in (fitc_result, mcherry_result):
+      self.assertEqual(list(result.data), self.identifiers)
+      self.assertTrue(all(0 <= value <= 260_000 for value in result.data.values()))
+    bright = await self.fluorescence.measure(BASELINE, gain=1000, enhanced_dynamic_range=True)
+    self.assertTrue(all(0 <= value <= 700_000_000 for value in bright.data.values()))
+    for refused in (
+      dict(flying_mode=True, enhanced_dynamic_range=True),  # `_check_arguments`
+      dict(focal_height=25.1),
+      dict(
+        wells=self.column_1,
+        measurement_settings=MeasurementSettings(well_scan="matrix", matrix_size=3),
+      ),
+    ):
+      with self.subTest(refused=refused), self.assertRaises(ValueError):
+        await self.fluorescence.measure(BASELINE, gain=1000, **refused)
+
+  async def test_a_matrix_reads_every_point_and_their_mean(self):
+    result = await self.fluorescence.measure(
+      BASELINE,
+      gain=1000,
+      wells=[self.plate.get_well("A1")],
+      measurement_settings=MeasurementSettings(well_scan="matrix", matrix_size=3),
+    )
+    assert result.matrix_points is not None
+    points = result.matrix_points["A1"]
+    self.assertEqual(len(points), 9)
+    self.assertEqual(result.data["A1"], sum(points) / 9)
+
+  async def test_earlier_results_are_returned_again(self):
+    earlier = await self.fluorescence.measure(BASELINE, gain=1000)
+    replayed = await self.fluorescence.measure(
+      BASELINE, gain=1000, wells=self.column_1, simulated_data=[earlier]
+    )
+    self.assertEqual(replayed.data, {well: earlier.data[well] for well in self.identifiers})
+    other = replace(BASELINE, emission_wavelength=600)
+    with self.assertRaises(ValueError):  # another fluorescence channel's result
+      await self.fluorescence.measure(other, gain=1000, simulated_data=[earlier])
+
+  async def test_spectra_filter_detection_and_z_profiles_are_not_simulated_yet(self):
+    for call in (
+      self.fluorescence.measure_spectrum(
+        scan="emission",
+        start_wavelength=510,
+        end_wavelength=600,
+        excitation_wavelength=485,
+        gain=1000,
+      ),
+      self.fluorescence.detect_all_filters(),
+      self.fluorescence.measure_z_profile(BASELINE, well=self.plate.get_well("A1")),
+    ):
+      with self.assertRaises(NotImplementedError):
+        await call
+
+
 if __name__ == "__main__":
   unittest.main()

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

@@ -269,6 +269,54 @@ def _decode_data_reply(self, reply: bytes, wells: int) -> Tuple[List[int], int,
 
   # -- measuring -----------------------------------------------------------------------------------
 
+  def _get_run_parameters(
+    self,
+    emission_wavelength: int,
+    emission_bandwidth: int,
+    gain: int,
+    focal_height: Optional[float],
+    normalization_time: float,
+    measurement_interval: float,
+    measurement: MeasurementSettings,
+    shake: Optional[ShakeSettings],
+    wells: Optional[List[Well]],
+    *,
+    timeout: Optional[float],
+  ) -> Tuple[Plate, List[Well], bytes]:
+    """The plate, the wells and the RUN parameters of a read, every argument checked.
+
+    Raises:
+      RuntimeError: if no plate is in the plate holder.
+      ValueError: if an argument is out of range (`_check_arguments`), or the settings cannot
+        read the wells of the plate (`_get_plate_and_wells`).
+    """
+    self._check_arguments(
+      emission_wavelength,
+      emission_bandwidth,
+      gain,
+      focal_height,
+      normalization_time,
+      measurement_interval,
+      measurement,
+      timeout=timeout,
+    )
+    plate, measured = self._get_plate_and_wells(measurement, shake, wells)
+    if focal_height is None:
+      focal_height = self._get_default_focal_height(measured)
+    parameters = self._compute_run_parameters(
+      plate,
+      measured,
+      measurement,
+      shake,
+      emission_wavelength=emission_wavelength,
+      emission_bandwidth=emission_bandwidth,
+      gain=gain,
+      focal_height=focal_height,
+      normalization_time=normalization_time,
+      measurement_interval=measurement_interval,
+    )
+    return plate, measured, parameters
+
   async def measure(
     self,
     *,
@@ -322,30 +370,17 @@ async def measure(
       if measurement_settings is None
       else measurement_settings
     )
-    self._check_arguments(
+    plate, wells, parameters = self._get_run_parameters(
       emission_wavelength,
       emission_bandwidth,
       gain,
       focal_height,
       normalization_time,
       measurement_interval,
       measurement,
-      timeout=timeout,
-    )
-    plate, wells = self._get_plate_and_wells(measurement, shake_settings, wells)
-    if focal_height is None:
-      focal_height = self._get_default_focal_height(wells)
-    parameters = self._compute_run_parameters(
-      plate,
-      wells,
-      measurement,
       shake_settings,
-      emission_wavelength=emission_wavelength,
-      emission_bandwidth=emission_bandwidth,
-      gain=gain,
-      focal_height=focal_height,
-      normalization_time=normalization_time,
-      measurement_interval=measurement_interval,
+      wells,
+      timeout=timeout,
     )
     timeout = self.default_read_timeout if timeout is None else timeout
     # TODO(device): no luminescent sample has been read yet, only an empty plate.

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

@@ -9,6 +9,7 @@
 import unittest
 from typing import Any, Dict, List
 
+from pylabrobot.bmg_labtech.clariostar.device import CLARIOstarPlus
 from pylabrobot.bmg_labtech.clariostar.driver.configuration_tests import decoded
 from pylabrobot.bmg_labtech.clariostar.driver.features.absorbance_tests import (
   RUN_REPLY,
@@ -30,6 +31,7 @@
   FakeFTDI,
   make_driver,
 )
+from pylabrobot.bmg_labtech.clariostar.driver.simulator import CLARIOstarSimulationLuminescence
 from pylabrobot.resources import Plate, cor_96_wellplate_360uL_Fb
 
 # The ground truth: every well, 580-80 nm, gain 3500, focal height 8.5 mm, settling 0.0 s,
@@ -231,3 +233,32 @@ def test_luminescence_is_built_only_when_reported(self):
       driver.configuration.luminescence_channels = channels
       driver._discover_features()
       self.assertEqual(isinstance(driver.luminescence, Luminescence), built)
+
+
+class SimulatedLuminescenceTests(unittest.IsolatedAsyncioTestCase):
+  async def asyncSetUp(self):
+    reader = CLARIOstarPlus(simulation=True)
+    await reader.setup()
+    self.plate = cor_96_wellplate_360uL_Fb("plate")
+    await reader.drawer.open()
+    reader.plate_holder.assign_child_resource(self.plate)
+    await reader.drawer.close()
+    luminescence = reader.luminescence
+    assert isinstance(luminescence, CLARIOstarSimulationLuminescence)
+    self.luminescence = luminescence
+
+  async def test_a_read_is_checked_and_reads_within_the_detectors_range(self):
+    result = await self.luminescence.measure(wells=self.plate.get_items("A1:H1"), gain=2000)
+    self.assertEqual(list(result.data), [f"{row}1" for row in "ABCDEFGH"])
+    self.assertTrue(all(0 <= value <= 210_000_000 for value in result.data.values()))
+    self.assertEqual((result.emission_wavelength, result.gain), (580, 2000))
+    with self.assertRaises(ValueError):  # `_check_arguments`
+      await self.luminescence.measure(measurement_interval=0.805)
+
+  async def test_earlier_results_are_returned_again(self):
+    earlier = await self.luminescence.measure()
+    column_1 = self.plate.get_items("A1:H1")
+    replayed = await self.luminescence.measure(wells=column_1, simulated_data=earlier)
+    self.assertEqual(replayed.data, {f"{row}1": earlier.data[f"{row}1"] for row in "ABCDEFGH"})
+    with self.assertRaises(ValueError):  # another emission band
+      await self.luminescence.measure(emission_wavelength=560, simulated_data=earlier)