diff --git a/pylabrobot/bmg_labtech/clariostar/__init__.py b/pylabrobot/bmg_labtech/clariostar/__init__.py index 7a4b14aeda5..38f84198aea 100644 --- a/pylabrobot/bmg_labtech/clariostar/__init__.py +++ b/pylabrobot/bmg_labtech/clariostar/__init__.py @@ -11,6 +11,7 @@ CLARIOstarDriver, CLARIOstarFluorescenceResult, CLARIOstarFluorescenceSpectrumResult, + CLARIOstarLuminescenceResult, CLARIOstarSimulationDriver, CLARIOstarStatus, CLARIOstarZProfileResult, @@ -20,6 +21,8 @@ Fluorescence, FluorescenceConfiguration, FrameError, + Luminescence, + LuminescenceConfiguration, MeasurementSettings, ShakeSettings, TemperatureController, diff --git a/pylabrobot/bmg_labtech/clariostar/device.py b/pylabrobot/bmg_labtech/clariostar/device.py index 456c83b50a7..f8c90ef8f6c 100644 --- a/pylabrobot/bmg_labtech/clariostar/device.py +++ b/pylabrobot/bmg_labtech/clariostar/device.py @@ -10,11 +10,12 @@ from .driver.features.absorbance import Absorbance from .driver.features.drawer import Drawer from .driver.features.fluorescence import Fluorescence +from .driver.features.luminescence import Luminescence from .driver.features.temperature_controller import TemperatureController from .driver.master import CLARIOstarDriver from .driver.simulator import CLARIOstarSimulationDriver -# Preliminary: the operating manuals' nominal size, rounded to the cm, the same for the CLARIOstar +# Preliminary: the nominal size, rounded to the cm, the same for the CLARIOstar # and the CLARIOstar Plus; not yet measured. SIZE_X = 450.0 SIZE_Y = 510.0 @@ -189,6 +190,11 @@ def fluorescence(self) -> Optional[Fluorescence]: """Fluorescence measurement, or None if the reader has none; known after `setup`.""" return self.driver.fluorescence + @property + def luminescence(self) -> Optional[Luminescence]: + """Luminescence measurement, or None if the reader has none; known after `setup`.""" + return self.driver.luminescence + async def setup(self) -> None: """Connect, initialize the reader (about 13 s), and check it is the model this device is. diff --git a/pylabrobot/bmg_labtech/clariostar/driver/__init__.py b/pylabrobot/bmg_labtech/clariostar/driver/__init__.py index d366a713c83..c8e2f6842a5 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/__init__.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/__init__.py @@ -8,10 +8,13 @@ CLARIOstarAbsorbanceSpectrumResult, CLARIOstarFluorescenceResult, CLARIOstarFluorescenceSpectrumResult, + CLARIOstarLuminescenceResult, CLARIOstarZProfileResult, Drawer, Fluorescence, FluorescenceConfiguration, + Luminescence, + LuminescenceConfiguration, MeasurementSettings, ShakeSettings, TemperatureController, diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/__init__.py b/pylabrobot/bmg_labtech/clariostar/driver/features/__init__.py index 6bd7b6b61a4..ebc7ab5c0a8 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/__init__.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/__init__.py @@ -12,5 +12,10 @@ Fluorescence, FluorescenceConfiguration, ) +from .luminescence import ( + CLARIOstarLuminescenceResult, + Luminescence, + LuminescenceConfiguration, +) from .measurement import MeasurementSettings, ShakeSettings from .temperature_controller import TemperatureController diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/absorbance.py b/pylabrobot/bmg_labtech/clariostar/driver/features/absorbance.py index 70254a5199e..8b199766db0 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/absorbance.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/absorbance.py @@ -76,7 +76,7 @@ class AbsorbanceConfiguration(MeasurementConfiguration): wavelength_range: Tuple[int, int] = (220, 1000) """Wavelengths a read accepts, in nm (the manufacturer's spectrometer range).""" maximum_wavelengths: int = 8 - """Wavelengths per read (the vendor software manual: up to 8; 8 read on hardware).""" + """Wavelengths per read (up to 8; 8 read on hardware).""" # ---------------------------------------- @@ -155,10 +155,10 @@ def __init__( # ---------------------------------------- # Water pathlength correction after the wavelength fields, in nm: two wavelengths, a third - # unused, and its bandwidth, as the vendor always sends them. + # unused, and its bandwidth, always sent. _PATHLENGTH_CORRECTION = (900.0, 993.0, 0.0, 10.0) - # Pause, u16 BE each: when (0 off, 1 before plate reading), then seconds. Off, the vendor sends - # 5 s; on, 5 s captured. + # Pause, u16 BE each: when (0 off, 1 before plate reading), then seconds; off, 5 s are sent, + # and on, 5 s is verified. _NO_PAUSE = bytes.fromhex("00000005") _PAUSE_BEFORE_READING = 0x0001 @@ -177,7 +177,7 @@ def _encode_wavelengths(*nanometres: float) -> bytes: # Decoding # ---------------------------------------- - # Data reply layout byte [6]: the only one captured, with the temperature controller on or off. + # Data reply layout byte [6]: the only one verified, with the temperature controller on or off. _DATA_LAYOUT = 0xA9 @staticmethod @@ -195,8 +195,8 @@ def _decode_data_reply(self, reply: bytes, wavelengths: List[int]) -> _Decoded: """The samples per wavelength and the references, in the reply's well order (row by row), the wavelengths' calibration pairs and the reference's. - The reply holds n + 3 groups of one u32 per well (the n samples, chromatics 2 and 3, the - reference), then n + 3 calibration pairs, in the same order. + The reply holds n + 3 groups of one u32 per well (the n samples, two groups not decoded, + the reference), then n + 3 calibration pairs, in the same order. Raises: ValueError: if the reply is incomplete or not the layout of this read. @@ -271,8 +271,8 @@ def _check_arguments( Raises: ValueError: listing every violation: not 1 to `maximum_wavelengths` wavelengths, a wavelength - not a whole nm within `wavelength_range`, an unknown report, a matrix well scan (captured - but not decoded), a timeout not positive. + not a whole nm within `wavelength_range`, an unknown report, a matrix well scan (not + decoded), a timeout not positive. """ errors = [] maximum = self.configuration.maximum_wavelengths @@ -350,7 +350,7 @@ async def measure( wells: the wells to measure, on the plate; None measures every well. wavelengths: 1 to `configuration.maximum_wavelengths` wavelengths, each a whole nm within `configuration.wavelength_range`; all read in one pass. - measurement_settings: how to read; None reads with the defaults (the vendor's). + measurement_settings: how to read; None reads with the defaults. shake_settings: shake the plate before the read; None does not shake. report: what the results hold: optical density, or the raw sample counts with the reference counts and calibration pairs that compute it (`CLARIOstarAbsorbanceResult`). diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/absorbance_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/features/absorbance_tests.py index 72be26c1b71..83d53ba7f6f 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/absorbance_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/absorbance_tests.py @@ -1,5 +1,5 @@ -"""Absorbance tests over a fake io: what the driver sends is pinned to vendor captures, the -optical densities to the vendor software's own export of the same reply.""" +"""Absorbance tests over a fake io: what the driver sends, and the optical densities it computes, +are pinned to the ground truth.""" import asyncio import gzip @@ -40,7 +40,7 @@ from pylabrobot.resources import Plate, Well, cor_96_wellplate_360uL_Fb, create_ordered_items_2d from pylabrobot.resources.biorad.plates import biorad_96_wellplate_200uL_Vb -# The vendor's default endpoint read: 600 nm, every well of its plate. +# The default read: 600 nm, every well of the plate. BASELINE_RUN_FRAME = bytes.fromhex( "0200900c0431ec2166059604602c561d060c0800ffffffffffffffffffffffff00000000000000000000000000000000" "00000000000000000000000000000000000000000a020000000000000000000000000000000000000000000000000000" @@ -60,14 +60,14 @@ "0100010000153f0d" ) -# Orbital, 3 mm, the vendor's 7 flashes, 0.1 s settling, on 6.58 mm wells. +# Orbital, 3 mm, 7 flashes, 0.1 s settling, on 6.58 mm wells. ORBITAL_RUN_FRAME = bytes.fromhex( "0200950c0431e82164059e04642c4a1d000c0800ffffffffffffffffffffffff00000000000000000000000000000000" "00000000000000000000000000000000000000000a320000000000000000000000000000000000000000000000000000" "00000000270f270f02030292000501177000000064232826ca0000006400000005020000000000010000000100070001" "000013cd0d" ) -# Spiral, 4 mm, the vendor's 15 flashes, unidirectional, no settling; its pause sends 0 s. +# Spiral, 4 mm, 15 flashes, unidirectional, no settling; its pause sends 0 s. SPIRAL_RUN_FRAME = bytes.fromhex( "0200950c0431e82164059e04642c4a1d000c0800ffffffffffffffffffffffff00000000000000000000000000000000" "00000000000000000000000000000000000000008a060000000000000000000000000000000000000000000000000000" @@ -82,7 +82,7 @@ "00000000270f270f02030292000501177000000064232826ca0000006400000000020000000000010000000100070001" "000013d10d" ) -# The vendor's pre-separators of five shaken reads, by (pattern, rpm, seconds). +# The pre-separators of five shaken reads, by (pattern, rpm, seconds). SHAKE_PRE_SEPARATORS = { ("orbital", 300, 5): "32000000000000000000000002000000000002000500000000000000000000", ("orbital", 500, 5): "32000000000000000000000002000000000004000500000000000000000000", @@ -91,7 +91,7 @@ ("linear", 300, 5): "32000000000000000000000002000000000102000500000000000000000000", } -# A 5 s pause before reading, otherwise the vendor's default read, on the captured plate. +# A 5 s pause before reading, otherwise the default read, on the ground-truth plate. PAUSE_RUN_FRAME = bytes.fromhex( "0200900c0431e82164059e04642c4a1d000c0800ffffffffffffffffffffffff00000000000000000000000000000000" "00000000000000000000000000000000000000000a020000000000000000000000000000000000000000000000000000" @@ -114,7 +114,7 @@ "000047900000000000297a0d" ) -# Spectrum RUNs as the vendor sent them, all on the captured plate: 300-700/1 nm, +# Spectrum RUNs, all on the ground-truth plate: 300-700/1 nm, # 400-600/1, 300-700/5, column 1, orbital, and the verification read: # 220-1000/1, spiral, unidirectional, 20 flashes, double orbital shaking, 5 s pause. SPECTRUM_300_700_1_RUN_FRAME = bytes.fromhex( @@ -153,7 +153,7 @@ # A timeout that ends a read right after its RUN, so a test can pin the RUN and the STOP. IMMEDIATE_TIMEOUT = 1e-9 -# Column 1 at 600 nm, unidirectional, on the captured plate: RUN_REPLY, STATUS_MEASURED and +# Column 1 at 600 nm, unidirectional, on the ground-truth plate: RUN_REPLY, STATUS_MEASURED and # COLUMN_1_DATA_REPLY are this read's replies. COLUMN_1_RUN_FRAME = bytes.fromhex( "0200900c0431e82164059e04642c4a1d000c080080080080080080080080080000000000000000000000000000000000" @@ -165,7 +165,7 @@ STOP_FRAME = bytes.fromhex("02000a0c0b000000230d") STOP_REPLY = bytes.fromhex("0200180c0125062600000b0000000000000000c00001430d") -# A 96-well read at 600 nm (verification plate) and the vendor software's export of it. +# A 96-well read at 600 nm (verification plate) and its ground-truth optical densities. VERIFICATION_DATA_REPLY = bytes.fromhex( "02064c0c020506260000a901880188001dffe20220030001006001000100000001010000000000f10000645800005e80" "00deccd0001278e10001ffe700008b42000070eb000070f500006f6400006b1c00cad94700dcfc9700e7389600005f28" @@ -202,7 +202,7 @@ "00004790000047830000477c0000475b0000475c0000474c000047400000473e0000472b00004758012ce63d00028ec6" "0076afa900028c2e0038e03700028ec3000048320000000001c4620d" ) -VERIFICATION_EXPORT = [ +VERIFICATION_GROUND_TRUTH = [ [2.885, 2.909, 0.126, 1.208, 2.173, 2.738, 2.829, 2.828, 2.834, 2.852, 0.165, 0.128], [0.114, 2.906, 2.865, 2.878, 2.95, 2.967, 2.842, 2.844, 2.752, 2.69, 0.142, 0.147], [0.147, 0.149, 0.097, 0.108, 0.141, 0.159, 0.149, 0.167, 2.308, 2.275, 0.132, 0.107], @@ -237,11 +237,11 @@ def baseline_plate() -> Plate: ) -def captured_96_well_plate() -> Plate: - """The captured plate's geometry, with the vendor database's 6.58 mm wells: +def ground_truth_96_well_plate() -> Plate: + """The ground-truth plate's geometry, with 6.58 mm wells: 127.76 x 85.48 mm, A1 centre 14.38 / 11.24 mm, 9 mm pitch.""" return Plate( - "captured", + "ground_truth", size_x=127.76, size_y=85.48, size_z=14.0, @@ -350,7 +350,7 @@ async def measure_with( def without_pause_seconds(payload: bytes) -> bytes: - """`payload` with the no-pause seconds set to v1's 5: the vendor sends 5 or 0.""" + """`payload` with the no-pause seconds set to v1's 5: the ground truth has 5 or 0.""" payload = bytearray(payload) pause_seconds = len(payload) - 17 if payload[pause_seconds - 3 : pause_seconds + 1] == bytes(4): @@ -360,14 +360,14 @@ def without_pause_seconds(payload: bytes) -> bytes: class RunParametersTests(unittest.TestCase): def test_orbital(self): - plate = captured_96_well_plate() + plate = ground_truth_96_well_plate() parameters = run_parameters( plate, plate.get_all_items(), [600], 7, well_scan="orbital", scan_diameter=3 ) self.assertEqual(b"\x04" + parameters, ftdi_extract_payload(ORBITAL_RUN_FRAME)) def test_spiral_but_its_pause_seconds(self): - plate = captured_96_well_plate() + plate = ground_truth_96_well_plate() payload = b"\x04" + run_parameters( plate, plate.get_all_items(), @@ -387,7 +387,7 @@ def test_spiral_but_its_pause_seconds(self): ) def test_shake_but_its_pause_seconds(self): - plate = captured_96_well_plate() + plate = ground_truth_96_well_plate() payload = b"\x04" + run_parameters( plate, plate.get_all_items(), @@ -407,7 +407,7 @@ def test_shake_but_its_pause_seconds(self): ) def test_shake_patterns_speeds_and_durations(self): - plate = captured_96_well_plate() + plate = ground_truth_96_well_plate() for (pattern, speed, duration), recorded in SHAKE_PRE_SEPARATORS.items(): parameters = run_parameters( plate, @@ -428,7 +428,7 @@ def test_five_wavelengths(self): class SettingsTests(unittest.TestCase): - def test_the_defaults_are_the_vendors_default_read(self): + def test_the_defaults_are_the_ground_truth(self): plate = baseline_plate() parameters = absorbance_feature()._compute_run_parameters( plate, plate.get_all_items(), [600], MeasurementSettings(), None @@ -476,7 +476,7 @@ def test_every_range_error_is_listed_at_once(self): class SpectrumRunParametersTests(unittest.TestCase): def spectrum_payload(self, start, end, step, flashes, wells=None, shake=None, **measurement): - plate = captured_96_well_plate() + plate = ground_truth_96_well_plate() wells = plate.get_all_items() if wells is None else wells(plate) return b"\x04" + absorbance_feature()._compute_spectrum_run_parameters( plate, wells, start, end, step, MeasurementSettings(flashes=flashes, **measurement), shake @@ -524,14 +524,14 @@ def test_out_of_range_spectra_are_refused(self): def load_verification_spectrum() -> Tuple[List[bytes], List[int], Dict[str, List[float]]]: - """The 220-1000 nm verification read: its 5 data pages as received, its wavelengths, and the - vendor software's export of it, optical density per well and wavelength.""" + """The 220-1000 nm verification read: its 5 data pages as received, its wavelengths, and its + ground-truth optical densities, per well and wavelength.""" path = os.path.join( os.path.dirname(__file__), "test_data", "absorbance_spectrum_verification.json.gz" ) with gzip.open(path, "rt") as f: data = json.load(f) - return [bytes.fromhex(page) for page in data["pages"]], data["wavelengths"], data["export"] + return [bytes.fromhex(page) for page in data["pages"]], data["wavelengths"], data["expected"] # The verification read's settings, as its RUN encodes them. @@ -543,10 +543,10 @@ def load_verification_spectrum() -> Tuple[List[bytes], List[int], Dict[str, List class SpectrumReadTests(unittest.IsolatedAsyncioTestCase): async def read_verification_spectrum(self, report="optical_density"): - pages, wavelengths, export = load_verification_spectrum() + pages, wavelengths, expected = load_verification_spectrum() io = FakeFTDI(chunk=4096) io.replies = [STATUS_IDLE, RUN_REPLY, STATUS_MEASURED, *pages] - absorbance = make_absorbance(io, captured_96_well_plate()) + absorbance = make_absorbance(io, ground_truth_96_well_plate()) result = await absorbance.measure_spectrum( start_wavelength=220, end_wavelength=1000, @@ -555,10 +555,10 @@ async def read_verification_spectrum(self, report="optical_density"): shake_settings=VERIFICATION_SPECTRUM_SHAKE, report=report, ) - return io, result, wavelengths, export + return io, result, wavelengths, expected - async def test_the_verification_read_matches_the_vendor_export(self): - io, result, wavelengths, export = await self.read_verification_spectrum() + async def test_the_verification_read_matches_the_ground_truth(self): + io, result, wavelengths, expected = await self.read_verification_spectrum() self.assertEqual( io.written, [STATUS_QUERY_FRAME, SPECTRUM_VERIFICATION_RUN_FRAME, STATUS_QUERY_FRAME] + [DATA_FRAME] * 5, @@ -567,16 +567,16 @@ async def test_the_verification_read_matches_the_vendor_export(self): self.assertEqual(list(result.data["A1"]), wavelengths) differences = [ abs(result.data[well][wavelength] - value) - for well, values in export.items() + for well, values in expected.items() for wavelength, value in zip(wavelengths, values) ] self.assertEqual(len(differences), 74_976) - self.assertLess(max(differences), 0.0006) # the export rounds to 3 decimals + self.assertLess(max(differences), 0.0006) # the ground truth rounds to 3 decimals self.assertEqual(result.temperature, 24.1) self.assertLessEqual(result.start_timestamp, result.end_timestamp) async def test_raw_counts_compute_the_optical_densities(self): - _, raw, wavelengths, export = await self.read_verification_spectrum(report="raw") + _, raw, wavelengths, expected = await self.read_verification_spectrum(report="raw") assert isinstance(raw, CLARIOstarAbsorbanceSpectrumResult) assert raw.references is not None and raw.calibration is not None assert raw.reference_calibration is not None @@ -585,8 +585,8 @@ async def test_raw_counts_compute_the_optical_densities(self): fraction = (raw.data[well][wavelength] / raw.calibration[wavelength][0]) * ( raw.reference_calibration[0] / raw.references[well] ) - expected = export[well][wavelengths.index(wavelength)] - self.assertAlmostEqual(-math.log10(fraction), expected, delta=0.0006) + expected_density = expected[well][wavelengths.index(wavelength)] + self.assertAlmostEqual(-math.log10(fraction), expected_density, delta=0.0006) def test_pages_out_of_order_or_missing_are_refused(self): pages = [ftdi_extract_payload(frame) for frame in load_verification_spectrum()[0]] @@ -598,7 +598,7 @@ def test_pages_out_of_order_or_missing_are_refused(self): async def test_a_timeout_not_positive_is_refused(self): io = FakeFTDI() - absorbance = make_absorbance(io, captured_96_well_plate()) + absorbance = make_absorbance(io, ground_truth_96_well_plate()) for timeout in (0, -1): with self.assertRaises(ValueError): await measure_with(absorbance, [600], timeout=timeout) @@ -613,7 +613,7 @@ class SimulatedSpectrumTests(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self): self.reader = CLARIOstarPlus(simulation=True) await self.reader.setup() - self.plate = captured_96_well_plate() + self.plate = ground_truth_96_well_plate() await self.reader.drawer.open() self.reader.plate_holder.assign_child_resource(self.plate) await self.reader.drawer.close() @@ -630,8 +630,8 @@ async def test_a_simulated_spectrum_reads_within_range(self): self.assertEqual(list(spectrum), list(range(400, 501, 10))) self.assertTrue(all(0.0 <= value <= 4.0 for value in spectrum.values())) - async def test_a_captured_spectrum_is_returned_again(self): - captured = await self.absorbance.measure_spectrum( + async def test_an_earlier_spectrum_is_returned_again(self): + earlier = await self.absorbance.measure_spectrum( start_wavelength=400, end_wavelength=500, step_size=10 ) replayed = await self.absorbance.measure_spectrum( @@ -639,23 +639,23 @@ async def test_a_captured_spectrum_is_returned_again(self): start_wavelength=400, end_wavelength=500, step_size=10, - simulated_data=captured, + simulated_data=earlier, ) - self.assertEqual(replayed.data, {well: captured.data[well] for well in COLUMN_1}) - with self.assertRaises(ValueError): # 405 nm was not captured + self.assertEqual(replayed.data, {well: earlier.data[well] for well in COLUMN_1}) + with self.assertRaises(ValueError): # 405 nm was not read await self.absorbance.measure_spectrum( - start_wavelength=400, end_wavelength=500, step_size=5, simulated_data=captured + start_wavelength=400, end_wavelength=500, step_size=5, simulated_data=earlier ) class OpticalDensityTests(unittest.TestCase): - def test_matches_the_vendor_export(self): + def test_matches_the_ground_truth(self): decoded = absorbance_feature()._decode_data_reply( ftdi_extract_payload(VERIFICATION_DATA_REPLY), [600] ) densities = _compute_readings(*decoded, "optical_density")[0] - exported = [value for row in VERIFICATION_EXPORT for value in row] - for density, value in zip(densities, exported): + expected = [value for row in VERIFICATION_GROUND_TRUTH for value in row] + for density, value in zip(densities, expected): self.assertAlmostEqual(density, value, delta=0.001) def test_a_reply_for_other_wavelengths_is_refused(self): @@ -669,7 +669,7 @@ class ReadTests(unittest.IsolatedAsyncioTestCase): async def test_read_column_1(self): io = FakeFTDI() io.replies = [STATUS_IDLE, RUN_REPLY, STATUS_MEASURING, STATUS_MEASURED, COLUMN_1_DATA_REPLY] - plate = captured_96_well_plate() + plate = ground_truth_96_well_plate() column_1 = plate.get_items("A1:H1") results = await measure_with( make_absorbance(io, plate), [600], wells=column_1, bidirectional=False @@ -686,7 +686,7 @@ async def test_the_first_poll_waits_after_the_run(self): # As the reader answered: a poll sent right after the RUN's reply went unanswered. io = FakeFTDI() io.replies = [STATUS_IDLE, RUN_REPLY, STATUS_MEASURING, STATUS_MEASURED, COLUMN_1_DATA_REPLY] - plate = captured_96_well_plate() + plate = ground_truth_96_well_plate() absorbance = make_absorbance(io, plate) absorbance._driver.default_poll_interval = 0.25 waits = [] @@ -722,7 +722,7 @@ async def test_a_timeout_stops_the_run(self): async def test_a_cancel_between_a_write_and_its_reply_leaves_the_link_clean(self): io = YieldingFTDI() - absorbance = make_absorbance(io, captured_96_well_plate()) + absorbance = make_absorbance(io, ground_truth_96_well_plate()) task = asyncio.create_task(absorbance.measure(wavelengths=[600], timeout=10)) io.cancel_on_status = task # cancelled after the first STATUS of the run is written with self.assertRaises(asyncio.CancelledError): @@ -773,20 +773,20 @@ async def test_the_ranges_are_the_configurations(self): await measure_with(absorbance, [600], **arguments) # type: ignore[arg-type] self.assertEqual(io.written, []) - async def test_raw_counts_compute_the_vendors_optical_densities(self): + async def test_raw_counts_compute_the_ground_truth_optical_densities(self): readings = {} for report in ("optical_density", "raw"): io = FakeFTDI() io.replies = [STATUS_IDLE, RUN_REPLY, STATUS_MEASURED, VERIFICATION_DATA_REPLY] (readings[report],) = await measure_with( - make_absorbance(io, captured_96_well_plate()), [600], report=report + make_absorbance(io, ground_truth_96_well_plate()), [600], report=report ) # type: ignore[arg-type] density, raw = readings["optical_density"], readings["raw"] self.assertIs(type(density), AbsorbanceResult) assert isinstance(raw, CLARIOstarAbsorbanceResult) assert raw.references is not None and raw.calibration and raw.reference_calibration - exported = [value for row in VERIFICATION_EXPORT for value in row] - for well, value in zip(density.data, exported): # both row by row + expected = [value for row in VERIFICATION_GROUND_TRUTH for value in row] + for well, value in zip(density.data, expected): # both row by row fraction = (raw.data[well] / raw.calibration[0]) * ( raw.reference_calibration[0] / raw.references[well] ) @@ -816,8 +816,8 @@ def test_the_read_timeout_default_is_checked_when_set(self): with self.assertRaises(ValueError): absorbance.default_read_timeout = 0 - async def test_scan_and_settling_bytes_are_the_vendors(self): - # (arguments, scan byte, settling byte) as the vendor sent them: the corners and + async def test_scan_and_settling_bytes_are_the_ground_truth(self): + # (arguments, scan byte, settling byte) as the ground truth has them: the corners and # directions, the pauses and three orbital reads (the same field). cases: List[Tuple[Dict[str, Any], int, int]] = [ ({}, 0x0A, 0x05), @@ -860,7 +860,7 @@ async def test_every_read_defaults_to_five_flashes(self): for spectrum in (False, True): io = FakeFTDI() io.replies = [STATUS_IDLE, RUN_REPLY, STATUS_MEASURING] - absorbance = make_absorbance(io, captured_96_well_plate()) + absorbance = make_absorbance(io, ground_truth_96_well_plate()) settings = MeasurementSettings(well_scan=well_scan) with self.assertRaises(TimeoutError): if spectrum: @@ -880,9 +880,9 @@ async def test_every_read_defaults_to_five_flashes(self): async def test_well_scans_out_of_range_send_nothing(self): io = FakeFTDI() - absorbance = make_absorbance(io, captured_96_well_plate()) + absorbance = make_absorbance(io, ground_truth_96_well_plate()) for arguments in ( - {"well_scan": "matrix", "matrix_size": 3}, # captured, its data not decoded + {"well_scan": "matrix", "matrix_size": 3}, # its data not decoded {"well_scan": "orbital", "scan_diameter": 0}, {"well_scan": "spiral", "scan_diameter": 7}, # wider than the 6.58 mm well {"well_scan": "spiral", "scan_diameter": 3.5}, @@ -1098,25 +1098,25 @@ async def asyncSetUp(self): assert isinstance(absorbance, CLARIOstarSimulationAbsorbance) self.absorbance = absorbance - async def test_captured_results_are_returned_again(self): - captured = await measure_with(self.absorbance, [450, 600]) - replayed = await measure_with(self.absorbance, [450, 600], simulated_data=captured) - self.assertEqual([r.data for r in replayed], [r.data for r in captured]) + async def test_earlier_results_are_returned_again(self): + earlier = await measure_with(self.absorbance, [450, 600]) + replayed = await measure_with(self.absorbance, [450, 600], simulated_data=earlier) + self.assertEqual([r.data for r in replayed], [r.data for r in earlier]) - async def test_a_subset_reads_its_wells_of_the_captured_plate(self): - (captured,) = await measure_with(self.absorbance, [600]) + async def test_a_subset_reads_its_wells_of_the_earlier_plate(self): + (earlier,) = await measure_with(self.absorbance, [600]) column_1 = self.plate.get_items("A1:H1") (replayed,) = await measure_with( - self.absorbance, [600], wells=column_1, simulated_data=[captured] + self.absorbance, [600], wells=column_1, simulated_data=[earlier] ) - self.assertEqual(replayed.data, {well: captured.data[well] for well in COLUMN_1}) + self.assertEqual(replayed.data, {well: earlier.data[well] for well in COLUMN_1}) async def test_raw_counts_need_simulated_data(self): with self.assertRaises(NotImplementedError): await measure_with(self.absorbance, [600], report="raw") - async def test_captured_raw_results_are_returned_again(self): - captured = CLARIOstarAbsorbanceResult( + async def test_earlier_raw_results_are_returned_again(self): + earlier = CLARIOstarAbsorbanceResult( data={well: 1000.0 for well in COLUMN_1}, wavelength=600, temperature=None, @@ -1131,15 +1131,15 @@ async def test_captured_raw_results_are_returned_again(self): [600], wells=self.plate.get_items("A1:H1"), report="raw", - simulated_data=[captured], + simulated_data=[earlier], ) assert isinstance(replayed, CLARIOstarAbsorbanceResult) self.assertEqual( (replayed.data, replayed.references, replayed.calibration, replayed.reference_calibration), - (captured.data, captured.references, captured.calibration, captured.reference_calibration), + (earlier.data, earlier.references, earlier.calibration, earlier.reference_calibration), ) with self.assertRaises(ValueError): - await measure_with(self.absorbance, [600], simulated_data=[captured]) + await measure_with(self.absorbance, [600], simulated_data=[earlier]) async def test_simulated_data_must_match_the_measurement(self): (whole_plate,) = await measure_with(self.absorbance, [600]) diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py index 511937787d0..c38030096df 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py @@ -1,9 +1,8 @@ """CLARIOstar fluorescence: plate reads, spectra, filter detection and z-profiles. -A plate read goes through one to five fluorescence channels in one pass (`FluorescenceChannel`, the -vendor's chromatics), from the top or the bottom optic. Excitation and emission each select their -wavelength through a monochromator or a filter. A spectrum scans the excitation or the emission -monochromator. +A plate read goes through one to five fluorescence channels in one pass (`FluorescenceChannel`), +from the top or the bottom optic. Excitation and emission each select their wavelength through a +monochromator or a filter. A spectrum scans the excitation or the emission monochromator. Filter detection scans every filter position; a z-profile reads one well over its focal heights, and with a target the reader fits the gain at its peak. """ @@ -69,8 +68,7 @@ class CLARIOstarFluorescenceResult(FluorescenceResult): gain: the detector gain, within `FluorescenceConfiguration.gain_range`. matrix_points: a matrix scan's counts, every measured point of each well in the order the reader sends them, keyed as ``data``; None for any other well scan. ``data`` then holds - the plain mean of each well's points; the vendor's own reduction is not verified (its - export of a 3 x 3 matrix reads 141 where the mean is 141.78). + the plain mean of each well's points. """ gain: int @@ -93,7 +91,7 @@ class CLARIOstarZProfileResult: target; None without a target. peak_focal_height: the focal height the firmware computes as the z-profile's maximum, in mm. temperature: temperature in degrees C; None: no z-scan with the temperature controller on is - captured, so the reply's temperature, if it holds one, is not decoded. + verified, so the reply's temperature, if it holds one, is not decoded. start_timestamp: when the z-scan started. end_timestamp: when its result was in. """ @@ -126,29 +124,29 @@ class FluorescenceConfiguration(MeasurementConfiguration): """Device facts of the fluorescence measurement.""" wavelength_range: Tuple[int, int] = (320, 840) - """Excitation and emission wavelengths, in nm (the operating manual's monochromator range with - an extended-red photomultiplier; up to 740 without, `Fluorescence._get_reader_configuration`).""" + """Excitation and emission wavelengths, in nm, with an extended-red photomultiplier; up to 740 + without (`Fluorescence._get_reader_configuration`).""" bandwidth_range: Tuple[int, int] = (8, 100) - """Excitation and emission bandwidths, in nm (the operating manual).""" + """Excitation and emission bandwidths, in nm.""" dichroic_range: Tuple[float, float] = (340.0, 760.0) - """Where the dichroic mirror splits excitation from emission, in nm (the operating manual; up - to 740 without an extended-red photomultiplier).""" + """Where the dichroic mirror splits excitation from emission, in nm; up to 740 without an + extended-red photomultiplier.""" gain_range: Tuple[int, int] = (0, 4095) - """Detector gain (the vendor software manual: 0 to 4095).""" + """Detector gain (0 to 4095).""" focal_height_range: Tuple[float, float] = (0.0, 25.0) - """Focal height, in mm, in 0.1 mm steps (the vendor software manual: up to 25 mm).""" + """Focal height, in mm, in 0.1 mm steps (up to 25 mm).""" excitation_filter_slots: int = 4 - """Excitation filter positions (the vendor software manual: up to 4).""" + """Excitation filter positions (up to 4).""" dichroic_filter_slots: int = 3 - """Dichroic filter positions (the vendor software manual: up to 3).""" + """Dichroic filter positions (up to 3).""" emission_filter_slots: int = 4 - """Emission filter positions (the vendor software manual: up to 4).""" + """Emission filter positions (up to 4).""" filter_slots_per_slide: int = 2 """Filter positions on each of the two excitation or emission slides (filter detection).""" - maximum_chromatics: int = 5 - """Fluorescence channels per read, the vendor's chromatics (the vendor software: up to 5).""" + maximum_fluorescence_channels: int = 5 + """Fluorescence channels per read (up to 5).""" maximum_spectrum_wavelengths: int = 1000 - """Wavelengths per spectrum (the vendor software manual: up to 1000 measurement points).""" + """Wavelengths per spectrum (up to 1000).""" # ---------------------------------------- @@ -159,13 +157,13 @@ class FluorescenceConfiguration(MeasurementConfiguration): def _compute_band_edges(wavelength: float, bandwidth: float) -> Tuple[int, int]: """The band's upper and lower edge in 0.1 nm: `wavelength` +/- half `bandwidth`. - The nominal edges; the vendor software sends each band narrower (14.6 nm of 15). + The nominal edges. """ return round((wavelength + bandwidth / 2) * 10), round((wavelength - bandwidth / 2) * 10) def _compute_dichroic(excitation: Tuple[int, int], emission: Tuple[int, int]) -> int: - """The default dichroic split in 0.1 nm, rounded half to even, as the vendor's. + """The default dichroic split in 0.1 nm, rounded half to even. Midway between the two bands' facing edges: the excitation band's top and the emission band's bottom, or the other way round with excitation above emission. Each band is (upper, lower) edge @@ -200,8 +198,8 @@ class Fluorescence(Measurement): _WELL_SCAN_CODE = 0x03 _TRAILER_START = 0x00 - # Without an extended-red photomultiplier reading fluorescence, the vendor software refuses - # excitation, emission or a dichroic split above 740 nm. + # Without an extended-red photomultiplier reading fluorescence, excitation, emission and a + # dichroic split above 740 nm are refused. _STANDARD_WAVELENGTH_RANGE = (320, 740) _STANDARD_DICHROIC_RANGE = (340.0, 740.0) @@ -235,9 +233,9 @@ def __init__( (`_get_reader_configuration`). """ super().__init__(driver, configuration or self._get_reader_configuration(driver.configuration)) - self._default_excitation_bandwidth = 15 # nm, the vendor's default - self._default_emission_bandwidth = 20 # nm, the vendor's default - self._default_optic_position: OpticPosition = "top" # the vendor's default + self._default_excitation_bandwidth = 15 # nm + self._default_emission_bandwidth = 20 # nm + self._default_optic_position: OpticPosition = "top" # The filters each slide holds, by slot: `detect_all_filters`, or assigned; None until then. self.excitation_filter_slide: Optional[ExcitationFilterSlide] = None self.dichroic_filter_slide: Optional[DichroicFilterSlide] = None @@ -256,7 +254,7 @@ def _check_default_bandwidth(self, name: str, value: int) -> None: @property def default_excitation_bandwidth(self) -> int: - """Excitation bandwidth, nm, of a monochromator read given none (the vendor's: 15).""" + """Excitation bandwidth, nm, of a monochromator read given none (15 by default).""" return self._default_excitation_bandwidth @default_excitation_bandwidth.setter @@ -267,7 +265,7 @@ def default_excitation_bandwidth(self, value: int) -> None: @property def default_emission_bandwidth(self) -> int: - """Emission bandwidth, nm, of a monochromator read given none (the vendor's: 20).""" + """Emission bandwidth, nm, of a monochromator read given none (20 by default).""" return self._default_emission_bandwidth @default_emission_bandwidth.setter @@ -292,7 +290,7 @@ def _check_optic_position(self, optic_position: str, name: str = "optic_position @property def default_optic_position(self) -> OpticPosition: - """The optic a read uses given none (the vendor's: top).""" + """The optic a read uses given none (top by default).""" return self._default_optic_position @default_optic_position.setter @@ -493,7 +491,7 @@ def _check_bands( f"dichroic_split_wavelength must lie between {lower} and {upper} nm, got {split}" ) if emission[0] < excitation[1]: - # TODO(device): no read with excitation above emission is captured. + # TODO(device): no read with excitation above emission is verified. self._warn_untested("a fluorescence read with excitation above emission") lowest_split, highest_split = self.configuration.dichroic_range if not lowest_split <= split <= highest_split: @@ -536,7 +534,7 @@ def _check_fluorescence_channel( fluorescence_channel.excitation_filter is not None or fluorescence_channel.emission_filter is not None ): - # No read with an excitation or emission filter and the dichroic mirror is captured. + # No read with an excitation or emission filter and the dichroic mirror is verified. errors.append("an excitation or emission filter needs a dichroic_filter") split = fluorescence_channel.dichroic_split_wavelength if split is not None and ( @@ -592,8 +590,8 @@ def _check_read( # After the focal height: the kind of read, discrete 00 or spectrum 02. _DISCRETE = 0x00 _SPECTRUM = 0x02 - # Between two of the vendor's chromatics, one per fluorescence channel. - _CHROMATIC_SEPARATOR = bytes.fromhex("00000c") + # Between two fluorescence channels' blocks. + _FLUORESCENCE_CHANNEL_SEPARATOR = bytes.fromhex("00000c") # The excitation and emission wavelength selectors, after the emission edges: a plate read's # monochromator or filter; a spectrum's scanned and fixed monochromators. _WAVELENGTH_SELECTORS: Dict[str, int] = { @@ -603,7 +601,7 @@ def _check_read( "scanned_monochromator": 0x02, "fixed_monochromator": 0x03, } - # Pause before reading: off. No fluorescence read with a pause is captured. + # Pause before reading: off. No fluorescence read with a pause is verified. _NO_PAUSE = bytes(3) def _compute_optics( @@ -683,7 +681,7 @@ def _encode_fluorescence_channel( gain: int, selectors: Optional[Tuple[int, int]] = None, ) -> bytes: - """One fluorescence channel's 17 bytes, the vendor's chromatic: gain u16 BE, the optics, `00`. + """One fluorescence channel's 17 bytes: gain u16 BE, the optics, `00`. Args: selectors: the wavelength selectors, emission then excitation; None for each one's @@ -704,7 +702,7 @@ def _compute_run_parameters( focal_height: float, kind: int, count: int, - chromatics: Sequence[bytes], + encoded_fluorescence_channels: Sequence[bytes], optics: int = 0, flying_mode: bool = False, enhanced_dynamic_range: bool = False, @@ -714,11 +712,11 @@ def _compute_run_parameters( The run start (`_compute_run_start`, with `optics`, `flying_mode`, `enhanced_dynamic_range` and `scan_byte`), the focal height in 0.01 mm, `kind`, `count` u16 BE and `00 00 00 00 00 0c`, - the `chromatics` (`_encode_fluorescence_channel`), no pause, the run end. + the `encoded_fluorescence_channels` (`_encode_fluorescence_channel`), no pause, the run end. Args: - kind: `_DISCRETE`, or `_SPECTRUM` with the start and end chromatics between which the reader - steps. + kind: `_DISCRETE`, or `_SPECTRUM` with the start and end fluorescence channels between + which the reader steps. count: the fluorescence channels of a discrete read, or a spectrum's wavelengths. """ return ( @@ -737,7 +735,7 @@ def _compute_run_parameters( + count.to_bytes(2, "big") + bytes(5) + b"\x0c" - + self._CHROMATIC_SEPARATOR.join(chromatics) + + self._FLUORESCENCE_CHANNEL_SEPARATOR.join(encoded_fluorescence_channels) + self._NO_PAUSE + self._compute_run_end(measurement.flashes) ) @@ -804,11 +802,11 @@ def _get_result_optics( # Filter detection # ---------------------------------------- - # Filter detection `24`, mode byte: the slide's code OR bit 5, which the vendor always sets. + # Filter detection `24`, mode byte: the slide's code OR bit 5, always set. _FILTER_SCAN_SLIDES: Dict[str, int] = {"excitation": 0x00, "emission": 0x01, "dichroic": 0x03} _FILTER_SCAN_BIT_5 = 0x20 - # The emission detector gains an emission scan sends before the slide positions, as the vendor - # sends them; excitation and dichroic scans send zeros. + # The emission detector gains an emission scan sends before the slide positions; excitation and + # dichroic scans send zeros. _EMISSION_SCAN_GAINS = (1100, 1200, 1100, 1150) # Seconds one position may take: excitation and dichroic 6.4 s at most, emission 48.6 s. _FILTER_SCAN_TIMEOUT = 20.0 @@ -874,10 +872,10 @@ def _compute_optical_filter( ) -> Optional[OpticalFilter]: """The excitation or emission filter a FILTER_RESULT describes; None for an empty position. - As the vendor decides: + By the result's fields: - a bandwidth: a band-pass between the edges; - else a first edge with a slope: a long-pass, open above; - - else a second edge with a slope: a short-pass, its band not stated (not captured). + - else a second edge with a slope: a short-pass, its band not stated (not verified). """ bandwidth, first_edge, second_edge, first_slope, second_slope = result if bandwidth > 0: @@ -885,7 +883,7 @@ def _compute_optical_filter( if first_edge > 0 and first_slope != 0: return OpticalFilter(slot, passbands=((first_edge / 10, None),)) if second_edge > 0 and second_slope != 0: - # TODO(device): no short-pass filter is captured; state its band once a reader reports one. + # TODO(device): no short-pass filter is verified; state its band once a reader reports one. logger.warning( "filter %d is a short-pass up to %s nm; its band is not stated", slot, second_edge / 10 ) @@ -898,15 +896,15 @@ def _compute_dichroic_filter( ) -> Optional[DichroicFilter]: """The dichroic a FILTER_RESULT describes; None for an empty position. - As the vendor decides: + By the result's fields: - a first edge: a long-pass that transmits from it, open above; - - else a second edge: a short-pass, its bands not stated (not captured). + - else a second edge: a short-pass, its bands not stated (not verified). """ _, first_edge, second_edge, _, _ = result if first_edge > 0: return DichroicFilter(slot, transmission_bands=((first_edge / 10, None),)) if second_edge > 0: - # TODO(device): no short-pass dichroic is captured; state its bands once a reader reports one. + # TODO(device): no short-pass dichroic is verified; state its bands once a reader reports one. logger.warning( "dichroic %d is a short-pass at %s nm; its bands are not stated", slot, second_edge / 10 ) @@ -918,7 +916,7 @@ def _compute_dichroic_filter( async def _detect_filter(self, slide: str, slot: int) -> bytes: """Scan one filter position (`24`) and return its FILTER_RESULT reply. - As the vendor: status until the data is ready (the reader may still be busy), SPECTRAL_DATA + Status until the data is ready (the reader may still be busy), SPECTRAL_DATA (`05 11`, not decoded) then FILTER_RESULT (`05 1b`), then status until idle. Any failure, a timeout or a cancelled call stops the scan (`0b 00`) first. @@ -973,7 +971,7 @@ async def _detect_filter(self, slide: str, slot: int) -> bytes: async def detect_all_filters( self, ) -> Tuple[ExcitationFilterSlide, DichroicFilterSlide, EmissionFilterSlide]: - """Detect the filters in every position, as the vendor's "Detect all filters". + """Detect the filters in every position. - One scan each: excitation 1 to 4, dichroic A to C, emission 5 to 8 (slots 1 to 4). - Excitation and dichroic positions take about 8 s each, emission about 49 s: about 4 minutes. @@ -987,8 +985,8 @@ async def detect_all_filters( Raises: RuntimeError: if the reader reports no spectrometer (absorbance), or no extended-red - photomultiplier for the emission scans, whose captured limits are its; if a scan ends - without its data. + photomultiplier for the emission scans, whose limits are verified only with one; if a + scan ends without its data. TimeoutError: if a position takes longer than its timeout (`_FILTER_SCAN_TIMEOUT`, `_EMISSION_FILTER_SCAN_TIMEOUT` for emission). ValueError: if a FILTER_RESULT is not one of its position. @@ -999,7 +997,7 @@ async def detect_all_filters( if device.extended_red_photomultiplier_tube is False: raise RuntimeError( "the reader reports no extended-red photomultiplier; the emission scans' limits are only " - "captured with one" + "verified with one" ) excitation_slide = ExcitationFilterSlide(self.configuration.excitation_filter_slots) dichroic_slide = DichroicFilterSlide(self.configuration.dichroic_filter_slots) @@ -1035,11 +1033,11 @@ async def detect_all_filters( # Z-profiles # ---------------------------------------- - # The gain definition bytes of `09`, the z-scan, as the vendor sends them. + # The gain definition bytes of `09`, the z-scan. _Z_PROFILE_GAIN_DEFINITION = bytes([0x04, 0x08]) - # The step between focal heights, in 0.01 mm: the only one captured. + # The step between focal heights, in 0.01 mm: the only one verified. _Z_PROFILE_RESOLUTION = 10 - # Before the optics, as the end of the RUN's chromatic header. + # Before the optics, as the end of a RUN's fluorescence channel header. _Z_PROFILE_OPTICS_HEADER = bytes.fromhex("0000000c") # Status reply byte 2 bit 5: the z-profile is ready. _Z_PROFILE_RESULT_READY = 0x20 @@ -1164,7 +1162,7 @@ def _decode_z_profile_result( peak_detector_percent=percent / 100 if percent else None, average_signal=int.from_bytes(reply[19:23], "big"), z_profile=z_profile, - # TODO(device): no z-scan with the temperature controller on is captured. + # TODO(device): no z-scan with the temperature controller on is verified. temperature=None, start_timestamp=start_timestamp, end_timestamp=end_timestamp, @@ -1274,7 +1272,7 @@ async def measure_z_profile( # Discrete reads # ---------------------------------------- - # Flashes per well flying mode reads with (the vendor software manual: 1 or 3). + # Flashes per well flying mode reads with (1 or 3). _FLYING_MODE_FLASHES = (1, 3) # Data reply layout byte [6], with the temperature controller on or off. _DATA_LAYOUT = 0xA1 @@ -1298,15 +1296,15 @@ def _check_arguments( The keywords are `measure`'s. Raises: - ValueError: listing every violation: not 1 to `maximum_chromatics` fluorescence channels, - not one gain per fluorescence channel, a fluorescence channel's + ValueError: listing every violation: not 1 to `maximum_fluorescence_channels` fluorescence + channels, not one gain per fluorescence channel, a fluorescence channel's (`_check_fluorescence_channel`, `_check_gain`; prefixed `fluorescence_channels[i]` when there are several), `_check_read`'s, `_check_optic_position`'s; flying mode with a well scan, with flashes other than `_FLYING_MODE_FLASHES`, with settling time or with enhanced dynamic range; enhanced dynamic range with orbital or spiral averaging. """ - maximum = self.configuration.maximum_chromatics + maximum = self.configuration.maximum_fluorescence_channels count = len(fluorescence_channels) errors = ( [] if 1 <= count <= maximum else [f"give 1 to {maximum} fluorescence channels, got {count}"] @@ -1359,7 +1357,7 @@ def _decode_data_reply( - Bytes 11-14: the ceiling a saturated count reads (700,000,000 with enhanced dynamic range). - Bytes 32-33: the temperature in 0.1 C, 0 when not monitored. - From byte 34: u32 BE counts, by fluorescence channel, then by well, a well's points - together (an absorbance matrix's layout; no fluorescence matrix of several wells captured). + together (an absorbance matrix's layout; no fluorescence matrix of several wells verified). Raises: ValueError: if the reply is incomplete or not the layout of this read. @@ -1501,13 +1499,13 @@ async def measure( Args: fluorescence_channels: one `FluorescenceChannel` (`pylabrobot.lib.fluorescence`), or 1 to - `configuration.maximum_chromatics` of them, each within `_check_fluorescence_channel`; a - bandwidth None is the default bandwidth, a filter given as a slot or a name the slides' - (`detect_all_filters`). + `configuration.maximum_fluorescence_channels` of them, each within + `_check_fluorescence_channel`; a bandwidth None is the default bandwidth, a filter given + as a slot or a name the slides' (`detect_all_filters`). gain: the detector gain within `configuration.gain_range`: one for every fluorescence channel, or one per fluorescence channel in the order given. wells: the wells to measure, on the plate; None measures every well. A matrix scan reads - one well (no fluorescence matrix of several wells is captured). + one well (no fluorescence matrix of several wells is verified). focal_height: mm above the plate's base, within `configuration.focal_height_range`, in 0.1 mm steps. None reads at `default_focal_height_percent_of_well_depth` (25: a quarter up the well, for low and moderate fills). For full wells, or the most signal, focus on a @@ -1552,7 +1550,7 @@ async def measure( 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 captured: its reply layout is unknown. + # 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, @@ -1562,7 +1560,7 @@ async def measure( focal_height=focal_height, kind=self._DISCRETE, count=len(filled), - chromatics=[ + encoded_fluorescence_channels=[ self._encode_fluorescence_channel(fluorescence_channel, gain) for fluorescence_channel, gain in zip(filled, gains) ], @@ -1611,7 +1609,7 @@ async def measure( # ---------------------------------------- _SCANS = ("excitation", "emission") - # A spectrum's scan byte: read well by well, in no reading direction, as the vendor sends it. + # A spectrum's scan byte: read well by well, in no reading direction. # A plate read's scan byte (`0a`) makes the reader send pages of another layout. _SPECTRUM_SCAN_BYTE = 0x00 # A spectrum page's layout byte [6]. @@ -1668,7 +1666,7 @@ def _check_spectrum_arguments( errors.append(f"{name} must be a whole {lowest} to {highest} nm, got {value!r}") ends_valid = False if step_size != 1: - errors.append(f"step_size must be 1 nm, the only one captured, got {step_size!r}") + errors.append(f"step_size must be 1 nm, the only one verified, got {step_size!r}") if ends_valid: maximum = self.configuration.maximum_spectrum_wavelengths count = end_wavelength - start_wavelength + 1 @@ -1806,7 +1804,7 @@ async def measure_spectrum( emission_bandwidth: nm; None for `default_emission_bandwidth`. gain: the detector gain within `configuration.gain_range`. wells: the wells to measure, on the plate; None measures every well. - step_size: nm between wavelengths: 1, the only one captured. + step_size: nm between wavelengths: 1, the only one verified. focal_height: mm above the plate's base, within `configuration.focal_height_range`, in 0.1 mm steps. None reads at `default_focal_height_percent_of_well_depth` (25: a quarter up the well, for low and moderate fills). For full wells, or the most signal, focus on a @@ -1866,7 +1864,7 @@ async def measure_spectrum( focal_height=focal_height, kind=self._SPECTRUM, count=len(wavelengths), - chromatics=[ + encoded_fluorescence_channels=[ self._encode_fluorescence_channel(start, gain, selectors), self._encode_fluorescence_channel(end, gain, selectors), ], diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py index cffa814dd90..9154b7d960c 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py @@ -1,4 +1,4 @@ -"""Fluorescence tests over a fake io: what the driver sends is pinned to the vendor's frames. +"""Fluorescence tests over a fake io: what the driver sends is pinned to the ground-truth frames. Except for the band edges, which this driver sends at their nominal values, and the scan byte's bit 4, which it leaves clear in flying mode, enhanced dynamic range and a matrix. @@ -53,7 +53,7 @@ ) from pylabrobot.resources import Plate, Well, cor_96_wellplate_360uL_Fb, create_ordered_items_2d -# The vendor's default read: 485/15 to 528/20 nm, gain 1000, 8.5 mm, 10 flashes. +# The default read: 485/15 to 528/20 nm, gain 1000, 8.5 mm, 10 flashes. BASELINE_RUN_FRAME = bytes.fromhex( "02009c0c0431ec2166059604602c561d060c0800ffffffffffffffffffffffff00000000000000000000000000000000" "00000000000000000000000000000000000000000a000000000000000000000000000000000000000000000000000000" @@ -146,8 +146,8 @@ "008e000000940000008f0000008a000000960000008c00000090000000930000008e00000089003fd70d" ) -# The same read as the vendor's software reports it, row by row. -TEMPERATURE_37_EXPORT = [ +# The same read's ground-truth counts, row by row. +TEMPERATURE_37_GROUND_TRUTH = [ [141, 143, 165, 157, 147, 152, 150, 147, 149, 148, 2757, 149], [154, 151, 157, 155, 145, 147, 163, 155, 1816, 174, 2125, 149], [2735, 2009, 181, 2159, 1928, 1844, 1777, 1729, 1799, 191, 2133, 153], @@ -332,7 +332,7 @@ def make_fluorescence(io: FakeFTDI, plate: Optional[Plate] = None) -> Fluorescen return Fluorescence(driver) -# The filters of the captured reader, as its filter detection reports them. +# The filters of the ground-truth reader, as its filter detection reports them. EX_TR = OpticalFilter(1, "Ex TR", passbands=((282.2, 371.8),)) LP_TR = DichroicFilter(1, "LP TR", transmission_bands=((421.8, None),)) EM_615_18 = OpticalFilter(1, "615-18", passbands=((604.9, 623.2),)) @@ -341,8 +341,8 @@ def make_fluorescence(io: FakeFTDI, plate: Optional[Plate] = None) -> Fluorescen # The default read's fluorescence channel; the reads below vary it or the read-wide settings. BASELINE = FluorescenceChannel(excitation_wavelength=485, emission_wavelength=528) TEN_FLASHES = MeasurementSettings(flashes=10) -CAPTURED_FOCAL_HEIGHT = 8.5 # mm, the vendor's default, every capture's -CAPTURED_READS: Dict[str, Tuple[bytes, FluorescenceChannel, Dict[str, Any]]] = { +GROUND_TRUTH_FOCAL_HEIGHT = 8.5 # mm, every ground-truth read's +GROUND_TRUTH_READS: Dict[str, Tuple[bytes, FluorescenceChannel, Dict[str, Any]]] = { "default": (BASELINE_RUN_FRAME, BASELINE, {}), "540_590_nm": ( WAVELENGTHS_540_590_RUN_FRAME, @@ -403,13 +403,13 @@ def make_fluorescence(io: FakeFTDI, plate: Optional[Plate] = None) -> Fluorescen def with_nominal_edges( - captured: bytes, fluorescence_channels: Sequence[FluorescenceChannel] + ground_truth: bytes, fluorescence_channels: Sequence[FluorescenceChannel] ) -> bytes: - """`captured`'s payload with each monochromator's band edges at their nominal values. + """`ground_truth`'s payload with each monochromator's band edges at their nominal values. - The vendor software sends each band narrower (14.6 nm of 15, 19.4 nm of 20). + The ground truth has each band narrower (14.6 nm of 15, 19.4 nm of 20). """ - payload = bytearray(ftdi_extract_payload(captured)) + payload = bytearray(ftdi_extract_payload(ground_truth)) # Each fluorescence channel is gain, the excitation edges, the dichroic, the emission edges, the # wavelength selectors and `00 00 0c`; the last is followed by 19 bytes. first = len(payload) - 36 - 20 * (len(fluorescence_channels) - 1) @@ -432,7 +432,7 @@ def with_nominal_edges( def without_equidistant_cycles(payload: bytes) -> bytes: - """`payload` with scan byte bit 4 clear: the vendor's equidistant kinetic cycles, which this + """`payload` with scan byte bit 4 clear: equidistant kinetic cycles, which this driver does not send (flying mode `1e`, enhanced dynamic range and a matrix `1a`).""" patched = bytearray(payload) patched[64] &= ~0x10 # the family byte, the 63-byte plate block, then the scan byte @@ -447,7 +447,7 @@ async def read( io.replies = [STATUS_IDLE, RUN_REPLY, STATUS_MEASURING, STATUS_MEASURED, reply] feature = make_fluorescence(io, fluorescence_plate()) return feature, await feature.measure( - fluorescence_channel, **{"gain": 1000, "focal_height": CAPTURED_FOCAL_HEIGHT, **arguments} + fluorescence_channel, **{"gain": 1000, "focal_height": GROUND_TRUTH_FOCAL_HEIGHT, **arguments} ) @@ -470,7 +470,7 @@ async def test_a_set_default_bandwidth_is_sent_and_checked(self): replace(BASELINE, emission_wavelength=600), measurement_settings=TEN_FLASHES, gain=1000, - focal_height=CAPTURED_FOCAL_HEIGHT, + focal_height=GROUND_TRUTH_FOCAL_HEIGHT, ) given = replace(BASELINE, emission_wavelength=600, emission_bandwidth=100) self.assertEqual( # emission 600 nm, 100 nm wide @@ -478,8 +478,8 @@ async def test_a_set_default_bandwidth_is_sent_and_checked(self): with_nominal_edges(EMISSION_BANDWIDTH_100_RUN_FRAME, [given]), ) - async def test_captured_reads_are_sent_with_nominal_band_edges(self): - for name, (captured, fluorescence_channel, arguments) in CAPTURED_READS.items(): + async def test_ground_truth_reads_are_sent_with_nominal_band_edges(self): + for name, (ground_truth, fluorescence_channel, arguments) in GROUND_TRUTH_READS.items(): with self.subTest(name): io = FakeFTDI() _, result = await read( @@ -489,7 +489,8 @@ async def test_captured_reads_are_sent_with_nominal_band_edges(self): **{"measurement_settings": TEN_FLASHES, **arguments}, ) self.assertEqual( - ftdi_extract_payload(io.written[1]), with_nominal_edges(captured, [fluorescence_channel]) + ftdi_extract_payload(io.written[1]), + with_nominal_edges(ground_truth, [fluorescence_channel]), ) # The result says what the read used: a filter's passband, or the monochromator's band. excitation_filter, emission_filter = ( @@ -512,14 +513,14 @@ async def test_captured_reads_are_sent_with_nominal_band_edges(self): class ReadTests(unittest.IsolatedAsyncioTestCase): - async def test_counts_are_the_vendors_export(self): + async def test_counts_are_the_ground_truth(self): _, result = await read(FakeFTDI(), TEMPERATURE_37_DATA_REPLY) - exported = { + expected = { f"{'ABCDEFGH'[row]}{column + 1}": float(value) - for row, values in enumerate(TEMPERATURE_37_EXPORT) + for row, values in enumerate(TEMPERATURE_37_GROUND_TRUTH) for column, value in enumerate(values) } - self.assertEqual(result.data, exported) + self.assertEqual(result.data, expected) self.assertEqual((result.excitation_wavelength, result.emission_wavelength), (485, 528)) self.assertEqual(result.temperature, 37.0) @@ -588,7 +589,8 @@ async def test_out_of_range_reads_send_nothing(self): self.assertRaises(ValueError), ): await feature.measure( - fluorescence_channel, **{"gain": 1000, "focal_height": CAPTURED_FOCAL_HEIGHT, **arguments} + fluorescence_channel, + **{"gain": 1000, "focal_height": GROUND_TRUTH_FOCAL_HEIGHT, **arguments}, ) with self.assertRaises(ValueError) as raised: await feature.measure(replace(BASELINE, emission_wavelength=528.0), gain=fractional_gain) @@ -599,7 +601,7 @@ async def test_out_of_range_reads_send_nothing(self): class ExcitationAboveEmissionTests(unittest.IsolatedAsyncioTestCase): async def test_a_read_is_sent_with_the_split_between_the_bands_and_warns(self): - """Not captured: excitation 600/15 nm above emission 500/20 nm; the split lies midway between + """No ground truth: excitation 600/15 nm above emission 500/20 nm; the split lies midway between the bands' facing edges, 510.0 and 592.5 nm: 551.2 nm.""" io = FakeFTDI() with patch.object(measurement_module.logger, "warning") as warning: @@ -628,7 +630,7 @@ async def test_column_1_as_the_reader_read_it(self): io.replies.append(bytes.fromhex(session["fluorescence_column_1_reply"])) plate = cor_96_wellplate_360uL_Fb("plate") result = await make_fluorescence(io, plate).measure( - BASELINE, wells=plate.get_items("A1:H1"), gain=1000, focal_height=CAPTURED_FOCAL_HEIGHT + BASELINE, wells=plate.get_items("A1:H1"), gain=1000, focal_height=GROUND_TRUTH_FOCAL_HEIGHT ) self.assertEqual(io.written[1], bytes.fromhex(session["fluorescence_column_1_run"])) self.assertEqual(list(result.data.values()), [68.0, 62.0, 72.0, 64.0, 67.0, 69.0, 67.0, 68.0]) @@ -675,26 +677,26 @@ async def test_a_well_without_a_floor_thickness_sends_nothing(self): @functools.lru_cache(maxsize=None) -def load_captures() -> Dict[str, Any]: - """Larger frames, as hex: replies with the same read as the vendor's software reports it. +def load_ground_truth() -> Dict[str, Any]: + """Larger frames, as hex: replies with the same read's ground-truth counts. Row by row, one grid per fluorescence channel; each spectrum's replies in order. """ - path = os.path.join(os.path.dirname(__file__), "test_data", "fluorescence_captures.json.gz") + path = os.path.join(os.path.dirname(__file__), "test_data", "fluorescence_ground_truth.json.gz") with gzip.open(path, "rt") as f: return dict(json.load(f)) -def keyed_by_well(export: List[List[int]]) -> Dict[str, float]: - """An export grid, row by row, keyed by well identifier.""" +def keyed_by_well(grid: List[List[int]]) -> Dict[str, float]: + """A grid of counts, row by row, keyed by well identifier.""" return { f"{'ABCDEFGHIJKLMNOP'[row]}{column + 1}": float(value) - for row, values in enumerate(export) + for row, values in enumerate(grid) for column, value in enumerate(values) } -async def read_capture( +async def read_ground_truth( reply: bytes, fluorescence_channels: Union[FluorescenceChannel, Sequence[FluorescenceChannel]] = BASELINE, plate: Optional[Plate] = None, @@ -707,15 +709,15 @@ async def read_capture( feature = make_fluorescence(io, fluorescence_plate() if plate is None else plate) feature._driver.configuration.bottom_measurement_available = bottom_measurement_available return io, await feature.measure( - fluorescence_channels, **{"gain": 1000, "focal_height": CAPTURED_FOCAL_HEIGHT, **arguments} + fluorescence_channels, **{"gain": 1000, "focal_height": GROUND_TRUTH_FOCAL_HEIGHT, **arguments} ) class BottomOpticTests(unittest.IsolatedAsyncioTestCase): - async def test_a_bottom_read_is_the_vendors(self): - capture = load_captures()["bottom_optic"] - io, result = await read_capture( - bytes.fromhex(capture["reply"]), + async def test_a_bottom_read_is_the_ground_truth(self): + ground_truth = load_ground_truth()["bottom_optic"] + io, result = await read_ground_truth( + bytes.fromhex(ground_truth["reply"]), bottom_measurement_available=True, optic_position="bottom", measurement_settings=TEN_FLASHES, @@ -723,19 +725,19 @@ async def test_a_bottom_read_is_the_vendors(self): self.assertEqual( ftdi_extract_payload(io.written[1]), with_nominal_edges(BOTTOM_RUN_FRAME, [BASELINE]) ) - self.assertEqual(result.data, keyed_by_well(capture["export"])) + self.assertEqual(result.data, keyed_by_well(ground_truth["expected"])) async def test_the_default_optic_position_is_read_with_and_checked(self): - capture = load_captures()["bottom_optic"] + ground_truth = load_ground_truth()["bottom_optic"] io = FakeFTDI(chunk=4096) io.replies = [STATUS_IDLE, RUN_REPLY, STATUS_MEASURING, STATUS_MEASURED] - io.replies.append(bytes.fromhex(capture["reply"])) + io.replies.append(bytes.fromhex(ground_truth["reply"])) feature = make_fluorescence(io, fluorescence_plate()) feature._driver.configuration.bottom_measurement_available = True self.assertEqual(feature.default_optic_position, "top") feature.default_optic_position = "bottom" await feature.measure( - BASELINE, measurement_settings=TEN_FLASHES, gain=1000, focal_height=CAPTURED_FOCAL_HEIGHT + BASELINE, measurement_settings=TEN_FLASHES, gain=1000, focal_height=GROUND_TRUTH_FOCAL_HEIGHT ) self.assertEqual( ftdi_extract_payload(io.written[1]), with_nominal_edges(BOTTOM_RUN_FRAME, [BASELINE]) @@ -752,7 +754,7 @@ async def test_no_bottom_optic_or_an_unknown_one_sends_nothing(self): feature._driver.configuration.bottom_measurement_available = available with self.subTest(position), self.assertRaises(ValueError): await feature.measure( # type: ignore[call-overload] - BASELINE, optic_position=position, gain=1000, focal_height=CAPTURED_FOCAL_HEIGHT + BASELINE, optic_position=position, gain=1000, focal_height=GROUND_TRUTH_FOCAL_HEIGHT ) self.assertEqual(io.written, []) @@ -769,23 +771,26 @@ async def test_no_bottom_optic_or_an_unknown_one_sends_nothing(self): class FluorescenceChannelsTests(unittest.IsolatedAsyncioTestCase): - async def test_two_fluorescence_channels_are_the_vendors_read(self): - capture = load_captures()["two_fluorescence_channels"] - io, results = await read_capture( - bytes.fromhex(capture["reply"]), DUAL_FLUORESCENCE_CHANNELS, measurement_settings=TEN_FLASHES + async def test_two_fluorescence_channels_are_the_ground_truth(self): + ground_truth = load_ground_truth()["two_fluorescence_channels"] + io, results = await read_ground_truth( + bytes.fromhex(ground_truth["reply"]), + DUAL_FLUORESCENCE_CHANNELS, + measurement_settings=TEN_FLASHES, ) expected = with_nominal_edges(DUAL_RUN_FRAME, DUAL_FLUORESCENCE_CHANNELS) self.assertEqual(ftdi_extract_payload(io.written[1]), expected) self.assertEqual( - [result.data for result in results], [keyed_by_well(grid) for grid in capture["export"]] + [result.data for result in results], + [keyed_by_well(grid) for grid in ground_truth["expected"]], ) self.assertEqual([result.excitation_wavelength for result in results], [485, 610]) self.assertEqual([result.emission_bandwidth for result in results], [20, 50]) async def test_each_fluorescence_channel_is_sent_with_its_own_gain(self): - capture = load_captures()["two_fluorescence_channels"] - io, results = await read_capture( - bytes.fromhex(capture["reply"]), + ground_truth = load_ground_truth()["two_fluorescence_channels"] + io, results = await read_ground_truth( + bytes.fromhex(ground_truth["reply"]), DUAL_FLUORESCENCE_CHANNELS, gain=[1000, 1500], measurement_settings=TEN_FLASHES, @@ -818,16 +823,16 @@ async def test_out_of_range_fluorescence_channels_send_nothing(self): class FlyingModeTests(unittest.IsolatedAsyncioTestCase): - async def test_flying_mode_is_the_vendors_read(self): - capture = load_captures()["flying_mode"] - io, result = await read_capture( - bytes.fromhex(capture["reply"]), + async def test_flying_mode_is_the_ground_truth(self): + ground_truth = load_ground_truth()["flying_mode"] + io, result = await read_ground_truth( + bytes.fromhex(ground_truth["reply"]), flying_mode=True, measurement_settings=MeasurementSettings(settling_time=0, flashes=1), ) expected = with_nominal_edges(FLYING_MODE_RUN_FRAME, [BASELINE]) self.assertEqual(ftdi_extract_payload(io.written[1]), without_equidistant_cycles(expected)) - self.assertEqual(result.data, keyed_by_well(capture["export"])) + self.assertEqual(result.data, keyed_by_well(ground_truth["expected"])) async def test_what_flying_mode_cannot_read_sends_nothing(self): io = FakeFTDI() @@ -845,22 +850,22 @@ async def test_what_flying_mode_cannot_read_sends_nothing(self): enhanced_dynamic_range=enhanced, measurement_settings=settings, gain=1000, - focal_height=CAPTURED_FOCAL_HEIGHT, + focal_height=GROUND_TRUTH_FOCAL_HEIGHT, ) self.assertEqual(io.written, []) class EnhancedDynamicRangeTests(unittest.IsolatedAsyncioTestCase): - async def test_enhanced_dynamic_range_is_the_vendors_read(self): - capture = load_captures()["enhanced_dynamic_range"] - reply = bytes.fromhex(capture["reply"]) + async def test_enhanced_dynamic_range_is_the_ground_truth(self): + ground_truth = load_ground_truth()["enhanced_dynamic_range"] + reply = bytes.fromhex(ground_truth["reply"]) with patch.object(measurement_module.logger, "warning") as warning: - io, result = await read_capture( + io, result = await read_ground_truth( reply, enhanced_dynamic_range=True, measurement_settings=TEN_FLASHES ) expected = with_nominal_edges(ENHANCED_DYNAMIC_RANGE_RUN_FRAME, [BASELINE]) self.assertEqual(ftdi_extract_payload(io.written[1]), without_equidistant_cycles(expected)) - self.assertEqual(result.data, keyed_by_well(capture["export"])) + self.assertEqual(result.data, keyed_by_well(ground_truth["expected"])) decode = make_fluorescence(FakeFTDI())._decode_data_reply _, ceiling, _ = decode(ftdi_extract_payload(reply), 1, 96) self.assertEqual(ceiling, 700_000_000) @@ -876,7 +881,7 @@ async def test_with_orbital_or_spiral_averaging_sends_nothing(self): enhanced_dynamic_range=True, measurement_settings=MeasurementSettings(well_scan=well_scan), # type: ignore[arg-type] gain=1000, - focal_height=CAPTURED_FOCAL_HEIGHT, + focal_height=GROUND_TRUTH_FOCAL_HEIGHT, ) self.assertEqual(io.written, []) @@ -884,7 +889,7 @@ async def test_with_orbital_or_spiral_averaging_sends_nothing(self): class MatrixTests(unittest.IsolatedAsyncioTestCase): async def test_a_matrix_returns_every_point(self): plate = fluorescence_plate() - io, result = await read_capture( + io, result = await read_ground_truth( MATRIX_DATA_REPLY, plate=plate, wells=[plate.get_well("A1")], @@ -896,7 +901,7 @@ async def test_a_matrix_returns_every_point(self): self.assertEqual(ftdi_extract_payload(io.written[1]), without_equidistant_cycles(expected)) points = [145.0, 147.0, 142.0, 136.0, 144.0, 144.0, 137.0, 142.0, 139.0] self.assertEqual(result.matrix_points, {"A1": points}) - # The plain mean: the driver's choice, not the vendor's (its export reads 141). + # The plain mean: the driver's choice (the ground truth reads 141). self.assertEqual(result.data, {"A1": sum(points) / 9}) async def test_a_matrix_of_several_wells_sends_nothing(self): @@ -904,13 +909,13 @@ async def test_a_matrix_of_several_wells_sends_nothing(self): plate = fluorescence_plate() feature = make_fluorescence(io, plate) settings = MeasurementSettings(well_scan="matrix", matrix_size=3) - with self.assertRaises(ValueError): # no fluorescence matrix of several wells is captured + with self.assertRaises(ValueError): # no fluorescence matrix of several wells is verified await feature.measure( BASELINE, wells=plate.get_items("A1:B1"), measurement_settings=settings, gain=1000, - focal_height=CAPTURED_FOCAL_HEIGHT, + focal_height=GROUND_TRUTH_FOCAL_HEIGHT, ) self.assertEqual(io.written, []) @@ -940,10 +945,10 @@ def plate_384() -> Plate: class PlateTests(unittest.IsolatedAsyncioTestCase): - async def test_384_wells_are_the_vendors_read(self): - capture = load_captures()["plate_384"] - io, result = await read_capture( - bytes.fromhex(capture["reply"]), + async def test_384_wells_are_the_ground_truth(self): + ground_truth = load_ground_truth()["plate_384"] + io, result = await read_ground_truth( + bytes.fromhex(ground_truth["reply"]), plate=plate_384(), focal_height=5.0, measurement_settings=TEN_FLASHES, @@ -951,7 +956,7 @@ async def test_384_wells_are_the_vendors_read(self): self.assertEqual( ftdi_extract_payload(io.written[1]), with_nominal_edges(PLATE_384_RUN_FRAME, [BASELINE]) ) - self.assertEqual(result.data, keyed_by_well(capture["export"])) + self.assertEqual(result.data, keyed_by_well(ground_truth["expected"])) async def test_a_single_well_as_the_reader_answered(self): io = FakeFTDI(chunk=4096) @@ -962,14 +967,14 @@ async def test_a_single_well_as_the_reader_answered(self): wells=[plate.get_well("A1")], measurement_settings=TEN_FLASHES, gain=1000, - focal_height=CAPTURED_FOCAL_HEIGHT, + focal_height=GROUND_TRUTH_FOCAL_HEIGHT, ) expected = with_nominal_edges(SINGLE_WELL_RUN_FRAME, [BASELINE]) self.assertEqual(ftdi_extract_payload(io.written[1]), expected) self.assertEqual(result.data, {"A1": 145.0}) -# The spectra's fixed excitation or emission and their scans, as the captures read them. +# The spectra's fixed excitation or emission and their scans, as the ground truth reads them. SPECTRUM_WELLS = ["A1", "B1", "C2", "D2"] EXCITATION_SPECTRUM: Dict[str, Any] = dict( scan="excitation", @@ -990,13 +995,13 @@ async def test_a_single_well_as_the_reader_answered(self): def spectrum_replies(name: str) -> List[bytes]: - """The reader's replies to capture `name`'s RUN, status polls and data requests, in order: each - page after a status with data ready, the last status idle.""" - return [bytes.fromhex(frame) for frame in load_captures()[name]["replies"]] + """The reader's replies to ground truth `name`'s RUN, status polls and data requests, in order: + each page after a status with data ready, the last status idle.""" + return [bytes.fromhex(frame) for frame in load_ground_truth()[name]["replies"]] def spectrum_pages(name: str) -> List[bytes]: - """The data pages of capture `name`, in the order received.""" + """The data pages of ground truth `name`, in the order received.""" replies = [ftdi_extract_payload(reply) for reply in spectrum_replies(name)] return [reply for reply in replies if reply[0] == 0x02] @@ -1021,11 +1026,11 @@ async def measure(self, replies: List[bytes], arguments: Dict[str, Any]): feature = make_fluorescence(io, plate) result = await feature.measure_spectrum( wells=[plate.get_well(well) for well in SPECTRUM_WELLS], - **{"gain": 1000, "focal_height": CAPTURED_FOCAL_HEIGHT, **arguments}, + **{"gain": 1000, "focal_height": GROUND_TRUTH_FOCAL_HEIGHT, **arguments}, ) return io, result - async def test_an_excitation_spectrum_is_the_vendors_read(self): + async def test_an_excitation_spectrum_is_the_ground_truth(self): io, result = await self.measure(spectrum_replies("excitation_spectrum"), EXCITATION_SPECTRUM) fluorescence_channels = [ FluorescenceChannel( @@ -1044,11 +1049,11 @@ async def test_an_excitation_spectrum_is_the_vendors_read(self): self.assertEqual((result.excitation_wavelength, result.excitation_bandwidth), (None, 10)) self.assertEqual((result.emission_wavelength, result.emission_bandwidth), (800, 20)) self.assertEqual(result.gain, 1000) - # The raw counts: the vendor's export is spectrally corrected. + # The raw counts: the ground truth's are spectrally corrected. self.assertEqual(counts_at(result, 400), {"A1": 41.0, "B1": 31.0, "C2": 22.0, "D2": 25.0}) self.assertEqual(list(counts_at(result, 620).values()), [46.0, 14.0, 17.0, 22.0]) - async def test_an_emission_spectrum_is_the_vendors_read(self): + async def test_an_emission_spectrum_is_the_ground_truth(self): io, result = await self.measure(spectrum_replies("emission_spectrum"), EMISSION_SPECTRUM) fluorescence_channels = [ FluorescenceChannel( @@ -1100,7 +1105,7 @@ async def test_a_short_page_stops_the_run_and_the_plate_is_read_again(self): replies = spectrum_replies("emission_spectrum") first_page = next(index for index, reply in enumerate(replies) if reply[4] == 0x02) # Not a reader's reply: the first page cut short, as an absorbance page arrived - # (ShortPageTests); no short fluorescence page is captured. + # (ShortPageTests); no short fluorescence page is verified. short = replies[:first_page] + [replies[first_page][:200], STOP_REPLY] io, result = await self.measure(short + replies, EMISSION_SPECTRUM) runs = [frame for frame in io.written if ftdi_extract_payload(frame)[0] == 0x04] @@ -1143,7 +1148,7 @@ async def test_out_of_range_spectra_send_nothing(self): for arguments in refused: with self.subTest(arguments=arguments), self.assertRaises(ValueError): await feature.measure_spectrum( - **{"gain": 1000, "focal_height": CAPTURED_FOCAL_HEIGHT, **arguments} + **{"gain": 1000, "focal_height": GROUND_TRUTH_FOCAL_HEIGHT, **arguments} ) self.assertEqual(io.written, []) @@ -1165,7 +1170,7 @@ def test_no_fluorescence_is_not_built(self): def test_the_wavelength_limits_follow_the_photomultipliers(self): # (extended red, dual, crossed over): the highest wavelength and dichroic split. for photomultipliers, highest in ( - ((True, True, False), (840, 760.0)), # the captured reader + ((True, True, False), (840, 760.0)), # the ground-truth reader ((True, True, True), (740, 740.0)), ((True, False, False), (840, 760.0)), ((False, False, False), (740, 740.0)), @@ -1193,7 +1198,9 @@ def load_filter_detection() -> List[Dict[str, Any]]: Each holds its `24` frame and the replies the driver reads, as hex: the status before the `24`, the reply to it, to `80` until data ready, to `05 11`, to `05 1b`, then the next status. """ - path = os.path.join(os.path.dirname(__file__), "test_data", "filter_detection_captures.json.gz") + path = os.path.join( + os.path.dirname(__file__), "test_data", "filter_detection_ground_truth.json.gz" + ) with gzip.open(path, "rt") as f: return list(json.load(f)["positions"]) @@ -1218,7 +1225,7 @@ def decode_filter_result(reply: bytes, slide: str) -> Tuple[int, int, int, int, def make_detecting_fluorescence(io: FakeFTDI) -> Fluorescence: - """A fluorescence feature of the captured reader, answered by the whole detection capture.""" + """A fluorescence feature of the ground-truth reader, answered by its whole filter detection.""" io.replies = [ bytes.fromhex(reply) for position in load_filter_detection() for reply in position["replies"] ] @@ -1254,7 +1261,7 @@ async def test_detection_as_the_reader_answered(self): self.assertEqual(io.replies, []) def test_long_pass_and_short_pass_filters_are_decoded(self): - """No long-pass or short-pass filter is captured: the replies are real results edited.""" + """No long-pass or short-pass filter has ground truth: the replies are real results edited.""" # Not a reader's reply: excitation 1's without its bandwidth, then also without its first edge. long_pass = bytearray(detection_result("excitation", 1)) long_pass[9:11] = bytes(2) @@ -1313,7 +1320,7 @@ async def test_no_spectrometer_or_extended_red_sends_nothing(self): def make_detected_slides(fluorescence_feature: Fluorescence) -> None: - """Give the feature the reader's slides as it detected them, named as the vendor names them.""" + """Give the feature the reader's slides as it detected them, with their filters' names.""" excitation = ExcitationFilterSlide(4) dichroic = DichroicFilterSlide(3) emission = EmissionFilterSlide(4) @@ -1336,7 +1343,7 @@ def make_detected_slides(fluorescence_feature: Fluorescence) -> None: async def read_with_slides(io: FakeFTDI, fluorescence_channel: FluorescenceChannel, **arguments): - """A whole-plate read through `fluorescence_channel`, gain 1000, the captured slides; feature, + """A whole-plate read through `fluorescence_channel`, gain 1000, the ground-truth slides; feature, result.""" io.replies = [ STATUS_IDLE, @@ -1348,12 +1355,12 @@ async def read_with_slides(io: FakeFTDI, fluorescence_channel: FluorescenceChann feature = make_fluorescence(io, fluorescence_plate()) make_detected_slides(feature) return feature, await feature.measure( - fluorescence_channel, **{"gain": 1000, "focal_height": CAPTURED_FOCAL_HEIGHT, **arguments} + fluorescence_channel, **{"gain": 1000, "focal_height": GROUND_TRUTH_FOCAL_HEIGHT, **arguments} ) class FilterSlideTests(unittest.IsolatedAsyncioTestCase): - async def test_slots_and_names_are_the_vendors_reads(self): + async def test_slots_and_names_are_the_ground_truth(self): """The three filters' read by slot and by name; the one-filter reads by slot.""" reads: Dict[str, Tuple[bytes, List[FluorescenceChannel]]] = { "three_filters": ( @@ -1374,14 +1381,14 @@ async def test_slots_and_names_are_the_vendors_reads(self): [FluorescenceChannel(excitation_filter=1, dichroic_filter=1, emission_wavelength=528)], ), } - for name, (captured, fluorescence_channels) in reads.items(): + for name, (ground_truth, fluorescence_channels) in reads.items(): for fluorescence_channel in fluorescence_channels: with self.subTest(name, fluorescence_channel=fluorescence_channel): io = FakeFTDI() _, result = await read_with_slides( io, fluorescence_channel, measurement_settings=TEN_FLASHES ) - expected = with_nominal_edges(captured, [fluorescence_channel]) + expected = with_nominal_edges(ground_truth, [fluorescence_channel]) self.assertEqual(ftdi_extract_payload(io.written[1]), expected) if fluorescence_channel.excitation_filter is not None: self.assertEqual(result.excitation_wavelength, (282.2 + 371.8) / 2) @@ -1389,8 +1396,8 @@ async def test_slots_and_names_are_the_vendors_reads(self): self.assertEqual(result.emission_bandwidth, 623.2 - 604.9) async def test_every_slot_is_its_detection_position(self): - """No RUN through slots 2 to 4 or dichroic B and C is captured: their RUN fields are the slide - positions of the detection's `24`.""" + """No RUN through slots 2 to 4 or dichroic B and C has ground truth: their RUN fields are the + slide positions of the detection's `24`.""" words = { (slide, slot): [int.from_bytes(request[i : i + 2], "big") for i in range(10, 28, 2)] for (slide, slot), request in zip( @@ -1447,24 +1454,24 @@ async def test_unknown_slots_and_names_send_nothing(self): self.assertEqual(io.written, []) -def load_z_profile_captures() -> Dict[str, Any]: +def load_z_profile_ground_truth() -> Dict[str, Any]: """The reader's z-scans: of A1 without a target, of D6 to 90 %, and a brightest-well search. Each holds its frame (`09`; the search `0c`, which the driver does not send, kept for its result) and the replies the driver reads, as hex: the status before it, the reply to it, to `80`, to `05 05`. """ - path = os.path.join(os.path.dirname(__file__), "test_data", "z_profile_captures.json.gz") + path = os.path.join(os.path.dirname(__file__), "test_data", "z_profile_ground_truth.json.gz") with gzip.open(path, "rt") as f: return dict(json.load(f)) -def with_nominal_z_scan_edges(captured: bytes) -> bytes: - """`captured`'s payload with 485/15 and 528/20 nm at their nominal edges (`with_nominal_edges`). +def with_nominal_z_scan_edges(ground_truth: bytes) -> bytes: + """`ground_truth`'s payload with 485/15 and 528/20 nm at their nominal edges. - The optics are the last 14 bytes. + As `with_nominal_edges`; the optics are the last 14 bytes. """ - payload = bytearray(ftdi_extract_payload(captured)) + payload = bytearray(ftdi_extract_payload(ground_truth)) for offset, edges in ((-14, _compute_band_edges(485, 15)), (-8, _compute_band_edges(528, 20))): for index, value in enumerate(edges): start = len(payload) + offset + 2 * index @@ -1473,18 +1480,19 @@ def with_nominal_z_scan_edges(captured: bytes) -> bytes: async def run_z_profile(name: str, well: str, **arguments): - """Capture `name`'s z-scan of `well` answered by its replies; the io, the capture, the result.""" - capture = load_z_profile_captures()[name] + """Ground truth `name`'s z-scan of `well` answered by its replies; the io, the ground truth, + the result.""" + ground_truth = load_z_profile_ground_truth()[name] io = FakeFTDI(chunk=4096) - io.replies = [bytes.fromhex(reply) for reply in capture["replies"]] + io.replies = [bytes.fromhex(reply) for reply in ground_truth["replies"]] plate = fluorescence_plate() feature = make_fluorescence(io, plate) result = await feature.measure_z_profile(BASELINE, well=plate.get_well(well), **arguments) - return io, capture, result + return io, ground_truth, result class ZProfileTests(unittest.IsolatedAsyncioTestCase): - async def test_z_scans_are_sent_and_decoded_as_the_vendors(self): + async def test_z_scans_are_sent_and_decoded_as_the_ground_truth(self): """A1 without a target and D6 to 90 %.""" z_scans = { "z_profile_without_target": ("A1", None, (15.0, None, "A1", None, None, 60920)), @@ -1492,10 +1500,10 @@ async def test_z_scans_are_sent_and_decoded_as_the_vendors(self): } for name, (well, target, expected) in z_scans.items(): with self.subTest(name): - io, capture, result = await run_z_profile(name, well, gain_target_percent=target) + io, ground_truth, result = await run_z_profile(name, well, gain_target_percent=target) self.assertEqual( ftdi_extract_payload(io.written[1]), - with_nominal_z_scan_edges(bytes.fromhex(capture["request"])), + with_nominal_z_scan_edges(bytes.fromhex(ground_truth["request"])), ) self.assertEqual( ( @@ -1521,7 +1529,7 @@ def test_a_height_below_0_is_decoded(self): """Not a reader's reply: the brightest-well search's result, last point and peak -0.60 mm.""" reply = bytearray( ftdi_extract_payload( - bytes.fromhex(load_z_profile_captures()["brightest_well_search"]["replies"][-1]) + bytes.fromhex(load_z_profile_ground_truth()["brightest_well_search"]["replies"][-1]) ) ) points = int.from_bytes(reply[23:25], "big") @@ -1538,7 +1546,7 @@ def test_a_height_below_0_is_decoded(self): def test_a_reply_not_of_this_plate_is_refused(self): """The brightest-well search's result, B11, cut short; then of column 13 (not a reply).""" reply = ftdi_extract_payload( - bytes.fromhex(load_z_profile_captures()["brightest_well_search"]["replies"][-1]) + bytes.fromhex(load_z_profile_ground_truth()["brightest_well_search"]["replies"][-1]) ) off_plate = bytearray(reply) off_plate[15] = 13 @@ -1550,9 +1558,9 @@ def test_a_reply_not_of_this_plate_is_refused(self): async def test_a_z_scan_ending_without_its_result_stops(self): """Not a reader's reply: an idle status with byte 2 bit 5 clear right after the `09`.""" - capture = load_z_profile_captures()["z_profile_without_target"] + ground_truth = load_z_profile_ground_truth()["z_profile_without_target"] io = FakeFTDI() - replies = [bytes.fromhex(reply) for reply in capture["replies"]] + replies = [bytes.fromhex(reply) for reply in ground_truth["replies"]] io.replies = [replies[0], replies[1], replies[0], STOP_REPLY] plate = fluorescence_plate() with self.assertRaises(RuntimeError): diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence.py b/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence.py new file mode 100644 index 00000000000..6e266a6451c --- /dev/null +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence.py @@ -0,0 +1,375 @@ +"""CLARIOstar luminescence: the light a sample emits, through the emission monochromator. + +A basic plate read: the top optic, each well at its centre, enhanced dynamic range on, one +emission band through the monochromator. The focal height, gain, emission band, normalization time, +measurement interval, settling time, reading direction and shaking are sent; the bytes between them +are fixed. +""" + +import logging +from dataclasses import dataclass, replace +from datetime import datetime +from typing import TYPE_CHECKING, List, Optional, Sequence, Tuple + +from pylabrobot.lib.luminescence.results import LuminescenceResult +from pylabrobot.resources import Plate, Well + +from .fluorescence import Fluorescence, _compute_band_edges +from .measurement import ( + Measurement, + MeasurementConfiguration, + MeasurementSettings, + ShakeSettings, + _get_measured_identifiers, + _get_readings_by_well, +) + +if TYPE_CHECKING: + from ..configuration import DeviceConfiguration + from ..master import CLARIOstarDriver + +logger = logging.getLogger(__name__) + + +# ---------------------------------------- +# Results and device facts +# ---------------------------------------- + + +@dataclass +class CLARIOstarLuminescenceResult(LuminescenceResult): + """A CLARIOstar luminescence result, with the detector gain it was read with. + + Attributes: + gain: the detector gain, within `LuminescenceConfiguration.gain_range`. + """ + + gain: int + + +@dataclass +class LuminescenceConfiguration(MeasurementConfiguration): + """Device facts of the luminescence measurement.""" + + wavelength_range: Tuple[int, int] = (320, 840) + """Emission wavelength, in nm: the emission monochromator's (up to 740 without the extended-red + photomultiplier on the luminescence path, `Luminescence._get_reader_configuration`).""" + bandwidth_range: Tuple[int, int] = (8, 100) + """Emission bandwidth, in nm: the emission monochromator's, as for fluorescence.""" + gain_range: Tuple[int, int] = (0, 4095) + """Detector gain, as for fluorescence (2000 and 3500 read on hardware).""" + focal_height_range: Tuple[float, float] = (0.0, 25.0) + """Focal height, in mm, in 0.1 mm steps (0 to 25, as for fluorescence).""" + time_range: Tuple[float, float] = (0.01, 655.35) + """Normalization time and measurement interval, in s, in 0.01 s steps: what their u16 field + holds. The reader's own limits are not known; 0.20, 0.80, 1.00 and 2.00 s read on hardware.""" + + +class Luminescence(Measurement): + """Luminescence measurement. + + Reached as `driver.luminescence`, built at setup only if the reader reports a luminescence + measurement channel. + """ + + configuration: LuminescenceConfiguration + + _MODE = 0x01 + # No `_WELL_SCAN_CODE`: a luminescence read is a point read, without a well-scan field. + _TRAILER_START = 0x00 + + @classmethod + def _get_reader_configuration(cls, device: "DeviceConfiguration") -> LuminescenceConfiguration: + """The luminescence facts of the reader `device` describes: its wavelength limits. + + The extended range needs the extended-red photomultiplier on the luminescence path: fitted, + and on a dual-photomultiplier reader crossed over; otherwise fluorescence's standard range. + Unknown hardware keeps the defaults. + """ + configuration = LuminescenceConfiguration() + standard = device.extended_red_photomultiplier_tube is False or bool( + device.dual_photomultiplier_tubes and device.photomultiplier_tube_crossover is False + ) + if standard: + return replace(configuration, wavelength_range=Fluorescence._STANDARD_WAVELENGTH_RANGE) + return configuration + + def __init__( + self, driver: "CLARIOstarDriver", configuration: Optional[LuminescenceConfiguration] = None + ): + """Build the luminescence feature of `driver`. + + Args: + driver: the driver to send commands through. + configuration: the luminescence's device facts; None for the reader's + (`_get_reader_configuration`). + """ + super().__init__(driver, configuration or self._get_reader_configuration(driver.configuration)) + self._default_focal_height_percent_of_well_depth = 25.0 + + # ---------------------------------------- + # Defaults + # ---------------------------------------- + + @property + def default_focal_height_percent_of_well_depth(self) -> float: + """The focal height of a read given none, in percent of the well depth above its floor.""" + return self._default_focal_height_percent_of_well_depth + + @default_focal_height_percent_of_well_depth.setter + def default_focal_height_percent_of_well_depth(self, value: float) -> None: + """Set the default, raising unless it is 0 to 100 %.""" + self._check_default_focal_height_percent(value) + self._default_focal_height_percent_of_well_depth = value + + def _get_default_focal_height(self, wells: Sequence[Well]) -> float: + """The focal height `default_focal_height_percent_of_well_depth` up `wells`, mm.""" + return self._compute_default_focal_height( + wells, self._default_focal_height_percent_of_well_depth, self.configuration.focal_height_range + ) + + # ---------------------------------------- + # Discrete reads + # ---------------------------------------- + + # -- checks -------------------------------------------------------------------------------------- + + def _check_arguments( + self, + emission_wavelength: int, + emission_bandwidth: int, + gain: int, + focal_height: Optional[float], + normalization_time: float, + measurement_interval: float, + measurement: MeasurementSettings, + *, + timeout: Optional[float] = None, + ) -> None: + """Raise if the emission band, gain, focal height, times or read settings are out of range. + + The arguments are `measure`'s. + + Raises: + ValueError: listing every violation: an emission wavelength not a whole nm within + `wavelength_range`, `_check_bandwidth`'s, `_check_gain`'s, `_check_focal_height`'s (None + is the default's, checked with the plate), a normalization time or measurement interval + not in 0.01 s steps within `time_range`, a well scan other than a point, a pause before + reading, a timeout not positive. + """ + errors = [] + lowest, highest = self.configuration.wavelength_range + if not isinstance(emission_wavelength, int) or not lowest <= emission_wavelength <= highest: + errors.append( + f"emission_wavelength must be a whole {lowest} to {highest} nm, got {emission_wavelength!r}" + ) + errors += self._check_bandwidth( + "emission_bandwidth", emission_bandwidth, self.configuration.bandwidth_range + ) + errors += self._check_gain(gain, self.configuration.gain_range) + if focal_height is not None: + errors += self._check_focal_height(focal_height, self.configuration.focal_height_range) + shortest, longest = self.configuration.time_range + for name, value in ( + ("normalization_time", normalization_time), + ("measurement_interval", measurement_interval), + ): + if not shortest <= value <= longest or abs(value * 100 - round(value * 100)) > 1e-9: + errors.append(f"{name} must be {shortest} to {longest} s in 0.01 s steps, got {value}") + if measurement.well_scan != "point": + # Only point reads of luminescence are verified. + errors.append( + f"luminescence reads each well at its centre (well_scan 'point'), got " + f"{measurement.well_scan!r}" + ) + if measurement.pause_before_reading is not None: + errors.append("a pause before reading is not supported for luminescence") + if timeout is not None and not timeout > 0: + errors.append(f"timeout must be positive, got {timeout}") + if errors: + raise ValueError("; ".join(errors)) + + # -- RUN parameters ------------------------------------------------------------------------------ + + # Between the focal height and the gain, and between the gain and the emission edges; not + # decoded. + _AFTER_FOCAL_HEIGHT = bytes.fromhex("0000010000") + _AFTER_GAIN = bytes.fromhex("000100010001") + # Between the emission edges and the normalization time; not decoded. + _AFTER_EMISSION = bytes.fromhex("00050000") + # Between the normalization time and the run end; not decoded. + _BEFORE_RUN_END = bytes(4) + + def _compute_run_parameters( + self, + plate: Plate, + wells: List[Well], + measurement: MeasurementSettings, + shake: Optional[ShakeSettings], + *, + emission_wavelength: int, + emission_bandwidth: int, + gain: int, + focal_height: float, + normalization_time: float, + measurement_interval: float, + ) -> bytes: + """The RUN parameters of a luminescence read. + + The run start (`_compute_run_start`, with enhanced dynamic range), the focal height in + 0.01 mm, the gain u16 BE, the emission band's nominal edges in 0.1 nm (`_compute_band_edges`), + the normalization time in 0.01 s, then the run end with the measurement interval in 0.01 s + where an absorbance or fluorescence read has its flashes. + """ + upper, lower = _compute_band_edges(emission_wavelength, emission_bandwidth) + return ( + self._compute_run_start(plate, wells, measurement, shake, enhanced_dynamic_range=True) + + round(focal_height * 100).to_bytes(2, "big") + + self._AFTER_FOCAL_HEIGHT + + gain.to_bytes(2, "big") + + self._AFTER_GAIN + + upper.to_bytes(2, "big") + + lower.to_bytes(2, "big") + + self._AFTER_EMISSION + + round(normalization_time * 100).to_bytes(2, "big") + + self._BEFORE_RUN_END + + self._compute_run_end(round(measurement_interval * 100)) + ) + + # -- decoding ------------------------------------------------------------------------------------ + + # Data reply layout byte [6], of every luminescence read on hardware. + _DATA_LAYOUT = 0x25 + + def _decode_data_reply(self, reply: bytes, wells: int) -> Tuple[List[int], int, Optional[float]]: + """The counts row by row, the detector's ceiling, and the temperature. + + - Bytes 7-8, 9-10 and 18-19: the values, one per well. + - Bytes 11-14: the ceiling a saturated count reads (210,000,000 on hardware). + - Bytes 32-33: the temperature in 0.1 C, 0 when not monitored. + - From byte 34: u32 BE counts, one per well. + + Raises: + ValueError: if the reply is incomplete or not the layout of this read. + """ + if len(reply) < 34: + raise ValueError(f"luminescence reply too short: {len(reply)} bytes") + if reply[6] != self._DATA_LAYOUT: + raise ValueError(f"luminescence reply layout 0x{reply[6]:02x} is not decoded") + announced = int.from_bytes(reply[7:9], "big") + if announced != wells or int.from_bytes(reply[18:20], "big") != wells: + raise ValueError(f"luminescence reply announces {announced} values; expected {wells} wells") + if int.from_bytes(reply[9:11], "big") != wells or len(reply) < 34 + 4 * wells: + raise ValueError("luminescence reply is incomplete") + counts = [int.from_bytes(reply[34 + 4 * i : 38 + 4 * i], "big") for i in range(wells)] + raw_temperature = int.from_bytes(reply[32:34], "big") + temperature = raw_temperature / 10 if raw_temperature else None + return counts, int.from_bytes(reply[11:15], "big"), temperature + + # -- measuring ----------------------------------------------------------------------------------- + + 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, + ) -> CLARIOstarLuminescenceResult: + """Measure the plate's luminescence through the emission monochromator. + + Reads from the top with enhanced dynamic range on, each well at its centre. Cancelling the call + stops the run on the reader. No luminescent sample has been read with this driver yet: the + first read warns. + + Args: + wells: the wells to measure, on the plate; None measures every well. + emission_wavelength: whole nm within `configuration.wavelength_range`. + emission_bandwidth: whole nm within `configuration.bandwidth_range`. + gain: the detector gain within `configuration.gain_range`. + focal_height: mm above the plate's base, within `configuration.focal_height_range`, in + 0.1 mm steps. None reads at `default_focal_height_percent_of_well_depth` (25: a quarter + up the well, for low and moderate fills). + normalization_time: s the counts are normalized to, in 0.01 s steps within + `configuration.time_range` (2.00 roughly doubled the counts of 1.00 on hardware). + measurement_interval: s each well is measured, in 0.01 s steps within + `configuration.time_range` (at 0.20 a column-1 read, 8 wells, took 9.5 s on hardware + instead of 14.4 s). + measurement_settings: how to read: a point read without a pause; its flashes do not apply. + None reads with the defaults but no settling time. + shake_settings: shake the plate before the read; None does not shake. + timeout: seconds the run may take; `default_read_timeout` when None. The defaults read a + 96-well plate in about 105 s. + + Returns: + the counts of the measured wells, with the emission band and the gain they were read with. + + Raises: + RuntimeError: if no plate is in the plate holder, or the run ends without its data. + ValueError: if an argument is out of range (`_check_arguments`), the settings cannot read + the wells of the plate, or the reply is not this read's (`_decode_data_reply`). + TimeoutError: if the run takes longer than `timeout` seconds. + """ + measurement = ( + MeasurementSettings(settling_time=0.0) + if measurement_settings is None + else measurement_settings + ) + self._check_arguments( + 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, + ) + timeout = self.default_read_timeout if timeout is None else timeout + # TODO(device): no luminescent sample has been read yet, only an empty plate. + self._warn_untested("luminescence") + logger.info( + "luminescence: %d/%d nm, gain %d, %.2f s per well, %d of %d wells of %s", + emission_wavelength, + emission_bandwidth, + gain, + measurement_interval, + len(wells), + plate.num_items, + plate.name, + ) + start_timestamp = datetime.now() + replies, end_timestamp = await self._run_measurement(parameters, timeout) + identifiers = _get_measured_identifiers(plate, wells) + counts, ceiling, temperature = self._decode_data_reply(replies[0], len(identifiers)) + self._warn_saturated("luminescence", counts, ceiling) + return CLARIOstarLuminescenceResult( + data=_get_readings_by_well(plate, wells, [float(count) for count in counts]), + emission_wavelength=emission_wavelength, + emission_bandwidth=emission_bandwidth, + gain=gain, + temperature=temperature, + start_timestamp=start_timestamp, + end_timestamp=end_timestamp, + ) diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence_tests.py new file mode 100644 index 00000000000..b3fa5afc47b --- /dev/null +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence_tests.py @@ -0,0 +1,222 @@ +"""Luminescence tests over a fake io: what the driver sends is pinned to the ground truth, but for +the plate's geometry (a PLR plate; the plate block is tested with the other modes) and the emission +band's edges, which this driver sends at their nominal values; the counts are the ground truth's. + +What every mode shares (plate block, scan and settling bytes, shaking, the run and its stop) is +tested with absorbance and fluorescence; only luminescence's own fields and reply are here. +""" + +import unittest +from typing import Any, Dict, List + +from pylabrobot.bmg_labtech.clariostar.driver.configuration_tests import decoded +from pylabrobot.bmg_labtech.clariostar.driver.features.absorbance_tests import ( + RUN_REPLY, + STATUS_MEASURED, + STATUS_MEASURING, +) +from pylabrobot.bmg_labtech.clariostar.driver.features.drawer_tests import make_plate_holder +from pylabrobot.bmg_labtech.clariostar.driver.features.fluorescence_tests import ( + TEMPERATURE_37_DATA_REPLY, +) +from pylabrobot.bmg_labtech.clariostar.driver.features.luminescence import Luminescence +from pylabrobot.bmg_labtech.clariostar.driver.features.measurement import ( + Measurement, + MeasurementSettings, +) +from pylabrobot.bmg_labtech.clariostar.driver.framing import ftdi_extract_payload +from pylabrobot.bmg_labtech.clariostar.driver.master_tests import ( + STATUS_IDLE, + FakeFTDI, + make_driver, +) +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, +# normalization 1.00 s, measurement interval 0.80 s. +LUMINESCENCE_RUN_FRAME = bytes.fromhex( + "02009a0c0431e82164059604602c521d040c0800ffffffffffffffffffffffff00000000000000000000000000000000" + "00000000000000000000000000000000000000000a014000000000000000000000000000000000000000000000000000" + "00000000270f270f01035200000100000dac0001000100011834151c0005000000640000000000000000000001000000" + "0100500001000012f30d" +) + +# Its data reply: layout 25, 96 values, ceiling 210,000,000, no temperature. +LUMINESCENCE_DATA_REPLY = bytes.fromhex( + "0201aa0c02050626000025006000600c845880010001006001000100000001010000000000000000001c0000001c0000" + "00150000001200000011000000210000002100000010000000170000001400000008000000280000001e000000120000" + "001f00000012000000120000002300000015000000110000001f0000001a00000010000000160000000b000000100000" + "001b0000000d0000001f0000000b000000190000002300000029000000170000000f0000001900000020000000210000" + "0010000000100000000c000000110000001f0000001b00000016000000100000001f0000001400000017000000100000" + "001500000012000000150000001e000000190000001f0000001200000025000000120000001b000000150000001a0000" + "00100000003300000024000000230000001900000016000000150000001200000011000000150000001b000000160000" + "00370000001e000000190000000f000000160000001a0000001b0000001a000000140000001000000019000000200000" + "0012000000100000001700000016000000140000001200000015000000120000001100000011000c790d" +) + +# The ground truth's counts, row by row. +LUMINESCENCE_GROUND_TRUTH = [ + [28, 28, 21, 18, 17, 33, 33, 16, 23, 20, 8, 40], + [30, 18, 31, 18, 18, 35, 21, 17, 31, 26, 16, 22], + [11, 16, 27, 13, 31, 11, 25, 35, 41, 23, 15, 25], + [32, 33, 16, 16, 12, 17, 31, 27, 22, 16, 31, 20], + [23, 16, 21, 18, 21, 30, 25, 31, 18, 37, 18, 27], + [21, 26, 16, 51, 36, 35, 25, 22, 21, 18, 17, 21], + [27, 22, 55, 30, 25, 15, 22, 26, 27, 26, 20, 16], + [25, 32, 18, 16, 23, 22, 20, 18, 21, 18, 17, 17], +] + + +def plate() -> Plate: + """The plate the tests read: a Corning 96 360 uL flat-bottom.""" + return cor_96_wellplate_360uL_Fb("plate") + + +def make_luminescence(io: FakeFTDI, plate: Plate) -> Luminescence: + """A luminescence feature whose drawer holds `plate` in its plate holder.""" + driver = make_driver(io) + driver.configuration = decoded() + driver.drawer.plate_holder = make_plate_holder() + driver.drawer.plate_holder.assign_child_resource(plate) + return Luminescence(driver) + + +def with_fields(ground_truth: bytes, fields: Dict[int, int]) -> bytes: + """`ground_truth`'s payload on `plate()`: its plate geometry, and each u16 in `fields` replaced. + + `fields` maps payload offsets to values. + """ + payload = bytearray(ftdi_extract_payload(ground_truth)) + payload[1:13] = Measurement._encode_plate_geometry(plate())[:12] + for offset, value in fields.items(): + payload[offset : offset + 2] = value.to_bytes(2, "big") + return bytes(payload) + + +# Payload offsets of the luminescence fields, after the plate block, scan byte, pre-separator, +# `27 0f 27 0f` and the settling byte. +FOCAL_HEIGHT = 101 +GAIN = 108 +EMISSION_EDGES = 116 +NORMALIZATION_TIME = 124 +MEASUREMENT_INTERVAL = 141 + +# mm, the ground truth's. +GROUND_TRUTH_FOCAL_HEIGHT = 8.5 + +# The ground truth as this driver sends it: the emission band at its nominal edges, 620.0 / +# 540.0 nm; the ground truth has 619.6 / 540.4. +NOMINAL_EDGES = {EMISSION_EDGES: 6200, EMISSION_EDGES + 2: 5400} + + +async def read(io: FakeFTDI, **arguments): + """A read of the whole plate answered by the ground truth's reply; the result.""" + io.replies = [STATUS_IDLE, RUN_REPLY, STATUS_MEASURING, STATUS_MEASURED, LUMINESCENCE_DATA_REPLY] + return await make_luminescence(io, plate()).measure( + **{"focal_height": GROUND_TRUTH_FOCAL_HEIGHT, **arguments} + ) + + +# The driver's asyncio.Lock needs an event loop on Python 3.9: build drivers inside one. +class LuminescenceTests(unittest.IsolatedAsyncioTestCase): + async def test_the_defaults_are_the_ground_truth(self): + io = FakeFTDI() + await read(io) + self.assertEqual( + ftdi_extract_payload(io.written[1]), with_fields(LUMINESCENCE_RUN_FRAME, NOMINAL_EDGES) + ) + + async def test_each_field_is_sent_where_the_ground_truth_has_it(self): + """Every luminescence field changed at once, to values a CLARIOstarPlus (firmware 1.35) + accepted.""" + io = FakeFTDI() + await read( + io, + focal_height=4.0, + gain=2000, + emission_wavelength=560, + emission_bandwidth=40, + normalization_time=2.0, + measurement_interval=0.2, + ) + fields = { + FOCAL_HEIGHT: 400, + GAIN: 2000, + EMISSION_EDGES: 5800, + EMISSION_EDGES + 2: 5400, + NORMALIZATION_TIME: 200, + MEASUREMENT_INTERVAL: 20, + } + self.assertEqual( + ftdi_extract_payload(io.written[1]), with_fields(LUMINESCENCE_RUN_FRAME, fields) + ) + + async def test_none_reads_a_quarter_up_the_well(self): + # Corning 96 360 uL: the well at 3.03 mm, its floor 0.50 mm, its depth 10.67 mm: 6.2 mm. + runs = [] + for focal_height in (None, 6.2): + io = FakeFTDI() + io.replies = [STATUS_IDLE, RUN_REPLY, STATUS_MEASURING, STATUS_MEASURED] + io.replies.append(LUMINESCENCE_DATA_REPLY) + feature = make_luminescence(io, cor_96_wellplate_360uL_Fb("plate")) + await feature.measure(focal_height=focal_height) + runs.append(io.written[1]) + self.assertEqual(runs[0], runs[1]) + with self.assertRaises(ValueError): + feature.default_focal_height_percent_of_well_depth = 101 + + async def test_counts_are_the_ground_truth(self): + result = await read(FakeFTDI()) + expected = { + f"{'ABCDEFGH'[row]}{column + 1}": float(value) + for row, values in enumerate(LUMINESCENCE_GROUND_TRUTH) + for column, value in enumerate(values) + } + self.assertEqual(result.data, expected) + self.assertIsNone(result.temperature) + with self.assertRaises(ValueError): # fluorescence's layout + make_luminescence(FakeFTDI(), plate())._decode_data_reply( + ftdi_extract_payload(TEMPERATURE_37_DATA_REPLY), 96 + ) + + async def test_what_luminescence_cannot_read_sends_nothing(self): + io = FakeFTDI() + feature = make_luminescence(io, plate()) + refused: List[Dict[str, Any]] = [ + dict(measurement_settings=MeasurementSettings(well_scan="orbital")), + dict(measurement_settings=MeasurementSettings(pause_before_reading=5)), + dict(measurement_interval=0.805), # not in 0.01 s steps + ] + for arguments in refused: + with self.assertRaises(ValueError): + await feature.measure(**arguments) + self.assertEqual(io.written, []) + + def test_the_wavelength_limits_follow_the_photomultipliers(self): + # (extended red, dual, crossed over): the highest emission wavelength. + for photomultipliers, highest in ( + ((True, True, False), 740), # the reader of the hardware reads + ((True, True, True), 840), + ((True, False, False), 840), + ((False, False, False), 740), + ): + with self.subTest(photomultipliers=photomultipliers): + driver = make_driver(FakeFTDI()) + driver.configuration = decoded() + driver.configuration.luminescence_channels = 1 + ( + driver.configuration.extended_red_photomultiplier_tube, + driver.configuration.dual_photomultiplier_tubes, + driver.configuration.photomultiplier_tube_crossover, + ) = photomultipliers + driver._discover_features() + assert driver.luminescence is not None + self.assertEqual(driver.luminescence.configuration.wavelength_range[1], highest) + + def test_luminescence_is_built_only_when_reported(self): + for channels, built in ((1, True), (0, False)): + driver = make_driver(FakeFTDI()) + driver.configuration = decoded() + driver.configuration.luminescence_channels = channels + driver._discover_features() + self.assertEqual(isinstance(driver.luminescence, Luminescence), built) diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/measurement.py b/pylabrobot/bmg_labtech/clariostar/driver/features/measurement.py index b0bc12f5dc6..f9208af59a9 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/measurement.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/measurement.py @@ -22,7 +22,7 @@ logger = logging.getLogger(__name__) -# The well mask holds 384 bits; a 1536-well plate has no captured encoding. +# The well mask holds 384 bits; a 1536-well plate has no verified encoding. _WELL_MASK_SIZE = 48 Corner = Literal["top_left", "top_right", "bottom_left", "bottom_right"] @@ -36,12 +36,12 @@ WellScan = Literal["point", "orbital", "spiral", "matrix"] # Pre-separator byte 0 bits: how each well is read (orbital `32`, spiral `06`, matrix `10`). _WELL_SCANS: Dict[str, int] = {"point": 0x00, "orbital": 0x30, "spiral": 0x04, "matrix": 0x10} -# The matrix sizes, points per side, the vendor software offers. +# The matrix sizes, points per side. _MATRIX_SIZES = (2, 3, 4, 5, 6, 7, 8, 9, 10, 15, 20, 25, 30) ShakePattern = Literal["orbital", "linear", "double_orbital"] # Shaking before reading, pre-separator: [12] = 02 before the (endpoint) read, [17] pattern, -# [18] speed code = rpm / 100 - 1, [20] seconds. Meander is not captured in a RUN. +# [18] speed code = rpm / 100 - 1, [20] seconds. Meander is not verified in a RUN. _SHAKE_PATTERNS: Dict[str, int] = {"orbital": 0x00, "linear": 0x01, "double_orbital": 0x02} @@ -61,17 +61,17 @@ class MeasurementConfiguration: """Device facts every measurement mode shares.""" settling_time_range: Tuple[float, float] = (0.0, 1.0) - """Seconds before each well is read (0, 0.1, 0.2, 0.5 and 1.0 captured).""" + """Seconds before each well is read (0, 0.1, 0.2, 0.5 and 1.0 verified).""" flashes_range: Tuple[int, int] = (1, 200) - """Flashes per well, every well scan (the vendor software manual: up to 200 per well).""" + """Flashes per well, every well scan (up to 200 per well).""" pause_range: Tuple[int, int] = (1, 3600) - """Pause before reading, seconds (the vendor software manual: up to 3600 s).""" + """Pause before reading, seconds (up to 3600 s).""" orbital_maximum_wells: int = 384 - """Orbital averaging works on plates of up to 384 wells (the vendor software manual: 6 to 384).""" + """Orbital averaging works on plates of up to 384 wells (6 to 384).""" matrix_maximum_wells: int = 384 - """A matrix scan works on plates of up to 384 wells (the vendor software: up to 384).""" + """A matrix scan works on plates of up to 384 wells (up to 384).""" shake_speed_range: Tuple[int, int] = (100, 700) - """Shaking before the read, rpm (the standalone shake's captured 100-700; 300, 500 in a RUN).""" + """Shaking before the read, rpm (100-700; 300 and 500 verified in a RUN).""" shake_duration_range: Tuple[int, int] = (1, 255) """Shaking before the read, seconds: one byte until a longer duration settles its order.""" @@ -143,8 +143,8 @@ class MeasurementSettings: diameter. flashes: whole flashes per well (per point of a matrix), within the configuration's `flashes_range`. - matrix_size: points per side of a matrix scan, one of 2 to 10, 15, 20, 25, 30 (the vendor - software's); None for any other well scan. + matrix_size: points per side of a matrix scan, one of 2 to 10, 15, 20, 25, 30; None + for any other well scan. Raises: ValueError: on construction, listing every field of the wrong kind (see `check` for the @@ -155,10 +155,10 @@ class MeasurementSettings: corner: Corner = "top_left" reading_direction: ReadingDirection = "column_major" bidirectional: bool = True - settling_time: float = 0.1 # the vendor's default + settling_time: float = 0.1 well_scan: WellScan = "point" - scan_diameter: int = 3 # the vendor's - flashes: int = 5 # as the vendor's default endpoint read + scan_diameter: int = 3 + flashes: int = 5 matrix_size: Optional[int] = None def __post_init__(self) -> None: @@ -194,7 +194,7 @@ def check( ) -> None: """Raise if these settings cannot read `wells` of `plate` on the reader. - The scan diameter's upper bound is inferred: 3 to 5 mm are captured, all on 6.58 mm wells. + The scan diameter's upper bound is inferred: 3 to 5 mm are verified, all on 6.58 mm wells. Raises: ValueError: listing every rule broken: a field outside the configuration's range, no wells, @@ -468,14 +468,14 @@ def _encode_plate_block(self, plate: Plate, wells: Sequence[Well]) -> bytes: # -- scan and settling bytes --------------------------------------------------------------------- - # Scan byte bit 2: flying mode. Bit 4, the vendor's equidistant kinetic cycles, stays 0. + # Scan byte bit 2: flying mode. Bit 4, equidistant kinetic cycles, stays 0. _FLYING_MODE = 0x04 def _compute_scan_byte(self, settings: MeasurementSettings, *, flying_mode: bool = False) -> int: """The scan byte of a read; `flying_mode` sets bit 2. Bit 7 unidirectional, bits 6-5 the start corner, bit 3 column-major, bit 2 flying mode, bit 1 - set. Captured: `0a` (default), `2a`, `4a`, `6a` (corners), `8a` (unidirectional), `02` + set. Verified: `0a` (default), `2a`, `4a`, `6a` (corners), `8a` (unidirectional), `02` (row-major), `0e` (flying mode). """ return ( @@ -576,8 +576,7 @@ def _compute_run_start( def _compute_run_end(self, flashes: int) -> bytes: """The RUN parameters of an endpoint read after the mode's own fields. - `_TRAILER_START`, the trailer, 1 cycle, the flashes (u16 BE), a 1 s cycle time, `00`, as the - vendor sends them. + `_TRAILER_START`, the trailer, 1 cycle, the flashes (u16 BE), a 1 s cycle time, `00`. """ return ( bytes([self._TRAILER_START]) @@ -619,7 +618,7 @@ async def _request_data(self, values: Optional[int]) -> List[bytes]: are in. A spectrum's data comes as several data replies, its pages, each sized to fit a frame (3 to 5 - in the captured spectra); each `05 02 00 00 00 00 00` returns the next. + in a spectrum); each `05 02 00 00 00 00 00` returns the next. Raises: FrameError: if a page arrives short or malformed. @@ -671,7 +670,7 @@ async def _run_measurement( ) -> Tuple[List[bytes], datetime]: """Send RUN and return the data replies, and when they were read. - As the vendor ends a read: status until the reader reports the data ready, then REQUEST DATA + A read ends with status until the reader reports the data ready, then REQUEST DATA for the whole data set (`05 02 00 00 00 00 00`), once or once per page, then status until idle. Data that comes a page at a time while the run continues is requested whenever the data ready bit is set, which each request clears, until all its pages are in. @@ -695,8 +694,8 @@ async def _run_measurement( read_at = datetime.now() start = time.monotonic() try: - # A poll sent right after the RUN's reply goes unanswered, as the reader answered; the - # vendor's first poll follows the reply after 46 ms or more. + # A poll sent right after the RUN's reply goes unanswered, as the reader answered: wait + # one poll interval first. await asyncio.sleep(self._driver.default_poll_interval) while True: status = await self._driver.send_command(self._driver.CommandFamily.STATUS, retries=5) diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/temperature_controller.py b/pylabrobot/bmg_labtech/clariostar/driver/features/temperature_controller.py index 3c46047d852..f0daf3a79df 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/temperature_controller.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/temperature_controller.py @@ -14,7 +14,7 @@ _OFF = 0x0000 _MONITOR = 0x0001 -# Setpoint range in degrees C, from the operating manual. 20.0 is refused (quit code 110). +# Setpoint range in degrees C. 20.0 is refused (quit code 110). _TEMPERATURE_RANGE = (25.0, 45.0) _EXTENDED_TEMPERATURE_RANGE = (10.0, 65.0) diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/absorbance_spectrum_verification.json.gz b/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/absorbance_spectrum_verification.json.gz index 8486b782468..a19ccf444ef 100644 Binary files a/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/absorbance_spectrum_verification.json.gz and b/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/absorbance_spectrum_verification.json.gz differ diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/filter_detection_captures.json.gz b/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/filter_detection_captures.json.gz deleted file mode 100644 index 965222288c1..00000000000 Binary files a/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/filter_detection_captures.json.gz and /dev/null differ diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/filter_detection_ground_truth.json.gz b/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/filter_detection_ground_truth.json.gz new file mode 100644 index 00000000000..51207df8b29 Binary files /dev/null and b/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/filter_detection_ground_truth.json.gz differ diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/fluorescence_captures.json.gz b/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/fluorescence_captures.json.gz deleted file mode 100644 index c6c5400d0c1..00000000000 Binary files a/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/fluorescence_captures.json.gz and /dev/null differ diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/fluorescence_ground_truth.json.gz b/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/fluorescence_ground_truth.json.gz new file mode 100644 index 00000000000..0767a20e3b2 Binary files /dev/null and b/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/fluorescence_ground_truth.json.gz differ diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/z_profile_captures.json.gz b/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/z_profile_captures.json.gz deleted file mode 100644 index bca82e4ea91..00000000000 Binary files a/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/z_profile_captures.json.gz and /dev/null differ diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/z_profile_ground_truth.json.gz b/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/z_profile_ground_truth.json.gz new file mode 100644 index 00000000000..56182426507 Binary files /dev/null and b/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/z_profile_ground_truth.json.gz differ diff --git a/pylabrobot/bmg_labtech/clariostar/driver/ftdi_queued_reads_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/ftdi_queued_reads_tests.py index de3e73dbbec..0ee953a428e 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/ftdi_queued_reads_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/ftdi_queued_reads_tests.py @@ -23,7 +23,7 @@ STATUS_MEASURED, STOP_FRAME, STOP_REPLY, - captured_96_well_plate, + ground_truth_96_well_plate, make_absorbance, ) from pylabrobot.bmg_labtech.clariostar.driver.framing import ftdi_extract_payload @@ -404,7 +404,7 @@ async def test_a_cancelled_spectrum_leaves_the_reads_running_and_the_link_clean( before = ([(0.0, reply)] for reply in (STATUS_IDLE, RUN_REPLY, STATUS_MEASURED)) chip = PacedChipModel(*before, packets, [(0.0, STOP_REPLY)], [(0.0, STATUS_IDLE)], pace=0.25) io, libusb = LibftdiFTDI(chip), FakeLibusb(chip) - absorbance = make_absorbance(io, captured_96_well_plate()) + absorbance = make_absorbance(io, ground_truth_96_well_plate()) driver = absorbance._driver reads = keep_reads_queued(driver, libusb) self.addCleanup(driver._stop_queued_reads) @@ -443,7 +443,7 @@ async def test_a_cancelled_spectrum_through_libftdi_leaves_the_link_clean(self): before = ([(0.0, reply)] for reply in (STATUS_IDLE, RUN_REPLY, STATUS_MEASURED)) chip = PacedChipModel(*before, packets, [(0.0, STOP_REPLY)], [(0.0, STATUS_IDLE)], pace=0.25) io = LibftdiFTDI(chip) - absorbance = make_absorbance(io, captured_96_well_plate()) + absorbance = make_absorbance(io, ground_truth_96_well_plate()) driver = absorbance._driver driver._PACKET_READ_TIMEOUT, driver._FRAME_STALL_TIMEOUT = 1.0, 0.3 self.assertIsNone(driver._queued_reads) diff --git a/pylabrobot/bmg_labtech/clariostar/driver/master.py b/pylabrobot/bmg_labtech/clariostar/driver/master.py index 6665b05e684..10a6db48b3d 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/master.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/master.py @@ -34,6 +34,7 @@ from .features.absorbance import Absorbance from .features.drawer import Drawer from .features.fluorescence import Fluorescence +from .features.luminescence import Luminescence from .features.temperature_controller import TemperatureController from .framing import ( FTDI_CR, @@ -91,8 +92,7 @@ class CLARIOstarDriver: # Seconds a command may keep the reader busy, when waited on (``read_timeout`` None). _default_read_timeout: float = 150.0 - # Seconds between status polls while waiting: with a ~38 ms exchange, close to the vendor - # software's polling (302 ms median over 39,373 polls). + # Seconds between status polls while waiting, with a ~38 ms exchange. default_poll_interval: float = 0.25 _FTDI_READ_CHUNK = 25 # bytes per read; a shorter read marks the end of a burst @@ -238,11 +238,12 @@ def __init__( # FTDI only: bulk-in transfers kept pending from setup to stop; None reads through libftdi. self._queued_reads: Optional[QueuedReads] = None self.drawer = Drawer(self) - # Built by `discover` only if the reader reports a temperature controller, absorbance, and - # a fluorescence channel. + # Built by `discover` only if the reader reports a temperature controller, absorbance, a + # fluorescence channel, and a luminescence channel. self.temperature_controller: Optional[TemperatureController] = None self.absorbance: Optional[Absorbance] = None self.fluorescence: Optional[Fluorescence] = None + self.luminescence: Optional[Luminescence] = None # -- FTDI interface -------------------------------------------------------- @@ -826,6 +827,8 @@ def _discover_features(self) -> None: self.absorbance = Absorbance(self) if self.configuration.fluorescence_channels and self.fluorescence is None: self.fluorescence = Fluorescence(self) + if self.configuration.luminescence_channels and self.luminescence is None: + self.luminescence = Luminescence(self) def _check_firmware_compatibility(self) -> None: """Log a warning if the firmware version has not been checked on hardware.""" @@ -845,7 +848,7 @@ async def discover(self) -> None: on hardware. """ versions = {"INIT reply": self.configuration.firmware_version} - # The stored parameters first, as the vendor software reads them right after INIT. + # The stored parameters first, right after INIT. try: parameters = await self.request_parameters() except ValueError as e: diff --git a/pylabrobot/bmg_labtech/clariostar/driver/master_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/master_tests.py index 6d73816698b..fdc60d37f12 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/master_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/master_tests.py @@ -51,7 +51,7 @@ "0001310007000000000000000071710d" ) -# Frames the vendor software exchanged with the same reader. +# Ground-truth frames of the same reader. # Idle at power-on, motors off. STATUS_MOTORS_OFF = bytes.fromhex("0200180c010500040000000000000000000000c00000f00d") # Stored parameters; the project and serial number, reply bytes 161-164, are zeroed and the checksum diff --git a/pylabrobot/bmg_labtech/clariostar/driver/simulator.py b/pylabrobot/bmg_labtech/clariostar/driver/simulator.py index 4b42f16720b..af75e4ed978 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/simulator.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/simulator.py @@ -184,7 +184,7 @@ async def measure( results: List[AbsorbanceResult] = [] if simulated_data is None: if report == "raw": - raise NotImplementedError("raw counts are not simulated; give captured simulated_data") + raise NotImplementedError("raw counts are not simulated; give simulated_data") measured = len({id(well) for well in wells}) logger.debug("%s read: simulation: random optical densities", SIMULATED_LINK) for wavelength in wavelengths: @@ -298,7 +298,7 @@ async def measure_spectrum( timestamp = datetime.now() if simulated_data is None: if report == "raw": - raise NotImplementedError("raw counts are not simulated; give captured simulated_data") + raise NotImplementedError("raw counts are not simulated; give simulated_data") logger.debug("%s read: simulation: random optical densities", SIMULATED_LINK) data = { identifier: { diff --git a/pylabrobot/lib/luminescence/__init__.py b/pylabrobot/lib/luminescence/__init__.py new file mode 100644 index 00000000000..44fc9935dfc --- /dev/null +++ b/pylabrobot/lib/luminescence/__init__.py @@ -0,0 +1 @@ +from .results import LuminescenceResult diff --git a/pylabrobot/lib/luminescence/results.py b/pylabrobot/lib/luminescence/results.py new file mode 100644 index 00000000000..19a496d0724 --- /dev/null +++ b/pylabrobot/lib/luminescence/results.py @@ -0,0 +1,28 @@ +"""The result of a luminescence measurement, the same for every plate reader.""" + +from dataclasses import dataclass +from datetime import datetime +from typing import Dict, Optional + + +@dataclass +class LuminescenceResult: + """Luminescence intensities of one emission band, from one read of a plate. + + Attributes: + data: intensities of the measured wells in the reader's counts, keyed by well identifier + (``"B3"``), row by row; wells not measured are not in it. + emission_wavelength: emission wavelength in nm; None if the read selects no wavelength (all + the light the sample emits). + emission_bandwidth: emission bandwidth in nm; None as for the wavelength. + temperature: temperature in degrees C, or ``None`` if the reader did not measure one. + start_timestamp: when the read started. + end_timestamp: when the read's data was in. + """ + + data: Dict[str, float] + emission_wavelength: Optional[float] + emission_bandwidth: Optional[float] + temperature: Optional[float] + start_timestamp: datetime + end_timestamp: datetime