From a60c294c9a0b46784f39307aa8647f2877c0325e Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Thu, 22 Oct 2026 16:38:48 +0100 Subject: [PATCH] v1 `CLARIOstar`: fluorescence grids, and the centres of any wells from them - `Fluorescence.measure_grid` reads the signal at every point of a 31 x 31 grid around one or two given points on the plate holder, at one focal height and gain, with the centre the reader finds in each; untested on hardware for points other than a plate's first and last well. - `Fluorescence.measure_well_centers` does so for any wells of the plate, two to a scan; without a focal height a z-scan of the first well gives it, and the gain too unless one is given. - The scan (`07`), its layout (`05 0d`) and the centres (`05 04`) are new; the plate block and the optics are the existing encodings, and the counts are u32 from byte 34. Co-Authored-By: Claude Opus 5.5 --- .../driver/features/fluorescence.py | 412 +++++++++++++++++- .../driver/features/fluorescence_tests.py | 192 +++++++- .../test_data/grid_ground_truth.json.gz | Bin 0 -> 6191 bytes .../bmg_labtech/clariostar/driver/master.py | 4 + 4 files changed, 604 insertions(+), 4 deletions(-) create mode 100644 pylabrobot/bmg_labtech/clariostar/driver/features/test_data/grid_ground_truth.json.gz diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py index f179da5e340..c915c298ab4 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py @@ -34,7 +34,7 @@ F, FilterSlide, ) -from pylabrobot.resources import Plate, Well +from pylabrobot.resources import Coordinate, Plate, Well from ..errors import FrameError from .measurement import ( @@ -108,6 +108,36 @@ class CLARIOstarZProfileResult: end_timestamp: datetime +@dataclass(frozen=True) +class CLARIOstarFluorescenceGridResult: + """One grid: the counts over a square grid of points around `position`. + + Attributes: + position: the grid's centre, x and y in mm (z 0): on the plate for a well, on the plate holder + for a given point, both from their left-front corner. + well: the identifier of the well at `position` (``"A1"``); None for a given point. + center: where the reader found the signal's centre, in the frame of `position`; None if it + found none. + spacing: mm between neighbouring points of the grid. + counts: the grid's counts, rows of points in the order the reader sends them; which way the + rows and columns run is not verified. + focal_height: the focal height the grid was read at, mm. + gain: the detector gain it was read at. + start_timestamp: when the scan started. + end_timestamp: when its last grid and the centres were in. + """ + + position: Coordinate + well: Optional[str] + center: Optional[Coordinate] + spacing: float + counts: List[List[int]] + focal_height: float + gain: int + start_timestamp: datetime + end_timestamp: datetime + + @dataclass class CLARIOstarFluorescenceSpectrumResult(FluorescenceSpectrumResult): """A CLARIOstar fluorescence spectrum, with the detector gain it was read with. @@ -1311,6 +1341,386 @@ async def measure_z_profile( reply, plate, start_timestamp=start_timestamp, end_timestamp=datetime.now() ) + # ---------------------------------------- + # Grids + # ---------------------------------------- + + # `07`'s first byte: the scan of a geometry's first and last well, the only one verified. + _GRID_SCAN = 0x02 + # The detection byte: top or bottom optic; its high nibble, the corner mode, stays 0. + _GRID_DETECTION: Dict[str, int] = {"top": 0x00, "bottom": 0x01} + # The gain target of the z-scan before a scan, in percent: 50 before every scan on hardware. + _GRID_GAIN_TARGET_PERCENT = 50.0 + # A grid reply's counts, u32 BE from this byte. + _GRID_COUNTS_START = 34 + # The plate holder's footprint given points are placed on, mm (the SLAS plate footprint). + _PLATE_HOLDER_SIZE = (127.76, 85.48) + + # -- checks -------------------------------------------------------------------------------------- + + def _check_grid( + self, + fluorescence_channel: FluorescenceChannel, + optic_position: str, + gain: Optional[int], + focal_height: Optional[float], + timeout: Optional[float], + ) -> List[str]: + """The violations of a scan's fluorescence channel, optic position, gain, focal height and + timeout; a gain or focal height None is not checked.""" + errors = self._check_fluorescence_channel(fluorescence_channel) + errors += self._check_optic_position(optic_position) + if gain is not None: + 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) + if timeout is not None and not timeout > 0: + errors.append(f"timeout must be positive, got {timeout}") + return errors + + # -- `07` parameters ----------------------------------------------------------------------------- + + @staticmethod + def _encode_grid_geometry(size: Tuple[float, float], points: Sequence[Coordinate]) -> bytes: + """A made-up plate's 14 bytes around one or two points, as `_encode_plate_geometry`. + + Its first well at the first point and its last at the second (or the first again), with 2 + columns and 1 row (1 and 1 for one point); Y measured from the back edge. + """ + first = points[0] + last = points[-1] + values = (size[0], size[1], first.x, size[1] - first.y, last.x, size[1] - last.y) + block = b"".join(round(value * 100).to_bytes(2, "big") for value in values) + return block + (bytes([2, 1]) if len(points) == 2 else bytes([1, 1])) + + def _compute_grid_parameters( + self, + geometry: bytes, + fluorescence_channel: FluorescenceChannel, + optic_position: str, + focal_height: float, + gain: int, + ) -> bytes: + """`07`'s 40 parameter bytes. + + `02`, the 14-byte geometry, the detection byte, `_Z_PROFILE_OPTICS_HEADER`, the optics + (`_encode_optics`), the focal height in 0.01 mm and the gain (u16 BE), then `00 00` (the + wavelength of an absorbance scan). + """ + optics = self._encode_optics( + *self._compute_optics(fluorescence_channel), + *self._compute_wavelength_selectors(fluorescence_channel), + ) + return ( + bytes([self._GRID_SCAN]) + + geometry + + bytes([self._GRID_DETECTION[optic_position]]) + + self._Z_PROFILE_OPTICS_HEADER + + optics + + round(focal_height * 100).to_bytes(2, "big") + + gain.to_bytes(2, "big") + + bytes(2) + ) + + # -- decoding ------------------------------------------------------------------------------------ + + @staticmethod + def _decode_grid_layout(reply: bytes) -> Tuple[int, float]: + """A scan's grid side and point spacing in mm, from its PLATE_MAP_CONFIG (`05 0d`) reply. + + Byte 6 the side, 7-8 the points (side squared), 9-10 the spacing in um, 11-12 the length of a + mask of the points used (0: every point). + + Raises: + ValueError: if the reply is cut short, or its points or mask are not a full square grid. + """ + if len(reply) < 13: + raise ValueError(f"PLATE_MAP_CONFIG reply too short, got {len(reply)} bytes") + side = reply[6] + points = int.from_bytes(reply[7:9], "big") + if points != side * side or int.from_bytes(reply[11:13], "big"): + raise ValueError(f"PLATE_MAP_CONFIG reply is not a full grid: side {side}, {points} points") + return side, int.from_bytes(reply[9:11], "big") / 1000 + + @classmethod + def _decode_grid(cls, reply: bytes, side: int) -> Tuple[Tuple[int, int], List[List[int]]]: + """A grid's well and counts, from its data reply. + + Bytes 7-8 the points, 23-24 and 25-26 the well's column and row (from 1, u16 BE), then the + counts, u32 BE from byte 34, `side` to a row. + + Raises: + ValueError: if the reply does not hold `side` squared counts. + """ + points = side * side + start = cls._GRID_COUNTS_START + if len(reply) < start + 4 * points or int.from_bytes(reply[7:9], "big") != points: + raise ValueError(f"grid reply does not hold {points} counts, got {len(reply)} bytes") + counts = [int.from_bytes(reply[i : i + 4], "big") for i in range(start, start + 4 * points, 4)] + well = (int.from_bytes(reply[23:25], "big"), int.from_bytes(reply[25:27], "big")) + return well, [counts[row : row + side] for row in range(0, points, side)] + + @staticmethod + def _decode_grid_centers( + reply: bytes, size_y: float + ) -> Tuple[Optional[Coordinate], Optional[Coordinate]]: + """The first and last well's centres the reader found, from its PLATE_MAP_XY (`05 04`) reply. + + X1, Y1, Xn, Yn at bytes 8-15, u16 BE in 0.01 mm, Y from the back edge of a plate `size_y` mm + deep; returned from the front edge. All zero: the reader found none. + + Raises: + ValueError: if the reply is cut short. + """ + if len(reply) < 16: + raise ValueError(f"PLATE_MAP_XY reply too short, got {len(reply)} bytes") + x1, y1, x2, y2 = (int.from_bytes(reply[i : i + 2], "big") / 100 for i in range(8, 16, 2)) + if not any((x1, y1, x2, y2)): + return None, None + return Coordinate(x1, size_y - y1, 0), Coordinate(x2, size_y - y2, 0) + + # -- measuring ----------------------------------------------------------------------------------- + + async def _run_grid( + self, parameters: bytes, timeout: float + ) -> Tuple[int, float, List[bytes], bytes]: + """Send a scan (`07`), collect its grids and the centres the reader found. + + PLATE_MAP_CONFIG right after the scan; then status until the reader is idle, a grid + (`05 02`) each time status byte 2 bit 0 is set; once idle, PLATE_MAP_XY. Any failure, a + timeout or a cancelled call stops it (`0b 00`) first. + + Returns: + the grid side, the point spacing in mm, the grid replies and the PLATE_MAP_XY reply. + + Raises: + TimeoutError: if the scan takes longer than `timeout` seconds. + """ + driver = self._driver + await driver.send_command(driver.CommandFamily.PLATE_MAP_SCAN, parameters=parameters) + start = time.monotonic() + grids: List[bytes] = [] + try: + layout = await driver.send_command( + driver.CommandFamily.REQUEST, + driver.Command.PLATE_MAP_CONFIG, + parameters=bytes(5), + when_idle=False, + ) + side, spacing = self._decode_grid_layout(layout) + while True: + await asyncio.sleep(driver.default_poll_interval) + status = await driver.send_command(driver.CommandFamily.STATUS, retries=5) + if len(status) > 2 and status[2] & self._DATA_READY: + grids.append(await self._request_page(retries=1)) + continue + if not driver._parse_status(status).busy: + break + if time.monotonic() - start >= timeout: + raise TimeoutError(f"the scan took longer than {timeout} s") + centers = await driver.send_command( + driver.CommandFamily.REQUEST, driver.Command.PLATE_MAP_XY, parameters=bytes(5) + ) + except BaseException: + await self._drain_and_stop("scan") + raise + return side, spacing, grids, centers + + async def _get_grids( + self, + parameters: bytes, + positions: Sequence[Coordinate], + wells: Sequence[Optional[str]], + size_y: float, + focal_height: float, + gain: int, + timeout: Optional[float], + ) -> List[CLARIOstarFluorescenceGridResult]: + """Run one scan of one or two `positions` and return a grid for each, in order. + + A grid that repeats the well before it is dropped: the reader may send a one-point scan's + point twice. + + Raises: + RuntimeError: if the scan returns another number of grids than `positions`. + """ + start_timestamp = datetime.now() + side, spacing, replies, centers_reply = await self._run_grid( + parameters, self.default_read_timeout if timeout is None else timeout + ) + grids: List[List[List[int]]] = [] + previous: Optional[Tuple[int, int]] = None + for reply in replies: + well, counts = self._decode_grid(reply, side) + if well != previous: + grids.append(counts) + previous = well + if len(grids) != len(positions): + raise RuntimeError(f"the scan returned {len(grids)} grids for {len(positions)} points") + centers = self._decode_grid_centers(centers_reply, size_y) + end_timestamp = datetime.now() + return [ + CLARIOstarFluorescenceGridResult( + position=position, + well=well, + center=center, + spacing=spacing, + counts=counts, + focal_height=focal_height, + gain=gain, + start_timestamp=start_timestamp, + end_timestamp=end_timestamp, + ) + for position, well, center, counts in zip(positions, wells, centers, grids) + ] + + async def measure_grid( + self, + fluorescence_channel: FluorescenceChannel, + *, + positions: Sequence[Coordinate], + gain: int, + focal_height: float, + optic_position: Optional[OpticPosition] = None, + timeout: Optional[float] = None, + ) -> List[CLARIOstarFluorescenceGridResult]: + """Measure the signal over a grid of points around each of one or two `positions`. + + - The reader reads a 31 x 31 grid around each point, at one focal height and gain, and finds + the signal's centre (about 50-80 s a point). + - The points need not be wells: any point the optic reaches on the plate holder. + - Cancelling the call stops it on the reader. + + Args: + fluorescence_channel: the optics, as for `measure` (`_check_fluorescence_channel`). + positions: one or two points, x and y in mm on the plate holder from its left-front corner. + gain: the detector gain, within `configuration.gain_range`. + focal_height: mm, in 0.1 mm steps within `configuration.focal_height_range`. + optic_position: "top" or "bottom"; `default_optic_position` when None. + timeout: seconds the scan may take; `default_read_timeout` when None. + + Returns: + a grid for each point, in order. + + Raises: + ValueError: if an argument is out of range, or a reply is not what the scan sends. + TimeoutError: if the scan takes longer than `timeout` seconds. + RuntimeError: if the scan returns another number of grids than points. + """ + filled = self._get_filled_fluorescence_channel(fluorescence_channel) + optic = self.default_optic_position if optic_position is None else optic_position + errors = self._check_grid(filled, optic, gain, focal_height, timeout) + size_x, size_y = self._PLATE_HOLDER_SIZE + if not 1 <= len(positions) <= 2: + errors.append(f"positions must hold one or two points, got {len(positions)}") + for point in positions: + if not (0 <= point.x <= size_x and 0 <= point.y <= size_y): + errors.append( + f"positions must lie on the plate holder (0-{size_x}, 0-{size_y}), got {point}" + ) + if errors: + raise ValueError("; ".join(errors)) + self._warn_untested("a grid at given positions") + geometry = self._encode_grid_geometry(self._PLATE_HOLDER_SIZE, positions) + parameters = self._compute_grid_parameters(geometry, filled, optic, focal_height, gain) + logger.info("grid: %s at %s", self._describe_fluorescence_channel(filled), positions) + return await self._get_grids( + parameters, positions, [None] * len(positions), size_y, focal_height, gain, timeout + ) + + async def measure_well_centers( + self, + fluorescence_channel: FluorescenceChannel, + *, + wells: Optional[Sequence[Well]] = None, + gain: Optional[int] = None, + focal_height: Optional[float] = None, + gain_target_percent: float = _GRID_GAIN_TARGET_PERCENT, + optic_position: Optional[OpticPosition] = None, + timeout: Optional[float] = None, + ) -> List[CLARIOstarFluorescenceGridResult]: + """Measure where the signal of each of `wells` is centred, from a grid of points around it. + + - Without `focal_height`, a z-scan of the first well (`measure_z_profile`) gives the focal + height, and the gain too unless `gain` is given; every well is then read at both. + - The wells are scanned two at a time (about 50-80 s a well). The plate's first and last well + together are the scan the reader is verified with; other wells are untested. + - Cancelling the call stops it on the reader. + + Args: + fluorescence_channel: the optics, as for `measure` (`_check_fluorescence_channel`). + wells: wells of the plate in the plate holder, any number; None for its first and last well. + gain: the detector gain, within `configuration.gain_range`; None for the z-scan's, which + needs `focal_height` None. + focal_height: mm, in 0.1 mm steps within `configuration.focal_height_range`; None for the + z-scan's peak, from the top optic only. + gain_target_percent: the z-scan's gain target when it fits the gain, above 0 up to 100. + optic_position: "top" or "bottom"; `default_optic_position` when None. + timeout: seconds each scan (and the z-scan) may take; `default_read_timeout` when None. + + Returns: + a grid for each well, in order, its centre and `center` relative to the plate. + + Raises: + RuntimeError: if no plate is in the plate holder, the z-scan fits no gain, or a scan returns + another number of grids than wells. + ValueError: if an argument is out of range, a well is not on the plate, or a reply is not + what the scan sends. + TimeoutError: if a scan takes longer than `timeout` seconds. + """ + filled = self._get_filled_fluorescence_channel(fluorescence_channel) + optic = self.default_optic_position if optic_position is None else optic_position + errors = self._check_grid(filled, optic, gain, focal_height, timeout) + if focal_height is not None and gain is None: + errors.append("give a gain with focal_height, or leave both to the z-scan") + if focal_height is None and optic != "top": + errors.append("the z-scan reads from the top: give focal_height for the bottom optic") + if not 0 < gain_target_percent <= 100: + errors.append(f"gain_target_percent must be above 0 up to 100, got {gain_target_percent}") + if wells is not None and len(wells) == 0: + errors.append("wells must hold at least one well") + if errors: + raise ValueError("; ".join(errors)) + plate, _ = self._get_plate_and_wells( + MeasurementSettings(), None, None if wells is None else list(wells) + ) + items = plate.get_all_items() + targets = [items[0], items[-1]] if wells is None else list(wells) + if focal_height is None: + z_profile = await self.measure_z_profile( + filled, + well=targets[0], + gain_target_percent=gain_target_percent if gain is None else None, + timeout=timeout, + ) + focal_height = z_profile.peak_focal_height + gain = z_profile.gain if gain is None else gain + if gain is None: + raise RuntimeError("the z-scan fitted no gain") + results: List[CLARIOstarFluorescenceGridResult] = [] + for i in range(0, len(targets), 2): + pair = targets[i : i + 2] + centres = [well.location + well.center() for well in pair if well.location is not None] + if len(centres) != len(pair): + raise ValueError(f"every well needs a location on {plate.name}") + if pair == [items[0], items[-1]]: + geometry = self._encode_plate_geometry(plate) + else: + self._warn_untested("a grid of wells other than the first and last") + geometry = self._encode_grid_geometry((plate.get_size_x(), plate.get_size_y()), centres) + parameters = self._compute_grid_parameters(geometry, filled, optic, focal_height, gain) + identifiers: List[Optional[str]] = [plate.get_child_identifier(well) for well in pair] + logger.info( + "well centres: %s at %s of %s", + self._describe_fluorescence_channel(filled), + ", ".join(str(identifier) for identifier in identifiers), + plate.name, + ) + results += await self._get_grids( + parameters, centres, identifiers, plate.get_size_y(), focal_height, gain, timeout + ) + return results + # ---------------------------------------- # Discrete reads # ---------------------------------------- diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py index d6313317015..c73fe8ac780 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py @@ -12,7 +12,7 @@ from dataclasses import replace from datetime import datetime from typing import Any, Dict, List, Optional, Sequence, Tuple, Union -from unittest.mock import patch +from unittest.mock import AsyncMock, patch from pylabrobot.bmg_labtech.clariostar.device import CLARIOstarPlus from pylabrobot.bmg_labtech.clariostar.driver.configuration_tests import decoded @@ -30,6 +30,7 @@ from pylabrobot.bmg_labtech.clariostar.driver.features.drawer_tests import make_plate_holder from pylabrobot.bmg_labtech.clariostar.driver.features.fluorescence import ( CLARIOstarFluorescenceSpectrumResult, + CLARIOstarZProfileResult, Fluorescence, _compute_band_edges, ) @@ -37,9 +38,10 @@ MeasurementSettings, ShakeSettings, ) -from pylabrobot.bmg_labtech.clariostar.driver.framing import ftdi_extract_payload +from pylabrobot.bmg_labtech.clariostar.driver.framing import ftdi_extract_payload, ftdi_wrap_payload from pylabrobot.bmg_labtech.clariostar.driver.master_tests import ( STATUS_IDLE, + STATUS_QUERY_FRAME, FakeFTDI, load_session_replies, make_driver, @@ -53,7 +55,13 @@ EmissionFilterSlide, ExcitationFilterSlide, ) -from pylabrobot.resources import Plate, Well, cor_96_wellplate_360uL_Fb, create_ordered_items_2d +from pylabrobot.resources import ( + Coordinate, + Plate, + Well, + cor_96_wellplate_360uL_Fb, + create_ordered_items_2d, +) # The default read: 485/15 to 528/20 nm, gain 1000, 8.5 mm, 10 flashes. BASELINE_RUN_FRAME = bytes.fromhex( @@ -1725,5 +1733,183 @@ async def test_spectra_filter_detection_and_z_profiles_are_not_simulated_yet(sel await call +def load_grid_ground_truth() -> Dict[str, Any]: + """The reader's scan of a 96-well plate's A1 and H12 from the top: its frame (`07`) and the + replies the driver reads, as hex: the status before it, the reply to it, PLATE_MAP_CONFIG, the + statuses and the two grids, PLATE_MAP_XY.""" + path = os.path.join(os.path.dirname(__file__), "test_data", "grid_ground_truth.json.gz") + with gzip.open(path, "rt") as f: + return dict(json.load(f)) + + +def grid_plate() -> Plate: + """The scanned plate's geometry: 127.76 x 85.48 mm, A1 centre 14.30 / 11.20 mm, H12 centre + 113.46 / 74.28 mm (from the back edge), 6.86 mm wells.""" + return Plate( + "scanned", + size_x=127.76, + size_y=85.48, + size_z=14.2, + ordered_items=create_ordered_items_2d( + Well, + num_items_x=12, + num_items_y=8, + dx=14.30 - 6.86 / 2, + dy=11.20 - 6.86 / 2, + dz=1.0, + item_dx=(113.46 - 14.30) / 11, + item_dy=(74.28 - 11.20) / 7, + size_x=6.86, + size_y=6.86, + size_z=10.67, + ), + ) + + +# The scan's optics: 488 nm excitation and 535 nm emission, split at 507.5 nm. +GRID_CHANNEL = FluorescenceChannel( + excitation_wavelength=488, + excitation_bandwidth=14, + emission_wavelength=535, + emission_bandwidth=30, + dichroic_split_wavelength=507.5, +) + + +def with_nominal_scan_edges(ground_truth: bytes) -> bytes: + """`ground_truth`'s frame with 488/14 and 535/30 nm at their nominal edges. + + The ground truth has each band narrower (13.6 nm of 14, 29.4 nm of 30); the optics start at + payload byte 21. + """ + payload = bytearray(ftdi_extract_payload(ground_truth)) + for offset, edges in ((21, _compute_band_edges(488, 14)), (27, _compute_band_edges(535, 30))): + for index, value in enumerate(edges): + payload[offset + 2 * index : offset + 2 * index + 2] = value.to_bytes(2, "big") + return ftdi_wrap_payload(bytes(payload)) + + +async def scan_ground_truth(**arguments): + """The ground truth's scan of A1 and H12 answered by its replies; the io and the result.""" + ground_truth = load_grid_ground_truth()["corner_scan_96_top"] + io = FakeFTDI(chunk=4096) + io.replies = [bytes.fromhex(reply) for reply in ground_truth["replies"]] + feature = make_fluorescence(io, grid_plate()) + feature._driver.default_poll_interval = 0 + results = await feature.measure_well_centers(GRID_CHANNEL, **arguments) + return io, ground_truth, results + + +class GridTests(unittest.IsolatedAsyncioTestCase): + async def test_a_scan_of_the_first_and_last_well_is_sent_and_decoded_as_the_ground_truth(self): + io, ground_truth, results = await scan_ground_truth(focal_height=15.0, gain=2239) + payloads = [ftdi_extract_payload(frame) for frame in io.written] + self.assertEqual( + io.written[:2], + [STATUS_QUERY_FRAME, with_nominal_scan_edges(bytes.fromhex(ground_truth["frame"]))], + ) + self.assertEqual(payloads[2], bytes.fromhex("050d0000000000")) + self.assertEqual(payloads.count(bytes.fromhex("05020000000000")), 2) + self.assertEqual(payloads[-1], bytes.fromhex("05040000000000")) + self.assertEqual([result.well for result in results], ["A1", "H12"]) + first, last = results + assert first.center is not None and last.center is not None + self.assertAlmostEqual(first.center.x, 14.45) + self.assertAlmostEqual(first.center.y, 85.48 - 11.35) + self.assertAlmostEqual(last.center.x, 113.33) + self.assertAlmostEqual(last.center.y, 85.48 - 74.00) + self.assertEqual((first.spacing, first.focal_height, first.gain), (0.33, 15.0, 2239)) + for result in results: + self.assertEqual([len(row) for row in result.counts], [31] * 31) + # Read from byte 34, every count is within the detector's 260,000. + self.assertLessEqual(max(max(row) for row in result.counts), 260_000) + grid = bytes.fromhex(ground_truth["replies"][5]) + self.assertEqual(first.counts[0][0], int.from_bytes(ftdi_extract_payload(grid)[34:38], "big")) + + async def test_given_points_make_up_a_plate_around_them(self): + feature = make_fluorescence(FakeFTDI()) + two = feature._encode_grid_geometry( + (127.76, 85.48), [Coordinate(20, 30, 0), Coordinate(50, 60, 0)] + ) + self.assertEqual(two.hex(), "31e8216407d015ac138809f40201") + one = feature._encode_grid_geometry((127.76, 85.48), [Coordinate(20, 30, 0)]) + self.assertEqual(one[4:12], two[4:8] * 2) + self.assertEqual(one[12:], bytes([1, 1])) + + async def test_given_points_are_checked_and_warned_untested(self): + feature = make_fluorescence(FakeFTDI()) + for positions in ([], [Coordinate(1, 1, 0)] * 3, [Coordinate(200, 10, 0)]): + with self.assertRaises(ValueError): + await feature.measure_grid(GRID_CHANNEL, positions=positions, gain=2239, focal_height=15.0) + with ( + patch.object(feature, "_get_grids", AsyncMock(return_value=[])), + self.assertLogs(measurement_module.logger, "WARNING") as logs, + ): + await feature.measure_grid( + GRID_CHANNEL, positions=[Coordinate(20, 30, 0)], gain=2239, focal_height=15.0 + ) + self.assertIn("has not been tested on hardware", logs.output[0]) + + async def test_a_z_scan_runs_only_without_a_focal_height(self): + plate = grid_plate() + feature = make_fluorescence(FakeFTDI(), plate) + z_profile = CLARIOstarZProfileResult( + well="A1", + gain=2239, + z_profile=[], + average_signal=0, + peak_counts=None, + peak_detector_percent=None, + peak_focal_height=14.9, + temperature=None, + start_timestamp=datetime.now(), + end_timestamp=datetime.now(), + ) + with ( + patch.object(feature, "measure_z_profile", AsyncMock(return_value=z_profile)) as z_scan, + patch.object(feature, "_get_grids", AsyncMock(return_value=[])) as scan, + ): + await feature.measure_well_centers(GRID_CHANNEL) + self.assertIs(z_scan.call_args.kwargs["well"], plate.get_well("A1")) + self.assertEqual(z_scan.call_args.kwargs["gain_target_percent"], 50.0) + self.assertEqual(scan.call_args.args[4:6], (14.9, 2239)) + await feature.measure_well_centers(GRID_CHANNEL, gain=1000) + self.assertIsNone(z_scan.call_args.kwargs["gain_target_percent"]) + self.assertEqual(scan.call_args.args[4:6], (14.9, 1000)) + await feature.measure_well_centers(GRID_CHANNEL, focal_height=10.0, gain=1000) + self.assertEqual(z_scan.call_count, 2) + self.assertEqual(scan.call_args.args[4:6], (10.0, 1000)) + + async def test_wells_are_scanned_two_at_a_time(self): + plate = grid_plate() + feature = make_fluorescence(FakeFTDI(), plate) + wells = [plate.get_well(name) for name in ("B2", "G11", "D6")] + with ( + patch.object(feature, "_get_grids", AsyncMock(return_value=[])) as scan, + self.assertLogs(measurement_module.logger, "WARNING"), + ): + await feature.measure_well_centers(GRID_CHANNEL, wells=wells, focal_height=15.0, gain=2239) + self.assertEqual([call.args[2] for call in scan.call_args_list], [["B2", "G11"], ["D6"]]) + first_pair, last = (call.args[0] for call in scan.call_args_list) + self.assertEqual((first_pair[13], first_pair[14]), (2, 1)) + self.assertEqual((last[13], last[14]), (1, 1)) + + async def test_invalid_arguments_send_nothing(self): + io = FakeFTDI() + feature = make_fluorescence(io, grid_plate()) + cases: List[Dict[str, Any]] = [ + dict(focal_height=15.0), # no gain + dict(optic_position="bottom"), # the z-scan reads from the top + dict(gain=5000, focal_height=15.0), + dict(gain=2239, focal_height=30.0), + dict(gain_target_percent=0), + dict(wells=[]), + ] + for arguments in cases: + with self.subTest(arguments=arguments), self.assertRaises(ValueError): + await feature.measure_well_centers(GRID_CHANNEL, **arguments) + self.assertEqual(io.written, []) + + if __name__ == "__main__": unittest.main() diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/grid_ground_truth.json.gz b/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/grid_ground_truth.json.gz new file mode 100644 index 0000000000000000000000000000000000000000..da3445742d261f312dc856f957f77f9f3f3c17de GIT binary patch literal 6191 zcmWMqYdjN*1Fc)qy2x*1?YGli0L51c*+Z3s@hz{pF6e8y}y#^4{^T z){g0vGvFuKckqAfR<4dGqte^P$HvCQM93mUtzk%%NNha=faTASw3>bbR>cQs6k%Gh z=H>_-5)x&X6yM1Ed!<4L{`4EA?qWzCO4NT)`K4Y;8)f9D)Gas_F~2CE6`=_NcueBY z`k}%4j~2&No| z3~}fDCzO0==~C6QDZE>q)F7umQ<;7{#;mf%77F7aQm+ma6bfa6xboy$RE+rHC?2zD z&<{1zPQ(K^$M45f#>YDo{*%7+1n-f4n0-meyq3RpF-?qWkbw%h4bI24oAZ>)E^b~l z5d&uQ68K@(>Uz7uSpv6xxZA%igket$uN6$Y|swi{cu(mqY+wJ~Y_6yV5Gq1x*{D3hS zmcdoLkdEO0Epi!5fF>3)vxLIO-IB>8AKZc#QYsT`ypv|C#PA;5#R$MkaUyvBsWuCp^p_^-!Mxt!TixJx4+D&@>>|)vkgENk9s%x zeA%|Paa%g`x>;+wRsItF`qxw_Ao-|xT4QC)IT82))vTp>4U?GPOXCaK-Oi(8rjYf0 zvuRfsN`ofEK7rJ2E;O=9bzEn9?0>@(9h(Y{f^)90h2}5fnjOpj95R7y7Fy5YHrk`}vuQ8hr1#clOaH5e=9^jYp)ciloV!?nwzn;Ub5?9NEl z!z^^Dgu8syo3qvF&y}QIJzaJNVm!U=PsLR>p8d0O3_*_88^lh*saBPR1OT>ynUXpk0=np2t;M zxJ;m&na9EtI5}*uIKHvCK0Ry?yPM4Yhnt|U5|jrZKv`%{lu&dX*4x6$Wri0TnRYN( z%aj*L-go;iuM_`_4BnKLEwr7e!WW`{Pe&ipj9{c6(cz%D%VymB79R4jaBX}W+`sa} z)JhBA{{=ubVMbfPMXhl$(n&crIh=dvA_&vBoUUBDAzS)@W6-bN-{?IyY_fKobD;|%)WB~Jt|GJeP6RvL#80-)8@`N4=uh<} zLY-I3P~&LPS0-qf_OwQ{Mzh^moRKMvzXHW29EffebS&k^37qJg2}e=#Dvco8RfJaS?$}9H^OhojFQ!@XDmV)BRBa=RFxEvpWTX$BoINp2nhKl2>3T^hysO(tA{?U6U5n6eN_i)A~-8y3p%x#XEJ8V2-bHnXKw zCK)>^0%=l1gooHf9Ty0uW{04(*d=B_9 zmC$4!UkRUu?2~78$;VAoDI-LnSCz%+hIYi+sNR#2Vp8BI$7 zub_QRxvuf^$LYFTiY6Fy^e=vFaiH)(sZih${(X0Ms{Pd*Vqt~(~bV~hI&*d7| zYX%cm@`jOA&iBp_%ma%-KMSbsj+}t^x}{h}vjuRYNy_JxsvG(A(k2D5%E(8;3Dd!L z-BFhZSuXpCx3p1a+Y$N0MDWpiZO~Sw;giI!96Ea%q632%uWtWJu+eMmNcIkSsJQ{3 zjXgMHyN*Q|pVSFA7rQ|8qzyd=7kpjI9$IgNze>1d>Z{BIAx5MgrO_e=f+b_PYd8~1 z2l&cn?3*FOvuh0O#u?+Ls6B;_f|5+1=3Nn{BE}!X%;8_`IgsYoWg}vuaMCx2c}6CG zwQRY-r0bQ^Muh4` zk^_ATMa;wMHwmjCfMTEV;gaRgp2QX*bhloiEa1Fp8Hf8N9hfBd2}t>B<`Qa}^EDel z__X?xSm3msn1sc}13XJzZDBGgd0iZ|<>J^j*-OE0*wv-Hlr}1a|7X#1&uz*Fuf%tJ z-)J)kICMH4nW49L0tj7tLe?U{jx{k&S>~jDOPestqxh%2DBeGvh~rGW-55@I=`V1! z?K{~uXi%$pyN;bNtt1> z!GY!PjK3%Mi#}#)-HUX7<(5yTT9orbceQ%)PURQ9N)Oaj>be}vo2gImX8$|tC5|MC`k0U&C=bs4WbwBG_8dcEE?}=qcL|h+J@t8ALC3!^%(7 zucgJwGa)@Dds2FCZvjn7lBn$GDKXFK4F>HCC64kx86!n`}U%@ zu?#r=X5qIX4VYv>$6NPztTkFYgOneGr&mvdCxA=Sds!s9R(QDJ{VDDR7`pE{4se*5 zV;1ykx&zhp$QTb`qy#<)x z;3p;Ab)QMaSp0B4uZ~IU{+Aev`!<&bq(2syxLt*GL6DUY29DhPjJ|8b+Nt_#T52$U8J6NP+8W zm*PjW=7l_#2PhA=U8xDo7Hwfg(5ssa^r(iFNh>|xV5oG*nwG*>@&1Lz&dMW{tIcj( z&5DB~m8L=*OB0lBr}a9kP}F9rYgX0ui^Q9^p>3?EMZ?ugN!5L@Lp+;Oic(gnTt919ZFQA7dzJrHnv$xNV`YlOgGb}DLiv}S;G0;tQEt1xYz&0zuxxDPx}ZK zTX#^sLwPt$9PFI-NtaLspI@^6rq{Zi(AexZfcSfSuBtLBWP;bQjZj70!pILKoq*hV zW8jpW%sG9wvvj|zq(vUFYr+*wj?u{PgiX z_}(X({2#f-O2W>B$ax!n2!Br@a_PMdOwaC2!pG(v7}r3(BA4ZF-k5tBqu}3|HXl8t z;%TOb^&utI&+`$z zIo7lRl6dm4sB7+%XQvkqIkpVoKz-3Y72EX7;B~ide-+IO8n1EZ_H}HHQRWmkHes=Y zW^^e>_CaWv6T;)qA;<4X?i%7UUE9fzlD>XS;tuqHR!gXlGr6PAwhQ|;gX!$!mrJi# zfuoD~C0jJP&g3`mnfDyFDlz5YE^H90|l)G-NGGTJFS|^|h|A|lwRv?Z5FPaS`&uC3s9|LMMksEzVc|@Kxn#YJWqT z<&_8~Yr*Khb2RNhiQTu1vRy5nZpOUa@`($NE2lFRs-%eWnpEU5YOIC)zios>jd}GJ zpYgt)8~*^3l;#73+bnA8O7&!kS89CpU2OFEp{@g+q34kJLh#v4&z7=0Of`895wy_d z)0}C&3u}Yjzjy=k@Wl=;QqUYLu}zpSxkhVnSatG!UQm&>7ZlTqOG$~|N5ynCT$nfp zL=^VwO+r#E%C$#pPpdr=#t{V^C0s!IUBRjiqv$%%5aaBib_fub6rh{e*H)`=01lL` zmNrlq!nP3jnA)iPc-{Gf&i8G|{_xuYP?$Jc2P62Z1(+JJXc!OLeG@-bR&?%C31kW} z?ofuWVoT#CAIn`GJ(osF5R0UdYhOEu1|ZcX#Bx|aa)9Kgt#6ZNwtv(-W%L<3Vs6T* zvXlOx=U@ortB{;ZOR@yTh3d9To8ZfhpS+0duEqpG32TGS ze8iN)qd&3#p%K9TQ44?LYyZGn>6Mp@%LuXWh5tCsm*j4n=K~~v%sZSWAEu(iXZ+zg z3IpnDMT&s5@}0D})XS_+#|8KkZaibfrRR$|8v9ql1oUcS(3Jx?o#U*ypA zI8}}coF_8n>KY;#IpY(YR1=furG7Ez;{%I24Y7diXMr^8SlGf_c>ONRosw1fV9AD< z?jz%4J1s``%9frm{_4WsC^#Z9n_4>WD)|`ZD3QMRaZ{)HVzv`26%qa+2Gr!40Uqsl zjVxF;U3y^vL4P} zg-0kpwEQc}TiB4Kiw4pLg~0Y~eH^-?I+>HSB)gO~Z5FLB2}}L8qcPVgxE{7#OuZMs zjPHG?u*p_@{(<;10NkH~MbKF$er|cd#19G>LOM~|D4Ybs#x?VhUQ1R zQ1CgYT&m9-(zx*UzllgVSvdXX&i%@B+eL2)TR~|l%<4o%1Vu6TpZ+nRsv~f3oYOPJ z`nFOMyY4Nw{$>Y!01@LUsi?r$coAtmZt)oe`AF;NC4ge25F9G9@m{{AsEUA;F-KCG zk0!ACk*|8L_f!Rm%0~-Q4!?8PE_7x#%$@voX>I8$_cS3V3;+gj2b2A%!dgxaf&lI4 zvzD;Vo#h3WLKH|yWTQUup$Qdt5{SRLOf@POV?Fkfrc2Fa9sR_?H9{$A^bUqiU; ziRZfqaOhSL4H-io=-?NWdjC~68kBzm8AXa3sW2zB`P>X(pE2-hq=(@Y7Ip)U1El^o z20xfRpnk_Yw*!-RXw^tJcGk34eWS#CxaO2QV7!99?pqE-}`xUWn{#} zzdiP*&$>MzPuq@37@6?85~)sx43?yf-HqF&fE6PbtK4{XY2c^yWXJQs)9rC#rO7)H z_Zy4FcdUKfo^VQd&-?H@Zu|}yASQ%L!M=LkW}1-6>p8@8j1TB#^W8#V8<{Eo9?^8? zOHX?6JG`wI{(-kVg2A*g+t0+8o8+d$8s>^_&QExn&TU`%&5nHyxT>1 z=1l1X_6Zg~TqGxP)T3I9@k8(7F!?Jq?P7c?`~2%;fnla^F7hHB)sZ>v7uW%8FC8t% z&=ZdN>G7Lz^jyHJmy#RRrHcu~BT`c3U2k^JkPYYx1Sn68_*L<(z{e@lzKaZ>B<=Sy zKx~U;OHLtCaf*I?fA zbP^$56PF3(W!CpZy4`Lfot{W*V1=% zYsb2m#f=oQb<%uZu#htNmbzJS>aD=8$8)uKbQ)(gPC6qfKcY45+ckPmi=gE0&SP;u zs3p2>*E#oX*K=nMQgPHB)A~=$#>zxl|o&!cW6#ZW7Z+I0SyQ{)34e^Fn~U zmBvO6?ATv-eLcWcF!8XYx9`pa<{fhDpcOf3Q&JWVa6kce`{$DKogzFVqtQ>be(X$H z1+`4NzBsq^O-f|hwztkfrVYYtOS3y}ViKoD_YryDv~#x6Zm+f4WApvTCg+}xlvKC= hXC39czt+MPt5UKxtB^Ho)@Zfs)_6Y*?eqTy{twNBg$Muu literal 0 HcmV?d00001 diff --git a/pylabrobot/bmg_labtech/clariostar/driver/master.py b/pylabrobot/bmg_labtech/clariostar/driver/master.py index 3da6760b1fc..ffe5de56236 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/master.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/master.py @@ -125,6 +125,7 @@ class CommandFamily: RUN = 0x04 REQUEST = 0x05 TEMPERATURE_CONTROLLER = 0x06 + PLATE_MAP_SCAN = 0x07 FOCUS_WELL = 0x09 STOP = 0x0B SET_PARAMETER = 0x0E @@ -141,9 +142,11 @@ class Command: DRAWER_OPEN = 0x01 # REQUEST DATA = 0x02 + PLATE_MAP_XY = 0x04 FOCUS_RESULT = 0x05 PARAMETERS = 0x07 FIRMWARE_INFO = 0x09 + PLATE_MAP_CONFIG = 0x0D SPECTRAL_DATA = 0x11 FILTER_RESULT = 0x1B # SET_PARAMETER; seen only writing the two-byte test-run counter @@ -157,6 +160,7 @@ class Command: CommandFamily.STOP, CommandFamily.STATUS, CommandFamily.TEMPERATURE_CONTROLLER, + CommandFamily.PLATE_MAP_SCAN, CommandFamily.FOCUS_WELL, CommandFamily.FILTER_SCAN, }