diff --git a/docs/_static/devices.json b/docs/_static/devices.json index 3cdc86a31a6..7e19a254508 100644 --- a/docs/_static/devices.json +++ b/docs/_static/devices.json @@ -1208,6 +1208,20 @@ "manager": "https://discuss.pylabrobot.org/u/rickwierenga", "oem": "https://lifesciences.tecan.com/multimode-plate-reader" }, + { + "id": "thermo-fisher-nanodrop-1000", + "vendor": "Thermo Fisher", + "name": "NanoDrop 1000", + "kind": "plate reader", + "capabilities": [ + "absorbance" + ], + "status": "basic", + "api": "pylabrobot.thermo_fisher.nanodrop_1000.ThermoFisherNanoDrop1000", + "api_version": "v1", + "code_slug": "thermo_fisher/nanodrop_1000", + "doc_slug": "thermo_fisher/nanodrop_1000/index" + }, { "id": "thermo-fisher-alps-300", "vendor": "Thermo Fisher", diff --git a/docs/user_guide/thermo_fisher/index.md b/docs/user_guide/thermo_fisher/index.md index 4c887655bbb..f5ea67777ee 100644 --- a/docs/user_guide/thermo_fisher/index.md +++ b/docs/user_guide/thermo_fisher/index.md @@ -4,4 +4,5 @@ :maxdepth: 1 alps/index +nanodrop_1000/index ``` diff --git a/docs/user_guide/thermo_fisher/nanodrop_1000/hello-world.ipynb b/docs/user_guide/thermo_fisher/nanodrop_1000/hello-world.ipynb new file mode 100644 index 00000000000..f9f481cb4b0 --- /dev/null +++ b/docs/user_guide/thermo_fisher/nanodrop_1000/hello-world.ipynb @@ -0,0 +1,79 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "nanodrop-intro", + "metadata": {}, + "source": "# Thermo Fisher NanoDrop 1000\n\nThe NanoDrop 1000 measures absorbance spectra from small-volume samples. Complete the [NanoDrop 1000 setup guide](setup.md) before connecting with PyLabRobot." + }, + { + "cell_type": "markdown", + "id": "nanodrop-connect-heading", + "metadata": {}, + "source": "## Connect" + }, + { + "cell_type": "code", + "execution_count": null, + "id": "nanodrop-connect", + "metadata": {}, + "outputs": [], + "source": "from pylabrobot.thermo_fisher.nanodrop_1000 import ThermoFisherNanoDrop1000\n\nnanodrop = ThermoFisherNanoDrop1000()\nawait nanodrop.setup()" + }, + { + "cell_type": "markdown", + "id": "nanodrop-blank-heading", + "metadata": {}, + "source": "## Blank\n\nClean the pedestals, load the blank solution, lower the sampling arm, and acquire the blank." + }, + { + "cell_type": "code", + "execution_count": null, + "id": "nanodrop-blank", + "metadata": {}, + "outputs": [], + "source": "await nanodrop.take_blank(integration_ms=20)" + }, + { + "cell_type": "markdown", + "id": "nanodrop-measure-heading", + "metadata": {}, + "source": "## Measure a sample\n\nClean the pedestals, load the sample, lower the sampling arm, and measure absorbance. The result contains matching wavelength and absorbance lists." + }, + { + "cell_type": "code", + "execution_count": null, + "id": "nanodrop-measure", + "metadata": {}, + "outputs": [], + "source": "wavelengths, absorbance = await nanodrop.measure_absorbance(integration_ms=20)\n\npeak_index = max(range(len(absorbance)), key=absorbance.__getitem__)\nprint(f\"Peak absorbance: {absorbance[peak_index]:.3f} at {wavelengths[peak_index]:.1f} nm\")" + }, + { + "cell_type": "markdown", + "id": "nanodrop-stop-heading", + "metadata": {}, + "source": "## Disconnect" + }, + { + "cell_type": "code", + "execution_count": null, + "id": "nanodrop-stop", + "metadata": {}, + "outputs": [], + "source": "await nanodrop.stop()" + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3 (ipykernel)", + "language": "python", + "name": "python3" + }, + "language_info": { + "name": "python", + "version": "3.11.0" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/docs/user_guide/thermo_fisher/nanodrop_1000/index.md b/docs/user_guide/thermo_fisher/nanodrop_1000/index.md new file mode 100644 index 00000000000..acd64af2c04 --- /dev/null +++ b/docs/user_guide/thermo_fisher/nanodrop_1000/index.md @@ -0,0 +1,15 @@ +# Thermo Fisher NanoDrop 1000 + +```{device-card} thermo-fisher-nanodrop-1000 +``` + +The NanoDrop 1000 measures absorbance spectra from small-volume samples. Start with the setup guide +to configure the USB connection, then follow the hello-world example to blank the instrument and +measure a sample. + +```{toctree} +:maxdepth: 1 + +setup +hello-world +``` diff --git a/docs/user_guide/thermo_fisher/nanodrop_1000/setup.md b/docs/user_guide/thermo_fisher/nanodrop_1000/setup.md new file mode 100644 index 00000000000..fa69f514668 --- /dev/null +++ b/docs/user_guide/thermo_fisher/nanodrop_1000/setup.md @@ -0,0 +1,49 @@ +# NanoDrop 1000 setup + +Install PyLabRobot with USB support: + +```bash +pip install "pylabrobot[usb]" +``` + +## Windows USB driver + +The NanoDrop 1000 must use a libusb-compatible driver before PyLabRobot can communicate with it. +Changing this driver prevents the official NanoDrop software from using the device until you switch +the driver back. + +Some Windows 10 and 11 systems report the NanoDrop as an unknown USB device because its older USB +controller does not provide a BOS descriptor. Only if Device Manager reports that problem, open an +administrator Command Prompt and run: + +```bat +reg add "HKLM\SYSTEM\CurrentControlSet\Control\usbflags\245710020002" /v SkipBOSDescriptorQuery /t REG_DWORD /d 1 /f +``` + +Unplug and reconnect the NanoDrop after changing the setting. + +Install the libusb driver with Zadig: + +1. Download and run [Zadig](https://zadig.akeo.ie/). +2. Select **Options > List All Devices**. +3. Select **NanoDrop 1000**. +4. Select **libusb-win32** as the replacement driver. +5. Click **Replace Driver** or **Install Driver**. + +## Restore the official driver + +To use the official NanoDrop software again: + +1. Open Device Manager. +2. Find the NanoDrop 1000 under **libusb-win32 devices**. +3. Select **Update driver > Browse my computer for drivers > Let me pick from a list**. +4. Select the original NanoDrop or Cypress EZ-USB driver. + +If you added the BOS descriptor registry setting and want to remove it, run this from an +administrator Command Prompt: + +```bat +reg delete "HKLM\SYSTEM\CurrentControlSet\Control\usbflags\245710020002" /v SkipBOSDescriptorQuery /f +``` + +Continue with the [NanoDrop 1000 hello world](hello-world.md). diff --git a/pylabrobot/io/usb.py b/pylabrobot/io/usb.py index eeb406e2385..9fe2853fb9e 100644 --- a/pylabrobot/io/usb.py +++ b/pylabrobot/io/usb.py @@ -228,7 +228,12 @@ def _read_packet( # No data available (yet), this will give a timeout error. Don't reraise. return None - async def read(self, timeout: Optional[int] = None, size: Optional[int] = None) -> bytes: + async def read( + self, + timeout: Optional[int] = None, + size: Optional[int] = None, + endpoint: Optional[int] = None, + ) -> bytes: """Read a response from the device. Args: @@ -236,6 +241,7 @@ async def read(self, timeout: Optional[int] = None, size: Optional[int] = None) timeout (specified by the `read_timeout` attribute). size: The maximum number of bytes to read. If `None`, read all available data until no more packets arrive. + endpoint: The endpoint address to read from. If `None`, use the configured read endpoint. """ if self.dev is None or self.read_endpoint is None: @@ -255,7 +261,7 @@ def read_or_timeout() -> bytes: last_packet: Optional[bytearray] = None while True: # read while we have data, and while the last packet is the max size. remaining = size - len(resp) if size is not None else None - last_packet = self._read_packet(size=remaining) + last_packet = self._read_packet(size=remaining, endpoint=endpoint) if last_packet is not None: resp += last_packet if self.read_endpoint is None: @@ -296,6 +302,29 @@ def read_or_timeout() -> bytes: ) return response + async def drain( + self, + endpoint: Optional[int] = None, + timeout: float = 0.05, + size: int = 512, + max_duration: float = 1, + ) -> None: + """Discard queued input from an endpoint until no packet arrives before the timeout.""" + if self.dev is None or self.read_endpoint is None: + raise RuntimeError(f"USB device for '{self.human_readable_device_name}' is not connected.") + + deadline = time.monotonic() + max_duration + loop = asyncio.get_running_loop() + while time.monotonic() < deadline: + packet = await loop.run_in_executor( + self.read_executor, + lambda: self._read_packet(size=size, timeout=timeout, endpoint=endpoint), + ) + if packet is None: + return + + raise TimeoutError(f"Timed out draining USB device '{self.human_readable_device_name}'.") + def get_available_devices(self) -> List["usb.core.Device"]: """Get a list of available devices that match the specified vendor and product IDs, and serial number and device_address if specified.""" @@ -561,7 +590,12 @@ async def write(self, data: bytes, timeout: Optional[float] = None): align_sequences(expected=next_command.data, actual=decoded) raise ValidationError("Data mismatch: difference was written to stdout.") - async def read(self, timeout: Optional[float] = None, size: Optional[int] = None) -> bytes: + async def read( + self, + timeout: Optional[float] = None, + size: Optional[int] = None, + endpoint: Optional[int] = None, + ) -> bytes: next_command = USBCommand(**self.cr.next_command()) if not ( next_command.module == "usb" @@ -574,6 +608,15 @@ async def read(self, timeout: Optional[float] = None, size: Optional[int] = None data = data[:size] return data + async def drain( + self, + endpoint: Optional[int] = None, + timeout: float = 0.05, + size: int = 512, + max_duration: float = 1, + ) -> None: + pass + def ctrl_transfer( self, bmRequestType: int, diff --git a/pylabrobot/thermo_fisher/nanodrop_1000/__init__.py b/pylabrobot/thermo_fisher/nanodrop_1000/__init__.py new file mode 100644 index 00000000000..36141ae5947 --- /dev/null +++ b/pylabrobot/thermo_fisher/nanodrop_1000/__init__.py @@ -0,0 +1,3 @@ +from .nanodrop_1000 import ThermoFisherNanoDrop1000 + +__all__ = ["ThermoFisherNanoDrop1000"] diff --git a/pylabrobot/thermo_fisher/nanodrop_1000/nanodrop_1000.py b/pylabrobot/thermo_fisher/nanodrop_1000/nanodrop_1000.py new file mode 100644 index 00000000000..b55c052ff4f --- /dev/null +++ b/pylabrobot/thermo_fisher/nanodrop_1000/nanodrop_1000.py @@ -0,0 +1,211 @@ +import asyncio +import logging +import math +from typing import List, Tuple + +from pylabrobot.io.usb import USB + +logger = logging.getLogger(__name__) + + +class ThermoFisherNanoDrop1000: + VID = 0x2457 + PID = 0x1002 + EP_OUT = 0x02 + EP_IN_HEAVY = 0x82 + EP_IN_COMM = 0x87 + + def __init__(self): + self.io = USB( + id_vendor=self.VID, + id_product=self.PID, + human_readable_device_name="Thermo Fisher NanoDrop 1000", + packet_read_timeout=0.05, + read_timeout=1, + read_endpoint_address=self.EP_IN_COMM, + write_endpoint_address=self.EP_OUT, + configuration_callback=self._configure_usb_device, + ) + self._connected = False + + self.coefficients = {} + self.dark_spectrum = None + self.blank_spectrum = None + + @classmethod + def _configure_usb_device(cls, device) -> None: + device.set_configuration() + device.clear_halt(cls.EP_OUT) + device.clear_halt(cls.EP_IN_HEAVY) + device.clear_halt(cls.EP_IN_COMM) + + async def setup(self): + """Initializes the USB connection.""" + logger.info("Connecting to NanoDrop 1000") + await self.io.setup(empty_buffer=False) + self._connected = True + + # Wake & Init + await self.send_command([0x08]) + await asyncio.sleep(0.1) + await self.send_command([0x01]) + await asyncio.sleep(0.2) + + await self._download_all_coefficients() + + async def stop(self): + """Safely powers down hardware and releases the USB port.""" + if self._connected: + try: + # Ensure lamp and magnet are off before disconnect + await self.send_command([0x03, 0x00]) + await self.send_command([0x0F, 0x00]) + except Exception: + logger.warning("Failed to power down the NanoDrop cleanly", exc_info=True) + await self.io.stop() + self._connected = False + logger.info("NanoDrop 1000 disconnected") + + self.coefficients = {} + + async def send_command(self, payload: List[int]): + """Generic transport method for writing to the command mailbox.""" + await self.io.write(bytes(payload)) + + async def read_comm(self, timeout=500) -> bytes: + """Reads from the 64-byte text/status endpoint.""" + return await self.io.read(timeout=timeout / 1000, size=64) + + async def read_heavy(self, packets=64, timeout=1000) -> bytearray: + """Reads bulk interleaved blocks from the main camera endpoint.""" + data_buffer = bytearray() + for _ in range(packets): + packet = await self.io.read( + timeout=timeout / 1000, + size=64, + endpoint=self.EP_IN_HEAVY, + ) + data_buffer.extend(packet) + return data_buffer + + async def flush_comm(self): + await self.io.drain(endpoint=self.EP_IN_COMM, timeout=0.05, size=64) + + async def flush_heavy(self): + await self.io.drain(endpoint=self.EP_IN_HEAVY, timeout=0.05, size=512) + + async def set_lamp(self, state: bool): + cmd = 0xFF if state else 0x00 + await self.send_command([0x03, cmd]) + + async def set_magnet(self, state: bool): + cmd = 0xFF if state else 0x00 + await self.send_command([0x0F, cmd]) + + async def _set_integration_time(self, ms: int): + if ms < 3: + ms = 3 + logger.warning("Integration time is too low; using 3 ms") + elif ms > 65535: + ms = 65535 + logger.warning("Integration time is too high; using 65535 ms") + + lsb = ms & 0xFF + msb = (ms >> 8) & 0xFF + await self.send_command([0x02, lsb, msb]) + + async def _download_all_coefficients(self): + logger.info("Downloading NanoDrop factory memory map") + await self.flush_comm() + + for index in range(1, 15): + if index == 5: + continue + await self.send_command([0x05, index]) + await asyncio.sleep(0.05) + try: + data = await self.read_comm() + text = bytearray(data[2:]).decode("ascii", errors="ignore").split("\x00")[0] + self.coefficients[index] = float(text) + except Exception: + logger.warning("Failed to read coefficient index %d", index, exc_info=True) + + def _calculate_x_axis(self) -> List[float]: + c0, c1 = self.coefficients.get(1, 0), self.coefficients.get(2, 0) + c2, c3 = self.coefficients.get(3, 0), self.coefficients.get(4, 0) + return [ + c0 + (c1 * pixel) + (c2 * (pixel**2)) + (c3 * (pixel**3)) for pixel in range(2048) + ] + + async def get_raw_spectrum(self) -> List[float]: + await self.flush_heavy() + await self.send_command([0x09]) + + data_buffer = await self.read_heavy() + + pixels = [] + for i in range(0, 4096, 128): + lsb_block = data_buffer[i : i + 64] + msb_block = data_buffer[i + 64 : i + 128] + for j in range(64): + pixels.append((msb_block[j] << 8) | lsb_block[j]) + + raw_intensities = [float(pixel) for pixel in pixels] + + # TODO [Future Work]: Optical Black Pixel Subtraction + # The first 25 pixels (0-24) are optically black. Calculate their average + # and subtract it from the entire array to correct for thermal baseline drift. + + # TODO [Future Work]: Non-Linearity Correction + # Apply the 7th-order polynomial using coefficients 6 through 13 to `raw_intensities` + # to ensure perfect photometric accuracy across the dynamic range. + + return raw_intensities + + async def take_blank(self, integration_ms=20): + await self._set_integration_time(integration_ms) + + await self.set_lamp(False) + await self.set_magnet(True) + await asyncio.sleep(0.2) + logger.info("Acquiring dark baseline") + self.dark_spectrum = await self.get_raw_spectrum() + + await self.set_lamp(True) + await asyncio.sleep(0.2) + logger.info("Acquiring blank baseline") + self.blank_spectrum = await self.get_raw_spectrum() + + await self.set_lamp(False) + await self.set_magnet(False) + logger.info("Blanking complete") + + async def measure_absorbance(self, integration_ms=20) -> Tuple[List[float], List[float]]: + if self.blank_spectrum is None or self.dark_spectrum is None: + raise ValueError("You must run take_blank() before measuring!") + + # TODO [Future Work]: Auto-Exposure Bracketing (HDR) + # Replace the static `integration_ms` with a loop that fires 8ms, 16ms, 32ms, etc. + # and mathematically stitches the optimal exposures together. + + await self._set_integration_time(integration_ms) + await self.set_magnet(True) + await self.set_lamp(True) + await asyncio.sleep(0.2) + + logger.info("Measuring sample") + sample_spectrum = await self.get_raw_spectrum() + + await self.set_lamp(False) + await self.set_magnet(False) + + absorbance = [ + -math.log10(max(sample - dark, 1) / max(blank - dark, 1)) + for sample, dark, blank in zip( + sample_spectrum, + self.dark_spectrum, + self.blank_spectrum, + ) + ] + + return self._calculate_x_axis(), absorbance