From 6b3e07bf913972ba236b1cde1eca767d876a1f40 Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Thu, 22 Oct 2026 13:13:18 +0100 Subject: [PATCH] v1 `CLARIOstar`: build the configuration from what the reader reports - Each block of the reader's hardware list (its INIT reply) is decoded on its own: one missing or too short leaves only its fields unknown, with a warning naming it. - A slide position without a whole number is left out of `monochromator_slides`. - The stored parameters are keyed by what each value is, grouped by what the values are for, with units; the eight that can be negative are decoded signed. - The serial number is read from the reader's stored parameters; a link opened by another serial warns. A saved configuration still holds the placeholder. - The fluorescence filter slots are the reader's slides' (each position not "open"); a reader whose slides are missing or of different sizes keeps the defaults. - A reader reporting no linear variable filters refuses fluorescence by wavelength and luminescence: it has no monochromator. Unknown lets the read through. - Setup reads the firmware's build date (REQUEST 09) and the build number (INIT), shown after the firmware version. - The setup summary prints every configuration field, True/False for each yes/no, the serial number after the model; a field not read shows None. Co-Authored-By: Claude Opus 5.5 --- .../bmg_labtech/clariostar/device_tests.py | 53 +++- .../clariostar/driver/configuration.py | 178 +++++++---- .../clariostar/driver/configuration_tests.py | 96 +++++- .../driver/features/fluorescence.py | 59 +++- .../driver/features/fluorescence_tests.py | 44 +++ .../driver/features/luminescence.py | 5 +- .../driver/features/luminescence_tests.py | 11 + .../clariostar/driver/features/measurement.py | 11 + .../features/temperature_controller_tests.py | 3 +- .../bmg_labtech/clariostar/driver/master.py | 267 +++++++++++++---- .../clariostar/driver/master_tests.py | 32 +- .../clariostar/driver/parameter_table.py | 281 ++++++++++-------- ...star_plus_1_35_temperature_controller.json | 2 + .../clariostar/driver/simulator.py | 4 + 14 files changed, 786 insertions(+), 260 deletions(-) diff --git a/pylabrobot/bmg_labtech/clariostar/device_tests.py b/pylabrobot/bmg_labtech/clariostar/device_tests.py index 9402852646f..508d81def09 100644 --- a/pylabrobot/bmg_labtech/clariostar/device_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/device_tests.py @@ -23,6 +23,7 @@ OFF_REPLY, ) from pylabrobot.bmg_labtech.clariostar.driver.master_tests import ( + FIRMWARE_INFO_REPLY, INIT_REPLY, PARAMETERS_REPLY, STATUS_BUSY, @@ -32,6 +33,7 @@ FakeFTDI, make_driver, ) +from pylabrobot.bmg_labtech.clariostar.driver.simulator_tests import decoded_with_parameters from pylabrobot.resources import Resource @@ -50,7 +52,54 @@ def test_the_setup_summary_names_the_reader(self): reader.driver.configuration = decoded() summary = reader.driver.format_setup_summary() self.assertTrue(summary.startswith("[reader_2] Connected on FakeFTDI 0x0403:0xbb68"), summary) - self.assertIn("Firmware: 1.35", summary) + self.assertIn(" Firmware: 1.35 (build 1)\n Mainboard revision: 2\n", summary) + self.assertIn( + " Detectors:\n - standard photomultiplier: True\n" + " - extended-red photomultiplier: True\n - dual photomultipliers: True\n" + " - crossover: False\n", + summary, + ) + # The INIT reply only: the stored parameters are not read. + self.assertIn(" Serial number: None\n", summary) + self.assertIn(" - absorbance: None\n", summary) + self.assertIn(" - bottom optic: None\n", summary) + reader.driver.configuration.photomultiplier_tube_crossover = True + self.assertIn(" - crossover: True\n", reader.driver.format_setup_summary()) + + async def test_the_setup_summary_shows_the_modes_and_filter_slots(self): + reader = CLARIOstar("reader", io=FakeFTDI()) + reader.driver.configuration = await decoded_with_parameters() + reader.driver._discover_features() + summary = reader.driver.format_setup_summary() + self.assertIn( + " Measurement modes:\n - absorbance: True\n - fluorescence: True\n" + " - luminescence: True\n - time-resolved fluorescence: True\n" + " - fluorescence polarization, visible: False\n" + " - fluorescence polarization, ultraviolet: False\n - AlphaScreen: False\n", + summary, + ) + self.assertIn(" Filter slots: excitation 4, dichroic 3, emission 4\n", summary) + reader.driver.configuration.monochromator_slides = None + self.assertIn("(defaults: the slides are not reported)", reader.driver.format_setup_summary()) + + async def test_the_setup_summary_shows_the_options(self): + reader = CLARIOstar("reader", io=FakeFTDI()) + c = await decoded_with_parameters() + c.serial_number = "123-4567" + c.high_speed_shaking_allowed = c.well_1536_plates_supported = True + c.pumps = ("XCalibur", None, "NextGen") + c.barcode_readers = {"front": "SE 923", "right": "CSE 600"} + reader.driver.configuration = c + summary = reader.driver.format_setup_summary() + self.assertIn(" Model: CLARIOstarPlus\n Serial number: 123-4567\n", summary) + self.assertIn(" High-speed shaking: True\n 1536-well plates: True\n", summary) + self.assertIn( + " Pumps:\n - pump 1: XCalibur\n - pump 2: False\n - pump 3: NextGen\n" + " Barcode readers:\n - front: SE 923\n - right: CSE 600\n" + " - optional 1: False\n - optional 2: False\n" + " Stacker: False\n Atmospheric control unit: False", + summary, + ) async def setup_reader(self, reader: CLARIOstarDevice) -> None: assert isinstance(reader.driver.io, FakeFTDI) @@ -61,6 +110,8 @@ async def setup_reader(self, reader: CLARIOstarDevice) -> None: STATUS_IDLE, STATUS_IDLE, PARAMETERS_REPLY, + STATUS_IDLE, + FIRMWARE_INFO_REPLY, STATUS_IDLE, # temperature monitoring: sensors off STATUS_IDLE, MONITOR_REPLY, diff --git a/pylabrobot/bmg_labtech/clariostar/driver/configuration.py b/pylabrobot/bmg_labtech/clariostar/driver/configuration.py index c8c2a3758ff..ef8b229161b 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/configuration.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/configuration.py @@ -2,9 +2,12 @@ import dataclasses import json +import logging from dataclasses import dataclass from typing import Any, Dict, Iterator, Optional, Tuple +logger = logging.getLogger(__name__) + INIT_REPLY_TYPE = 0x13 # The firmware whose replies are decoded, as (lowest, highest) firmware number x 1000: the 1.x line. @@ -12,18 +15,32 @@ DECODED_FIRMWARE = (1300, 1999) _BARCODE_READER_TYPES = {0: "SE 923", 1: "CSE 600", 2: "SE 955", 3: "SE 965"} -_BARCODE_READER_POSITIONS = ("front", "right", "optional 1", "optional 2") +BARCODE_READER_POSITIONS = ("front", "right", "optional 1", "optional 2") + + +# The INIT reply's blocks by id, named after the module each describes. +_BLOCK_NAMES = { + 0x01: "main board", + 0x09: "detectors", + 0x0E: "temperature controller", + 0x0F: "pumps", + 0x11: "stacker", + 0x12: "barcode readers", + 0x1F: "spectrometer", + 0x2F: "slides", + 0x31: "atmospheric control unit", +} def _bit(value: int, bit: int) -> bool: return bool(value >> bit & 1) -def _subsystems(payload: bytes) -> Iterator[Tuple[int, bytes]]: - """Yield the INIT reply's blocks, one per subsystem: `[id][length u16 BE][data]`, from byte 6. +def _blocks(payload: bytes) -> Iterator[Tuple[int, bytes]]: + """Yield the INIT reply's blocks, one per module: `[id][length u16 BE][data]`, from byte 6. Raises: - ValueError: if the payload is not an INIT reply or its subsystems do not end with it. + ValueError: if the payload is not an INIT reply or its blocks do not end with it. """ if not payload or payload[0] != INIT_REPLY_TYPE: raise ValueError(f"not an INIT reply: {payload[:1].hex()}") @@ -32,20 +49,21 @@ def _subsystems(payload: bytes) -> Iterator[Tuple[int, bytes]]: length = int.from_bytes(payload[offset + 1 : offset + 3], "big") data = payload[offset + 3 : offset + 3 + length] if offset + 3 > len(payload) or len(data) != length: - raise ValueError(f"INIT reply cut short in the subsystem at byte {offset}") + raise ValueError(f"INIT reply cut short in the block at byte {offset}") yield payload[offset], data offset += 3 + length -def _subsystem(subsystems: Dict[int, bytes], subsystem_id: int, length: int) -> bytes: - """Return a subsystem's data, at least `length` bytes long. +def _block(blocks: Dict[int, bytes], block_id: int, length: int) -> bytes: + """Return a block's data, at least `length` bytes long. Raises: - ValueError: if the reply has no such subsystem, or a shorter one. + ValueError: if the reply has no such block, or a shorter one. """ - data = subsystems.get(subsystem_id, b"") + data = blocks.get(block_id, b"") if len(data) < length: - raise ValueError(f"INIT reply has no subsystem 0x{subsystem_id:02x} of {length} bytes") + name = _BLOCK_NAMES.get(block_id, "unknown") + raise ValueError(f"INIT reply has no {name} block (0x{block_id:02x}) of {length} bytes") return data @@ -57,39 +75,51 @@ def _pump_type(types: int, pump: int) -> str: def _parse_monochromator_slides(text: str) -> Dict[str, Dict[int, str]]: - """Parse `NAME:1=open,2=Filter 1,...` lines into {slide: {position: name}}.""" + """Parse `NAME:1=open,2=Filter 1,...` lines into {slide: {position: name}}. + + An entry without a whole position number is left out. + """ slides: Dict[str, Dict[int, str]] = {} for line in text.splitlines(): name, _, positions = line.partition(":") if positions: + entries = (entry.partition("=") for entry in positions.split(",")) slides[name.strip()] = { - int(number): label - for number, _, label in (entry.partition("=") for entry in positions.split(",")) + int(number): label for number, _, label in entries if number.strip().isdigit() } return slides @dataclass class DeviceConfiguration: - """What the reader reported about itself. None until `initialize` and `discover` have read it. + """What the reader reported about itself. - The fitted hardware comes from the INIT reply, the rest from the stored parameters. + The fitted hardware comes from the INIT reply (`update_from_init_reply`), block by block; the + rest from the stored parameters (`update_from_parameters`). A field is None until read, + or when its block is missing from the reply. """ model: Optional[str] = None - # Never read from the reader; a saved file holds `SERIAL_NUMBER_PLACEHOLDER`. + # The reader's serial, from its stored parameters; a saved file holds `SERIAL_NUMBER_PLACEHOLDER`. serial_number: Optional[str] = None mainboard_revision: Optional[int] = None firmware_version: Optional[str] = None + firmware_build: Optional[int] = None + # The day the firmware was compiled, ISO "YYYY-MM-DD": the same on every reader running it. + firmware_build_date: Optional[str] = None extended_red_photomultiplier_tube: Optional[bool] = None dual_photomultiplier_tubes: Optional[bool] = None - # Dual photomultipliers swapped (cross-over): fluorescence on the standard one. + # The dual photomultipliers swapped (cross-over): fluorescence on the standard one. photomultiplier_tube_crossover: Optional[bool] = None + # What it counts is not verified: each slide reports 3 positions. excitation_slide_positions: Optional[int] = None spectrometer: Optional[bool] = None + # The reader's slides by name, each with its positions: {slide: {position: name}}. monochromator_slides: Optional[Dict[str, Dict[int, str]]] = None + # The monochromators: one bit for excitation, dichroic and emission, read as all or none. linear_variable_filters: Optional[bool] = None temperature_controller: Optional[bool] = None + # What they count is not verified. temperature_controller_channels: Optional[int] = None # The type of injector pumps 1-3, None where not built in. pumps: Optional[Tuple[Optional[str], Optional[str], Optional[str]]] = None @@ -104,75 +134,115 @@ class DeviceConfiguration: absorbance_available: Optional[bool] = None alpha_screen_available: Optional[bool] = None fluorescence_polarization_available: Optional[bool] = None + # Polarization option 2: ultraviolet as well. fluorescence_polarization_ultraviolet_available: Optional[bool] = None bottom_measurement_available: Optional[bool] = None + # The temperature controller from 10 instead of 25 C, up to 65 instead of 45 C. extended_temperature_range: Optional[bool] = None high_speed_shaking_allowed: Optional[bool] = None well_1536_plates_supported: Optional[bool] = None def update_from_parameters(self, parameters: Dict[str, int]) -> None: - """Fill the fields the stored parameters carry, from `CLARIOstarDriver.request_parameters`.""" - self.fluorescence_channels = parameters["No. of fluorescence measurement channels"] - self.luminescence_channels = parameters["No. of luminescence measurement channels"] + """Fill the fields the stored parameters carry, from + `CLARIOstarDriver.request_parameters`.""" + self.fluorescence_channels = parameters["fluorescence_measurement_channel_count"] + self.luminescence_channels = parameters["luminescence_measurement_channel_count"] self.time_resolved_fluorescence_channels = parameters[ - "No. of time-resolved fluorescence meas. channels" + "time_resolved_fluorescence_measurement_channel_count" ] - self.absorbance_available = parameters["Absorbance measurement option"] != 0 - self.alpha_screen_available = parameters["AlphaScreen measurement option"] != 0 + self.absorbance_available = parameters["absorbance_measurement_option"] != 0 + self.alpha_screen_available = parameters["alpha_screen_measurement_option"] != 0 # 0 none, 1 visible range, 2 ultraviolet and visible range. - polarization = parameters["Fluorescence polarization measurement option"] + polarization = parameters["fluorescence_polarization_measurement_option"] self.fluorescence_polarization_available = polarization != 0 self.fluorescence_polarization_ultraviolet_available = polarization == 2 - self.bottom_measurement_available = parameters["Bottom measurement option"] != 0 - self.extended_temperature_range = parameters["Extended incubator"] != 0 - self.high_speed_shaking_allowed = parameters["High speed shaking allowed"] != 0 - self.well_1536_plates_supported = parameters["1536 well measurements enabled"] != 0 - - def update_from_init_reply(self, payload: bytes) -> None: - """Fill the fields the INIT reply carries. - - Raises: - ValueError: if the payload is not a whole INIT reply, or comes from firmware outside - `DECODED_FIRMWARE`; the main board fields are filled either way. - """ - subsystems = dict(_subsystems(payload)) - main_board = _subsystem(subsystems, 0x01, 5) - firmware = int.from_bytes(main_board[0:2], "big") - self.firmware_version = f"{firmware / 1000:.2f}" - self.mainboard_revision = main_board[4] - self.model = "CLARIOstarPlus" if main_board[4] >= 2 else "CLARIOstar" - if not DECODED_FIRMWARE[0] <= firmware <= DECODED_FIRMWARE[1]: - raise ValueError(f"INIT reply of firmware {self.firmware_version} is not decoded here") - - measurement = _subsystem(subsystems, 0x09, 1)[0] + self.bottom_measurement_available = parameters["bottom_measurement_option"] != 0 + self.extended_temperature_range = parameters["extended_incubator"] != 0 + self.high_speed_shaking_allowed = parameters["high_speed_shaking_allowed"] != 0 + self.well_1536_plates_supported = parameters["well_1536_measurements_enabled"] != 0 + + def _update_from_detectors(self, blocks: Dict[int, bytes]) -> None: + """The photomultipliers and the excitation slide, from block `09` byte 0.""" + measurement = _block(blocks, 0x09, 1)[0] self.dual_photomultiplier_tubes = _bit(measurement, 2) self.extended_red_photomultiplier_tube = _bit(measurement, 1) or _bit(measurement, 2) self.photomultiplier_tube_crossover = _bit(measurement, 1) and _bit(measurement, 2) self.excitation_slide_positions = 4 if _bit(measurement, 3) else 3 - self.spectrometer = _bit(_subsystem(subsystems, 0x1F, 1)[0], 0) - text, _, slide_info = _subsystem(subsystems, 0x2F, 1).partition(b"\x00") + def _update_from_spectrometer(self, blocks: Dict[int, bytes]) -> None: + """The absorbance spectrometer, from block `1F` bit 0.""" + self.spectrometer = _bit(_block(blocks, 0x1F, 1)[0], 0) + + def _update_from_monochromator_slides(self, blocks: Dict[int, bytes]) -> None: + """The slides' named positions and the linear variable filters, from block `2F`.""" + text, _, slide_info = _block(blocks, 0x2F, 1).partition(b"\x00") self.monochromator_slides = _parse_monochromator_slides(text.decode("ascii", errors="replace")) self.linear_variable_filters = bool(slide_info) and _bit(slide_info[0], 0) - temperature_controller = _subsystem(subsystems, 0x0E, 1)[0] + def _update_from_temperature_controller(self, blocks: Dict[int, bytes]) -> None: + """The temperature controller and its channels, from block `0E`.""" + temperature_controller = _block(blocks, 0x0E, 1)[0] self.temperature_controller = _bit(temperature_controller, 0) self.temperature_controller_channels = (temperature_controller & 0x70) >> 4 - pumps, pump_types = _subsystem(subsystems, 0x0F, 2)[:2] + def _update_from_pumps(self, blocks: Dict[int, bytes]) -> None: + """The injector pumps 1-3 and their types, from block `0F`.""" + pumps, pump_types = _block(blocks, 0x0F, 2)[:2] pump_1, pump_2, pump_3 = ( _pump_type(pump_types, pump) if _bit(pumps, pump) else None for pump in range(3) ) self.pumps = (pump_1, pump_2, pump_3) - self.stacker_connected = _bit(_subsystem(subsystems, 0x11, 1)[0], 0) - readers, reader_types = _subsystem(subsystems, 0x12, 2)[:2] + def _update_from_stacker(self, blocks: Dict[int, bytes]) -> None: + """The stacker, from block `11` bit 0.""" + self.stacker_connected = _bit(_block(blocks, 0x11, 1)[0], 0) + + def _update_from_barcode_readers(self, blocks: Dict[int, bytes]) -> None: + """The barcode readers by position and type, from block `12`.""" + readers, reader_types = _block(blocks, 0x12, 2)[:2] self.barcode_readers = { position: _BARCODE_READER_TYPES[reader_types >> (2 * index) & 0x03] - for index, position in enumerate(_BARCODE_READER_POSITIONS) + for index, position in enumerate(BARCODE_READER_POSITIONS) if _bit(readers, index) } - self.atmospheric_control_unit_connected = _bit(_subsystem(subsystems, 0x31, 1)[0], 0) + + def _update_from_atmospheric_control_unit(self, blocks: Dict[int, bytes]) -> None: + """The atmospheric control unit, from block `31` bit 0.""" + self.atmospheric_control_unit_connected = _bit(_block(blocks, 0x31, 1)[0], 0) + + def update_from_init_reply(self, payload: bytes) -> None: + """Fill the fields the INIT reply carries. + + Each block is decoded on its own: one missing or shorter than its fields leaves them None + and logs a warning; the others are filled. + + Raises: + ValueError: if the payload is not a whole INIT reply, or comes from firmware outside + `DECODED_FIRMWARE`; the main board fields are filled either way. + """ + blocks = dict(_blocks(payload)) + main_board = _block(blocks, 0x01, 5) + firmware = int.from_bytes(main_board[0:2], "big") + self.firmware_version = f"{firmware / 1000:.2f}" + self.firmware_build = main_board[2] + self.mainboard_revision = main_board[4] + self.model = "CLARIOstarPlus" if main_board[4] >= 2 else "CLARIOstar" + if not DECODED_FIRMWARE[0] <= firmware <= DECODED_FIRMWARE[1]: + raise ValueError(f"INIT reply of firmware {self.firmware_version} is not decoded here") + for update in ( + self._update_from_detectors, + self._update_from_spectrometer, + self._update_from_monochromator_slides, + self._update_from_temperature_controller, + self._update_from_pumps, + self._update_from_stacker, + self._update_from_barcode_readers, + self._update_from_atmospheric_control_unit, + ): + try: + update(blocks) + except ValueError as error: + logger.warning("%s; its fields stay unknown", error) # What is fitted: compared between a declared configuration and the reader at setup. Firmware is diff --git a/pylabrobot/bmg_labtech/clariostar/driver/configuration_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/configuration_tests.py index 95218508030..080b31df97b 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/configuration_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/configuration_tests.py @@ -4,11 +4,16 @@ import os import tempfile import unittest +from typing import Dict, List, Optional, Tuple from unittest.mock import patch +from pylabrobot.bmg_labtech.clariostar.driver import configuration as configuration_module from pylabrobot.bmg_labtech.clariostar.driver import master as master_module -from pylabrobot.bmg_labtech.clariostar.driver.configuration import DeviceConfiguration -from pylabrobot.bmg_labtech.clariostar.driver.framing import ftdi_extract_payload +from pylabrobot.bmg_labtech.clariostar.driver.configuration import ( + DeviceConfiguration, + _parse_monochromator_slides, +) +from pylabrobot.bmg_labtech.clariostar.driver.framing import ftdi_extract_payload, ftdi_wrap_payload from pylabrobot.bmg_labtech.clariostar.driver.master_tests import ( FIRMWARE_INFO_REPLY, INIT_REPLY, @@ -19,6 +24,12 @@ FakeFTDI, make_driver, ) +from pylabrobot.bmg_labtech.clariostar.driver.parameter_table import ( + PARAMETER_TABLE, + REPLY_OFFSET, + SERIAL_NUMBER_ADDRESSES, + SIGNED_PARAMETERS, +) INIT_PAYLOAD = ftdi_extract_payload(INIT_REPLY) @@ -71,6 +82,27 @@ def test_undecoded_firmware_fills_only_the_main_board(self): ) self.assertIsNone(configuration.spectrometer) + def test_a_missing_block_leaves_only_its_fields_unknown(self): + # Not a reader's reply: the recorded one without its block `2F`, the slides. + payload, offset = bytearray(INIT_PAYLOAD), 6 + while payload[offset] != 0x2F: + offset += 3 + int.from_bytes(payload[offset + 1 : offset + 3], "big") + del payload[offset : offset + 3 + int.from_bytes(payload[offset + 1 : offset + 3], "big")] + configuration = DeviceConfiguration() + with patch.object(configuration_module.logger, "warning") as warning: + configuration.update_from_init_reply(bytes(payload)) + self.assertIn("0x2f", warning.call_args.args[1].args[0]) + self.assertIsNone(configuration.monochromator_slides) + self.assertIsNone(configuration.linear_variable_filters) + complete = decoded() + for field in ("spectrometer", "temperature_controller", "pumps", "barcode_readers"): + self.assertEqual(getattr(configuration, field), getattr(complete, field), field) + self.assertTrue(configuration.temperature_controller) + + def test_an_unreadable_slide_position_is_left_out(self): + slides = _parse_monochromator_slides("BSP:1=open,x=Filter 1,3=Filter 2\nEM_SP:1=open") + self.assertEqual(slides, {"BSP": {1: "open", 3: "Filter 2"}, "EM_SP": {1: "open"}}) + def test_not_an_init_reply(self): with self.assertRaises(ValueError): DeviceConfiguration().update_from_init_reply(ftdi_extract_payload(STATUS_IDLE)) @@ -90,7 +122,7 @@ async def test_initialize_reads_the_configuration(self): async def test_parameters_agreeing_with_init_do_not_warn(self): io = FakeFTDI() - io.replies = [STATUS_IDLE, PARAMETERS_REPLY] + io.replies = [STATUS_IDLE, PARAMETERS_REPLY, STATUS_IDLE, FIRMWARE_INFO_REPLY] driver = make_driver(io) driver.configuration = decoded() with patch.object(master_module.logger, "warning") as warning: @@ -99,7 +131,7 @@ async def test_parameters_agreeing_with_init_do_not_warn(self): async def test_parameters_disagreeing_with_init_warn(self): io = FakeFTDI() - io.replies = [STATUS_IDLE, PARAMETERS_REPLY] + io.replies = [STATUS_IDLE, PARAMETERS_REPLY, STATUS_IDLE, FIRMWARE_INFO_REPLY] driver = make_driver(io) driver.configuration.firmware_version = "1.34" with self.assertLogs(master_module.logger, "WARNING") as logs: @@ -116,15 +148,31 @@ def make_driver(self, replies_of_parameters: int = 1): io.replies = [STATUS_IDLE, PARAMETERS_REPLY] * replies_of_parameters return make_driver(io), io + def test_the_parameter_table_keys_and_addresses(self): + keys = [key for key, _, _ in PARAMETER_TABLE] + self.assertEqual(len(keys), len(set(keys))) + owners: Dict[int, List[str]] = {} + for key, address, size in PARAMETER_TABLE: + self.assertLessEqual(address + size, 256, key) + for byte in range(address, address + size): + owners.setdefault(byte, []).append(key) + for address in SERIAL_NUMBER_ADDRESSES: + self.assertNotIn(address, owners) + self.assertEqual([byte for byte, owner in owners.items() if len(owner) > 1], []) + self.assertLessEqual(set(SIGNED_PARAMETERS), set(keys)) + async def test_request_parameters_from_real_reply(self): driver, io = self.make_driver() parameters = await driver.request_parameters() self.assertEqual(io.written, [STATUS_QUERY_FRAME, PARAMETERS_FRAME]) - self.assertEqual(parameters["Firmware version"], 1350) - self.assertEqual(parameters["No. of fluorescence measurement channels"], 1) - self.assertEqual(parameters["Fluorescence AutoGain gain value level 1 (PMT1)"], 494) - self.assertNotIn("Serial number of the reader", parameters) - self.assertNotIn("Project number", parameters) + self.assertEqual(parameters["firmware_version"], 1350) + self.assertEqual(parameters["fluorescence_measurement_channel_count"], 1) + self.assertEqual(parameters["fluorescence_auto_gain_level_1_photomultiplier_tube_1"], 494) + self.assertEqual(parameters["top_bottom_slide_offset"], -5) + self.assertEqual(parameters["x_transport_lock_position"], -168) + self.assertEqual(parameters["temperature_sensor_1_offset"], 20) + self.assertNotIn("serial_number", parameters) + self.assertNotIn("project_number", parameters) async def test_convenience_methods(self): driver, _ = self.make_driver(replies_of_parameters=4) @@ -148,6 +196,36 @@ async def test_configuration_from_parameters(self): self.assertFalse(c.high_speed_shaking_allowed) self.assertFalse(c.well_1536_plates_supported) + async def test_discover_reads_the_serial_number_and_checks_the_link(self): + # Not a reader's reply: the recorded one, its serial cleaned to zeros, given made-up ones. + def reply_with_serial(first: int, second: int) -> bytes: + payload = bytearray(ftdi_extract_payload(PARAMETERS_REPLY)) + for address, value in zip(SERIAL_NUMBER_ADDRESSES, (first, second)): + payload[address + REPLY_OFFSET : address + REPLY_OFFSET + 2] = value.to_bytes(2, "little") + return ftdi_wrap_payload(bytes(payload)) + + cases: List[Tuple[Tuple[int, int], Dict[str, Optional[str]], str, bool]] = [ + ((123, 4567), {"device_id": None}, "123-4567", False), + ((123, 4567), {"device_id": "123-4567"}, "123-4567", False), + ((123, 4567), {"device_id": "999-0001"}, "123-4567", True), + ((12, 345), {"device_id": "012-0345"}, "012-0345", False), # padded as the link has it + ((123, 4567), {"device_id": "1:4", "usb_address": "1:4"}, "123-4567", False), # by address + ] + for serial, link, expected, warns in cases: + io = FakeFTDI() + io.replies = [STATUS_IDLE, reply_with_serial(*serial), STATUS_IDLE, FIRMWARE_INFO_REPLY] + driver = make_driver(io) + driver.configuration = decoded() + with ( + self.subTest(link=link), + patch.object(io, "serialize", return_value={"vid": 0x0403, "pid": 0xBB68, **link}), + patch.object(master_module.logger, "warning") as warning, + ): + await driver.discover() + self.assertEqual(driver.configuration.serial_number, expected) + mismatch = any("serial number" in str(call.args[0]) for call in warning.call_args_list) + self.assertEqual(mismatch, warns) + async def test_other_reply_is_not_decoded(self): io = FakeFTDI() io.replies = [STATUS_IDLE, FIRMWARE_INFO_REPLY] diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py index c38030096df..d449859c69c 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py @@ -136,13 +136,13 @@ class FluorescenceConfiguration(MeasurementConfiguration): focal_height_range: Tuple[float, float] = (0.0, 25.0) """Focal height, in mm, in 0.1 mm steps (up to 25 mm).""" excitation_filter_slots: int = 4 - """Excitation filter positions (up to 4).""" + """Excitation filter positions (up to 4); the reader's (`Fluorescence._compute_filter_slots`).""" dichroic_filter_slots: int = 3 - """Dichroic filter positions (up to 3).""" + """Dichroic filter positions (up to 3); the reader's.""" emission_filter_slots: int = 4 - """Emission filter positions (up to 4).""" + """Emission filter positions (up to 4); the reader's.""" filter_slots_per_slide: int = 2 - """Filter positions on each of the two excitation or emission slides (filter detection).""" + """Filter positions on each of the two excitation or emission slides; the reader's.""" maximum_fluorescence_channels: int = 5 """Fluorescence channels per read (up to 5).""" maximum_spectrum_wavelengths: int = 1000 @@ -203,14 +203,53 @@ class Fluorescence(Measurement): _STANDARD_WAVELENGTH_RANGE = (320, 740) _STANDARD_DICHROIC_RANGE = (340.0, 740.0) + # The reader's slides holding filters: two for excitation, two for emission, one for dichroics. + _EXCITATION_SLIDES = ("EX_SP", "EX_LP") + _EMISSION_SLIDES = ("EM_SP", "EM_LP") + _DICHROIC_SLIDE = "BSP" + + @classmethod + def _compute_filter_slots( + cls, slides: Optional[Dict[str, Dict[int, str]]] + ) -> Optional[Tuple[int, int]]: + """The filters per excitation or emission slide, and the dichroic filters, in `slides`. + + A slide's positions other than "open" hold filters. None keeps the defaults: a slide not + reported, or excitation and emission slides of different sizes (one size serves them all). + """ + names = (*cls._EXCITATION_SLIDES, *cls._EMISSION_SLIDES, cls._DICHROIC_SLIDE) + if slides is None or any(name not in slides for name in names): + return None + filters = { + name: sum(label.strip().lower() != "open" for label in slides[name].values()) + for name in names + } + per_slide = {filters[name] for name in (*cls._EXCITATION_SLIDES, *cls._EMISSION_SLIDES)} + if len(per_slide) != 1 or not filters[cls._DICHROIC_SLIDE]: + logger.warning("filter slides of %s are not decoded; the default filter slots stay", filters) + return None + (size,) = per_slide + return (size, filters[cls._DICHROIC_SLIDE]) if size else None + @classmethod def _get_reader_configuration(cls, device: "DeviceConfiguration") -> FluorescenceConfiguration: - """The fluorescence facts of the reader `device` describes: its wavelength limits. + """The fluorescence facts of the reader `device` describes: wavelength limits, filter slots. The extended range needs the extended-red photomultiplier on the fluorescence path: fitted, - and on a dual-photomultiplier reader not crossed over. Unknown hardware keeps the defaults. + and on a dual-photomultiplier reader not crossed over. The filter slots are its slides' + (`_compute_filter_slots`). Unknown hardware keeps the defaults. """ configuration = FluorescenceConfiguration() + slots = cls._compute_filter_slots(device.monochromator_slides) + if slots is not None: + per_slide, dichroic = slots + configuration = replace( + configuration, + excitation_filter_slots=2 * per_slide, + dichroic_filter_slots=dichroic, + emission_filter_slots=2 * per_slide, + filter_slots_per_slide=per_slide, + ) standard = device.extended_red_photomultiplier_tube is False or bool( device.dual_photomultiplier_tubes and device.photomultiplier_tube_crossover ) @@ -418,8 +457,8 @@ def _check_excitation_or_emission_on_reader( """The violations of excitation or emission on the reader. Neither a wavelength nor a filter (unless `scanned`), a wavelength not a whole nm within - `wavelength_range`, a bandwidth `_check_bandwidth` refuses, a filter slot outside its filter - slots. + `wavelength_range` or on a reader without a monochromator (`_check_monochromator`), a + bandwidth `_check_bandwidth` refuses, a filter slot outside its filter slots. Args: excitation_or_emission: "excitation" or "emission", the prefix of the fields named in the @@ -436,6 +475,10 @@ def _check_excitation_or_emission_on_reader( ] return [] errors = [] + if wavelength is not None or scanned: + errors += self._check_monochromator( + f"{excitation_or_emission}_wavelength; give {excitation_or_emission}_filter" + ) if wavelength is None: if not scanned: errors.append( diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py index 9154b7d960c..c1ca2923805 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py @@ -1191,6 +1191,50 @@ def test_the_wavelength_limits_follow_the_photomultipliers(self): (configuration.wavelength_range[1], configuration.dichroic_range[1]), highest ) + def test_the_filter_slots_are_the_readers_slides(self): + def slots(slides): + device = decoded() + device.monochromator_slides = slides + configuration = Fluorescence._get_reader_configuration(device) + return ( + configuration.excitation_filter_slots, + configuration.dichroic_filter_slots, + configuration.emission_filter_slots, + configuration.filter_slots_per_slide, + ) + + self.assertEqual(slots(decoded().monochromator_slides), (4, 3, 4, 2)) # the reader read + one_each = {name: {1: "open", 2: "Ex 482-16"} for name in ("EX_SP", "EX_LP", "EM_SP", "EM_LP")} + self.assertEqual( + slots({**one_each, "BSP": {1: "open", 2: "LP 504", 3: "LP 560"}}), (2, 2, 2, 1) + ) + defaults = (4, 3, 4, 2) + self.assertEqual(slots(None), defaults) + self.assertEqual(slots(one_each), defaults) # no dichroic slide + with patch.object(fluorescence.logger, "warning") as warning: + uneven = {**one_each, "EM_LP": {1: "open", 2: "a", 3: "b"}, "BSP": {1: "open", 2: "c"}} + self.assertEqual(slots(uneven), defaults) + warning.assert_called_once() + + async def test_without_a_monochromator_only_filters_read(self): + """Not a reader's configuration: linear variable filters reported absent.""" + io = FakeFTDI() + feature = make_fluorescence(io, fluorescence_plate()) + feature._driver.configuration.linear_variable_filters = False + with self.assertRaisesRegex(ValueError, "no monochromator"): + await feature.measure(BASELINE, gain=1000, focal_height=GROUND_TRUTH_FOCAL_HEIGHT) + with self.assertRaisesRegex(ValueError, "no monochromator"): + await feature.measure_spectrum( + **EMISSION_SPECTRUM, gain=1000, focal_height=GROUND_TRUTH_FOCAL_HEIGHT + ) + self.assertEqual(io.written, []) + through_filters = FluorescenceChannel( + excitation_filter=EX_TR, dichroic_filter=LP_TR, emission_filter=EM_615_18 + ) + self.assertEqual(feature._check_fluorescence_channel(through_filters), []) + feature._driver.configuration.linear_variable_filters = None # unknown: let through + self.assertEqual(feature._check_fluorescence_channel(BASELINE), []) + def load_filter_detection() -> List[Dict[str, Any]]: """The reader's filter detection, its 11 positions in order. diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence.py b/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence.py index 6e266a6451c..be2d710b9bc 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence.py @@ -151,13 +151,14 @@ def _check_arguments( The arguments are `measure`'s. Raises: - ValueError: listing every violation: an emission wavelength not a whole nm within + ValueError: listing every violation: a reader without a monochromator + (`_check_monochromator`), 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 = [] + errors = self._check_monochromator("luminescence") lowest, highest = self.configuration.wavelength_range if not isinstance(emission_wavelength, int) or not lowest <= emission_wavelength <= highest: errors.append( diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence_tests.py index b3fa5afc47b..80bef859bfa 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence_tests.py @@ -192,6 +192,17 @@ async def test_what_luminescence_cannot_read_sends_nothing(self): await feature.measure(**arguments) self.assertEqual(io.written, []) + async def test_without_a_monochromator_nothing_is_sent(self): + """Not a reader's configuration: linear variable filters reported absent.""" + io = FakeFTDI() + feature = make_luminescence(io, plate()) + feature._driver.configuration.linear_variable_filters = False + with self.assertRaisesRegex(ValueError, "no monochromator"): + await feature.measure(focal_height=GROUND_TRUTH_FOCAL_HEIGHT) + self.assertEqual(io.written, []) + feature._driver.configuration.linear_variable_filters = None # unknown: let through + self.assertEqual(feature._check_monochromator("luminescence"), []) + def test_the_wavelength_limits_follow_the_photomultipliers(self): # (extended red, dual, crossed over): the highest emission wavelength. for photomultipliers, highest in ( diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/measurement.py b/pylabrobot/bmg_labtech/clariostar/driver/features/measurement.py index f9208af59a9..fe262622d11 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/measurement.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/measurement.py @@ -325,6 +325,17 @@ def _check_bandwidth(name: str, bandwidth: float, bandwidth_range: Tuple[int, in return [f"{name} must be a whole {narrowest} to {widest} nm, got {bandwidth!r}"] return [] + def _check_monochromator(self, what: str) -> List[str]: + """The violation of `what` through a monochromator on a reader that reports none. + + The monochromators are linear variable filters. Refused only when the reader reports them + absent; unknown lets the read through. One bit for excitation, dichroic and emission + together; read as all or none (no reader without them is verified). + """ + if self._driver.configuration.linear_variable_filters is False: + return [f"the reader reports no monochromator, which {what} needs"] + return [] + @staticmethod def _check_gain(gain: int, gain_range: Tuple[int, int]) -> List[str]: """The violation of a gain: not whole within `gain_range`.""" diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/temperature_controller_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/features/temperature_controller_tests.py index 68abbd403a3..d6d112e5edf 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/temperature_controller_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/temperature_controller_tests.py @@ -15,6 +15,7 @@ TemperatureController, ) from pylabrobot.bmg_labtech.clariostar.driver.master_tests import ( + FIRMWARE_INFO_REPLY, PARAMETERS_REPLY, REFUSED_TEMPERATURE, STATUS_IDLE, @@ -45,7 +46,7 @@ def make_temperature_controller(io: FakeFTDI, extended: bool = False) -> Tempera class GatingTests(unittest.IsolatedAsyncioTestCase): async def discover_with(self, temperature_controller_fitted: bool): io = FakeFTDI() - io.replies = [STATUS_IDLE, PARAMETERS_REPLY] + io.replies = [STATUS_IDLE, PARAMETERS_REPLY, STATUS_IDLE, FIRMWARE_INFO_REPLY] driver = make_driver(io) driver.configuration = decoded() driver.configuration.temperature_controller = temperature_controller_fitted diff --git a/pylabrobot/bmg_labtech/clariostar/driver/master.py b/pylabrobot/bmg_labtech/clariostar/driver/master.py index 10a6db48b3d..8a797c940ad 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/master.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/master.py @@ -17,6 +17,7 @@ from pylabrobot.io.validation_utils import LOG_LEVEL_IO from .configuration import ( + BARCODE_READER_POSITIONS, DECODED_FIRMWARE, FITTED_FIELDS, DeviceConfiguration, @@ -54,10 +55,17 @@ usb_wrap_payload, ) from .ftdi_queued_reads import LATENCY_TIMER, QueuedReads -from .parameter_table import PARAMETER_TABLE, REPLY_OFFSET +from .parameter_table import ( + PARAMETER_TABLE, + REPLY_OFFSET, + SERIAL_NUMBER_ADDRESSES, + SIGNED_PARAMETERS, +) logger = logging.getLogger(__name__) +_MONTHS = ("Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec") + @dataclass(frozen=True) class CLARIOstarStatus: @@ -84,6 +92,11 @@ def _configure_usb_device(dev: Any) -> None: dev.set_configuration() +def _flag(channels: Optional[int]) -> Optional[bool]: + """Whether a reported channel count is non-zero; None if it was not read.""" + return None if channels is None else bool(channels) + + class CLARIOstarDriver: """Owns the CLARIOstar connection, the command layer, and the reader's features. @@ -130,6 +143,7 @@ class Command: DATA = 0x02 FOCUS_RESULT = 0x05 PARAMETERS = 0x07 + FIRMWARE_INFO = 0x09 SPECTRAL_DATA = 0x11 FILTER_RESULT = 0x1B # SET_PARAMETER; seen only writing the two-byte test-run counter @@ -731,77 +745,143 @@ async def initialize(self) -> None: timeout=self.default_read_timeout, poll_interval=self.default_poll_interval ) - async def request_parameters(self) -> Dict[str, int]: - """Request the reader's stored parameters, by the reader's own names (see ``PARAMETER_TABLE``). - - The reader answers all of them in one reply. Its project and serial number are left out. + async def _request_parameters_reply(self) -> bytes: + """Request the reader's stored parameters (REQUEST `07`) and return the whole reply. Raises: - ValueError: if the reply is not a whole parameters reply, or comes from firmware outside - ``DECODED_FIRMWARE``. + ValueError: if the reply is not a whole stored parameters reply. """ payload = await self.send_command( self.CommandFamily.REQUEST, self.Command.PARAMETERS, parameters=b"\x00\x00\x00\x00\x00" ) if not payload or payload[0] != self.Command.PARAMETERS or len(payload) < 260: - raise ValueError(f"not a whole parameters reply: {len(payload)} bytes") + raise ValueError(f"not a whole stored parameters reply: {len(payload)} bytes") + return payload + + @staticmethod + def _decode_parameters(payload: bytes) -> Dict[str, int]: + """A stored parameters reply's values, by key (see ``PARAMETER_TABLE``). + + Raises: + ValueError: if the reply comes from firmware outside ``DECODED_FIRMWARE``. + """ parameters = { name: int.from_bytes( - payload[address + REPLY_OFFSET : address + REPLY_OFFSET + size], "little" + payload[address + REPLY_OFFSET : address + REPLY_OFFSET + size], + "little", + signed=name in SIGNED_PARAMETERS, ) for name, address, size in PARAMETER_TABLE } - if not DECODED_FIRMWARE[0] <= parameters["Firmware version"] <= DECODED_FIRMWARE[1]: - raise ValueError(f"parameters of firmware {parameters['Firmware version']} are not decoded") + if not DECODED_FIRMWARE[0] <= parameters["firmware_version"] <= DECODED_FIRMWARE[1]: + raise ValueError( + f"stored parameters of firmware {parameters['firmware_version']} are not decoded" + ) return parameters + @staticmethod + def _decode_serial_number(payload: bytes) -> str: + """A stored parameters reply's serial number, "NNN-NNNN" (``SERIAL_NUMBER_ADDRESSES``).""" + first, second = ( + int.from_bytes(payload[address + REPLY_OFFSET : address + REPLY_OFFSET + 2], "little") + for address in SERIAL_NUMBER_ADDRESSES + ) + return f"{first:03d}-{second:04d}" + + def _get_link_serial_number(self) -> Optional[str]: + """The serial number the reader was opened by: the FTDI chip's or the USB device's; None if + it was opened by address.""" + fields = self.io.serialize() + if fields.get("usb_address") or fields.get("device_address"): + return None + return fields.get("device_id") or fields.get("serial_number") + + async def request_parameters(self) -> Dict[str, int]: + """Request the reader's stored parameters, by key (see ``PARAMETER_TABLE``). + + The reader answers all of them in one reply. Its serial number is left out: `discover` reads + it. + + Raises: + ValueError: if the reply is not a whole stored parameters reply, or comes from firmware + outside ``DECODED_FIRMWARE``. + """ + return self._decode_parameters(await self._request_parameters_reply()) + async def request_fluorescence_channels(self) -> int: """Request the number of fluorescence measurement channels.""" - return (await self.request_parameters())["No. of fluorescence measurement channels"] + return (await self.request_parameters())["fluorescence_measurement_channel_count"] async def request_luminescence_channels(self) -> int: """Request the number of luminescence measurement channels.""" - return (await self.request_parameters())["No. of luminescence measurement channels"] + return (await self.request_parameters())["luminescence_measurement_channel_count"] async def request_time_resolved_fluorescence_channels(self) -> int: """Request the number of time-resolved fluorescence measurement channels.""" parameters = await self.request_parameters() - return parameters["No. of time-resolved fluorescence meas. channels"] + return parameters["time_resolved_fluorescence_measurement_channel_count"] async def request_absorbance_available(self) -> bool: """Request whether absorbance measurements are available.""" - return (await self.request_parameters())["Absorbance measurement option"] != 0 + return (await self.request_parameters())["absorbance_measurement_option"] != 0 async def request_alpha_screen_available(self) -> bool: """Request whether AlphaScreen measurements are available.""" - return (await self.request_parameters())["AlphaScreen measurement option"] != 0 + return (await self.request_parameters())["alpha_screen_measurement_option"] != 0 async def request_fluorescence_polarization_available(self) -> bool: """Request whether fluorescence polarization measurements are available.""" parameters = await self.request_parameters() - return parameters["Fluorescence polarization measurement option"] != 0 + return parameters["fluorescence_polarization_measurement_option"] != 0 async def request_bottom_measurement_available(self) -> bool: """Request whether measurements from below the plate are available.""" - return (await self.request_parameters())["Bottom measurement option"] != 0 + return (await self.request_parameters())["bottom_measurement_option"] != 0 async def request_extended_temperature_range(self) -> bool: """Request whether the extended temperature range (10 to 65 C) is built in.""" - return (await self.request_parameters())["Extended incubator"] != 0 + return (await self.request_parameters())["extended_incubator"] != 0 async def request_high_speed_shaking_allowed(self) -> bool: """Request whether shaking at 800-1100 rpm is allowed.""" - return (await self.request_parameters())["High speed shaking allowed"] != 0 + return (await self.request_parameters())["high_speed_shaking_allowed"] != 0 async def request_well_1536_plates_supported(self) -> bool: """Request whether the reader is set up for 1536-well plates, an upgrade fitted to it.""" - return (await self.request_parameters())["1536 well measurements enabled"] != 0 + return (await self.request_parameters())["well_1536_measurements_enabled"] != 0 + + @staticmethod + def _decode_firmware_build_date(payload: bytes) -> str: + """A firmware info reply's compile date ("Nov 20 2020" at bytes 8-19), as "YYYY-MM-DD". + + Raises: + ValueError: if the payload is not a firmware info reply, or its date is not one. + """ + if not payload or payload[0] != 0x0A or len(payload) < 20: + raise ValueError(f"not a firmware info reply: {payload[:1].hex()}, {len(payload)} bytes") + text = payload[8:20].split(b"\x00")[0].decode("ascii", errors="replace") + try: + month, day, year = text.split() + return f"{int(year):04d}-{_MONTHS.index(month) + 1:02d}-{int(day):02d}" + except ValueError: + raise ValueError(f"not a firmware build date: {text!r}") from None + + async def request_firmware_build_date(self) -> str: + """Request the day the reader's firmware was compiled (REQUEST `09`), as "YYYY-MM-DD". + + Raises: + ValueError: if the reply is not a firmware info reply with a date. + """ + payload = await self.send_command( + self.CommandFamily.REQUEST, self.Command.FIRMWARE_INFO, parameters=b"\x00" * 5 + ) + return self._decode_firmware_build_date(payload) async def _set_parameter(self, name: str, value: int) -> None: """Write one of the reader's stored parameters (see ``PARAMETER_TABLE``). Args: - name: the parameter, by the reader's own name. + name: the parameter's key in ``PARAMETER_TABLE``. value: the new value, unsigned, within the parameter's size. Raises: @@ -841,21 +921,35 @@ def _check_firmware_compatibility(self) -> None: ) async def discover(self) -> None: - """Read the stored parameters into ``configuration``. + """Read the stored parameters into ``configuration``, the serial number with them, and the + firmware build date. - Fields stay unchanged when the reply is not decoded. The firmware version is reported twice + Fields stay unchanged when a reply is not decoded. The firmware version is reported twice (INIT reply, stored parameters); logs a warning if they disagree, or for a version not checked - on hardware. + on hardware, and if the serial number is not the one the reader was opened by. """ versions = {"INIT reply": self.configuration.firmware_version} # The stored parameters first, right after INIT. try: - parameters = await self.request_parameters() + payload = await self._request_parameters_reply() + parameters = self._decode_parameters(payload) except ValueError as e: logger.warning("stored parameters not decoded (%s)", e) else: - versions["stored parameters"] = f"{parameters['Firmware version'] / 1000:.2f}" + versions["stored parameters"] = f"{parameters['firmware_version'] / 1000:.2f}" self.configuration.update_from_parameters(parameters) + self.configuration.serial_number = self._decode_serial_number(payload) + opened_by = self._get_link_serial_number() + if opened_by is not None and opened_by != self.configuration.serial_number: + logger.warning( + "the reader's serial number %s is not the one it was opened by, %s", + self.configuration.serial_number, + opened_by, + ) + try: + self.configuration.firmware_build_date = await self.request_firmware_build_date() + except ValueError as e: + logger.warning("firmware build date not read (%s)", e) reported = {source: version for source, version in versions.items() if version is not None} if len(set(reported.values())) > 1: logger.warning( @@ -924,8 +1018,8 @@ def _describe_link(self) -> str: return link + (f" ({', '.join(named)})" if named else "") def format_setup_summary(self) -> str: - """One block describing the reader that was found: how it is reached, its model and firmware, - and what is fitted. + """One block describing the reader that was found: how it is reached and every field of its + configuration; a field that was not read shows None. Returns: A multi-line summary, or a note that setup has not run. @@ -933,39 +1027,96 @@ def format_setup_summary(self) -> str: c = self.configuration if c.model is None: return f"[{self.name}] not discovered yet" - temperature = "none" - if c.temperature_controller: - temperature = "10-65 C" if c.extended_temperature_range else "25-45 C" - measurements = [ - label - for label, available in ( - ("absorbance", c.absorbance_available), - ("fluorescence", c.fluorescence_channels), - ("luminescence", c.luminescence_channels), - ("time-resolved fluorescence", c.time_resolved_fluorescence_channels), - ("fluorescence polarization", c.fluorescence_polarization_available), - ("AlphaScreen", c.alpha_screen_available), - ) - if available - ] - fitted = [ - label - for label, present in ( - ("pumps", c.pumps and any(c.pumps)), - ("stacker", c.stacker_connected), - ("barcode readers", c.barcode_readers), - ("atmospheric control unit", c.atmospheric_control_unit_connected), - ) - if present - ] + standard = c.extended_red_photomultiplier_tube + if standard is not None: + standard = not c.extended_red_photomultiplier_tube or bool(c.dual_photomultiplier_tubes) + pumps = c.pumps or (None, None, None) + barcode_readers = c.barcode_readers + groups: Tuple[Tuple[str, Tuple[Tuple[str, Any], ...]], ...] = ( + ( + "Measurement modes", + ( + ("absorbance", c.absorbance_available), + ("fluorescence", _flag(c.fluorescence_channels)), + ("luminescence", _flag(c.luminescence_channels)), + ("time-resolved fluorescence", _flag(c.time_resolved_fluorescence_channels)), + ("fluorescence polarization, visible", c.fluorescence_polarization_available), + ( + "fluorescence polarization, ultraviolet", + c.fluorescence_polarization_ultraviolet_available, + ), + ("AlphaScreen", c.alpha_screen_available), + ), + ), + ( + "Detectors", + ( + ("standard photomultiplier", standard), + ("extended-red photomultiplier", c.extended_red_photomultiplier_tube), + ("dual photomultipliers", c.dual_photomultiplier_tubes), + ("crossover", c.photomultiplier_tube_crossover), + ), + ), + ( + "Optics", + ( + ("monochromators", c.linear_variable_filters), + ("spectrometer", c.spectrometer), + ("bottom optic", c.bottom_measurement_available), + ), + ), + ) + details = [f"build {c.firmware_build}"] if c.firmware_build is not None else [] + details += [f"built {c.firmware_build_date}"] if c.firmware_build_date else [] + firmware_details = f" ({', '.join(details)})" if details else "" lines = [ f"[{self.name}] Connected on {self._describe_link()}", f" Model: {c.model}", - f" Firmware: {c.firmware_version or 'unknown'}", - f" Measurements: {', '.join(measurements) or 'none reported'}", - f" Temperature control: {temperature}", - f" Fitted: {', '.join(fitted) or 'nothing else'}", + f" Serial number: {c.serial_number}", + f" Firmware: {c.firmware_version}{firmware_details}", + f" Mainboard revision: {c.mainboard_revision}", ] + + def add_group(title: str, rows: Tuple[Tuple[str, Any], ...]) -> None: + lines.append(f" {title}:") + lines.extend(f" - {label}: {value}" for label, value in rows) + + for title, rows in groups: + add_group(title, rows) + if self.fluorescence is not None: + slots = self.fluorescence.configuration + defaults = Fluorescence._compute_filter_slots(c.monochromator_slides) is None + lines.append( + f" Filter slots: excitation {slots.excitation_filter_slots}, dichroic " + f"{slots.dichroic_filter_slots}, emission {slots.emission_filter_slots}" + + (" (defaults: the slides are not reported)" if defaults else "") + ) + add_group( + "Temperature control", + ( + ("temperature controller", c.temperature_controller), + ("channels", c.temperature_controller_channels), + ("extended range (10-65 C)", c.extended_temperature_range), + ), + ) + lines.append(f" High-speed shaking: {c.high_speed_shaking_allowed}") + lines.append(f" 1536-well plates: {c.well_1536_plates_supported}") + add_group( + "Pumps", + tuple( + (f"pump {index}", None if c.pumps is None else kind or False) + for index, kind in enumerate(pumps, 1) + ), + ) + add_group( + "Barcode readers", + tuple( + (position, None if barcode_readers is None else barcode_readers.get(position, False)) + for position in BARCODE_READER_POSITIONS + ), + ) + lines.append(f" Stacker: {c.stacker_connected}") + lines.append(f" Atmospheric control unit: {c.atmospheric_control_unit_connected}") return "\n".join(lines) async def setup(self) -> None: diff --git a/pylabrobot/bmg_labtech/clariostar/driver/master_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/master_tests.py index fdc60d37f12..5bcc8253f95 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/master_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/master_tests.py @@ -305,11 +305,32 @@ async def test_command_byte_rules(self): class FirmwareTests(unittest.IsolatedAsyncioTestCase): async def test_discover_applies_the_stored_parameters(self): io = FakeFTDI() - io.replies = [STATUS_IDLE, PARAMETERS_REPLY] + io.replies = [STATUS_IDLE, PARAMETERS_REPLY, STATUS_IDLE, FIRMWARE_INFO_REPLY] driver = make_driver(io) await driver.discover() self.assertEqual(driver.configuration.extended_temperature_range, False) + async def test_discover_reads_the_firmware_build_date(self): + io = FakeFTDI() + io.replies = [STATUS_IDLE, PARAMETERS_REPLY, STATUS_IDLE, FIRMWARE_INFO_REPLY] + driver = make_driver(io) + driver.configuration.model = "CLARIOstarPlus" + driver.configuration.firmware_build = 1 + await driver.discover() + self.assertEqual(io.written[-1], FIRMWARE_INFO_FRAME) + build_date = driver.configuration.firmware_build_date + self.assertRegex(build_date or "", r"^\d{4}-\d{2}-\d{2}$") + self.assertIn(f"Firmware: None (build 1, built {build_date})", driver.format_setup_summary()) + + async def test_a_reply_without_a_build_date_warns(self): + io = FakeFTDI() + io.replies = [STATUS_IDLE, PARAMETERS_REPLY, STATUS_IDLE, PARAMETERS_REPLY] + driver = make_driver(io) + with self.assertLogs(master_module.logger, "WARNING") as logs: + await driver.discover() + self.assertTrue(any("firmware build date not read" in line for line in logs.output)) + self.assertIsNone(driver.configuration.firmware_build_date) + def test_unconfirmed_firmware_warns(self): driver = make_driver(FakeFTDI()) driver.configuration.firmware_version = "9.99" @@ -325,10 +346,15 @@ async def test_command_waits_until_the_device_is_idle(self): STATUS_BUSY, STATUS_IDLE, PARAMETERS_REPLY, + STATUS_IDLE, + FIRMWARE_INFO_REPLY, ] driver = make_driver(io) await driver.discover() - self.assertEqual(io.written, [STATUS_QUERY_FRAME] * 3 + [PARAMETERS_FRAME]) + self.assertEqual( + io.written, + [STATUS_QUERY_FRAME] * 3 + [PARAMETERS_FRAME, STATUS_QUERY_FRAME, FIRMWARE_INFO_FRAME], + ) async def test_when_idle_false_sends_at_once(self): io = FakeFTDI() @@ -394,6 +420,8 @@ async def test_warning_is_logged_once_and_not_raised(self): STATUS_WARNING, STATUS_IDLE, PARAMETERS_REPLY, + STATUS_IDLE, + FIRMWARE_INFO_REPLY, ] driver = make_driver(io) with self.assertLogs(master_module.logger, "WARNING") as logs: diff --git a/pylabrobot/bmg_labtech/clariostar/driver/parameter_table.py b/pylabrobot/bmg_labtech/clariostar/driver/parameter_table.py index dd0a5599639..824fa208457 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/parameter_table.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/parameter_table.py @@ -1,134 +1,165 @@ -"""The reader's stored parameters: name, address and size in bytes, read with REQUEST 0x07. +"""The reader's stored parameters: key, address and size in bytes, read with REQUEST 0x07. -The names are the reader's own. The project and serial number are stored too and left out here. +Each key describes the stored value; the table is grouped by what the values are for. The serial +number is not in the table: it is the project number and the reader's number, at +`SERIAL_NUMBER_ADDRESSES`. """ from typing import Tuple -# A parameter at address n is reply byte n + 7; values are unsigned, little endian. +# A parameter at address n is reply byte n + 7; values are little endian. REPLY_OFFSET = 7 +# The serial number "NNN-NNNN": the project number (u16 at 154), then the reader's number (156). +SERIAL_NUMBER_ADDRESSES = (154, 156) + PARAMETER_TABLE: Tuple[Tuple[str, int, int], ...] = ( - ("EEPROM parameter list already initialized", 0, 1), - ("Max. number of allowed temp. errors before warning", 3, 1), - ("Exhibition demo mode", 22, 2), - ("Extended incubator", 134, 1), - ("Use microplate sensor", 33, 1), - ("Skip monochromator calibration", 10, 1), - ("No additional K factor determination after gain adj.", 44, 1), - ("Use ultra fast XY motor profile", 38, 1), - ("High speed shaking allowed", 139, 1), - ("No calibration of XY transport system after shaking", 15, 1), - ("Reader is integration unit", 145, 1), - ("No. of fluorescence measurement channels", 132, 1), - ("Excitation source for fluorescence testruns", 58, 1), - ("No. of time-resolved fluorescence meas. channels", 142, 1), - ("Fluorescence polarization measurement option", 143, 1), - ("No. of luminescence measurement channels", 128, 1), - ("Mode for internal online blank correction (FI)", 61, 1), - ("Mode for internal online blank correction (LUM)", 2, 1), - ("Absorbance measurement option", 133, 1), - ("0/100% absorbance data measurement mode", 31, 1), - ("Intensity value for flash lamp HV power supply (ABS)", 12, 2), - ("AlphaScreen measurement option", 135, 1), - ("Excitation source for AlphaScreen testruns", 59, 1), - ("Flash lamp stress warning limit", 80, 2), - ("Alternative flash rate", 136, 2), - ("Alternative flash rate methods", 138, 1), - ("Bottom measurement option", 141, 1), - ("PMT configuration", 129, 1), - ("Raw data correction factor IR excitation", 66, 2), - ("Raw data correction factor IR emission", 68, 2), - ("1536 well measurements enabled", 140, 1), - ("Excitation optic slide not built in", 144, 1), - ("Fluorescence AutoGain gain value level 1 (PMT1)", 160, 2), - ("Fluorescence AutoGain gain value level 2 (PMT1)", 198, 2), - ("Fluorescence AutoGain gain value level 3 (PMT1)", 202, 2), - ("Fluorescence AutoGain gain value level 4 (PMT1)", 204, 2), - ("Fluorescence AutoGain gain value level 5 (PMT1)", 250, 2), - ("Luminescence AutoGain gain value level 1 (PMT1)", 246, 2), - ("Luminescence AutoGain gain value level 2 (PMT1)", 248, 2), - ("Fluorescence AutoGain gain value level 1 (PMT2)", 90, 2), - ("Fluorescence AutoGain gain value level 2 (PMT2)", 92, 2), - ("Fluorescence AutoGain gain value level 3 (PMT2)", 94, 2), - ("Fluorescence AutoGain gain value level 4 (PMT2)", 96, 2), - ("Fluorescence AutoGain gain value level 5 (PMT2)", 98, 2), - ("Luminescence AutoGain gain value level 1 (PMT2)", 100, 2), - ("Luminescence AutoGain gain value level 2 (PMT2)", 102, 2), - ("AutoGain calibration procedure enabled", 82, 1), - ("Fan 1 speed", 39, 2), - ("Fan 2 speed", 76, 2), - ("Fan 2 busy speed", 210, 1), - ("Fan 2 idle speed", 211, 1), - ("Fan 2 cover open speed", 212, 1), - ("Fan 2 power-on speed", 213, 1), - ("Fan 2 power-on time", 214, 2), - ("Fan 2 after test run speed", 216, 1), - ("Fan 2 after test run time", 218, 2), - ("Fan 2 speed when target temp. > threshold temp. 1", 232, 1), - ("Threshold temperature 1 for fan speed regulation", 234, 2), - ("Fan 2 speed when target temp. > threshold temp. 2", 233, 1), - ("Threshold temperature 2 for fan speed regulation", 236, 2), - ("Fan 2 speed after software termination", 238, 1), - ("Fan 2 after software termination time", 240, 2), - ("Syringe volume pump 1", 35, 1), - ("Syringe volume pump 2", 36, 1), - ("Pull back volume after injection (air volume)", 34, 1), - ("Prime speed pump 1", 4, 1), - ("Prime speed pump 2", 5, 1), - ("Invert dispensing mode for pump 1", 6, 1), - ("Invert dispensing mode for pump 2", 7, 1), - ("Stroke volume for pump 1 priming", 8, 1), - ("Stroke volume for pump 2 priming", 9, 1), - ("Distance between injection needles and microplate", 16, 1), - ("Distance between aperture spoon and microplate", 30, 1), - ("Lense protection mode", 60, 1), - ("Focal length top optical system [1/10 mm]", 170, 2), - ("Focal length bottom optical system [1/10 mm]", 206, 2), - ("Thickness of the microplate carrier [0.01mm]", 208, 2), - ("Top focus z height correction for filters type 1", 70, 1), - ("Top focus z height correction for filters type 2", 71, 1), - ("Top focus z height correction for filters type 3", 72, 1), - ("Top focus z height correction for filters type 4", 73, 1), - ("Temperature control 1 offset", 166, 2), - ("Temperature control 2 offset", 168, 2), - ("Temperature sensor 1 offset", 243, 3), - ("Temperature sensor 2 offset", 244, 3), - ("X offset 1 (standard top measurement position)", 150, 2), - ("Y offset 1 (standard top measurement position)", 152, 2), - ("X offset 2 (standard bottom measurement position)", 188, 2), - ("Y offset 2 (standard bottom measurement position)", 190, 2), - ("X offset 3 (spectrometer measurement position)", 192, 2), - ("Y offset 3 (spectrometer measurement position)", 194, 2), - ("X offset 6 (pump 1+2 injection position)", 228, 2), - ("Y offset 6 (pump 1+2 injection position)", 230, 2), - ("Z offset for optic focus system top", 164, 2), - ("Z offset for optic focus system bottom", 176, 2), - ("Offset for Z motor plate detection / needle transp.", 200, 2), - ("Z offset for aperture fine adjustment", 217, 1), - ("Offset for top/bottom switching slide", 162, 2), - ("Offset for PMT slide", 196, 2), - ("X transport lock position", 172, 2), - ("Y transport lock position", 174, 2), - ("Offset Scan: measurement values threshold", 45, 1), - ("Front plate barcode Y position", 48, 2), - ("Front plate barcode scanning resolution", 54, 2), - ("Right plate barcode X position", 50, 2), - ("Right plate barcode scanning resolution", 56, 2), - ("X position for plate carrier under stack 1", 178, 2), - ("Y position for plate carrier under stack 1", 180, 2), - ("X position for plate carrier under stack 2", 182, 2), - ("Y position for plate carrier under stack 2", 184, 2), - ("XY movement order", 86, 1), - ("Measure stop times of single actions on wells", 40, 1), - ("Data logging mode", 42, 1), - ("Data logging mode 2", 43, 1), - ("Store reference channel values as channel B values", 85, 1), - ("Flash off (dark measurement)", 84, 1), - ("Top/bottom switch position", 64, 1), - ("Disable door sensor monitoring", 29, 1), - ("Number of calibration errors that have occurred", 14, 1), - ("Gain value for focus adjustment procedure", 74, 2), - ("Firmware version", 158, 2), - ("Number of test runs performed", 18, 4), + # The reader's own counters and markers. + ("eeprom_parameter_list_initialized", 0, 1), + ("calibration_error_count", 14, 1), + ("test_run_count", 18, 4), + ("firmware_version", 158, 2), + # Fitted options. + ("luminescence_measurement_channel_count", 128, 1), + ("photomultiplier_tube_configuration", 129, 1), + ("fluorescence_measurement_channel_count", 132, 1), + ("absorbance_measurement_option", 133, 1), + ("extended_incubator", 134, 1), + ("alpha_screen_measurement_option", 135, 1), + ("high_speed_shaking_allowed", 139, 1), + ("well_1536_measurements_enabled", 140, 1), + ("bottom_measurement_option", 141, 1), + ("time_resolved_fluorescence_measurement_channel_count", 142, 1), + ("fluorescence_polarization_measurement_option", 143, 1), + ("no_excitation_optic_slide", 144, 1), + ("integration_unit", 145, 1), + # Measurement. + ("luminescence_online_blank_correction_mode", 2, 1), + ("skip_monochromator_calibration", 10, 1), + ("absorbance_flash_lamp_high_voltage_intensity", 12, 2), # 0 brightest to 4095 dimmest + ("absorbance_0_100_percent_measurement_mode", 31, 1), + ("measure_well_action_stop_times", 40, 1), + ("no_k_factor_determination_after_gain_adjustment", 44, 1), + ("offset_scan_measurement_threshold", 45, 1), # percent + ("fluorescence_test_run_excitation_source", 58, 1), + ("alpha_screen_test_run_excitation_source", 59, 1), + ("lens_protection_mode", 60, 1), + ("fluorescence_intensity_online_blank_correction_mode", 61, 1), + ("top_bottom_switch_position", 64, 1), + ("focus_adjustment_gain", 74, 2), + ("flash_lamp_stress_warning_limit", 80, 2), # 0.1 Hz + ("auto_gain_calibration_enabled", 82, 1), + ("flash_off_dark_measurement", 84, 1), + ("store_reference_channel_as_channel_b", 85, 1), + ("alternative_flash_rate", 136, 2), # Hz + ("alternative_flash_rate_methods", 138, 1), # bits + # Auto-gain levels, by photomultiplier tube. + ("fluorescence_auto_gain_level_1_photomultiplier_tube_1", 160, 2), + ("fluorescence_auto_gain_level_2_photomultiplier_tube_1", 198, 2), + ("fluorescence_auto_gain_level_3_photomultiplier_tube_1", 202, 2), + ("fluorescence_auto_gain_level_4_photomultiplier_tube_1", 204, 2), + ("fluorescence_auto_gain_level_5_photomultiplier_tube_1", 250, 2), + ("luminescence_auto_gain_level_1_photomultiplier_tube_1", 246, 2), + ("luminescence_auto_gain_level_2_photomultiplier_tube_1", 248, 2), + ("fluorescence_auto_gain_level_1_photomultiplier_tube_2", 90, 2), + ("fluorescence_auto_gain_level_2_photomultiplier_tube_2", 92, 2), + ("fluorescence_auto_gain_level_3_photomultiplier_tube_2", 94, 2), + ("fluorescence_auto_gain_level_4_photomultiplier_tube_2", 96, 2), + ("fluorescence_auto_gain_level_5_photomultiplier_tube_2", 98, 2), + ("luminescence_auto_gain_level_1_photomultiplier_tube_2", 100, 2), + ("luminescence_auto_gain_level_2_photomultiplier_tube_2", 102, 2), + # Optics: focus, slides and correction factors. + ("aperture_spoon_to_plate_distance", 30, 1), # 0.1 mm + ("do_not_use_smoothed_focus_curve", 41, 1), + ("infrared_excitation_raw_data_correction_factor", 66, 2), # x 0.01 + ("infrared_emission_raw_data_correction_factor", 68, 2), # x 0.01 + ("top_focus_z_correction_filter_type_1", 70, 1), # 0.01 mm + ("top_focus_z_correction_filter_type_2", 71, 1), # 0.01 mm + ("top_focus_z_correction_filter_type_3", 72, 1), # 0.01 mm + ("top_focus_z_correction_filter_type_4", 73, 1), # 0.01 mm + ("top_bottom_slide_offset", 162, 2), # steps + ("top_focus_z_offset", 164, 2), # steps + ("top_optic_focal_length", 170, 2), # 0.1 mm + ("bottom_focus_z_offset", 176, 2), # steps + ("photomultiplier_tube_slide_offset", 196, 2), # steps + ("bottom_optic_focal_length", 206, 2), # 0.1 mm + ("aperture_fine_adjustment_z_offset", 217, 1), # 0.1 mm + # Plate transport: positions and offsets. + ("no_xy_transport_calibration_after_shaking", 15, 1), + ("use_plate_sensor", 33, 1), + ("use_ultra_fast_xy_motor_profile", 38, 1), + ("xy_movement_order", 86, 1), + ("top_measurement_position_x_offset", 150, 2), # steps + ("top_measurement_position_y_offset", 152, 2), # steps + ("x_transport_lock_position", 172, 2), + ("y_transport_lock_position", 174, 2), + ("stack_1_carrier_x_position", 178, 2), + ("stack_1_carrier_y_position", 180, 2), + ("stack_2_carrier_x_position", 182, 2), + ("stack_2_carrier_y_position", 184, 2), + ("bottom_measurement_position_x_offset", 188, 2), # steps + ("bottom_measurement_position_y_offset", 190, 2), # steps + ("spectrometer_position_x_offset", 192, 2), # steps + ("spectrometer_position_y_offset", 194, 2), # steps + ("z_plate_detection_needle_transport_offset", 200, 2), # steps + ("plate_carrier_thickness", 208, 2), # 0.01 mm + ("injection_position_x_offset", 228, 2), + ("injection_position_y_offset", 230, 2), + # Pumps and injection. + ("pump_1_prime_speed", 4, 1), + ("pump_2_prime_speed", 5, 1), + ("pump_1_invert_dispensing_mode", 6, 1), + ("pump_2_invert_dispensing_mode", 7, 1), + ("pump_1_prime_stroke_volume", 8, 1), # percent of the syringe volume + ("pump_2_prime_stroke_volume", 9, 1), # percent of the syringe volume + ("injection_needle_to_plate_distance", 16, 1), # 0.1 mm + ("do_not_use_prime_sensors", 17, 1), + ("injection_pull_back_air_volume", 34, 1), # ul + ("pump_1_syringe_volume", 35, 1), + ("pump_2_syringe_volume", 36, 1), + ("inverted_smart_dispensing_mode", 79, 1), + # Barcode readers. + ("front_barcode_y_position", 48, 2), # steps + ("right_barcode_x_position", 50, 2), # steps + ("front_barcode_scanning_resolution", 54, 2), # steps per mm + ("right_barcode_scanning_resolution", 56, 2), # steps per mm + # Fans and temperature. + ("maximum_temperature_errors_before_warning", 3, 1), + ("power_on_incubation", 24, 2), # 0 off, 1 temperature monitoring only, else the target in 0.1 C + ("fan_1_speed", 39, 1), # percent, 0 controlled by the reader + ("fan_2_speed", 76, 2), # percent, above 100 controlled by the reader + ("temperature_control_1_offset", 166, 2), # 0.1 C + ("temperature_control_2_offset", 168, 2), # 0.1 C + ("fan_2_busy_speed", 210, 1), # percent + ("fan_2_idle_speed", 211, 1), # percent + ("fan_2_cover_open_speed", 212, 1), # percent + ("fan_2_power_on_speed", 213, 1), # percent + ("fan_2_power_on_time", 214, 2), # s + ("fan_2_after_test_run_speed", 216, 1), # percent + ("fan_2_after_test_run_time", 218, 2), # s + ("fan_2_speed_above_threshold_temperature_1", 232, 1), # percent + ("fan_2_speed_above_threshold_temperature_2", 233, 1), # percent + ("fan_threshold_temperature_1", 234, 2), # 0.1 C + ("fan_threshold_temperature_2", 236, 2), # 0.1 C + ("fan_2_after_software_termination_speed", 238, 1), # percent + ("fan_2_after_software_termination_time", 240, 2), # s + ("temperature_sensor_1_offset", 243, 1), # 0.1 C + ("temperature_sensor_2_offset", 244, 1), # 0.1 C + # Device. + ("exhibition_demo_mode", 22, 2), # bits + ("disable_door_sensor_monitoring", 29, 1), + ("data_logging_mode", 42, 1), # bits + ("data_logging_mode_2", 43, 1), # bits +) + +# The parameters that can be negative; the rest are unsigned. +SIGNED_PARAMETERS = ( + "top_bottom_slide_offset", + "temperature_control_1_offset", + "temperature_control_2_offset", + "x_transport_lock_position", + "y_transport_lock_position", + "spectrometer_position_x_offset", + "temperature_sensor_1_offset", + "temperature_sensor_2_offset", ) diff --git a/pylabrobot/bmg_labtech/clariostar/driver/recordings/clariostar_plus_1_35_temperature_controller.json b/pylabrobot/bmg_labtech/clariostar/driver/recordings/clariostar_plus_1_35_temperature_controller.json index 75436eb96b9..643bbc14ba3 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/recordings/clariostar_plus_1_35_temperature_controller.json +++ b/pylabrobot/bmg_labtech/clariostar/driver/recordings/clariostar_plus_1_35_temperature_controller.json @@ -4,6 +4,8 @@ "serial_number": "000-0000", "mainboard_revision": 2, "firmware_version": "1.35", + "firmware_build": 1, + "firmware_build_date": null, "extended_red_photomultiplier_tube": true, "dual_photomultiplier_tubes": true, "photomultiplier_tube_crossover": false, diff --git a/pylabrobot/bmg_labtech/clariostar/driver/simulator.py b/pylabrobot/bmg_labtech/clariostar/driver/simulator.py index af75e4ed978..8f7d4e9345f 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/simulator.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/simulator.py @@ -463,6 +463,10 @@ async def request_well_1536_plates_supported(self) -> bool: """Whether the reader is set up for 1536-well plates, from the configuration.""" return _require_read(self.configuration.well_1536_plates_supported, "1536-well plates") + async def request_firmware_build_date(self) -> str: + """The day the firmware was compiled, from the configuration.""" + return _require_read(self.configuration.firmware_build_date, "firmware build date") + async def discover(self) -> None: """Read the configuration from ``declared_configuration_json``.""" if self.declared_configuration_json is None: