diff --git a/pylabrobot/bmg_labtech/clariostar/__init__.py b/pylabrobot/bmg_labtech/clariostar/__init__.py index 903c76d6d91..7a4b14aeda5 100644 --- a/pylabrobot/bmg_labtech/clariostar/__init__.py +++ b/pylabrobot/bmg_labtech/clariostar/__init__.py @@ -13,6 +13,7 @@ CLARIOstarFluorescenceSpectrumResult, CLARIOstarSimulationDriver, CLARIOstarStatus, + CLARIOstarZProfileResult, CommandRefusedError, DeviceConfiguration, Drawer, diff --git a/pylabrobot/bmg_labtech/clariostar/driver/__init__.py b/pylabrobot/bmg_labtech/clariostar/driver/__init__.py index af46dab35b2..d366a713c83 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/__init__.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/__init__.py @@ -8,6 +8,7 @@ CLARIOstarAbsorbanceSpectrumResult, CLARIOstarFluorescenceResult, CLARIOstarFluorescenceSpectrumResult, + CLARIOstarZProfileResult, Drawer, Fluorescence, FluorescenceConfiguration, diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/__init__.py b/pylabrobot/bmg_labtech/clariostar/driver/features/__init__.py index a1017936edf..6bd7b6b61a4 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/__init__.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/__init__.py @@ -8,6 +8,7 @@ from .fluorescence import ( CLARIOstarFluorescenceResult, CLARIOstarFluorescenceSpectrumResult, + CLARIOstarZProfileResult, Fluorescence, FluorescenceConfiguration, ) diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py index 7af2c42cb13..511937787d0 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py @@ -1,12 +1,16 @@ -"""CLARIOstar fluorescence: plate reads and spectra. +"""CLARIOstar fluorescence: plate reads, spectra, filter detection and z-profiles. A plate read goes through one to five fluorescence channels in one pass (`FluorescenceChannel`, the vendor's chromatics), from the top or the bottom optic. Excitation and emission each select their wavelength through a monochromator or a filter. A spectrum scans the excitation or the emission monochromator. +Filter detection scans every filter position; a z-profile reads one well over its focal heights, +and with a target the reader fits the gain at its peak. """ +import asyncio import logging +import time from dataclasses import dataclass, replace from datetime import datetime from typing import ( @@ -24,6 +28,13 @@ from pylabrobot.lib.fluorescence.channels import FluorescenceChannel from pylabrobot.lib.fluorescence.results import FluorescenceResult, FluorescenceSpectrumResult from pylabrobot.lib.optics.filters import DichroicFilter, OpticalFilter +from pylabrobot.lib.optics.holders import ( + DichroicFilterSlide, + EmissionFilterSlide, + ExcitationFilterSlide, + F, + FilterSlide, +) from pylabrobot.resources import Plate, Well from ..errors import FrameError @@ -66,6 +77,39 @@ class CLARIOstarFluorescenceResult(FluorescenceResult): matrix_points: Optional[Dict[str, List[float]]] = None +@dataclass(frozen=True) +class CLARIOstarZProfileResult: + """The z-scan of one well as the firmware reports it. + + Attributes: + well: the identifier of the scanned well (``"D6"``). + gain: the detector gain the firmware fits at the peak for the target; None without a target. + z_profile: (focal height in mm, counts) from the highest to the lowest point; the counts at a + gain the firmware chooses per well: for the shape, not comparable across wells. + average_signal: the firmware's average of the z-profile's five points around + `peak_focal_height`, counts. + peak_counts: the counts at `peak_focal_height` and `gain`; None without a target. + peak_detector_percent: `peak_counts` in percent of the detector's 260,000 counts, as the + target; None without a target. + peak_focal_height: the focal height the firmware computes as the z-profile's maximum, in mm. + temperature: temperature in degrees C; None: no z-scan with the temperature controller on is + captured, so the reply's temperature, if it holds one, is not decoded. + start_timestamp: when the z-scan started. + end_timestamp: when its result was in. + """ + + well: str + gain: Optional[int] + z_profile: List[Tuple[float, int]] + average_signal: int + peak_counts: Optional[int] + peak_detector_percent: Optional[float] + peak_focal_height: float + temperature: Optional[float] + start_timestamp: datetime + end_timestamp: datetime + + @dataclass class CLARIOstarFluorescenceSpectrumResult(FluorescenceSpectrumResult): """A CLARIOstar fluorescence spectrum, with the detector gain it was read with. @@ -194,10 +238,14 @@ def __init__( self._default_excitation_bandwidth = 15 # nm, the vendor's default self._default_emission_bandwidth = 20 # nm, the vendor's default self._default_optic_position: OpticPosition = "top" # the vendor's default + # The filters each slide holds, by slot: `detect_all_filters`, or assigned; None until then. + self.excitation_filter_slide: Optional[ExcitationFilterSlide] = None + self.dichroic_filter_slide: Optional[DichroicFilterSlide] = None + self.emission_filter_slide: Optional[EmissionFilterSlide] = None self._default_focal_height_percent_of_well_depth = 25.0 # ---------------------------------------- - # Defaults + # Defaults and filters from the slides # ---------------------------------------- def _check_default_bandwidth(self, name: str, value: int) -> None: @@ -272,21 +320,84 @@ def _get_default_focal_height(self, wells: Sequence[Well]) -> float: wells, self._default_focal_height_percent_of_well_depth, self.configuration.focal_height_range ) - def _get_fluorescence_channel_with_default_bandwidths( + @staticmethod + def _get_resolved_filter(value: Union[F, int, str, None]) -> Optional[F]: + """A fluorescence channel's filter, once `_get_filled_fluorescence_channel` has resolved it. + + Raises: + TypeError: if `value` is still a slot or a name. + """ + if isinstance(value, (int, str)): + raise TypeError(f"a filter must be resolved through its filter slide, got {value!r}") + return value + + @staticmethod + def _get_filter( + filter_slide: Optional[FilterSlide[F]], value: Union[F, int, str], slide: str + ) -> F: + """The filter `value` names: itself, the slide's filter at a slot, or its filter of a name. + + An empty slot within the slide is an anonymous filter (`FilterSlide.__getitem__`). + + Args: + slide: "excitation", "dichroic" or "emission", the prefix of the field named in the errors. + + Raises: + ValueError: if a slot or a name is given without a slide, a slot is outside the slide, or + no filter of the slide has the name. + """ + if not isinstance(value, (int, str)): + return value + if filter_slide is None: + raise ValueError( + f"{slide}_filter as a slot or a name needs a {slide} filter slide (run detect_all_filters " + f"or assign {slide}_filter_slide), got {value!r}" + ) + if isinstance(value, str): + try: + return filter_slide.by_name(value) + except KeyError as error: + raise ValueError(error.args[0]) from None + if not 1 <= value <= filter_slide.maximum_slots: + raise ValueError( + f"{slide}_filter slot must be 1 to {filter_slide.maximum_slots}, got {value}" + ) + return filter_slide[value] + + def _get_filled_fluorescence_channel( self, fluorescence_channel: FluorescenceChannel ) -> FluorescenceChannel: - """A copy of `fluorescence_channel`, excitation and emission bandwidths given the default. + """A copy of `fluorescence_channel` with its filters from the slides and its bandwidths filled. - Only excitation or emission without a filter or a bandwidth gets the default. + - A filter given as a slot or a name is its slide's (`_get_filter`). + - Excitation or emission without a filter or a bandwidth has the default bandwidth. + + Raises: + ValueError: if a slot or a name has no slide, a slot is outside its slide, or no filter of + the slide has the name. """ + excitation_filter = fluorescence_channel.excitation_filter + if excitation_filter is not None: + excitation_filter = self._get_filter( + self.excitation_filter_slide, excitation_filter, "excitation" + ) + dichroic_filter = fluorescence_channel.dichroic_filter + if dichroic_filter is not None: + dichroic_filter = self._get_filter(self.dichroic_filter_slide, dichroic_filter, "dichroic") + emission_filter = fluorescence_channel.emission_filter + if emission_filter is not None: + emission_filter = self._get_filter(self.emission_filter_slide, emission_filter, "emission") excitation_bandwidth = fluorescence_channel.excitation_bandwidth - if excitation_bandwidth is None and fluorescence_channel.excitation_filter is None: + if excitation_bandwidth is None and excitation_filter is None: excitation_bandwidth = self._default_excitation_bandwidth emission_bandwidth = fluorescence_channel.emission_bandwidth - if emission_bandwidth is None and fluorescence_channel.emission_filter is None: + if emission_bandwidth is None and emission_filter is None: emission_bandwidth = self._default_emission_bandwidth return replace( fluorescence_channel, + excitation_filter=excitation_filter, + dichroic_filter=dichroic_filter, + emission_filter=emission_filter, excitation_bandwidth=excitation_bandwidth, emission_bandwidth=emission_bandwidth, ) @@ -398,7 +509,8 @@ def _check_fluorescence_channel( a wavelength and a bandwidth. Args: - fluorescence_channel: the fluorescence channel, its bandwidths filled. + fluorescence_channel: the fluorescence channel, its filters resolved and its bandwidths + filled. scanned: a spectrum's scanned excitation or emission, which has no wavelength. Returns: @@ -406,20 +518,21 @@ def _check_fluorescence_channel( refuses, an excitation or emission filter without a `dichroic_filter`, a split not in 0.1 nm steps, a dichroic slot outside `dichroic_filter_slots`, `_check_bands`'s. """ + dichroic_filter = self._get_resolved_filter(fluorescence_channel.dichroic_filter) errors = self._check_excitation_or_emission_on_reader( "excitation", fluorescence_channel.excitation_wavelength, fluorescence_channel.excitation_bandwidth, - fluorescence_channel.excitation_filter, + self._get_resolved_filter(fluorescence_channel.excitation_filter), scanned=scanned == "excitation", ) + self._check_excitation_or_emission_on_reader( "emission", fluorescence_channel.emission_wavelength, fluorescence_channel.emission_bandwidth, - fluorescence_channel.emission_filter, + self._get_resolved_filter(fluorescence_channel.emission_filter), scanned=scanned == "emission", ) - if fluorescence_channel.dichroic_filter is None and ( + if dichroic_filter is None and ( fluorescence_channel.excitation_filter is not None or fluorescence_channel.emission_filter is not None ): @@ -430,9 +543,9 @@ def _check_fluorescence_channel( not isinstance(split, (int, float)) or abs(split * 10 - round(split * 10)) > 1e-9 ): errors.append(f"dichroic_split_wavelength must be nm in 0.1 nm steps, got {split!r}") - if fluorescence_channel.dichroic_filter is not None: + if dichroic_filter is not None: slots = self.configuration.dichroic_filter_slots - slot = fluorescence_channel.dichroic_filter.slot + slot = dichroic_filter.slot if not isinstance(slot, int) or not 1 <= slot <= slots: errors.append(f"dichroic_filter slot must be 1 to {slots}, got {slot!r}") if ( @@ -448,7 +561,7 @@ def _check_fluorescence_channel( fluorescence_channel.emission_wavelength, fluorescence_channel.emission_bandwidth, split=split, - dichroic_filter=fluorescence_channel.dichroic_filter, + dichroic_filter=dichroic_filter, ) return errors @@ -507,27 +620,28 @@ def _compute_optics( ValueError: if excitation or emission has neither a filter nor a wavelength and bandwidth. """ per_slide = self.configuration.filter_slots_per_slide + excitation_filter = self._get_resolved_filter(fluorescence_channel.excitation_filter) + emission_filter = self._get_resolved_filter(fluorescence_channel.emission_filter) + dichroic_filter = self._get_resolved_filter(fluorescence_channel.dichroic_filter) excitation_wavelength = fluorescence_channel.excitation_wavelength excitation_bandwidth = fluorescence_channel.excitation_bandwidth emission_wavelength = fluorescence_channel.emission_wavelength emission_bandwidth = fluorescence_channel.emission_bandwidth - if fluorescence_channel.excitation_filter is not None: - excitation = _compute_filter_positions(fluorescence_channel.excitation_filter.slot, per_slide) + if excitation_filter is not None: + excitation = _compute_filter_positions(excitation_filter.slot, per_slide) elif excitation_wavelength is not None and excitation_bandwidth is not None: excitation = _compute_band_edges(excitation_wavelength, excitation_bandwidth) else: raise ValueError("give excitation_wavelength and excitation_bandwidth, or excitation_filter") - if fluorescence_channel.emission_filter is not None: - first, second = _compute_filter_positions( - fluorescence_channel.emission_filter.slot, per_slide - ) + if emission_filter is not None: + first, second = _compute_filter_positions(emission_filter.slot, per_slide) emission = (second, first) elif emission_wavelength is not None and emission_bandwidth is not None: emission = _compute_band_edges(emission_wavelength, emission_bandwidth) else: raise ValueError("give emission_wavelength and emission_bandwidth, or emission_filter") - if fluorescence_channel.dichroic_filter is not None: - dichroic = fluorescence_channel.dichroic_filter.slot + 1 + if dichroic_filter is not None: + dichroic = dichroic_filter.slot + 1 elif fluorescence_channel.dichroic_split_wavelength is not None: dichroic = round(fluorescence_channel.dichroic_split_wavelength * 10) else: @@ -631,11 +745,11 @@ def _compute_run_parameters( # -- the log and the results --------------------------------------------------------------------- def _describe_fluorescence_channel( - self, fluorescence_channel: FluorescenceChannel, gain: int + self, fluorescence_channel: FluorescenceChannel, gain: Optional[int] = None ) -> str: - """One fluorescence channel of the read, for the log: excitation, emission and gain. + """One fluorescence channel of the read, for the log: excitation, emission and the gain. - Excitation and emission each as wavelength and bandwidth, or filter. + Excitation and emission each as wavelength and bandwidth, or filter; the gain only when given. """ def describe( @@ -652,14 +766,14 @@ def describe( excitation = describe( fluorescence_channel.excitation_wavelength, fluorescence_channel.excitation_bandwidth, - fluorescence_channel.excitation_filter, + self._get_resolved_filter(fluorescence_channel.excitation_filter), ) emission = describe( fluorescence_channel.emission_wavelength, fluorescence_channel.emission_bandwidth, - fluorescence_channel.emission_filter, + self._get_resolved_filter(fluorescence_channel.emission_filter), ) - return f"{excitation} -> {emission}, gain {gain}" + return f"{excitation} -> {emission}" + ("" if gain is None else f", gain {gain}") def _get_result_optics( self, fluorescence_channel: FluorescenceChannel @@ -669,8 +783,8 @@ def _get_result_optics( Excitation's and emission's filter passband, or monochromator wavelength and the bandwidth the read used. """ - excitation_filter = fluorescence_channel.excitation_filter - emission_filter = fluorescence_channel.emission_filter + excitation_filter = self._get_resolved_filter(fluorescence_channel.excitation_filter) + emission_filter = self._get_resolved_filter(fluorescence_channel.emission_filter) optics: Dict[str, Optional[float]] = {} if excitation_filter is not None: optics["excitation_wavelength"] = excitation_filter.center_wavelength @@ -686,6 +800,476 @@ def _get_result_optics( optics["emission_bandwidth"] = fluorescence_channel.emission_bandwidth return optics + # ---------------------------------------- + # Filter detection + # ---------------------------------------- + + # Filter detection `24`, mode byte: the slide's code OR bit 5, which the vendor always sets. + _FILTER_SCAN_SLIDES: Dict[str, int] = {"excitation": 0x00, "emission": 0x01, "dichroic": 0x03} + _FILTER_SCAN_BIT_5 = 0x20 + # The emission detector gains an emission scan sends before the slide positions, as the vendor + # sends them; excitation and dichroic scans send zeros. + _EMISSION_SCAN_GAINS = (1100, 1200, 1100, 1150) + # Seconds one position may take: excitation and dichroic 6.4 s at most, emission 48.6 s. + _FILTER_SCAN_TIMEOUT = 20.0 + _EMISSION_FILTER_SCAN_TIMEOUT = 100.0 + # FILTER_RESULT byte 6: the slide's code OR 0x80, which brings the slopes and the curve's range. + _FILTER_RESULT_LAYOUT = 0x80 + + # -- `24` parameters ----------------------------------------------------------------------------- + + def _compute_filter_scan_parameters(self, slide: str, slot: int) -> bytes: + """`24`'s parameters for one filter position: the mode, then 13 words u16 BE. + + - The first four: `_EMISSION_SCAN_GAINS` for emission, zeros otherwise. + - Then the slide positions: two slits, EX_SP, EX_LP, BSP, EM_SP, EM_LP, EM_SLIT, EX_SLIT + (the reader's slides), 1 their home. + - A filter's positions are the RUN's (`_compute_filter_positions`): excitation (EX_SP, EX_LP), + dichroic BSP = slot + 1, emission (EM_LP, EM_SP). + """ + per_slide = self.configuration.filter_slots_per_slide + excitation, dichroic, emission = (1, 1), 1, (1, 1) + if slide == "excitation": + excitation = _compute_filter_positions(slot, per_slide) + elif slide == "dichroic": + dichroic = slot + 1 + else: + long_pass, short_pass = _compute_filter_positions(slot, per_slide) + emission = (short_pass, long_pass) + gains = self._EMISSION_SCAN_GAINS if slide == "emission" else (0, 0, 0, 0) + words = (*gains, 1, 1, *excitation, dichroic, *emission, 1, 1) + mode = self._FILTER_SCAN_SLIDES[slide] | self._FILTER_SCAN_BIT_5 + return bytes([mode]) + b"".join(word.to_bytes(2, "big") for word in words) + + # -- decoding ------------------------------------------------------------------------------------ + + def _decode_filter_result(self, reply: bytes, slide: str) -> Tuple[int, int, int, int, int]: + """A FILTER_RESULT (`05 1b`) reply's bandwidth, two edges and two slopes. + + - The edges are the band's thresholds; bandwidth and edges in 0.1 nm. + - Bytes 7-14: peak, bandwidth, edge 1, edge 2 (u16 BE). + - Bytes 15-18: the slopes (s16 BE); 19-22: the curve's range. + - Bytes 23-24: the point count n, then n points of 8 bytes and 8 bytes `00`. + + Raises: + ValueError: if the reply is not a whole FILTER_RESULT of `slide`. + """ + layout = self._FILTER_SCAN_SLIDES[slide] | self._FILTER_RESULT_LAYOUT + if len(reply) < 25 or reply[6] != layout: + raise ValueError(f"not a filter result of the {slide} slide, got {reply[:7].hex()}") + points = int.from_bytes(reply[23:25], "big") + if len(reply) < 25 + 8 * points + 8: + raise ValueError(f"filter result cut short: {points} points, got {len(reply)} bytes") + bandwidth, first_edge, second_edge = ( + int.from_bytes(reply[offset : offset + 2], "big") for offset in (9, 11, 13) + ) + first_slope, second_slope = ( + int.from_bytes(reply[offset : offset + 2], "big", signed=True) for offset in (15, 17) + ) + return bandwidth, first_edge, second_edge, first_slope, second_slope + + @staticmethod + def _compute_optical_filter( + slot: int, result: Tuple[int, int, int, int, int] + ) -> Optional[OpticalFilter]: + """The excitation or emission filter a FILTER_RESULT describes; None for an empty position. + + As the vendor decides: + - a bandwidth: a band-pass between the edges; + - else a first edge with a slope: a long-pass, open above; + - else a second edge with a slope: a short-pass, its band not stated (not captured). + """ + bandwidth, first_edge, second_edge, first_slope, second_slope = result + if bandwidth > 0: + return OpticalFilter(slot, passbands=((first_edge / 10, second_edge / 10),)) + if first_edge > 0 and first_slope != 0: + return OpticalFilter(slot, passbands=((first_edge / 10, None),)) + if second_edge > 0 and second_slope != 0: + # TODO(device): no short-pass filter is captured; state its band once a reader reports one. + logger.warning( + "filter %d is a short-pass up to %s nm; its band is not stated", slot, second_edge / 10 + ) + return OpticalFilter(slot) + return None + + @staticmethod + def _compute_dichroic_filter( + slot: int, result: Tuple[int, int, int, int, int] + ) -> Optional[DichroicFilter]: + """The dichroic a FILTER_RESULT describes; None for an empty position. + + As the vendor decides: + - a first edge: a long-pass that transmits from it, open above; + - else a second edge: a short-pass, its bands not stated (not captured). + """ + _, first_edge, second_edge, _, _ = result + if first_edge > 0: + return DichroicFilter(slot, transmission_bands=((first_edge / 10, None),)) + if second_edge > 0: + # TODO(device): no short-pass dichroic is captured; state its bands once a reader reports one. + logger.warning( + "dichroic %d is a short-pass at %s nm; its bands are not stated", slot, second_edge / 10 + ) + return DichroicFilter(slot) + return None + + # -- detecting ----------------------------------------------------------------------------------- + + async def _detect_filter(self, slide: str, slot: int) -> bytes: + """Scan one filter position (`24`) and return its FILTER_RESULT reply. + + As the vendor: status until the data is ready (the reader may still be busy), SPECTRAL_DATA + (`05 11`, not decoded) then FILTER_RESULT (`05 1b`), then status until idle. Any failure, a + timeout or a cancelled call stops the scan (`0b 00`) first. + + Args: + slide: "excitation", "dichroic" or "emission". + slot: the filter position on that slide, from 1. + + Raises: + TimeoutError: if the position takes longer than its timeout (`_FILTER_SCAN_TIMEOUT`, + `_EMISSION_FILTER_SCAN_TIMEOUT` for emission). + RuntimeError: if the scan ends without reporting its data ready. + """ + timeout = ( + self._EMISSION_FILTER_SCAN_TIMEOUT if slide == "emission" else self._FILTER_SCAN_TIMEOUT + ) + parameters = self._compute_filter_scan_parameters(slide, slot) + driver = self._driver + await driver.send_command(driver.CommandFamily.FILTER_SCAN, parameters=parameters) + start = time.monotonic() + try: + await asyncio.sleep(driver.default_poll_interval) + while True: + status = await driver.send_command(driver.CommandFamily.STATUS, retries=5) + if len(status) > 2 and status[2] & self._DATA_READY: + break + if not driver._parse_status(status).busy: + raise RuntimeError(f"the {slide} filter scan ended without reporting its data ready") + if time.monotonic() - start >= timeout: + raise TimeoutError(f"the {slide} filter scan took longer than {timeout} s") + await asyncio.sleep(driver.default_poll_interval) + await driver.send_command( + driver.CommandFamily.REQUEST, + driver.Command.SPECTRAL_DATA, + parameters=bytes(5), + when_idle=False, + ) + reply = await driver.send_command( + driver.CommandFamily.REQUEST, + driver.Command.FILTER_RESULT, + parameters=bytes(5), + when_idle=False, + ) + await driver._wait_until_ready( + timeout=max(timeout - (time.monotonic() - start), 0), + poll_interval=driver.default_poll_interval, + ) + return reply + except BaseException: + await self._drain_and_stop(f"{slide} filter scan") + raise + + async def detect_all_filters( + self, + ) -> Tuple[ExcitationFilterSlide, DichroicFilterSlide, EmissionFilterSlide]: + """Detect the filters in every position, as the vendor's "Detect all filters". + + - One scan each: excitation 1 to 4, dichroic A to C, emission 5 to 8 (slots 1 to 4). + - Excitation and dichroic positions take about 8 s each, emission about 49 s: about 4 minutes. + - Detected filters have no name, their edges in nm as the reader states them; an empty + position holds none. + - The slides replace `excitation_filter_slide`, `dichroic_filter_slide` and + `emission_filter_slide` once every position is scanned. + + Returns: + the excitation, dichroic and emission filter slides. + + Raises: + RuntimeError: if the reader reports no spectrometer (absorbance), or no extended-red + photomultiplier for the emission scans, whose captured limits are its; if a scan ends + without its data. + TimeoutError: if a position takes longer than its timeout (`_FILTER_SCAN_TIMEOUT`, + `_EMISSION_FILTER_SCAN_TIMEOUT` for emission). + ValueError: if a FILTER_RESULT is not one of its position. + """ + device = self._driver.configuration + if device.spectrometer is False or device.absorbance_available is False: + raise RuntimeError("the reader reports no spectrometer, which filter detection scans with") + if device.extended_red_photomultiplier_tube is False: + raise RuntimeError( + "the reader reports no extended-red photomultiplier; the emission scans' limits are only " + "captured with one" + ) + excitation_slide = ExcitationFilterSlide(self.configuration.excitation_filter_slots) + dichroic_slide = DichroicFilterSlide(self.configuration.dichroic_filter_slots) + emission_slide = EmissionFilterSlide(self.configuration.emission_filter_slots) + for slot in range(1, excitation_slide.maximum_slots + 1): + reply = await self._detect_filter("excitation", slot) + optical_filter = self._compute_optical_filter( + slot, self._decode_filter_result(reply, "excitation") + ) + if optical_filter is not None: + excitation_slide.register(optical_filter) + for slot in range(1, dichroic_slide.maximum_slots + 1): + reply = await self._detect_filter("dichroic", slot) + dichroic_filter = self._compute_dichroic_filter( + slot, self._decode_filter_result(reply, "dichroic") + ) + if dichroic_filter is not None: + dichroic_slide.register(dichroic_filter) + for slot in range(1, emission_slide.maximum_slots + 1): + reply = await self._detect_filter("emission", slot) + optical_filter = self._compute_optical_filter( + slot, self._decode_filter_result(reply, "emission") + ) + if optical_filter is not None: + emission_slide.register(optical_filter) + logger.info("filter detection: %r, %r, %r", excitation_slide, dichroic_slide, emission_slide) + self.excitation_filter_slide = excitation_slide + self.dichroic_filter_slide = dichroic_slide + self.emission_filter_slide = emission_slide + return excitation_slide, dichroic_slide, emission_slide + + # ---------------------------------------- + # Z-profiles + # ---------------------------------------- + + # The gain definition bytes of `09`, the z-scan, as the vendor sends them. + _Z_PROFILE_GAIN_DEFINITION = bytes([0x04, 0x08]) + # The step between focal heights, in 0.01 mm: the only one captured. + _Z_PROFILE_RESOLUTION = 10 + # Before the optics, as the end of the RUN's chromatic header. + _Z_PROFILE_OPTICS_HEADER = bytes.fromhex("0000000c") + # Status reply byte 2 bit 5: the z-profile is ready. + _Z_PROFILE_RESULT_READY = 0x20 + + # -- checks -------------------------------------------------------------------------------------- + + def _check_z_profile_arguments( + self, + fluorescence_channel: FluorescenceChannel, + maximum_focal_height: float, + gain_target_percent: Optional[float], + *, + timeout: Optional[float] = None, + ) -> None: + """Raise if the fluorescence channel, maximum focal height, target or timeout is out of range. + + Raises: + ValueError: listing every violation: the fluorescence channel's + (`_check_fluorescence_channel`), a maximum focal height not in 0.1 mm steps above 0 within + `focal_height_range` (the reader refuses 0), a target not above 0 up to 100 %, a timeout + not positive. + """ + errors = self._check_fluorescence_channel(fluorescence_channel) + lowest, highest = self.configuration.focal_height_range + if ( + not lowest < maximum_focal_height <= highest + or not maximum_focal_height > 0 + or abs(maximum_focal_height * 10 - round(maximum_focal_height * 10)) > 1e-9 + ): + errors.append( + f"maximum_focal_height must be above {max(lowest, 0)} up to {highest} mm in 0.1 mm " + f"steps, got {maximum_focal_height}" + ) + if gain_target_percent is not None and not 0 < gain_target_percent <= 100: + errors.append(f"gain_target_percent must be above 0 up to 100, got {gain_target_percent}") + if timeout is not None and not timeout > 0: + errors.append(f"timeout must be positive, got {timeout}") + if errors: + raise ValueError("; ".join(errors)) + + # -- `09` parameters ----------------------------------------------------------------------------- + + def _compute_z_profile_parameters( + self, + plate: Plate, + well: Well, + fluorescence_channel: FluorescenceChannel, + maximum_focal_height: float, + gain_target_percent: Optional[float], + ) -> bytes: + """`09`'s 42 parameter bytes. + + The plate's geometry, the well's column and row (from 1), the gain definition, the resolution, + `maximum_focal_height`, the target in 0.01 % (0 without), then the optics: + `_Z_PROFILE_OPTICS_HEADER` and a RUN fluorescence channel's bytes without its gain and + closing `00` (`_encode_optics`). + """ + index = next(i for i, item in enumerate(plate.get_all_items()) if item is well) + rows = plate.num_items_y + target = 0 if gain_target_percent is None else round(gain_target_percent * 100) + optics = self._encode_optics( + *self._compute_optics(fluorescence_channel), + *self._compute_wavelength_selectors(fluorescence_channel), + ) + return ( + self._encode_plate_geometry(plate) + + bytes([index // rows + 1, index % rows + 1]) + + self._Z_PROFILE_GAIN_DEFINITION + + self._Z_PROFILE_RESOLUTION.to_bytes(2, "big") + + round(maximum_focal_height * 100).to_bytes(2, "big") + + target.to_bytes(2, "big") + + self._Z_PROFILE_OPTICS_HEADER + + optics + ) + + # -- decoding ------------------------------------------------------------------------------------ + + @staticmethod + def _decode_z_profile_result( + reply: bytes, plate: Plate, *, start_timestamp: datetime, end_timestamp: datetime + ) -> CLARIOstarZProfileResult: + """The z-profile of a FOCUS_RESULT (`05 05`) reply. + + - Bytes 7-8: the gain; 9-12 the peak counts (u32 BE); 13-14 their percent of the detector's + range in 0.01 %. + - Bytes 15 and 16: the well's column and row (from 1). + - Bytes 17-18: the peak focal height in 0.01 mm (s16 BE); 19-22 the z-profile's average + around it. + - Bytes 23-24: the point count n, then n points (s32 BE height in 0.01 mm, which can be below + 0; u32 BE counts). + - Gain, peak counts and percent are 0 without a target. + + Raises: + ValueError: if the reply is cut short or names a well not on the plate. + """ + if len(reply) < 25: + raise ValueError(f"FOCUS_RESULT reply too short, got {len(reply)} bytes") + points = int.from_bytes(reply[23:25], "big") + if len(reply) < 25 + 8 * points: + raise ValueError(f"FOCUS_RESULT reply cut short: {points} points, got {len(reply)} bytes") + column, row = reply[15], reply[16] + if not (1 <= column <= plate.num_items_x and 1 <= row <= plate.num_items_y): + raise ValueError( + f"FOCUS_RESULT reply names a well not on the plate, got column {column}, row {row}" + ) + well = plate.get_all_items()[(column - 1) * plate.num_items_y + row - 1] + gain = int.from_bytes(reply[7:9], "big") + peak_counts = int.from_bytes(reply[9:13], "big") + percent = int.from_bytes(reply[13:15], "big") + z_profile = [ + ( + int.from_bytes(reply[offset : offset + 4], "big", signed=True) / 100, + int.from_bytes(reply[offset + 4 : offset + 8], "big"), + ) + for offset in range(25, 25 + 8 * points, 8) + ] + return CLARIOstarZProfileResult( + peak_focal_height=int.from_bytes(reply[17:19], "big", signed=True) / 100, + gain=gain or None, + well=plate.get_child_identifier(well), + peak_counts=peak_counts or None, + peak_detector_percent=percent / 100 if percent else None, + average_signal=int.from_bytes(reply[19:23], "big"), + z_profile=z_profile, + # TODO(device): no z-scan with the temperature controller on is captured. + temperature=None, + start_timestamp=start_timestamp, + end_timestamp=end_timestamp, + ) + + # -- measuring ----------------------------------------------------------------------------------- + + async def _run_z_profile(self, parameters: bytes, timeout: float) -> bytes: + """Send a z-scan (`09`) and return its FOCUS_RESULT reply. + + As `_run_measurement`: status until the reader is idle, its stage at status byte 7 and the + gain it tries at 9-10; the result must be ready (`_Z_PROFILE_RESULT_READY`), then + `05 05 00 00 00 00 00`. Any failure, a timeout or a cancelled call stops it (`0b 00`) first. + + Raises: + TimeoutError: if the z-scan takes longer than `timeout` seconds. + RuntimeError: if it ends without reporting its result ready. + """ + driver = self._driver + await driver.send_command(driver.CommandFamily.FOCUS_WELL, parameters=parameters) + start = time.monotonic() + stage: Optional[int] = None + try: + await asyncio.sleep(driver.default_poll_interval) + while True: + status = await driver.send_command(driver.CommandFamily.STATUS, retries=5) + if not driver._parse_status(status).busy: + break + if len(status) > 10 and status[7] != stage: + stage = status[7] + logger.debug("z-scan: stage %d, gain %d", stage, int.from_bytes(status[9:11], "big")) + if time.monotonic() - start >= timeout: + raise TimeoutError(f"the z-scan took longer than {timeout} s") + await asyncio.sleep(driver.default_poll_interval) + if not (len(status) > 2 and status[2] & self._Z_PROFILE_RESULT_READY): + raise RuntimeError("the z-scan ended without reporting its result ready") + return await driver.send_command( + driver.CommandFamily.REQUEST, + driver.Command.FOCUS_RESULT, + parameters=bytes(5), + when_idle=False, + ) + except BaseException: + await self._drain_and_stop("z-scan") + raise + + async def measure_z_profile( + self, + fluorescence_channel: FluorescenceChannel, + *, + well: Well, + gain_target_percent: Optional[float] = None, + maximum_focal_height: float = 15.0, + timeout: Optional[float] = None, + ) -> CLARIOstarZProfileResult: + """Measure the z-profile of `well` from the top and, with a target, fit the gain at its peak. + + - The reader reads the well from `maximum_focal_height` down in 0.1 mm steps to a lowest + height it chooses (0.7 mm and -0.6 mm on hardware), and computes the peak (about 8 s). + - With `gain_target_percent` it then fits the gain at the peak until the well reads that + percent of the detector's 260,000 counts (about 14 s in all). + - For the brightest of several wells, `measure` them first and pass the brightest. + - Cancelling the call stops it on the reader. + + Args: + fluorescence_channel: the optics, as for `measure` (`_check_fluorescence_channel`); a + bandwidth None is the default bandwidth, a filter given as a slot or a name the slides'. + well: the well to scan, on the plate in the plate holder. + gain_target_percent: above 0 up to 100; None measures the z-profile only. + maximum_focal_height: the height the scan starts at, mm above 0 in 0.1 mm steps within + `configuration.focal_height_range`. + timeout: seconds the z-scan may take; `default_read_timeout` when None. + + Returns: + the z-profile, its peak, and the gain with a target. + + Raises: + RuntimeError: if no plate is in the plate holder, or the z-scan ends without a result. + ValueError: if an argument is out of range (`_check_z_profile_arguments`, `_get_filter`), + the well is not on the plate, or the reply is not a z-profile. + TimeoutError: if the z-scan takes longer than `timeout` seconds. + """ + filled = self._get_filled_fluorescence_channel(fluorescence_channel) + self._check_z_profile_arguments( + filled, maximum_focal_height, gain_target_percent, timeout=timeout + ) + plate, _ = self._get_plate_and_wells(MeasurementSettings(), None, [well]) + parameters = self._compute_z_profile_parameters( + plate, well, filled, maximum_focal_height, gain_target_percent + ) + logger.info( + "z-profile: %s on %s of %s, gain target %s %%", + self._describe_fluorescence_channel(filled), + plate.get_child_identifier(well), + plate.name, + gain_target_percent, + ) + start_timestamp = datetime.now() + reply = await self._run_z_profile( + parameters, self.default_read_timeout if timeout is None else timeout + ) + return self._decode_z_profile_result( + reply, plate, start_timestamp=start_timestamp, end_timestamp=datetime.now() + ) + # ---------------------------------------- # Discrete reads # ---------------------------------------- @@ -851,8 +1435,7 @@ def _get_fluorescence_channels_and_gains( ) -> Tuple[List[FluorescenceChannel], List[int]]: """`measure`'s fluorescence channels, filled, and a gain for each. - Each gets the default bandwidths (`_get_fluorescence_channel_with_default_bandwidths`); one gain - is given for all, or one each. + Each is filled by `_get_filled_fluorescence_channel`; one gain is given for all, or one each. """ given = ( [fluorescence_channels] @@ -860,8 +1443,7 @@ def _get_fluorescence_channels_and_gains( else list(fluorescence_channels) ) filled = [ - self._get_fluorescence_channel_with_default_bandwidths(fluorescence_channel) - for fluorescence_channel in given + self._get_filled_fluorescence_channel(fluorescence_channel) for fluorescence_channel in given ] gains = list(gain) if isinstance(gain, Sequence) else [gain] * len(filled) return filled, gains @@ -920,14 +1502,16 @@ async def measure( Args: fluorescence_channels: one `FluorescenceChannel` (`pylabrobot.lib.fluorescence`), or 1 to `configuration.maximum_chromatics` of them, each within `_check_fluorescence_channel`; a - bandwidth None is the default bandwidth. + bandwidth None is the default bandwidth, a filter given as a slot or a name the slides' + (`detect_all_filters`). gain: the detector gain within `configuration.gain_range`: one for every fluorescence channel, or one per fluorescence channel in the order given. wells: the wells to measure, on the plate; None measures every well. A matrix scan reads one well (no fluorescence matrix of several wells is captured). focal_height: mm above the plate's base, within `configuration.focal_height_range`, in 0.1 mm steps. None reads at `default_focal_height_percent_of_well_depth` (25: a quarter - up the well, for low and moderate fills). + up the well, for low and moderate fills). For full wells, or the most signal, focus on a + representative well with `measure_z_profile` and pass its `peak_focal_height`. optic_position: read from the top, or from the bottom if the reader has one; None for `default_optic_position`. flying_mode: read each well without stopping: point reads of 1 or 3 flashes with no @@ -945,9 +1529,9 @@ async def measure( Raises: RuntimeError: if no plate is in the plate holder, or the run ends without its data. - ValueError: if an argument is out of range (`_check_arguments`), a matrix scan has several - wells, the settings cannot read the wells of the plate, or the reply is not this read's - (`_decode_data_reply`). + ValueError: if an argument is out of range (`_check_arguments`), a filter's slot or name has + no slide or is not on it (`_get_filter`), a matrix scan has several wells, the settings + cannot read the wells of the plate, or the reply is not this read's (`_decode_data_reply`). TimeoutError: if the run takes longer than `timeout` seconds. """ measurement = MeasurementSettings() if measurement_settings is None else measurement_settings @@ -1225,7 +1809,8 @@ async def measure_spectrum( step_size: nm between wavelengths: 1, the only one captured. focal_height: mm above the plate's base, within `configuration.focal_height_range`, in 0.1 mm steps. None reads at `default_focal_height_percent_of_well_depth` (25: a quarter - up the well, for low and moderate fills). + up the well, for low and moderate fills). For full wells, or the most signal, focus on a + representative well with `measure_z_profile` and pass its `peak_focal_height`. measurement_settings: how to read, without a pause, a matrix or another corner, reading direction or bidirectional setting; None reads with the defaults. shake_settings: shake the plate before the read; None does not shake. @@ -1244,7 +1829,7 @@ async def measure_spectrum( FrameError: if a data page arrives short on the second read too. """ measurement = MeasurementSettings() if measurement_settings is None else measurement_settings - fixed = self._get_fluorescence_channel_with_default_bandwidths( + fixed = self._get_filled_fluorescence_channel( FluorescenceChannel( excitation_wavelength=excitation_wavelength, excitation_bandwidth=excitation_bandwidth, diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py index 6ae17d750f5..cffa814dd90 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py @@ -10,10 +10,12 @@ import os import unittest from dataclasses import replace +from datetime import datetime from typing import Any, Dict, List, Optional, Sequence, Tuple, Union from unittest.mock import patch from pylabrobot.bmg_labtech.clariostar.driver.configuration_tests import decoded +from pylabrobot.bmg_labtech.clariostar.driver.features import fluorescence from pylabrobot.bmg_labtech.clariostar.driver.features import measurement as measurement_module from pylabrobot.bmg_labtech.clariostar.driver.features.absorbance_tests import ( COLUMN_1_DATA_REPLY, @@ -44,6 +46,11 @@ from pylabrobot.lib.fluorescence.channels import FluorescenceChannel from pylabrobot.lib.fluorescence.results import FluorescenceSpectrumResult from pylabrobot.lib.optics.filters import DichroicFilter, OpticalFilter +from pylabrobot.lib.optics.holders import ( + DichroicFilterSlide, + EmissionFilterSlide, + ExcitationFilterSlide, +) from pylabrobot.resources import Plate, Well, cor_96_wellplate_360uL_Fb, create_ordered_items_2d # The vendor's default read: 485/15 to 528/20 nm, gain 1000, 8.5 mm, 10 flashes. @@ -492,47 +499,18 @@ async def test_captured_reads_are_sent_with_nominal_band_edges(self): self.assertEqual( result.excitation_bandwidth, excitation_filter.bandwidth - if excitation_filter is not None + if isinstance(excitation_filter, OpticalFilter) else fluorescence_channel.excitation_bandwidth or 15, ) self.assertEqual( result.emission_bandwidth, emission_filter.bandwidth - if emission_filter is not None + if isinstance(emission_filter, OpticalFilter) else fluorescence_channel.emission_bandwidth or 20, ) self.assertEqual(result.gain, arguments.get("gain", 1000)) -class FilterPositionTests(unittest.IsolatedAsyncioTestCase): - async def test_every_slot_moves_its_slides(self): - """Excitation (slide 1, slide 2), emission (slide 2, slide 1), dichroic slot + 1; 1 is a - slide's home, 2 its first filter (the filter detection capture). Only slot 1 and dichroic A - are a reader's (F-Lf01); the others are predictions no reader has answered.""" - excitation = {1: (2, 1), 2: (3, 1), 3: (1, 2), 4: (1, 3)} - emission = {1: (1, 2), 2: (1, 3), 3: (2, 1), 4: (3, 1)} - for slot in range(1, 5): - for dichroic in range(1, 4): - with self.subTest(slot=slot, dichroic=dichroic): - io = FakeFTDI() - await read( - io, - TEMPERATURE_37_DATA_REPLY, - FluorescenceChannel( - excitation_filter=OpticalFilter(slot), - dichroic_filter=DichroicFilter(dichroic), - emission_filter=OpticalFilter(slot), - ), - ) - payload = ftdi_extract_payload(io.written[1]) - fluorescence_channel = len(payload) - 36 - fields = [ - int.from_bytes(payload[fluorescence_channel + i : fluorescence_channel + i + 2], "big") - for i in (2, 4, 6, 8, 10) - ] - self.assertEqual(fields, [*excitation[slot], dichroic + 1, *emission[slot]]) - - class ReadTests(unittest.IsolatedAsyncioTestCase): async def test_counts_are_the_vendors_export(self): _, result = await read(FakeFTDI(), TEMPERATURE_37_DATA_REPLY) @@ -1209,5 +1187,408 @@ def test_the_wavelength_limits_follow_the_photomultipliers(self): ) +def load_filter_detection() -> List[Dict[str, Any]]: + """The reader's filter detection, its 11 positions in order. + + Each holds its `24` frame and the replies the driver reads, as hex: the status before the `24`, + the reply to it, to `80` until data ready, to `05 11`, to `05 1b`, then the next status. + """ + path = os.path.join(os.path.dirname(__file__), "test_data", "filter_detection_captures.json.gz") + with gzip.open(path, "rt") as f: + return list(json.load(f)["positions"]) + + +# The detection's positions, in its order (excitation 1-4, dichroic A-C, emission 5-8). +DETECTION_POSITIONS = [ + *(("excitation", slot) for slot in range(1, 5)), + *(("dichroic", slot) for slot in range(1, 4)), + *(("emission", slot) for slot in range(1, 5)), +] + + +def detection_result(slide: str, slot: int) -> bytes: + """The payload of the FILTER_RESULT the reader sent for a position, the reply to `05 1b`.""" + position = load_filter_detection()[DETECTION_POSITIONS.index((slide, slot))] + return ftdi_extract_payload(bytes.fromhex(position["replies"][-2])) + + +def decode_filter_result(reply: bytes, slide: str) -> Tuple[int, int, int, int, int]: + """`Fluorescence._decode_filter_result` of a feature without a plate.""" + return make_fluorescence(FakeFTDI())._decode_filter_result(reply, slide) + + +def make_detecting_fluorescence(io: FakeFTDI) -> Fluorescence: + """A fluorescence feature of the captured reader, answered by the whole detection capture.""" + io.replies = [ + bytes.fromhex(reply) for position in load_filter_detection() for reply in position["replies"] + ] + return make_fluorescence(io) + + +class FilterDetectionTests(unittest.IsolatedAsyncioTestCase): + async def test_detection_as_the_reader_answered(self): + """The reader's filters: excitation 1, dichroic A and emission 5, every other position empty.""" + io = FakeFTDI(chunk=4096) + feature = make_detecting_fluorescence(io) + feature.excitation_filter_slide = ExcitationFilterSlide(4) # a slide the detection replaces + feature.excitation_filter_slide.register(OpticalFilter(3, "old")) + excitation, dichroic, emission = await feature.detect_all_filters() + self.assertIs(feature.excitation_filter_slide, excitation) + self.assertIs(feature.dichroic_filter_slide, dichroic) + self.assertIs(feature.emission_filter_slide, emission) + self.assertEqual(excitation[1], OpticalFilter(1, passbands=((282.2, 371.8),))) + self.assertEqual(dichroic[1], DichroicFilter(1, transmission_bands=((421.8, None),))) + self.assertEqual(emission[1], OpticalFilter(1, passbands=((604.9, 623.2),))) + self.assertEqual(repr(excitation), "ExcitationFilterSlide(4 slots, {1: (unnamed)})") + self.assertEqual(repr(dichroic), "DichroicFilterSlide(3 slots, {1: (unnamed)})") + self.assertEqual(repr(emission), "EmissionFilterSlide(4 slots, {1: (unnamed)})") + # Each position: status, `24`, status until data ready, `05 11` and `05 1b` without a status + # before them (the reader still busy), status until idle. + expected: List[bytes] = [] + for position in load_filter_detection(): + polls = len(position["replies"]) - 5 + expected += [b"\x80", ftdi_extract_payload(bytes.fromhex(position["request"]))] + expected += [b"\x80"] * polls + [bytes.fromhex("05110000000000")] + expected += [bytes.fromhex("051b0000000000"), b"\x80"] + self.assertEqual([ftdi_extract_payload(frame) for frame in io.written], expected) + self.assertEqual(io.replies, []) + + def test_long_pass_and_short_pass_filters_are_decoded(self): + """No long-pass or short-pass filter is captured: the replies are real results edited.""" + # Not a reader's reply: excitation 1's without its bandwidth, then also without its first edge. + long_pass = bytearray(detection_result("excitation", 1)) + long_pass[9:11] = bytes(2) + result = decode_filter_result(bytes(long_pass), "excitation") + self.assertEqual( + Fluorescence._compute_optical_filter(1, result), OpticalFilter(1, passbands=((282.2, None),)) + ) + long_pass[11:13] = bytes(2) + result = decode_filter_result(bytes(long_pass), "excitation") + with patch.object(fluorescence.logger, "warning") as warning: + self.assertEqual(Fluorescence._compute_optical_filter(1, result), OpticalFilter(1)) + warning.assert_called_once() # a short-pass, its band not stated + # Not a reader's reply: dichroic A's with its edge moved to the second. + short_pass = bytearray(detection_result("dichroic", 1)) + short_pass[13:15], short_pass[11:13] = short_pass[11:13], bytes(2) + result = decode_filter_result(bytes(short_pass), "dichroic") + with patch.object(fluorescence.logger, "warning") as warning: + self.assertEqual(Fluorescence._compute_dichroic_filter(1, result), DichroicFilter(1)) + warning.assert_called_once() + + def test_a_reply_not_of_its_position_is_refused(self): + reply = detection_result("excitation", 1) + for wrong, slide in ( + (reply, "emission"), + (reply[:-9], "excitation"), # Not a reader's reply: cut short. + ): + with self.subTest(slide=slide, length=len(wrong)), self.assertRaises(ValueError): + decode_filter_result(wrong, slide) + + async def test_a_scan_ending_without_data_stops(self): + """Not a reader's sequence: the idle status before the first `24`, again after it.""" + first = load_filter_detection()[0]["replies"] + io = FakeFTDI(chunk=4096) + io.replies = [bytes.fromhex(first[0]), bytes.fromhex(first[1]), bytes.fromhex(first[0])] + io.replies.append(STOP_REPLY) + feature = make_fluorescence(io) + with self.assertRaises(RuntimeError): + await feature.detect_all_filters() + self.assertEqual(io.written[-1], STOP_FRAME) + self.assertIsNone(feature.excitation_filter_slide) + + async def test_no_spectrometer_or_extended_red_sends_nothing(self): + for field in ("spectrometer", "absorbance_available", "extended_red_photomultiplier_tube"): + io = FakeFTDI() + feature = make_fluorescence(io) + configuration = feature._driver.configuration + if field == "spectrometer": + configuration.spectrometer = False + elif field == "absorbance_available": + configuration.absorbance_available = False + else: + configuration.extended_red_photomultiplier_tube = False + with self.subTest(field), self.assertRaises(RuntimeError): + await feature.detect_all_filters() + self.assertEqual(io.written, []) + + +def make_detected_slides(fluorescence_feature: Fluorescence) -> None: + """Give the feature the reader's slides as it detected them, named as the vendor names them.""" + excitation = ExcitationFilterSlide(4) + dichroic = DichroicFilterSlide(3) + emission = EmissionFilterSlide(4) + for filter_slide, slide, name in ( + (excitation, "excitation", "Ex TR"), + (emission, "emission", "615-18"), + ): + detected = Fluorescence._compute_optical_filter( + 1, decode_filter_result(detection_result(slide, 1), slide) + ) + assert detected is not None + filter_slide.register(replace(detected, name=name)) + result = decode_filter_result(detection_result("dichroic", 1), "dichroic") + mirror = Fluorescence._compute_dichroic_filter(1, result) + assert mirror is not None + dichroic.register(replace(mirror, name="LP TR")) + fluorescence_feature.excitation_filter_slide = excitation + fluorescence_feature.dichroic_filter_slide = dichroic + fluorescence_feature.emission_filter_slide = emission + + +async def read_with_slides(io: FakeFTDI, fluorescence_channel: FluorescenceChannel, **arguments): + """A whole-plate read through `fluorescence_channel`, gain 1000, the captured slides; feature, + result.""" + io.replies = [ + STATUS_IDLE, + RUN_REPLY, + STATUS_MEASURING, + STATUS_MEASURED, + TEMPERATURE_37_DATA_REPLY, + ] + feature = make_fluorescence(io, fluorescence_plate()) + make_detected_slides(feature) + return feature, await feature.measure( + fluorescence_channel, **{"gain": 1000, "focal_height": CAPTURED_FOCAL_HEIGHT, **arguments} + ) + + +class FilterSlideTests(unittest.IsolatedAsyncioTestCase): + async def test_slots_and_names_are_the_vendors_reads(self): + """The three filters' read by slot and by name; the one-filter reads by slot.""" + reads: Dict[str, Tuple[bytes, List[FluorescenceChannel]]] = { + "three_filters": ( + FILTERS_RUN_FRAME, + [ + FluorescenceChannel(excitation_filter=1, dichroic_filter=1, emission_filter=1), + FluorescenceChannel( + excitation_filter="Ex TR", dichroic_filter="LP TR", emission_filter="615-18" + ), + ], + ), + "emission_filter": ( + EMISSION_FILTER_RUN_FRAME, + [FluorescenceChannel(excitation_wavelength=400, dichroic_filter=1, emission_filter=1)], + ), + "excitation_filter": ( + EXCITATION_FILTER_RUN_FRAME, + [FluorescenceChannel(excitation_filter=1, dichroic_filter=1, emission_wavelength=528)], + ), + } + for name, (captured, fluorescence_channels) in reads.items(): + for fluorescence_channel in fluorescence_channels: + with self.subTest(name, fluorescence_channel=fluorescence_channel): + io = FakeFTDI() + _, result = await read_with_slides( + io, fluorescence_channel, measurement_settings=TEN_FLASHES + ) + expected = with_nominal_edges(captured, [fluorescence_channel]) + self.assertEqual(ftdi_extract_payload(io.written[1]), expected) + if fluorescence_channel.excitation_filter is not None: + self.assertEqual(result.excitation_wavelength, (282.2 + 371.8) / 2) + if fluorescence_channel.emission_filter is not None: + self.assertEqual(result.emission_bandwidth, 623.2 - 604.9) + + async def test_every_slot_is_its_detection_position(self): + """No RUN through slots 2 to 4 or dichroic B and C is captured: their RUN fields are the slide + positions of the detection's `24`.""" + words = { + (slide, slot): [int.from_bytes(request[i : i + 2], "big") for i in range(10, 28, 2)] + for (slide, slot), request in zip( + DETECTION_POSITIONS, + (ftdi_extract_payload(bytes.fromhex(p["request"])) for p in load_filter_detection()), + ) + } + for slot in range(1, 5): + for dichroic in range(1, 4): + with self.subTest(slot=slot, dichroic=dichroic): + io = FakeFTDI() + await read_with_slides( + io, + FluorescenceChannel( + excitation_filter=slot, dichroic_filter=dichroic, emission_filter=slot + ), + ) + payload = ftdi_extract_payload(io.written[1]) + fluorescence_channel = len(payload) - 36 + fields = [ + int.from_bytes(payload[fluorescence_channel + i : fluorescence_channel + i + 2], "big") + for i in (2, 4, 6, 8, 10) + ] + # The detection's words: two slits, EX_SP, EX_LP, BSP, EM_SP, EM_LP, ... + excitation = words[("excitation", slot)][2:4] + mirror = words[("dichroic", dichroic)][4] + emission = words[("emission", slot)][5:7] + self.assertEqual(fields, [*excitation, mirror, *emission]) + + async def test_unknown_slots_and_names_send_nothing(self): + io = FakeFTDI() + feature = make_fluorescence(io, fluorescence_plate()) + unresolved = FluorescenceChannel( + excitation_filter=1, dichroic_filter=LP_TR, emission_wavelength=528 + ) + with self.assertRaisesRegex(ValueError, "detect_all_filters"): + await feature.measure(unresolved, gain=1000) + make_detected_slides(feature) + refused: List[Tuple[Union[OpticalFilter, int, str], Union[DichroicFilter, int, str]]] = [ + (5, 1), + (0, 1), + ("Ex XX", 1), + (1, 4), + (1, "LP XX"), + ] + for excitation_filter, dichroic_filter in refused: + fluorescence_channel = FluorescenceChannel( + excitation_filter=excitation_filter, + dichroic_filter=dichroic_filter, + emission_wavelength=528, + ) + with self.subTest(fluorescence_channel=fluorescence_channel), self.assertRaises(ValueError): + await feature.measure(fluorescence_channel, gain=1000) + self.assertEqual(io.written, []) + + +def load_z_profile_captures() -> Dict[str, Any]: + """The reader's z-scans: of A1 without a target, of D6 to 90 %, and a brightest-well search. + + Each holds its frame (`09`; the search `0c`, which the driver does not send, kept for its result) + and the replies the driver reads, as hex: the status before it, the reply to it, to `80`, to + `05 05`. + """ + path = os.path.join(os.path.dirname(__file__), "test_data", "z_profile_captures.json.gz") + with gzip.open(path, "rt") as f: + return dict(json.load(f)) + + +def with_nominal_z_scan_edges(captured: bytes) -> bytes: + """`captured`'s payload with 485/15 and 528/20 nm at their nominal edges (`with_nominal_edges`). + + The optics are the last 14 bytes. + """ + payload = bytearray(ftdi_extract_payload(captured)) + for offset, edges in ((-14, _compute_band_edges(485, 15)), (-8, _compute_band_edges(528, 20))): + for index, value in enumerate(edges): + start = len(payload) + offset + 2 * index + payload[start : start + 2] = value.to_bytes(2, "big") + return bytes(payload) + + +async def run_z_profile(name: str, well: str, **arguments): + """Capture `name`'s z-scan of `well` answered by its replies; the io, the capture, the result.""" + capture = load_z_profile_captures()[name] + io = FakeFTDI(chunk=4096) + io.replies = [bytes.fromhex(reply) for reply in capture["replies"]] + plate = fluorescence_plate() + feature = make_fluorescence(io, plate) + result = await feature.measure_z_profile(BASELINE, well=plate.get_well(well), **arguments) + return io, capture, result + + +class ZProfileTests(unittest.IsolatedAsyncioTestCase): + async def test_z_scans_are_sent_and_decoded_as_the_vendors(self): + """A1 without a target and D6 to 90 %.""" + z_scans = { + "z_profile_without_target": ("A1", None, (15.0, None, "A1", None, None, 60920)), + "z_profile_with_gain_target": ("D6", 90, (10.3, 1968, "D6", 231622, 89.08, 43243)), + } + for name, (well, target, expected) in z_scans.items(): + with self.subTest(name): + io, capture, result = await run_z_profile(name, well, gain_target_percent=target) + self.assertEqual( + ftdi_extract_payload(io.written[1]), + with_nominal_z_scan_edges(bytes.fromhex(capture["request"])), + ) + self.assertEqual( + ( + result.peak_focal_height, + result.gain, + result.well, + result.peak_counts, + result.peak_detector_percent, + result.average_signal, + ), + expected, + ) + self.assertEqual(len(result.z_profile), 144) + self.assertEqual(result.z_profile[0][0], 15.0) + self.assertEqual(result.z_profile[-1][0], 0.7) + self.assertLessEqual(result.start_timestamp, result.end_timestamp) + self.assertIsNone(result.temperature) + # `05 05` follows the idle status without one of its own. + self.assertEqual(ftdi_extract_payload(io.written[-1]), bytes.fromhex("05050000000000")) + self.assertEqual(io.replies, []) + + def test_a_height_below_0_is_decoded(self): + """Not a reader's reply: the brightest-well search's result, last point and peak -0.60 mm.""" + reply = bytearray( + ftdi_extract_payload( + bytes.fromhex(load_z_profile_captures()["brightest_well_search"]["replies"][-1]) + ) + ) + points = int.from_bytes(reply[23:25], "big") + last = 25 + 8 * (points - 1) + reply[last : last + 4] = (-60).to_bytes(4, "big", signed=True) + reply[17:19] = (-60).to_bytes(2, "big", signed=True) # the peak there too + now = datetime.now() + result = Fluorescence._decode_z_profile_result( + bytes(reply), fluorescence_plate(), start_timestamp=now, end_timestamp=now + ) + self.assertEqual(result.z_profile[-1][0], -0.6) + self.assertEqual(result.peak_focal_height, -0.6) + + def test_a_reply_not_of_this_plate_is_refused(self): + """The brightest-well search's result, B11, cut short; then of column 13 (not a reply).""" + reply = ftdi_extract_payload( + bytes.fromhex(load_z_profile_captures()["brightest_well_search"]["replies"][-1]) + ) + off_plate = bytearray(reply) + off_plate[15] = 13 + for wrong in (reply[:-8], bytes(off_plate)): + with self.subTest(length=len(wrong)), self.assertRaises(ValueError): + Fluorescence._decode_z_profile_result( + wrong, fluorescence_plate(), start_timestamp=datetime.now(), end_timestamp=datetime.now() + ) + + async def test_a_z_scan_ending_without_its_result_stops(self): + """Not a reader's reply: an idle status with byte 2 bit 5 clear right after the `09`.""" + capture = load_z_profile_captures()["z_profile_without_target"] + io = FakeFTDI() + replies = [bytes.fromhex(reply) for reply in capture["replies"]] + io.replies = [replies[0], replies[1], replies[0], STOP_REPLY] + plate = fluorescence_plate() + with self.assertRaises(RuntimeError): + await make_fluorescence(io, plate).measure_z_profile(BASELINE, well=plate.get_well("A1")) + self.assertEqual(io.written[-1], STOP_FRAME) + + async def test_out_of_range_z_scans_send_nothing(self): + io = FakeFTDI() + plate = fluorescence_plate() + feature = make_fluorescence(io, plate) + refused: List[Tuple[FluorescenceChannel, Dict[str, Any]]] = [ + (BASELINE, dict(maximum_focal_height=0.0)), # the reader refuses it + (BASELINE, dict(maximum_focal_height=25.1)), + (BASELINE, dict(maximum_focal_height=8.55)), + (BASELINE, dict(gain_target_percent=0)), + (BASELINE, dict(gain_target_percent=101)), + (BASELINE, dict(timeout=0)), + (replace(BASELINE, excitation_wavelength=None), {}), + ( + FluorescenceChannel(excitation_filter=1, dichroic_filter=1, emission_wavelength=528), + {}, + ), # no slides + ] + for fluorescence_channel, arguments in refused: + with ( + self.subTest(fluorescence_channel=fluorescence_channel, arguments=arguments), + self.assertRaises(ValueError), + ): + await feature.measure_z_profile( + fluorescence_channel, well=plate.get_well("A1"), **arguments + ) + other = Plate("other", 127.8, 85.5, 14.0, ordered_items={}) + with self.assertRaises(ValueError): + await feature.measure_z_profile(BASELINE, well=other) # type: ignore[arg-type] + self.assertEqual(io.written, []) + + if __name__ == "__main__": unittest.main() diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/measurement.py b/pylabrobot/bmg_labtech/clariostar/driver/features/measurement.py index a2b9ad36ef5..b0bc12f5dc6 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/measurement.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/measurement.py @@ -427,14 +427,12 @@ def _get_plate_and_wells( # -- plate block --------------------------------------------------------------------------------- @staticmethod - def _encode_plate_block(plate: Plate, wells: Sequence[Well]) -> bytes: - """The 63-byte plate block: geometry in 0.01 mm (u16 BE), columns, rows, `00`, well mask. + def _encode_plate_geometry(plate: Plate) -> bytes: + """The plate's 14 bytes: size, first and last well centre in 0.01 mm (u16 BE), columns, rows. - Y is measured from the plate's back edge. The mask bit of a well is row * columns + column, - most significant bit first. + Y is measured from the plate's back edge. """ items = plate.get_all_items() - columns, rows = plate.num_items_x, plate.num_items_y def center_from_back(well: Well): if well.location is None: @@ -446,15 +444,27 @@ def center_from_back(well: Well): last_x, last_y = center_from_back(items[-1]) geometry = (plate.get_size_x(), plate.get_size_y(), first_x, first_y, last_x, last_y) block = b"".join(round(value * 100).to_bytes(2, "big") for value in geometry) - block += bytes([columns, rows, 0x00]) + return block + bytes([plate.num_items_x, plate.num_items_y]) + + @staticmethod + def _encode_well_mask(plate: Plate, wells: Sequence[Well]) -> bytes: + """The 48-byte well mask, a bit per well of `wells`. + The bit of a well is row * columns + column, most significant bit first. + """ + items = plate.get_all_items() + columns, rows = plate.num_items_x, plate.num_items_y mask = bytearray(_WELL_MASK_SIZE) selected = {id(well) for well in wells} for index, well in enumerate(items): # column-major: A1, B1, ..., A2, ... if id(well) in selected: bit = (index % rows) * columns + index // rows mask[bit // 8] |= 0x80 >> (bit % 8) - return block + bytes(mask) + return bytes(mask) + + def _encode_plate_block(self, plate: Plate, wells: Sequence[Well]) -> bytes: + """The 63-byte plate block: `_encode_plate_geometry`, `00`, then `_encode_well_mask`.""" + return self._encode_plate_geometry(plate) + b"\x00" + self._encode_well_mask(plate, wells) # -- scan and settling bytes --------------------------------------------------------------------- @@ -631,6 +641,26 @@ async def _request_data(self, values: Optional[int]) -> List[bytes]: if received >= values: return replies + async def _drain_and_stop(self, what: str) -> None: + """Drain the reply still streaming, send STOP (`0b 00`) and drop what STOP left. + + A refused STOP only warns. + + Args: + what: what is stopped, for the log. + """ + logger.warning("stopping the %s", what) + try: + # The reader sends a reply it has started to its end: drain it before STOP, and drop + # whatever STOP left after it, so the next command reads its own reply. + await self._driver._drain_rx() + await self._driver.send_command( + self._driver.CommandFamily.STOP, parameters=b"\x00", when_idle=False + ) + await self._driver._purge_rx() + except (CommandRefusedError, FrameError) as e: # e.g. the run ended meanwhile (214) + logger.warning("STOP not accepted: %s", e) + async def _run_measurement( self, parameters: bytes, @@ -692,15 +722,5 @@ async def _run_measurement( raise TimeoutError(f"the measurement took longer than {timeout} s") await asyncio.sleep(self._driver.default_poll_interval) except BaseException: # a timeout, a cancelled call, a bad reply: stop the run first - logger.warning("stopping the measurement run") - try: - # The reader sends a reply it has started to its end: drain it before STOP, and drop - # whatever STOP left after it, so the next command reads its own reply. - await self._driver._drain_rx() - await self._driver.send_command( - self._driver.CommandFamily.STOP, parameters=b"\x00", when_idle=False - ) - await self._driver._purge_rx() - except (CommandRefusedError, FrameError) as e: # e.g. the run ended meanwhile (214) - logger.warning("STOP not accepted: %s", e) + await self._drain_and_stop("measurement run") raise diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/filter_detection_captures.json.gz b/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/filter_detection_captures.json.gz new file mode 100644 index 00000000000..965222288c1 Binary files /dev/null and b/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/filter_detection_captures.json.gz differ diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/z_profile_captures.json.gz b/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/z_profile_captures.json.gz new file mode 100644 index 00000000000..bca82e4ea91 Binary files /dev/null and b/pylabrobot/bmg_labtech/clariostar/driver/features/test_data/z_profile_captures.json.gz differ diff --git a/pylabrobot/bmg_labtech/clariostar/driver/master.py b/pylabrobot/bmg_labtech/clariostar/driver/master.py index 13dd70a0757..6665b05e684 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/master.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/master.py @@ -112,8 +112,10 @@ class CommandFamily: RUN = 0x04 REQUEST = 0x05 TEMPERATURE_CONTROLLER = 0x06 + FOCUS_WELL = 0x09 STOP = 0x0B SET_PARAMETER = 0x0E + FILTER_SCAN = 0x24 STATUS = 0x80 class Command: @@ -126,17 +128,23 @@ class Command: DRAWER_OPEN = 0x01 # REQUEST DATA = 0x02 + FOCUS_RESULT = 0x05 PARAMETERS = 0x07 + SPECTRAL_DATA = 0x11 + FILTER_RESULT = 0x1B # SET_PARAMETER; seen only writing the two-byte test-run counter SET_PARAMETER_WORD = 0x0B - # Families whose payload is the family byte plus parameters, without a command byte. + # Families whose payload is the family byte plus parameters, without a command byte; FILTER_SCAN's + # first parameter is a mode bit field. _NO_COMMAND_FAMILIES = frozenset( { CommandFamily.RUN, CommandFamily.STOP, CommandFamily.STATUS, CommandFamily.TEMPERATURE_CONTROLLER, + CommandFamily.FOCUS_WELL, + CommandFamily.FILTER_SCAN, } ) diff --git a/pylabrobot/lib/fluorescence/channels.py b/pylabrobot/lib/fluorescence/channels.py index f587a8a1be0..35192e94ffa 100644 --- a/pylabrobot/lib/fluorescence/channels.py +++ b/pylabrobot/lib/fluorescence/channels.py @@ -5,7 +5,7 @@ """ from dataclasses import dataclass -from typing import List, Optional +from typing import List, Optional, Union from pylabrobot.lib.optics.filters import DichroicFilter, OpticalFilter @@ -14,7 +14,7 @@ def _check_excitation_or_emission( excitation_or_emission: str, wavelength: Optional[float], bandwidth: Optional[float], - optical_filter: Optional[OpticalFilter], + optical_filter: Optional[Union[OpticalFilter, int, str]], ) -> List[str]: """The violations of excitation or emission. @@ -34,7 +34,7 @@ def _check_excitation_or_emission( def _check_dichroic( - dichroic_filter: Optional[DichroicFilter], split_wavelength: Optional[float] + dichroic_filter: Optional[Union[DichroicFilter, int, str]], split_wavelength: Optional[float] ) -> List[str]: """The violations of the dichroic: a fixed dichroic with a tunable split, a split not positive.""" errors = [] @@ -60,23 +60,26 @@ class FluorescenceChannel: Attributes: excitation_wavelength: nm through a monochromator or a tunable source; None with a filter. excitation_bandwidth: nm about `excitation_wavelength`; None with a filter or the default. - excitation_filter: the excitation filter instead of a wavelength. + excitation_filter: the excitation filter instead of a wavelength: a filter, or its slot or name + on the device's slides. dichroic_split_wavelength: nm where a tunable dichroic splits excitation from emission; None otherwise. - dichroic_filter: a fixed dichroic instead of a split. + dichroic_filter: a fixed dichroic instead of a split: a filter, or its slot or name on the + device's slides. emission_wavelength: nm through a monochromator; None with a filter. emission_bandwidth: nm about `emission_wavelength`; None with a filter or the default. - emission_filter: the emission filter instead of a wavelength. + emission_filter: the emission filter instead of a wavelength: a filter, or its slot or name on + the device's slides. """ excitation_wavelength: Optional[float] = None excitation_bandwidth: Optional[float] = None - excitation_filter: Optional[OpticalFilter] = None + excitation_filter: Optional[Union[OpticalFilter, int, str]] = None dichroic_split_wavelength: Optional[float] = None - dichroic_filter: Optional[DichroicFilter] = None + dichroic_filter: Optional[Union[DichroicFilter, int, str]] = None emission_wavelength: Optional[float] = None emission_bandwidth: Optional[float] = None - emission_filter: Optional[OpticalFilter] = None + emission_filter: Optional[Union[OpticalFilter, int, str]] = None def __post_init__(self) -> None: """Raise listing every component of the wrong combination or sign. diff --git a/pylabrobot/lib/optics/holders.py b/pylabrobot/lib/optics/holders.py index 9ebfb884a8d..a95e954c24c 100644 --- a/pylabrobot/lib/optics/holders.py +++ b/pylabrobot/lib/optics/holders.py @@ -9,7 +9,7 @@ F = TypeVar("F", bound=_FilterBase) -class _FilterSlideBase(Generic[F]): +class FilterSlide(Generic[F]): """The filters a device holds on one side, by slot. Attributes: @@ -74,21 +74,21 @@ def __repr__(self) -> str: return f"{type(self).__name__}({self.maximum_slots} slots, {{{', '.join(held)}}})" -class ExcitationFilterSlide(_FilterSlideBase[OpticalFilter]): +class ExcitationFilterSlide(FilterSlide[OpticalFilter]): """The excitation filters a device holds, by slot.""" _CATEGORY = "excitation" _FILTER_CLASS = OpticalFilter -class EmissionFilterSlide(_FilterSlideBase[OpticalFilter]): +class EmissionFilterSlide(FilterSlide[OpticalFilter]): """The emission filters a device holds, by slot.""" _CATEGORY = "emission" _FILTER_CLASS = OpticalFilter -class DichroicFilterSlide(_FilterSlideBase[DichroicFilter]): +class DichroicFilterSlide(FilterSlide[DichroicFilter]): """The dichroic filters a device holds, by slot.""" _CATEGORY = "dichroic"