diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py index d449859c69c..f179da5e340 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py +++ b/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", diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py index c1ca2923805..d6313317015 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py +++ b/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() diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence.py b/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence.py index be2d710b9bc..6c5f7090888 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence.py +++ b/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,7 +370,7 @@ 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, @@ -330,22 +378,9 @@ async def measure( 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. diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence_tests.py index 80bef859bfa..ad4324bf46d 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence_tests.py +++ b/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) diff --git a/pylabrobot/bmg_labtech/clariostar/driver/simulator.py b/pylabrobot/bmg_labtech/clariostar/driver/simulator.py index 8f7d4e9345f..17f658ff28e 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/simulator.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/simulator.py @@ -4,11 +4,31 @@ import os import random from datetime import datetime -from typing import Callable, Dict, List, Literal, Optional, Sequence, TypeVar, Union +from typing import ( + Any, + Callable, + Dict, + List, + Literal, + Optional, + Sequence, + Tuple, + TypeVar, + Union, + overload, +) from pylabrobot.io.ftdi import FTDI from pylabrobot.io.usb import USB from pylabrobot.lib.absorbance.results import AbsorbanceResult, AbsorbanceSpectrumResult +from pylabrobot.lib.fluorescence.channels import FluorescenceChannel +from pylabrobot.lib.fluorescence.results import FluorescenceResult +from pylabrobot.lib.luminescence.results import LuminescenceResult +from pylabrobot.lib.optics.holders import ( + DichroicFilterSlide, + EmissionFilterSlide, + ExcitationFilterSlide, +) from pylabrobot.resources import Plate, Well from .configuration import configuration_from_json, read_configuration @@ -20,6 +40,14 @@ _get_spectrum_wavelengths, ) from .features.drawer import Drawer +from .features.fluorescence import ( + CLARIOstarFluorescenceResult, + CLARIOstarFluorescenceSpectrumResult, + CLARIOstarZProfileResult, + Fluorescence, + OpticPosition, +) +from .features.luminescence import CLARIOstarLuminescenceResult, Luminescence from .features.measurement import ( MeasurementSettings, ShakeSettings, @@ -39,6 +67,11 @@ # The manufacturer's absorbance range and resolution, in OD. _OPTICAL_DENSITY_RANGE = (0.0, 4.0) _OPTICAL_DENSITY_RESOLUTION = 3 # decimals: 0.001 OD +# The detector's ceilings the reader states in its replies: fluorescence, fluorescence with +# enhanced dynamic range, luminescence. +_FLUORESCENCE_CEILING = 260_000 +_ENHANCED_DYNAMIC_RANGE_CEILING = 700_000_000 +_LUMINESCENCE_CEILING = 210_000_000 def _require_read(value: Optional[T], name: str) -> T: @@ -48,6 +81,21 @@ def _require_read(value: Optional[T], name: str) -> T: return value +def _get_measured_values( + plate: Plate, wells: List[Well], values: Dict[str, float], what: str +) -> Dict[str, float]: + """The values of the measured wells in `values`, keyed by well identifier, row by row. + + Raises: + ValueError: if `values` has no value for a measured well. + """ + identifiers = _get_measured_identifiers(plate, wells) + missing = [identifier for identifier in identifiers if identifier not in values] + if missing: + raise ValueError(f"{what} has no value for {missing}") + return {identifier: float(values[identifier]) for identifier in identifiers} + + class CLARIOstarSimulationDrawer(Drawer): """The drawer of a simulated reader: a move puts the plate holder where the drawer goes at once. @@ -119,21 +167,6 @@ class CLARIOstarSimulationAbsorbance(Absorbance): """Absorbance on a simulated reader: the read is checked and encoded, and every well reads a random optical density within the reader's range, or its value in `simulated_data`.""" - @staticmethod - def _get_measured_values( - plate: Plate, wells: List[Well], values: Dict[str, float], what: str - ) -> Dict[str, float]: - """The values of the measured wells in `values`, keyed by well identifier, row by row. - - Raises: - ValueError: if `values` has no value for a measured well. - """ - identifiers = _get_measured_identifiers(plate, wells) - missing = [identifier for identifier in identifiers if identifier not in values] - if missing: - raise ValueError(f"{what} has no value for {missing}") - return {identifier: float(values[identifier]) for identifier in identifiers} - async def measure( self, *, @@ -209,13 +242,11 @@ async def measure( logger.debug("%s read: simulation: simulated_data", SIMULATED_LINK) for result in simulated_data: what = f"simulated_data at {result.wavelength} nm" - data = self._get_measured_values(plate, wells, result.data, what) + data = _get_measured_values(plate, wells, result.data, what) if isinstance(result, CLARIOstarAbsorbanceResult): if result.references is None: raise ValueError(f"{what} is raw but holds no references") - references = self._get_measured_values( - plate, wells, result.references, what + " references" - ) + references = _get_measured_values(plate, wells, result.references, what + " references") results.append( CLARIOstarAbsorbanceResult( data=data, @@ -332,7 +363,7 @@ async def measure_spectrum( if isinstance(simulated_data, CLARIOstarAbsorbanceSpectrumResult): if simulated_data.references is None or simulated_data.calibration is None: raise ValueError("simulated_data is raw but holds no references or calibration") - references = self._get_measured_values( + references = _get_measured_values( plate, wells, simulated_data.references, "simulated_data references" ) return CLARIOstarAbsorbanceSpectrumResult( @@ -351,6 +382,273 @@ async def measure_spectrum( ) +class CLARIOstarSimulationFluorescence(Fluorescence): + """Fluorescence on a simulated reader: each read is checked and encoded, nothing is sent. + + Every well reads random counts within the detector's range, or its counts in `simulated_data`. + Spectra, filter detection and z-profiles are not simulated yet. + """ + + def _get_simulated_counts( + self, + plate: Plate, + wells: List[Well], + fluorescence_channels: Sequence[FluorescenceChannel], + simulated_data: Sequence[FluorescenceResult], + points: int, + ) -> List[List[List[int]]]: + """Each fluorescence channel's counts of the measured wells in `simulated_data`, by point. + + Raises: + ValueError: if `simulated_data` is not one result per fluorescence channel with its + optics, with every measured well, and for a matrix scan its points. + """ + if len(simulated_data) != len(fluorescence_channels): + raise ValueError( + f"simulated_data holds {len(simulated_data)} results for " + f"{len(fluorescence_channels)} fluorescence channels" + ) + identifiers = _get_measured_identifiers(plate, wells) + counts = [] + for index, (fluorescence_channel, result) in enumerate( + zip(fluorescence_channels, simulated_data) + ): + what = f"simulated_data[{index}]" + optics = self._get_result_optics(fluorescence_channel) + given = {name: getattr(result, name) for name in optics} + if given != optics: + raise ValueError(f"{what} holds {given}, measuring {optics}") + if points > 1: + matrix = result.matrix_points if isinstance(result, CLARIOstarFluorescenceResult) else None + if matrix is None or any(len(matrix.get(well, [])) != points for well in identifiers): + raise ValueError(f"{what} has no {points} matrix points for every measured well") + counts.append([[round(value) for value in matrix[well]] for well in identifiers]) + else: + values = _get_measured_values(plate, wells, result.data, what) + counts.append([[round(values[well])] for well in identifiers]) + return counts + + @overload + async def measure( + self, + fluorescence_channels: FluorescenceChannel, + *, + gain: int, + wells: Optional[List[Well]] = ..., + focal_height: Optional[float] = ..., + optic_position: Optional[OpticPosition] = ..., + flying_mode: bool = ..., + enhanced_dynamic_range: bool = ..., + measurement_settings: Optional[MeasurementSettings] = ..., + shake_settings: Optional[ShakeSettings] = ..., + timeout: Optional[float] = ..., + simulated_data: Optional[Sequence[FluorescenceResult]] = ..., + ) -> CLARIOstarFluorescenceResult: ... + + @overload + async def measure( + self, + fluorescence_channels: Sequence[FluorescenceChannel], + *, + gain: Union[int, Sequence[int]], + wells: Optional[List[Well]] = ..., + focal_height: Optional[float] = ..., + optic_position: Optional[OpticPosition] = ..., + flying_mode: bool = ..., + enhanced_dynamic_range: bool = ..., + measurement_settings: Optional[MeasurementSettings] = ..., + shake_settings: Optional[ShakeSettings] = ..., + timeout: Optional[float] = ..., + simulated_data: Optional[Sequence[FluorescenceResult]] = ..., + ) -> List[CLARIOstarFluorescenceResult]: ... + + async def measure( + self, + fluorescence_channels: Union[FluorescenceChannel, Sequence[FluorescenceChannel]], + *, + gain: Union[int, Sequence[int]], + wells: Optional[List[Well]] = None, + focal_height: Optional[float] = None, + optic_position: Optional[OpticPosition] = None, + flying_mode: bool = False, + enhanced_dynamic_range: bool = False, + measurement_settings: Optional[MeasurementSettings] = None, + shake_settings: Optional[ShakeSettings] = None, + timeout: Optional[float] = None, + simulated_data: Optional[Sequence[FluorescenceResult]] = None, + ) -> Union[CLARIOstarFluorescenceResult, List[CLARIOstarFluorescenceResult]]: + """Check and encode the read; every measured well reads random counts. + + The temperature is the temperature controller's target, as the simulated status reports it. + + Args: + fluorescence_channels, gain, wells, focal_height, optic_position, flying_mode, + enhanced_dynamic_range, measurement_settings, shake_settings, timeout: as + `Fluorescence.measure`. + simulated_data: results an earlier `measure` returned, one per fluorescence channel with + its optics, on a reader or in simulation; each measured well reads its counts there (for + a matrix scan, its points) instead of random ones. + + Raises: + RuntimeError: if no plate is in the plate holder. + ValueError: as `Fluorescence.measure` for the arguments, or if `simulated_data` does not + match the read (`_get_simulated_counts`). + """ + measurement = MeasurementSettings() if measurement_settings is None else measurement_settings + if optic_position is None: + optic_position = self._default_optic_position + filled, gains = self._get_fluorescence_channels_and_gains(fluorescence_channels, gain) + plate, wells, _ = 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, + ) + matrix = measurement.well_scan == "matrix" + points = measurement.matrix_size**2 if matrix and measurement.matrix_size else 1 + identifiers = _get_measured_identifiers(plate, wells) + logger.info( + "fluorescence: %d fluorescence channels, %d of %d wells of %s, %s scan, %s optic", + len(filled), + len(identifiers), + plate.num_items, + plate.name, + measurement.well_scan, + optic_position, + ) + temperature = (await self._driver.request_status()).temperature_bottom + timestamp = datetime.now() + if simulated_data is None: + logger.debug("%s read: simulation: random counts", SIMULATED_LINK) + ceiling = _ENHANCED_DYNAMIC_RANGE_CEILING if enhanced_dynamic_range else _FLUORESCENCE_CEILING + counts = [ + [[random.randint(0, ceiling) for _ in range(points)] for _ in identifiers] for _ in filled + ] + else: + logger.debug("%s read: simulation: simulated_data", SIMULATED_LINK) + counts = self._get_simulated_counts(plate, wells, filled, simulated_data, points) + results = self._get_results( + filled, + gains, + counts, + identifiers, + matrix=matrix, + temperature=temperature, + start_timestamp=timestamp, + end_timestamp=timestamp, + ) + return results[0] if isinstance(fluorescence_channels, FluorescenceChannel) else results + + async def measure_spectrum( + self, *args: Any, **kwargs: Any + ) -> CLARIOstarFluorescenceSpectrumResult: + """Raise: a fluorescence spectrum is not simulated yet.""" + raise NotImplementedError("a fluorescence spectrum is not simulated yet") + + async def detect_all_filters( + self, *args: Any, **kwargs: Any + ) -> Tuple[ExcitationFilterSlide, DichroicFilterSlide, EmissionFilterSlide]: + """Raise: filter detection is not simulated yet.""" + raise NotImplementedError("filter detection is not simulated yet") + + async def measure_z_profile(self, *args: Any, **kwargs: Any) -> CLARIOstarZProfileResult: + """Raise: a z-profile is not simulated yet.""" + raise NotImplementedError("a z-profile is not simulated yet") + + +class CLARIOstarSimulationLuminescence(Luminescence): + """Luminescence on a simulated reader: each read is checked and encoded, nothing is sent; every + well reads random counts within the detector's range, or its counts in `simulated_data`.""" + + async def measure( + self, + *, + wells: Optional[List[Well]] = None, + emission_wavelength: int = 580, + emission_bandwidth: int = 80, + gain: int = 3500, + focal_height: Optional[float] = None, + normalization_time: float = 1.0, + measurement_interval: float = 0.8, + measurement_settings: Optional[MeasurementSettings] = None, + shake_settings: Optional[ShakeSettings] = None, + timeout: Optional[float] = None, + simulated_data: Optional[LuminescenceResult] = None, + ) -> CLARIOstarLuminescenceResult: + """Check and encode the read; every measured well reads random counts. + + Args: + wells, emission_wavelength, emission_bandwidth, gain, focal_height, normalization_time, + measurement_interval, measurement_settings, shake_settings, timeout: as + `Luminescence.measure`. + simulated_data: a result an earlier `measure` returned with the same emission band, on a + reader or in simulation; each measured well reads its counts there instead. + + Raises: + RuntimeError: if no plate is in the plate holder. + ValueError: as `Luminescence.measure` for the arguments, or if `simulated_data` is of + another emission band or has no value for a measured well. + """ + measurement = ( + MeasurementSettings(settling_time=0.0) + if measurement_settings is None + else measurement_settings + ) + plate, wells, _ = self._get_run_parameters( + emission_wavelength, + emission_bandwidth, + gain, + focal_height, + normalization_time, + measurement_interval, + measurement, + shake_settings, + wells, + timeout=timeout, + ) + identifiers = _get_measured_identifiers(plate, wells) + logger.info( + "luminescence: %d/%d nm, gain %d, %d of %d wells of %s", + emission_wavelength, + emission_bandwidth, + gain, + len(identifiers), + plate.num_items, + plate.name, + ) + temperature = (await self._driver.request_status()).temperature_bottom + timestamp = datetime.now() + if simulated_data is None: + logger.debug("%s read: simulation: random counts", SIMULATED_LINK) + data = { + identifier: float(random.randint(0, _LUMINESCENCE_CEILING)) for identifier in identifiers + } + else: + band = (simulated_data.emission_wavelength, simulated_data.emission_bandwidth) + if band != (emission_wavelength, emission_bandwidth): + raise ValueError( + f"simulated_data is of {band[0]}/{band[1]} nm, measuring " + f"{emission_wavelength}/{emission_bandwidth} nm" + ) + logger.debug("%s read: simulation: simulated_data", SIMULATED_LINK) + data = _get_measured_values(plate, wells, simulated_data.data, "simulated_data") + return CLARIOstarLuminescenceResult( + data=data, + emission_wavelength=emission_wavelength, + emission_bandwidth=emission_bandwidth, + gain=gain, + temperature=temperature, + start_timestamp=timestamp, + end_timestamp=timestamp, + ) + + class CLARIOstarSimulationDriver(CLARIOstarDriver): """A CLARIOstar driver with no reader behind it. @@ -482,6 +780,10 @@ def _discover_features(self) -> None: self.temperature_controller = CLARIOstarSimulationTemperatureController(self) if self.configuration.absorbance_available and self.absorbance is None: self.absorbance = CLARIOstarSimulationAbsorbance(self) + if self.configuration.fluorescence_channels and self.fluorescence is None: + self.fluorescence = CLARIOstarSimulationFluorescence(self) + if self.configuration.luminescence_channels and self.luminescence is None: + self.luminescence = CLARIOstarSimulationLuminescence(self) def _describe_link(self) -> str: """The simulated link, and the configuration it answers from."""